Add a table factory that can read DB with both of PlainTable and BlockBasedTable in it

Summary: The new table factory is used if users want to convert a DB from one table format to the other. A user can use this table to open a DB written using one table format and write new files to another table format.

Test Plan: add a unit test

Reviewers: haobo, igor

Reviewed By: igor

Subscribers: dhruba, ljin, yhchiang, leveldb

Differential Revision: https://reviews.facebook.net/D19017

HISTORY.md

@@ -8,6 +8,7 @@
 ### New Features
 * Hash index for block-based table will be materialized and reconstructed more efficiently. Previously hash index is constructed by scanning the whole table during every table open.
 * FIFO compaction style
+* Add AdaptiveTableFactory, which is used to convert from a DB of PlainTable to BlockBasedTabe, or vise versa. It can be created using NewAdaptiveTableFactory()
 
 ## 3.0.0 (05/05/2014)
 

db/plain_table_db_test.cc

@@ -847,6 +847,47 @@ TEST(PlainTableDBTest, CompactionTrigger) {
   ASSERT_EQ(NumTableFilesAtLevel(1), 1);
 }
 
+TEST(PlainTableDBTest, AdaptiveTable) {
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+
+  options.table_factory.reset(NewPlainTableFactory());
+  DestroyAndReopen(&options);
+
+  ASSERT_OK(Put("1000000000000foo", "v1"));
+  ASSERT_OK(Put("0000000000000bar", "v2"));
+  ASSERT_OK(Put("1000000000000foo", "v3"));
+  dbfull()->TEST_FlushMemTable();
+
+  options.create_if_missing = false;
+  std::shared_ptr<TableFactory> dummy_factory;
+  options.table_factory.reset(
+      NewAdaptiveTableFactory(dummy_factory, dummy_factory, false));
+  Reopen(&options);
+  ASSERT_EQ("v3", Get("1000000000000foo"));
+  ASSERT_EQ("v2", Get("0000000000000bar"));
+
+  ASSERT_OK(Put("2000000000000foo", "v4"));
+  ASSERT_OK(Put("3000000000000bar", "v5"));
+  dbfull()->TEST_FlushMemTable();
+  ASSERT_EQ("v4", Get("2000000000000foo"));
+  ASSERT_EQ("v5", Get("3000000000000bar"));
+
+  Reopen(&options);
+  ASSERT_EQ("v3", Get("1000000000000foo"));
+  ASSERT_EQ("v2", Get("0000000000000bar"));
+  ASSERT_EQ("v4", Get("2000000000000foo"));
+  ASSERT_EQ("v5", Get("3000000000000bar"));
+
+  options.table_factory.reset(NewBlockBasedTableFactory());
+  Reopen(&options);
+  ASSERT_NE("v3", Get("1000000000000foo"));
+
+  options.table_factory.reset(NewPlainTableFactory());
+  Reopen(&options);
+  ASSERT_NE("v5", Get("3000000000000bar"));
+}
+
 }  // namespace rocksdb
 
 int main(int argc, char** argv) {

include/rocksdb/table.h

@@ -203,4 +203,19 @@ class TableFactory {
       WritableFile* file, CompressionType compression_type) const = 0;
 };
 
+#ifndef ROCKSDB_LITE
+// Create a special table factory that can open both of block based table format
+// and plain table, based on setting inside the SST files. It should be used to
+// convert a DB from one table format to another.
+// @block_based_table_factory:  block based table factory to use. If NULL, use
+//                              a default one.
+// @plain_table_factory: plain table factory to use. If NULL, use a default one.
+// @table_factory_to_write: the table factory used when writing to new files.
+extern TableFactory* NewAdaptiveTableFactory(
+    std::shared_ptr<TableFactory> block_based_table_factory = nullptr,
+    std::shared_ptr<TableFactory> plain_table_factory = nullptr,
+    std::shared_ptr<TableFactory> table_factory_to_write = nullptr);
+
+#endif  // ROCKSDB_LITE
+
 }  // namespace rocksdb

table/adaptive_table_factory.cc

