Remove the terrible hack in for flush_block_policy_factory

Summary:
Previous code is too convoluted and I must be drunk for letting
such code to be written without a second thought.

Thanks to the discussion with @sdong, I added the `Options` when
generating the flusher, thus avoiding the tricks.

Just FYI: I resisted to add Options in flush_block_policy.h since I
wanted to avoid cyclic dependencies: FlushBlockPolicy dpends on Options
and Options also depends FlushBlockPolicy... While I appreciate my
effort to prevent it, the old design turns out creating more troubles than
it tried to avoid.

Test Plan: ran ./table_test

Reviewers: sdong

Reviewed By: sdong

CC: sdong, leveldb

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

db/version_set.cc

@@ -1553,8 +1553,7 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu,
     // only one thread can be here at the same time
     if (!new_manifest_filename.empty()) {
       unique_ptr<WritableFile> descriptor_file;
-      s = env_->NewWritableFile(new_manifest_filename,
+      s = env_->NewWritableFile(new_manifest_filename, &descriptor_file,
-                                &descriptor_file,
                                 storage_options_.AdaptForLogWrite());
       if (s.ok()) {
         descriptor_log_.reset(new log::Writer(std::move(descriptor_file)));

include/rocksdb/flush_block_policy.h

@@ -11,6 +11,7 @@ namespace rocksdb {
 
 class Slice;
 class BlockBuilder;
+struct Options;
 
 // FlushBlockPolicy provides a configurable way to determine when to flush a
 // block in the block based tables,
@@ -36,29 +37,22 @@ class FlushBlockPolicyFactory {
   // Callers must delete the result after any database that is using the
   // result has been closed.
   virtual FlushBlockPolicy* NewFlushBlockPolicy(
-      const BlockBuilder& data_block_builder) const = 0;
+      const Options& options, const BlockBuilder& data_block_builder) const = 0;
 
   virtual ~FlushBlockPolicyFactory() { }
 };
 
 class FlushBlockBySizePolicyFactory : public FlushBlockPolicyFactory {
  public:
-  FlushBlockBySizePolicyFactory(const uint64_t block_size,
+  FlushBlockBySizePolicyFactory() {}
-                                const uint64_t block_size_deviation) :
-      block_size_(block_size),
-      block_size_deviation_(block_size_deviation) {
-  }
 
   virtual const char* Name() const override {
     return "FlushBlockBySizePolicyFactory";
   }
 
   virtual FlushBlockPolicy* NewFlushBlockPolicy(
+      const Options& options,
       const BlockBuilder& data_block_builder) const override;
-
- private:
-  const uint64_t block_size_;
-  const uint64_t block_size_deviation_;
 };
 
 }  // rocksdb

table/block_based_table_builder.cc

@@ -87,8 +87,8 @@ struct BlockBasedTableBuilder::Rep {
         filter_block(opt.filter_policy == nullptr
                          ? nullptr
                          : new FilterBlockBuilder(opt, &internal_comparator)),
-        flush_block_policy(
+        flush_block_policy(flush_block_policy_factory->NewFlushBlockPolicy(
-            flush_block_policy_factory->NewFlushBlockPolicy(data_block)) {}
+            options, data_block)) {}
 };
 
 BlockBasedTableBuilder::BlockBasedTableBuilder(

table/block_based_table_factory.cc

@@ -18,6 +18,15 @@
 
 namespace rocksdb {
 
+BlockBasedTableFactory::BlockBasedTableFactory(
+    const BlockBasedTableOptions& table_options)
+    : table_options_(table_options) {
+  if (table_options_.flush_block_policy_factory == nullptr) {
+    table_options_.flush_block_policy_factory.reset(
+        new FlushBlockBySizePolicyFactory());
+  }
+}
+
 Status BlockBasedTableFactory::NewTableReader(
     const Options& options, const EnvOptions& soptions,
     const InternalKeyComparator& internal_comparator,
@@ -32,34 +41,12 @@ TableBuilder* BlockBasedTableFactory::NewTableBuilder(
     const Options& options, const InternalKeyComparator& internal_comparator,
     WritableFile* file, CompressionType compression_type) const {
   auto flush_block_policy_factory =
-    table_options_.flush_block_policy_factory.get();
+      table_options_.flush_block_policy_factory.get();
-
-  // if flush block policy factory is not set, we'll create the default one
-  // from the options.
-  //
-  // NOTE: we cannot pre-cache the "default block policy factory" because
-  // `FlushBlockBySizePolicyFactory` takes `options.block_size` and
-  // `options.block_size_deviation` as parameters, which may be different
-  // every time.
-  if (flush_block_policy_factory == nullptr) {
-    flush_block_policy_factory =
-        new FlushBlockBySizePolicyFactory(options.block_size,
-                                          options.block_size_deviation);
-  }
 
   auto table_builder =
       new BlockBasedTableBuilder(options, internal_comparator, file,
                                  flush_block_policy_factory, compression_type);
 
-  // Delete flush_block_policy_factory only when it's just created from the
-  // options.
-  // We can safely delete flush_block_policy_factory since it will only be used
-  // during the construction of `BlockBasedTableBuilder`.
-  if (flush_block_policy_factory != 
-      table_options_.flush_block_policy_factory.get()) {
-    delete flush_block_policy_factory;
-  }
-
   return table_builder;
 }
 

table/block_based_table_factory.h

@@ -26,8 +26,7 @@ class BlockBasedTableBuilder;
 class BlockBasedTableFactory : public TableFactory {
  public:
   explicit BlockBasedTableFactory(
-      const BlockBasedTableOptions& table_options = BlockBasedTableOptions())
+      const BlockBasedTableOptions& table_options = BlockBasedTableOptions());
-      : table_options_(table_options) {}
 
   ~BlockBasedTableFactory() {}
 

table/flush_block_policy.cc

@@ -3,6 +3,7 @@
 //  LICENSE file in the root directory of this source tree. An additional grant
 //  of patent rights can be found in the PATENTS file in the same directory.
 
+#include "rocksdb/options.h"
 #include "rocksdb/flush_block_policy.h"
 #include "rocksdb/slice.h"
 #include "table/block_builder.h"
@@ -61,10 +62,9 @@ class FlushBlockBySizePolicy : public FlushBlockPolicy {
 };
 
 FlushBlockPolicy* FlushBlockBySizePolicyFactory::NewFlushBlockPolicy(
-    const BlockBuilder& data_block_builder) const {
+    const Options& options, const BlockBuilder& data_block_builder) const {
-  return new FlushBlockBySizePolicy(block_size_,
+  return new FlushBlockBySizePolicy(
-                                    block_size_deviation_,
+      options.block_size, options.block_size_deviation, data_block_builder);
-                                    data_block_builder);
 }
 
 }  // namespace rocksdb

table/table_test.cc

@@ -689,8 +689,7 @@ class Harness {
     switch (args.type) {
       case BLOCK_BASED_TABLE_TEST:
         table_options.flush_block_policy_factory.reset(
-            new FlushBlockBySizePolicyFactory(options_.block_size,
+            new FlushBlockBySizePolicyFactory());
-                                              options_.block_size_deviation));
         options_.table_factory.reset(new BlockBasedTableFactory(table_options));
         constructor_ = new TableConstructor(options_.comparator);
         break;