+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#ifndef ROCKSDB_LITE
+#include "table/adaptive_table_factory.h"
+
+#include "table/format.h"
+
+namespace rocksdb {
+
+AdaptiveTableFactory::AdaptiveTableFactory(
+    std::shared_ptr<TableFactory> block_based_table_factory,
+    std::shared_ptr<TableFactory> plain_table_factory,
+    std::shared_ptr<TableFactory> table_factory_to_write)
+    : block_based_table_factory_(block_based_table_factory),
+      plain_table_factory_(plain_table_factory),
+      table_factory_to_write_(table_factory_to_write) {
+  if (!plain_table_factory_) {
+    plain_table_factory_.reset(NewPlainTableFactory());
+  }
+  if (!block_based_table_factory_) {
+    block_based_table_factory_.reset(NewBlockBasedTableFactory());
+  }
+  if (!table_factory_to_write_) {
+    table_factory_to_write_ = block_based_table_factory_;
+  }
+}
+
+extern const uint64_t kPlainTableMagicNumber;
+extern const uint64_t kLegacyPlainTableMagicNumber;
+extern const uint64_t kBlockBasedTableMagicNumber;
+extern const uint64_t kLegacyBlockBasedTableMagicNumber;
+
+Status AdaptiveTableFactory::NewTableReader(
+    const Options& options, const EnvOptions& soptions,
+    const InternalKeyComparator& icomp, unique_ptr<RandomAccessFile>&& file,
+    uint64_t file_size, unique_ptr<TableReader>* table) const {
+  Footer footer;
+  auto s = ReadFooterFromFile(file.get(), file_size, &footer);
+  if (!s.ok()) {
+    return s;
+  }
+  if (footer.table_magic_number() == kPlainTableMagicNumber ||
+      footer.table_magic_number() == kLegacyPlainTableMagicNumber) {
+    return plain_table_factory_->NewTableReader(
+        options, soptions, icomp, std::move(file), file_size, table);
+  } else if (footer.table_magic_number() == kBlockBasedTableMagicNumber ||
+      footer.table_magic_number() == kLegacyBlockBasedTableMagicNumber) {
+    return block_based_table_factory_->NewTableReader(
+        options, soptions, icomp, std::move(file), file_size, table);
+  } else {
+    return Status::NotSupported("Unidentified table format");
+  }
+}
+
+TableBuilder* AdaptiveTableFactory::NewTableBuilder(
+    const Options& options, const InternalKeyComparator& internal_comparator,
+    WritableFile* file, CompressionType compression_type) const {
+  return table_factory_to_write_->NewTableBuilder(options, internal_comparator,
+                                                  file, compression_type);
+}
+
+extern TableFactory* NewAdaptiveTableFactory(
+    std::shared_ptr<TableFactory> block_based_table_factory,
+    std::shared_ptr<TableFactory> plain_table_factory,
+    std::shared_ptr<TableFactory> table_factory_to_write) {
+  return new AdaptiveTableFactory(
+      block_based_table_factory, plain_table_factory, table_factory_to_write);
+}
+
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE

table/adaptive_table_factory.h

+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#pragma once
+
+#ifndef ROCKSDB_LITE
+
+#include "rocksdb/options.h"
+#include "rocksdb/table.h"
+
+namespace rocksdb {
+
+struct Options;
+struct EnvOptions;
+
+using std::unique_ptr;
+class Status;
+class RandomAccessFile;
+class WritableFile;
+class Table;
+class TableBuilder;
+
+class AdaptiveTableFactory : public TableFactory {
+ public:
+  ~AdaptiveTableFactory() {}
+
+  explicit AdaptiveTableFactory(
+      std::shared_ptr<TableFactory> block_based_table_factory,
+      std::shared_ptr<TableFactory> plain_table_factory,
+      std::shared_ptr<TableFactory> table_factory_to_write);
+  const char* Name() const override { return "AdaptiveTableFactory"; }
+  Status NewTableReader(const Options& options, const EnvOptions& soptions,
+                        const InternalKeyComparator& internal_comparator,
+                        unique_ptr<RandomAccessFile>&& file, uint64_t file_size,
+                        unique_ptr<TableReader>* table) const override;
+  TableBuilder* NewTableBuilder(const Options& options,
+                                const InternalKeyComparator& icomparator,
+                                WritableFile* file,
+                                CompressionType compression_type) const
+      override;
+
+ private:
+  std::shared_ptr<TableFactory> block_based_table_factory_;
+  std::shared_ptr<TableFactory> plain_table_factory_;
+  std::shared_ptr<TableFactory> table_factory_to_write_;
+};
+
+}  // namespace rocksdb
+#endif  // ROCKSDB_LITE