Bump version to 5.0.2

Summary:
Cockroachdb exposed this bug in #1778. The bug happens when a compaction's output files are ended due to exceeding max_compaction_bytes. In that case we weren't taking into account the next file's start key when deciding how far to extend the current file's max_key. This caused the non-overlapping key-range invariant to be violated.

Note this was correctly handled for the usual case of cutting compaction output, which is file size exceeding max_output_file_size. I am not sure why these are two separate code paths, but we can consider refactoring it to prevent such errors in the future.
Closes https://github.com/facebook/rocksdb/pull/1784

Differential Revision: D4430235

Pulled By: ajkr

fbshipit-source-id: 80af748

CMakeLists.txt

@@ -502,6 +502,7 @@ set(TESTS
         db/db_iter_test.cc
         db/db_log_iter_test.cc
         db/db_memtable_test.cc
+        db/db_merge_operator_test.cc
         db/db_options_test.cc
         db/db_properties_test.cc
         db/db_table_properties_test.cc

HISTORY.md

@@ -1,10 +1,4 @@
 # Rocksdb Change Log
-## Unreleased
-### Public API Change
-* Options.level0_stop_writes_trigger default value changes from 24 to 32.
-* New compaction filter API: CompactionFilter::FilterV2(). Allows to drop ranges of keys.
-* Removed flashcache support.
-
 ## 5.0.0 (11/17/2016)
 ### Public API Change
 * Options::max_bytes_for_level_multiplier is now a double along with all getters and setters.
@@ -13,6 +7,10 @@
 * Support dynamically change `delayed_write_rate` option via SetDBOptions().
 * Options::allow_concurrent_memtable_write and Options::enable_write_thread_adaptive_yield are now true by default.
 * Remove Tickers::SEQUENCE_NUMBER to avoid confusion if statistics object is shared among RocksDB instance. Alternatively DB::GetLatestSequenceNumber() can be used to get the same value.
+* Options.level0_stop_writes_trigger default value changes from 24 to 32.
+* New compaction filter API: CompactionFilter::FilterV2(). Allows to drop ranges of keys.
+* Removed flashcache support.
+* DB::AddFile() is deprecated and is replaced with DB::IngestExternalFile(). DB::IngestExternalFile() remove all the restrictions that existed for DB::AddFile.
 
 ### New Features
 * Add avoid_flush_during_shutdown option, which speeds up DB shutdown by not flushing unpersisted data (i.e. with disableWAL = true). Unpersisted data will be lost. The options is dynamically changeable via SetDBOptions().

Makefile

@@ -309,6 +309,7 @@ TESTS = \
 	db_inplace_update_test \
 	db_iterator_test \
 	db_memtable_test \
+	db_merge_operator_test \
 	db_options_test \
 	db_range_del_test \
 	db_sst_test \
@@ -996,6 +997,9 @@ db_iterator_test: db/db_iterator_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHA
 db_memtable_test: db/db_memtable_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
+db_merge_operator_test: db/db_merge_operator_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(AM_LINK)
+
 db_options_test: db/db_options_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 

db/compaction_iterator.cc

@@ -437,9 +437,7 @@ void CompactionIterator::NextFromInput() {
                                 bottommost_level_);
       merge_out_iter_.SeekToFirst();
 
-      if (merge_helper_->FilteredUntil(&skip_until)) {
-        need_skip = true;
-      } else if (merge_out_iter_.Valid()) {
+      if (merge_out_iter_.Valid()) {
         // NOTE: key, value, and ikey_ refer to old entries.
         //       These will be correctly set below.
         key_ = merge_out_iter_.key();
@@ -460,6 +458,10 @@ void CompactionIterator::NextFromInput() {
         // coming after the merges
         has_current_user_key_ = false;
         pinned_iters_mgr_.ReleasePinnedData();
+
+        if (merge_helper_->FilteredUntil(&skip_until)) {
+          need_skip = true;
+        }
       }
     } else {
       // 1. new user key -OR-

db/compaction_iterator_test.cc

@@ -272,8 +272,8 @@ TEST_F(CompactionIteratorTest, CompactionFilterSkipUntil) {
         return Decision::kKeep;
       }
       if (k == "i") {
-        EXPECT_EQ(ValueType::kValue, t);
-        EXPECT_EQ("iv95", v);
+        EXPECT_EQ(ValueType::kMergeOperand, t);
+        EXPECT_EQ("im95", v);
         *skip_until = "z";
         return Decision::kRemoveAndSkipUntil;
       }
@@ -299,10 +299,10 @@ TEST_F(CompactionIteratorTest, CompactionFilterSkipUntil) {
        test::KeyStr("f", 30, kTypeMerge),  // skip to "g+"
        test::KeyStr("f", 25, kTypeValue), test::KeyStr("g", 90, kTypeValue),
        test::KeyStr("h", 91, kTypeValue),  // keep
-       test::KeyStr("i", 95, kTypeValue),  // skip to "z"
+       test::KeyStr("i", 95, kTypeMerge),  // skip to "z"
        test::KeyStr("j", 99, kTypeValue)},
       {"av50", "am45", "bv60", "bv40", "cv35", "dm70", "em71", "fm65", "fm30",
-       "fv25", "gv90", "hv91", "iv95", "jv99"},
+       "fv25", "gv90", "hv91", "im95", "jv99"},
       {}, {}, kMaxSequenceNumber, &merge_op, &filter);
 
   // Compaction should output just "a", "e" and "h" keys.

db/compaction_job.cc

@@ -772,9 +772,9 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
                    key, sub_compact->current_output_file_size) &&
                sub_compact->builder != nullptr) {
       CompactionIterationStats range_del_out_stats;
-      status =
-          FinishCompactionOutputFile(input->status(), sub_compact,
-                                     range_del_agg.get(), &range_del_out_stats);
+      status = FinishCompactionOutputFile(input->status(), sub_compact,
+                                          range_del_agg.get(),
+                                          &range_del_out_stats, &key);
       RecordDroppedKeys(range_del_out_stats,
                         &sub_compact->compaction_job_stats);
       if (!status.ok()) {
@@ -853,9 +853,6 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
       }
     }
 
-    Status input_status = input->status();
-    c_iter->Next();
-
     // Close output file if it is big enough
     // TODO(aekmekji): determine if file should be closed earlier than this
     // during subcompactions (i.e. if output size, estimated by input size, is
@@ -864,6 +861,9 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
     if (sub_compact->compaction->output_level() != 0 &&
         sub_compact->current_output_file_size >=
             sub_compact->compaction->max_output_file_size()) {
+      Status input_status = input->status();
+      c_iter->Next();
+
       const Slice* next_key = nullptr;
       if (c_iter->Valid()) {
         next_key = &c_iter->key();
@@ -879,6 +879,8 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
         // files.
         sub_compact->compression_dict = std::move(compression_dict);
       }
+    } else {
+      c_iter->Next();
     }
   }
 

db/db_impl.cc

@@ -2939,6 +2939,12 @@ Status DBImpl::WaitForFlushMemTable(ColumnFamilyData* cfd) {
     if (shutting_down_.load(std::memory_order_acquire)) {
       return Status::ShutdownInProgress();
     }
+    if (cfd->IsDropped()) {
+      // FlushJob cannot flush a dropped CF, if we did not break here
+      // we will loop forever since cfd->imm()->NumNotFlushed() will never
+      // drop to zero
+      return Status::InvalidArgument("Cannot flush a dropped CF");
+    }
     bg_cv_.Wait();
   }
   if (!bg_error_.ok()) {
@@ -6493,13 +6499,22 @@ Status DBImpl::IngestExternalFile(
 
     num_running_ingest_file_++;
 
+    // We cannot ingest a file into a dropped CF
+    if (cfd->IsDropped()) {
+      status = Status::InvalidArgument(
+          "Cannot ingest an external file into a dropped CF");
+    }
+
     // Figure out if we need to flush the memtable first
-    bool need_flush = false;
-    status = ingestion_job.NeedsFlush(&need_flush);
-    if (status.ok() && need_flush) {
-      mutex_.Unlock();
-      status = FlushMemTable(cfd, FlushOptions(), true /* writes_stopped */);
-      mutex_.Lock();
+    if (status.ok()) {
+      bool need_flush = false;
+      status = ingestion_job.NeedsFlush(&need_flush);
+
+      if (status.ok() && need_flush) {
+        mutex_.Unlock();
+        status = FlushMemTable(cfd, FlushOptions(), true /* writes_stopped */);
+        mutex_.Lock();
+      }
     }
 
     // Run the ingestion job
@@ -6541,9 +6556,35 @@ Status DBImpl::IngestExternalFile(
   // Cleanup
   ingestion_job.Cleanup(status);
 
+  if (status.ok()) {
+    NotifyOnExternalFileIngested(cfd, ingestion_job);
+  }
+
   return status;
 }
 
+void DBImpl::NotifyOnExternalFileIngested(
+    ColumnFamilyData* cfd, const ExternalSstFileIngestionJob& ingestion_job) {
+#ifndef ROCKSDB_LITE
+  if (immutable_db_options_.listeners.empty()) {
+    return;
+  }
+
+  for (const IngestedFileInfo& f : ingestion_job.files_to_ingest()) {
+    ExternalFileIngestionInfo info;
+    info.cf_name = cfd->GetName();
+    info.external_file_path = f.external_file_path;
+    info.internal_file_path = f.internal_file_path;
+    info.global_seqno = f.assigned_seqno;
+    info.table_properties = f.table_properties;
+    for (auto listener : immutable_db_options_.listeners) {
+      listener->OnExternalFileIngested(this, info);
+    }
+  }
+
+#endif
+}
+
 void DBImpl::WaitForIngestFile() {
   mutex_.AssertHeld();
   while (num_running_ingest_file_ > 0) {

db/db_impl.h

@@ -23,6 +23,7 @@
 #include "db/column_family.h"
 #include "db/compaction_job.h"
 #include "db/dbformat.h"
+#include "db/external_sst_file_ingestion_job.h"
 #include "db/flush_job.h"
 #include "db/flush_scheduler.h"
 #include "db/internal_stats.h"
@@ -556,6 +557,9 @@ class DBImpl : public DB {
   void NotifyOnMemTableSealed(ColumnFamilyData* cfd,
                               const MemTableInfo& mem_table_info);
 
+  void NotifyOnExternalFileIngested(
+      ColumnFamilyData* cfd, const ExternalSstFileIngestionJob& ingestion_job);
+
   void NewThreadStatusCfInfo(ColumnFamilyData* cfd) const;
 
   void EraseThreadStatusCfInfo(ColumnFamilyData* cfd) const;

db/db_iter.cc

@@ -518,6 +518,7 @@ void DBIter::MergeValuesNewToOld() {
                              iter_->IsValuePinned() /* operand_pinned */);
 
   ParsedInternalKey ikey;
+  Status s;
   for (iter_->Next(); iter_->Valid(); iter_->Next()) {
     if (!ParseKey(&ikey)) {
       // skip corrupted key
@@ -538,9 +539,12 @@ void DBIter::MergeValuesNewToOld() {
       // final result in saved_value_. We are done!
       // ignore corruption if there is any.
       const Slice val = iter_->value();
-      MergeHelper::TimedFullMerge(merge_operator_, ikey.user_key, &val,
-                                  merge_context_.GetOperands(), &saved_value_,
-                                  logger_, statistics_, env_, &pinned_value_);
+      s = MergeHelper::TimedFullMerge(
+          merge_operator_, ikey.user_key, &val, merge_context_.GetOperands(),
+          &saved_value_, logger_, statistics_, env_, &pinned_value_);
+      if (!s.ok()) {
+        status_ = s;
+      }
       // iter_ is positioned after put
       iter_->Next();
       return;
@@ -559,9 +563,12 @@ void DBIter::MergeValuesNewToOld() {
   // a deletion marker.
   // feed null as the existing value to the merge operator, such that
   // client can differentiate this scenario and do things accordingly.
-  MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(), nullptr,
-                              merge_context_.GetOperands(), &saved_value_,
-                              logger_, statistics_, env_, &pinned_value_);
+  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(), nullptr,
+                                  merge_context_.GetOperands(), &saved_value_,
+                                  logger_, statistics_, env_, &pinned_value_);
+  if (!s.ok()) {
+    status_ = s;
+  }
 }
 
 void DBIter::Prev() {
@@ -742,6 +749,7 @@ bool DBIter::FindValueForCurrentKey() {
     FindParseableKey(&ikey, kReverse);
   }
 
+  Status s;
   switch (last_key_entry_type) {
     case kTypeDeletion:
     case kTypeSingleDeletion:
@@ -753,16 +761,16 @@ bool DBIter::FindValueForCurrentKey() {
       if (last_not_merge_type == kTypeDeletion ||
           last_not_merge_type == kTypeSingleDeletion ||
           last_not_merge_type == kTypeRangeDeletion) {
-        MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(),
-                                    nullptr, merge_context_.GetOperands(),
-                                    &saved_value_, logger_, statistics_, env_,
-                                    &pinned_value_);
+        s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(),
+                                        nullptr, merge_context_.GetOperands(),
+                                        &saved_value_, logger_, statistics_,
+                                        env_, &pinned_value_);
       } else {
         assert(last_not_merge_type == kTypeValue);
-        MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(),
-                                    &pinned_value_,
-                                    merge_context_.GetOperands(), &saved_value_,
-                                    logger_, statistics_, env_, &pinned_value_);
+        s = MergeHelper::TimedFullMerge(
+            merge_operator_, saved_key_.GetKey(), &pinned_value_,
+            merge_context_.GetOperands(), &saved_value_, logger_, statistics_,
+            env_, &pinned_value_);
       }
       break;
     case kTypeValue:
@@ -773,6 +781,9 @@ bool DBIter::FindValueForCurrentKey() {
       break;
   }
   valid_ = true;
+  if (!s.ok()) {
+    status_ = s;
+  }
   return true;
 }
 
@@ -818,13 +829,15 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
     FindParseableKey(&ikey, kForward);
   }
 
+  Status s;
   if (!iter_->Valid() ||
       !user_comparator_->Equal(ikey.user_key, saved_key_.GetKey()) ||
       ikey.type == kTypeDeletion || ikey.type == kTypeSingleDeletion ||
       range_del_agg_.ShouldDelete(ikey)) {
-    MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(), nullptr,
-                                merge_context_.GetOperands(), &saved_value_,
-                                logger_, statistics_, env_, &pinned_value_);
+    s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(),
+                                    nullptr, merge_context_.GetOperands(),
+                                    &saved_value_, logger_, statistics_, env_,
+                                    &pinned_value_);
     // Make iter_ valid and point to saved_key_
     if (!iter_->Valid() ||
         !user_comparator_->Equal(ikey.user_key, saved_key_.GetKey())) {
@@ -832,14 +845,20 @@ bool DBIter::FindValueForCurrentKeyUsingSeek() {
       RecordTick(statistics_, NUMBER_OF_RESEEKS_IN_ITERATION);
     }
     valid_ = true;
+    if (!s.ok()) {
+      status_ = s;
+    }
     return true;
   }
 
   const Slice& val = iter_->value();
-  MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(), &val,
-                              merge_context_.GetOperands(), &saved_value_,
-                              logger_, statistics_, env_, &pinned_value_);
+  s = MergeHelper::TimedFullMerge(merge_operator_, saved_key_.GetKey(), &val,
+                                  merge_context_.GetOperands(), &saved_value_,
+                                  logger_, statistics_, env_, &pinned_value_);
   valid_ = true;
+  if (!s.ok()) {
+    status_ = s;
+  }
   return true;
 }
 

db/db_merge_operator_test.cc

@@ -0,0 +1,98 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under the BSD-style license found in the
+//  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 <string>
+#include <vector>
+
+#include "db/db_test_util.h"
+#include "db/forward_iterator.h"
+#include "port/stack_trace.h"
+#include "utilities/merge_operators.h"
+
+namespace rocksdb {
+
+// Test merge operator functionality.
+class DBMergeOperatorTest : public DBTestBase {
+ public:
+  DBMergeOperatorTest() : DBTestBase("/db_merge_operator_test") {}
+};
+
+// A test merge operator mimics put but also fails if one of merge operands is
+// "corrupted".
+class TestPutOperator : public MergeOperator {
+ public:
+  virtual bool FullMergeV2(const MergeOperationInput& merge_in,
+                           MergeOperationOutput* merge_out) const override {
+    if (merge_in.existing_value != nullptr &&
+        *(merge_in.existing_value) == "corrupted") {
+      return false;
+    }
+    for (auto value : merge_in.operand_list) {
+      if (value == "corrupted") {
+        return false;
+      }
+    }
+    merge_out->existing_operand = merge_in.operand_list.back();
+    return true;
+  }
+
+  virtual const char* Name() const override { return "TestPutOperator"; }
+};
+
+TEST_F(DBMergeOperatorTest, MergeErrorOnRead) {
+  Options options;
+  options.create_if_missing = true;
+  options.merge_operator.reset(new TestPutOperator());
+  Reopen(options);
+  ASSERT_OK(Merge("k1", "v1"));
+  ASSERT_OK(Merge("k1", "corrupted"));
+  std::string value;
+  ASSERT_TRUE(db_->Get(ReadOptions(), "k1", &value).IsCorruption());
+  VerifyDBInternal({{"k1", "corrupted"}, {"k1", "v1"}});
+}
+
+TEST_F(DBMergeOperatorTest, MergeErrorOnWrite) {
+  Options options;
+  options.create_if_missing = true;
+  options.merge_operator.reset(new TestPutOperator());
+  options.max_successive_merges = 3;
+  Reopen(options);
+  ASSERT_OK(Merge("k1", "v1"));
+  ASSERT_OK(Merge("k1", "v2"));
+  // Will trigger a merge when hitting max_successive_merges and the merge
+  // will fail. The delta will be inserted nevertheless.
+  ASSERT_OK(Merge("k1", "corrupted"));
+  // Data should stay unmerged after the error.
+  VerifyDBInternal({{"k1", "corrupted"}, {"k1", "v2"}, {"k1", "v1"}});
+}
+
+TEST_F(DBMergeOperatorTest, MergeErrorOnIteration) {
+  Options options;
+  options.create_if_missing = true;
+  options.merge_operator.reset(new TestPutOperator());
+
+  DestroyAndReopen(options);
+  ASSERT_OK(Merge("k1", "v1"));
+  ASSERT_OK(Merge("k1", "corrupted"));
+  ASSERT_OK(Put("k2", "v2"));
+  VerifyDBFromMap({{"k1", ""}, {"k2", "v2"}}, nullptr, false,
+                  {{"k1", Status::Corruption()}});
+  VerifyDBInternal({{"k1", "corrupted"}, {"k1", "v1"}, {"k2", "v2"}});
+
+  DestroyAndReopen(options);
+  ASSERT_OK(Merge("k1", "v1"));
+  ASSERT_OK(Put("k2", "v2"));
+  ASSERT_OK(Merge("k2", "corrupted"));
+  VerifyDBFromMap({{"k1", "v1"}, {"k2", ""}}, nullptr, false,
+                  {{"k2", Status::Corruption()}});
+  VerifyDBInternal({{"k1", "v1"}, {"k2", "corrupted"}, {"k2", "v2"}});
+}
+
+}  // namespace rocksdb
+
+int main(int argc, char** argv) {
+  rocksdb::port::InstallStackTraceHandler();
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

db/db_range_del_test.cc

@@ -112,6 +112,59 @@ TEST_F(DBRangeDelTest, CompactionOutputFilesExactlyFilled) {
   db_->ReleaseSnapshot(snapshot);
 }
 
+TEST_F(DBRangeDelTest, MaxCompactionBytesCutsOutputFiles) {
+  // Ensures range deletion spanning multiple compaction output files that are
+  // cut by max_compaction_bytes will have non-overlapping key-ranges.
+  // https://github.com/facebook/rocksdb/issues/1778
+  const int kNumFiles = 2, kNumPerFile = 1 << 8, kBytesPerVal = 1 << 12;
+  Options opts = CurrentOptions();
+  opts.comparator = test::Uint64Comparator();
+  opts.disable_auto_compactions = true;
+  opts.level0_file_num_compaction_trigger = kNumFiles;
+  opts.max_compaction_bytes = kNumPerFile * kBytesPerVal;
+  opts.memtable_factory.reset(new SpecialSkipListFactory(kNumPerFile));
+  // Want max_compaction_bytes to trigger the end of compaction output file, not
+  // target_file_size_base, so make the latter much bigger
+  opts.target_file_size_base = 100 * opts.max_compaction_bytes;
+  Reopen(opts);
+
+  // snapshot protects range tombstone from dropping due to becoming obsolete.
+  const Snapshot* snapshot = db_->GetSnapshot();
+
+  // It spans the whole key-range, thus will be included in all output files
+  ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(),
+                             GetNumericStr(0),
+                             GetNumericStr(kNumFiles * kNumPerFile - 1)));
+  Random rnd(301);
+  for (int i = 0; i < kNumFiles; ++i) {
+    std::vector<std::string> values;
+    // Write 1MB (256 values, each 4K)
+    for (int j = 0; j < kNumPerFile; j++) {
+      values.push_back(RandomString(&rnd, kBytesPerVal));
+      ASSERT_OK(Put(GetNumericStr(kNumPerFile * i + j), values[j]));
+    }
+    // extra entry to trigger SpecialSkipListFactory's flush
+    ASSERT_OK(Put(GetNumericStr(kNumPerFile), ""));
+    dbfull()->TEST_WaitForFlushMemTable();
+    ASSERT_EQ(i + 1, NumTableFilesAtLevel(0));
+  }
+
+  dbfull()->TEST_CompactRange(0, nullptr, nullptr, nullptr,
+                              true /* disallow_trivial_move */);
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+  ASSERT_GE(NumTableFilesAtLevel(1), 2);
+
+  std::vector<std::vector<FileMetaData>> files;
+  dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(), &files);
+
+  for (size_t i = 0; i < files[1].size() - 1; ++i) {
+    ASSERT_TRUE(InternalKeyComparator(opts.comparator)
+                    .Compare(files[1][i].largest, files[1][i + 1].smallest) <
+                0);
+  }
+  db_->ReleaseSnapshot(snapshot);
+}
+
 TEST_F(DBRangeDelTest, FlushRangeDelsSameStartKey) {
   db_->Put(WriteOptions(), "b1", "val");
   ASSERT_OK(

db/db_test_util.cc

@@ -8,6 +8,7 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 
 #include "db/db_test_util.h"
+#include "db/forward_iterator.h"
 
 namespace rocksdb {
 
@@ -501,6 +502,15 @@ Status DBTestBase::Put(int cf, const Slice& k, const Slice& v,
   }
 }
 
+Status DBTestBase::Merge(const Slice& k, const Slice& v, WriteOptions wo) {
+  return db_->Merge(wo, k, v);
+}
+
+Status DBTestBase::Merge(int cf, const Slice& k, const Slice& v,
+                         WriteOptions wo) {
+  return db_->Merge(wo, handles_[cf], k, v);
+}
+
 Status DBTestBase::Delete(const std::string& k) {
   return db_->Delete(WriteOptions(), k);
 }
@@ -1089,11 +1099,18 @@ std::vector<std::uint64_t> DBTestBase::ListTableFiles(Env* env,
 }
 
 void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
-                                 size_t* total_reads_res, bool tailing_iter) {
+                                 size_t* total_reads_res, bool tailing_iter,
+                                 std::map<std::string, Status> status) {
   size_t total_reads = 0;
 
   for (auto& kv : true_data) {
-    ASSERT_EQ(Get(kv.first), kv.second);
+    Status s = status[kv.first];
+    if (s.ok()) {
+      ASSERT_EQ(Get(kv.first), kv.second);
+    } else {
+      std::string value;
+      ASSERT_EQ(s, db_->Get(ReadOptions(), kv.first, &value));
+    }
     total_reads++;
   }
 
@@ -1106,21 +1123,40 @@ void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
     // Verify Iterator::Next()
     iter_cnt = 0;
     auto data_iter = true_data.begin();
+    Status s;
     for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) {
       ASSERT_EQ(iter->key().ToString(), data_iter->first);
-      ASSERT_EQ(iter->value().ToString(), data_iter->second);
+      Status current_status = status[data_iter->first];
+      if (!current_status.ok()) {
+        s = current_status;
+      }
+      ASSERT_EQ(iter->status(), s);
+      if (current_status.ok()) {
+        ASSERT_EQ(iter->value().ToString(), data_iter->second);
+      }
       iter_cnt++;
       total_reads++;
     }
     ASSERT_EQ(data_iter, true_data.end()) << iter_cnt << " / "
                                           << true_data.size();
+    delete iter;
 
     // Verify Iterator::Prev()
+    // Use a new iterator to make sure its status is clean.
+    iter = db_->NewIterator(ro);
     iter_cnt = 0;
+    s = Status::OK();
     auto data_rev = true_data.rbegin();
     for (iter->SeekToLast(); iter->Valid(); iter->Prev(), data_rev++) {
       ASSERT_EQ(iter->key().ToString(), data_rev->first);
-      ASSERT_EQ(iter->value().ToString(), data_rev->second);
+      Status current_status = status[data_rev->first];
+      if (!current_status.ok()) {
+        s = current_status;
+      }
+      ASSERT_EQ(iter->status(), s);
+      if (current_status.ok()) {
+        ASSERT_EQ(iter->value().ToString(), data_rev->second);
+      }
       iter_cnt++;
       total_reads++;
     }
@@ -1134,7 +1170,6 @@ void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
       ASSERT_EQ(kv.second, iter->value().ToString());
       total_reads++;
     }
-
     delete iter;
   }
 
@@ -1176,6 +1211,25 @@ void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
   }
 }
 
+void DBTestBase::VerifyDBInternal(
+    std::vector<std::pair<std::string, std::string>> true_data) {
+  Arena arena;
+  InternalKeyComparator icmp(last_options_.comparator);
+  RangeDelAggregator range_del_agg(icmp, {});
+  auto iter = dbfull()->NewInternalIterator(&arena, &range_del_agg);
+  iter->SeekToFirst();
+  for (auto p : true_data) {
+    ASSERT_TRUE(iter->Valid());
+    ParsedInternalKey ikey;
+    ASSERT_TRUE(ParseInternalKey(iter->key(), &ikey));
+    ASSERT_EQ(p.first, ikey.user_key);
+    ASSERT_EQ(p.second, iter->value());
+    iter->Next();
+  };
+  ASSERT_FALSE(iter->Valid());
+  iter->~InternalIterator();
+}
+
 #ifndef ROCKSDB_LITE
 
 uint64_t DBTestBase::GetNumberOfSstFilesForColumnFamily(

db/db_test_util.h

@@ -699,6 +699,12 @@ class DBTestBase : public testing::Test {
   Status Put(int cf, const Slice& k, const Slice& v,
              WriteOptions wo = WriteOptions());
 
+  Status Merge(const Slice& k, const Slice& v,
+               WriteOptions wo = WriteOptions());
+
+  Status Merge(int cf, const Slice& k, const Slice& v,
+               WriteOptions wo = WriteOptions());
+
   Status Delete(const std::string& k);
 
   Status Delete(int cf, const std::string& k);
@@ -827,7 +833,11 @@ class DBTestBase : public testing::Test {
 
   void VerifyDBFromMap(std::map<std::string, std::string> true_data,
                        size_t* total_reads_res = nullptr,
-                       bool tailing_iter = false);
+                       bool tailing_iter = false,
+                       std::map<std::string, Status> status = {});
+
+  void VerifyDBInternal(
+      std::vector<std::pair<std::string, std::string>> true_data);
 
 #ifndef ROCKSDB_LITE
   uint64_t GetNumberOfSstFilesForColumnFamily(DB* db,

db/external_sst_file_ingestion_job.cc

@@ -38,6 +38,15 @@ Status ExternalSstFileIngestionJob::Prepare(
     files_to_ingest_.push_back(file_to_ingest);
   }
 
+  for (const IngestedFileInfo& f : files_to_ingest_) {
+    if (f.cf_id !=
+            TablePropertiesCollectorFactory::Context::kUnknownColumnFamily &&
+        f.cf_id != cfd_->GetID()) {
+      return Status::InvalidArgument(
+          "External file column family id dont match");
+    }
+  }
+
   const Comparator* ucmp = cfd_->internal_comparator().user_comparator();
   auto num_files = files_to_ingest_.size();
   if (num_files == 0) {
@@ -302,8 +311,14 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
   file_to_ingest->num_entries = props->num_entries;
 
   ParsedInternalKey key;
-  std::unique_ptr<InternalIterator> iter(
-      table_reader->NewIterator(ReadOptions()));
+  ReadOptions ro;
+  // During reading the external file we can cache blocks that we read into
+  // the block cache, if we later change the global seqno of this file, we will
+  // have block in cache that will include keys with wrong seqno.
+  // We need to disable fill_cache so that we read from the file without
+  // updating the block cache.
+  ro.fill_cache = false;
+  std::unique_ptr<InternalIterator> iter(table_reader->NewIterator(ro));
 
   // Get first (smallest) key from file
   iter->SeekToFirst();
@@ -325,6 +340,10 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
   }
   file_to_ingest->largest_user_key = key.user_key.ToString();
 
+  file_to_ingest->cf_id = static_cast<uint32_t>(props->column_family_id);
+
+  file_to_ingest->table_properties = *props;
+
   return status;
 }
 

db/external_sst_file_ingestion_job.h

@@ -36,6 +36,10 @@ struct IngestedFileInfo {
   uint64_t file_size;
   // total number of keys in external file
   uint64_t num_entries;
+  // Id of column family this file shoule be ingested into
+  uint32_t cf_id;
+  // TableProperties read from external file
+  TableProperties table_properties;
   // Version of external file
   int version;
 
@@ -96,6 +100,10 @@ class ExternalSstFileIngestionJob {
 
   VersionEdit* edit() { return &edit_; }
 
+  const autovector<IngestedFileInfo>& files_to_ingest() const {
+    return files_to_ingest_;
+  }
+
  private:
   // Open the external file and populate `file_to_ingest` with all the
   // external information we need to ingest this file.

db/external_sst_file_test.cc

@@ -15,7 +15,7 @@ namespace rocksdb {
 class ExternalSSTFileTest : public DBTestBase {
  public:
   ExternalSSTFileTest() : DBTestBase("/external_sst_file_test") {
-    sst_files_dir_ = test::TmpDir(env_) + "/sst_files/";
+    sst_files_dir_ = dbname_ + "/sst_files/";
     DestroyAndRecreateExternalSSTFilesDir();
   }
 
@@ -28,7 +28,8 @@ class ExternalSSTFileTest : public DBTestBase {
       const Options options,
       std::vector<std::pair<std::string, std::string>> data, int file_id = -1,
       bool allow_global_seqno = false, bool sort_data = false,
-      std::map<std::string, std::string>* true_data = nullptr) {
+      std::map<std::string, std::string>* true_data = nullptr,
+      ColumnFamilyHandle* cfh = nullptr) {
     // Generate a file id if not provided
     if (file_id == -1) {
       file_id = last_file_id_ + 1;
@@ -51,7 +52,8 @@ class ExternalSSTFileTest : public DBTestBase {
       data.resize(uniq_iter - data.begin());
     }
     std::string file_path = sst_files_dir_ + ToString(file_id);
-    SstFileWriter sst_file_writer(EnvOptions(), options, options.comparator);
+    SstFileWriter sst_file_writer(EnvOptions(), options, options.comparator,
+                                  cfh);
 
     Status s = sst_file_writer.Open(file_path);
     if (!s.ok()) {
@@ -69,7 +71,11 @@ class ExternalSSTFileTest : public DBTestBase {
     if (s.ok()) {
       IngestExternalFileOptions ifo;
       ifo.allow_global_seqno = allow_global_seqno;
-      s = db_->IngestExternalFile({file_path}, ifo);
+      if (cfh) {
+        s = db_->IngestExternalFile(cfh, {file_path}, ifo);
+      } else {
+        s = db_->IngestExternalFile({file_path}, ifo);
+      }
     }
 
     if (s.ok() && true_data) {
@@ -84,25 +90,29 @@ class ExternalSSTFileTest : public DBTestBase {
   Status GenerateAndAddExternalFile(
       const Options options, std::vector<std::pair<int, std::string>> data,
       int file_id = -1, bool allow_global_seqno = false, bool sort_data = false,
-      std::map<std::string, std::string>* true_data = nullptr) {
+      std::map<std::string, std::string>* true_data = nullptr,
+      ColumnFamilyHandle* cfh = nullptr) {
     std::vector<std::pair<std::string, std::string>> file_data;
     for (auto& entry : data) {
       file_data.emplace_back(Key(entry.first), entry.second);
     }
     return GenerateAndAddExternalFile(options, file_data, file_id,
-                                      allow_global_seqno, sort_data, true_data);
+                                      allow_global_seqno, sort_data, true_data,
+                                      cfh);
   }
 
   Status GenerateAndAddExternalFile(
       const Options options, std::vector<int> keys, int file_id = -1,
       bool allow_global_seqno = false, bool sort_data = false,
-      std::map<std::string, std::string>* true_data = nullptr) {
+      std::map<std::string, std::string>* true_data = nullptr,
+      ColumnFamilyHandle* cfh = nullptr) {
     std::vector<std::pair<std::string, std::string>> file_data;
     for (auto& k : keys) {
       file_data.emplace_back(Key(k), Key(k) + ToString(file_id));
     }
     return GenerateAndAddExternalFile(options, file_data, file_id,
-                                      allow_global_seqno, sort_data, true_data);
+                                      allow_global_seqno, sort_data, true_data,
+                                      cfh);
   }
 
   Status DeprecatedAddFile(const std::vector<std::string>& files,
@@ -1780,6 +1790,151 @@ TEST_F(ExternalSSTFileTest, DirtyExit) {
   ASSERT_NOK(sst_file_writer->Finish());
 }
 
+TEST_F(ExternalSSTFileTest, FileWithCFInfo) {
+  Options options = CurrentOptions();
+  CreateAndReopenWithCF({"koko", "toto"}, options);
+
+  SstFileWriter sfw_default(EnvOptions(), options, options.comparator,
+                            handles_[0]);
+  SstFileWriter sfw_cf1(EnvOptions(), options, options.comparator, handles_[1]);
+  SstFileWriter sfw_cf2(EnvOptions(), options, options.comparator, handles_[2]);
+  SstFileWriter sfw_unknown(EnvOptions(), options, options.comparator);
+
+  // default_cf.sst
+  const std::string cf_default_sst = sst_files_dir_ + "/default_cf.sst";
+  ASSERT_OK(sfw_default.Open(cf_default_sst));
+  ASSERT_OK(sfw_default.Add("K1", "V1"));
+  ASSERT_OK(sfw_default.Add("K2", "V2"));
+  ASSERT_OK(sfw_default.Finish());
+
+  // cf1.sst
+  const std::string cf1_sst = sst_files_dir_ + "/cf1.sst";
+  ASSERT_OK(sfw_cf1.Open(cf1_sst));
+  ASSERT_OK(sfw_cf1.Add("K3", "V1"));
+  ASSERT_OK(sfw_cf1.Add("K4", "V2"));
+  ASSERT_OK(sfw_cf1.Finish());
+
+  // cf_unknown.sst
+  const std::string unknown_sst = sst_files_dir_ + "/cf_unknown.sst";
+  ASSERT_OK(sfw_unknown.Open(unknown_sst));
+  ASSERT_OK(sfw_unknown.Add("K5", "V1"));
+  ASSERT_OK(sfw_unknown.Add("K6", "V2"));
+  ASSERT_OK(sfw_unknown.Finish());
+
+  IngestExternalFileOptions ifo;
+
+  // SST CF dont match
+  ASSERT_NOK(db_->IngestExternalFile(handles_[0], {cf1_sst}, ifo));
+  // SST CF dont match
+  ASSERT_NOK(db_->IngestExternalFile(handles_[2], {cf1_sst}, ifo));
+  // SST CF match
+  ASSERT_OK(db_->IngestExternalFile(handles_[1], {cf1_sst}, ifo));
+
+  // SST CF dont match
+  ASSERT_NOK(db_->IngestExternalFile(handles_[1], {cf_default_sst}, ifo));
+  // SST CF dont match
+  ASSERT_NOK(db_->IngestExternalFile(handles_[2], {cf_default_sst}, ifo));
+  // SST CF match
+  ASSERT_OK(db_->IngestExternalFile(handles_[0], {cf_default_sst}, ifo));
+
+  // SST CF unknown
+  ASSERT_OK(db_->IngestExternalFile(handles_[1], {unknown_sst}, ifo));
+  // SST CF unknown
+  ASSERT_OK(db_->IngestExternalFile(handles_[2], {unknown_sst}, ifo));
+  // SST CF unknown
+  ASSERT_OK(db_->IngestExternalFile(handles_[0], {unknown_sst}, ifo));
+
+  // Cannot ingest a file into a dropped CF
+  ASSERT_OK(db_->DropColumnFamily(handles_[1]));
+  ASSERT_NOK(db_->IngestExternalFile(handles_[1], {unknown_sst}, ifo));
+
+  // CF was not dropped, ok to Ingest
+  ASSERT_OK(db_->IngestExternalFile(handles_[2], {unknown_sst}, ifo));
+}
+
+class TestIngestExternalFileListener : public EventListener {
+ public:
+  void OnExternalFileIngested(DB* db,
+                              const ExternalFileIngestionInfo& info) override {
+    ingested_files.push_back(info);
+  }
+
+  std::vector<ExternalFileIngestionInfo> ingested_files;
+};
+
+TEST_F(ExternalSSTFileTest, IngestionListener) {
+  Options options = CurrentOptions();
+  TestIngestExternalFileListener* listener =
+      new TestIngestExternalFileListener();
+  options.listeners.emplace_back(listener);
+  CreateAndReopenWithCF({"koko", "toto"}, options);
+
+  // Ingest into default cf
+  ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2}, -1, true, true, nullptr,
+                                       handles_[0]));
+  ASSERT_EQ(listener->ingested_files.size(), 1);
+  ASSERT_EQ(listener->ingested_files.back().cf_name, "default");
+  ASSERT_EQ(listener->ingested_files.back().global_seqno, 0);
+  ASSERT_EQ(listener->ingested_files.back().table_properties.column_family_id,
+            0);
+  ASSERT_EQ(listener->ingested_files.back().table_properties.column_family_name,
+            "default");
+
+  // Ingest into cf1
+  ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2}, -1, true, true, nullptr,
+                                       handles_[1]));
+  ASSERT_EQ(listener->ingested_files.size(), 2);
+  ASSERT_EQ(listener->ingested_files.back().cf_name, "koko");
+  ASSERT_EQ(listener->ingested_files.back().global_seqno, 0);
+  ASSERT_EQ(listener->ingested_files.back().table_properties.column_family_id,
+            1);
+  ASSERT_EQ(listener->ingested_files.back().table_properties.column_family_name,
+            "koko");
+
+  // Ingest into cf2
+  ASSERT_OK(GenerateAndAddExternalFile(options, {1, 2}, -1, true, true, nullptr,
+                                       handles_[2]));
+  ASSERT_EQ(listener->ingested_files.size(), 3);
+  ASSERT_EQ(listener->ingested_files.back().cf_name, "toto");
+  ASSERT_EQ(listener->ingested_files.back().global_seqno, 0);
+  ASSERT_EQ(listener->ingested_files.back().table_properties.column_family_id,
+            2);
+  ASSERT_EQ(listener->ingested_files.back().table_properties.column_family_name,
+            "toto");
+}
+
+TEST_F(ExternalSSTFileTest, SnapshotInconsistencyBug) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+  const int kNumKeys = 10000;
+
+  // Insert keys using normal path and take a snapshot
+  for (int i = 0; i < kNumKeys; i++) {
+    ASSERT_OK(Put(Key(i), Key(i) + "_V1"));
+  }
+  const Snapshot* snap = db_->GetSnapshot();
+
+  // Overwrite all keys using IngestExternalFile
+  std::string sst_file_path = sst_files_dir_ + "file1.sst";
+  SstFileWriter sst_file_writer(EnvOptions(), options, options.comparator);
+  ASSERT_OK(sst_file_writer.Open(sst_file_path));
+  for (int i = 0; i < kNumKeys; i++) {
+    ASSERT_OK(sst_file_writer.Add(Key(i), Key(i) + "_V2"));
+  }
+  ASSERT_OK(sst_file_writer.Finish());
+
+  IngestExternalFileOptions ifo;
+  ifo.move_files = true;
+  ASSERT_OK(db_->IngestExternalFile({sst_file_path}, ifo));
+
+  for (int i = 0; i < kNumKeys; i++) {
+    ASSERT_EQ(Get(Key(i), snap), Key(i) + "_V1");
+    ASSERT_EQ(Get(Key(i)), Key(i) + "_V2");
+  }
+
+  db_->ReleaseSnapshot(snap);
+}
+
 #endif  // ROCKSDB_LITE
 
 }  // namespace rocksdb

db/flush_job.cc

@@ -277,7 +277,8 @@ Status FlushJob::WriteLevel0Table() {
           NewMergingIterator(&cfd_->internal_comparator(), &memtables[0],
                              static_cast<int>(memtables.size()), &arena));
       std::unique_ptr<InternalIterator> range_del_iter(NewMergingIterator(
-          &cfd_->internal_comparator(), &range_del_iters[0],
+          &cfd_->internal_comparator(),
+          range_del_iters.empty() ? nullptr : &range_del_iters[0],
           static_cast<int>(range_del_iters.size())));
       Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
           "[%s] [JOB %d] Level-0 flush table #%" PRIu64 ": started",

include/rocksdb/listener.h

@@ -170,6 +170,20 @@ struct MemTableInfo {
 
 };
 
+struct ExternalFileIngestionInfo {
+  // the name of the column family
+  std::string cf_name;
+  // Path of the file outside the DB
+  std::string external_file_path;
+  // Path of the file inside the DB
+  std::string internal_file_path;
+  // The global sequence number assigned to keys in this file
+  SequenceNumber global_seqno;
+  // Table properties of the table being flushed
+  TableProperties table_properties;
+};
+
+
 // EventListener class contains a set of call-back functions that will
 // be called when specific RocksDB event happens such as flush.  It can
 // be used as a building block for developing custom features such as
@@ -291,6 +305,15 @@ class EventListener {
   virtual void OnColumnFamilyHandleDeletionStarted(ColumnFamilyHandle* handle) {
   }
 
+  // A call-back function for RocksDB which will be called after an external
+  // file is ingested using IngestExternalFile.
+  //
+  // Note that the this function will run on the same thread as
+  // IngestExternalFile(), if this function is blocked, IngestExternalFile()
+  // will be blocked from finishing.
+  virtual void OnExternalFileIngested(
+      DB* /*db*/, const ExternalFileIngestionInfo& /*info*/) {}
+
   virtual ~EventListener() {}
 };
 

include/rocksdb/sst_file_writer.h

@@ -7,6 +7,7 @@
 #include <string>
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
+#include "rocksdb/table_properties.h"
 #include "rocksdb/types.h"
 
 namespace rocksdb {
@@ -43,8 +44,12 @@ struct ExternalSstFileInfo {
 // All keys in files generated by SstFileWriter will have sequence number = 0
 class SstFileWriter {
  public:
+  // User can pass `column_family` to specify that the the generated file will
+  // be ingested into this column_family, note that passing nullptr means that
+  // the column_family is unknown.
   SstFileWriter(const EnvOptions& env_options, const Options& options,
-                const Comparator* user_comparator);
+                const Comparator* user_comparator,
+                ColumnFamilyHandle* column_family = nullptr);
 
   ~SstFileWriter();
 

include/rocksdb/version.h

@@ -6,7 +6,7 @@
 
 #define ROCKSDB_MAJOR 5
 #define ROCKSDB_MINOR 0
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 2
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

src.mk

@@ -96,7 +96,7 @@ LIB_SOURCES =                                                   \
   util/compaction_job_stats_impl.cc                             \
   util/concurrent_arena.cc                                      \
   util/crc32c.cc                                                \
-	util/db_options.cc                                            \
+  util/db_options.cc                                            \
   util/delete_scheduler.cc                                      \
   util/dynamic_bloom.cc                                         \
   util/env.cc                                                   \
@@ -126,7 +126,7 @@ LIB_SOURCES =                                                   \
   util/perf_level.cc                                            \
   util/random.cc                                                \
   util/rate_limiter.cc                                          \
-	util/sharded_cache.cc       																	\
+  util/sharded_cache.cc                                         \
   util/slice.cc                                                 \
   util/sst_file_manager_impl.cc                                 \
   util/statistics.cc                                            \
@@ -223,18 +223,20 @@ MAIN_SOURCES =                                                    \
   db/dbformat_test.cc                                                   \
   db/db_iter_test.cc                                                    \
   db/db_test.cc                                                         \
-  db/db_block_cache_test.cc						\
+  db/db_block_cache_test.cc                                             \
   db/db_io_failure_test.cc                                              \
   db/db_bloom_filter_test.cc                                            \
   db/db_compaction_filter_test.cc                                       \
   db/db_compaction_test.cc                                              \
   db/db_dynamic_level_test.cc                                           \
-  db/db_flush_test.cc							\
+  db/db_flush_test.cc                                                    \
   db/db_inplace_update_test.cc                                          \
-  db/db_iterator_test.cc						\
+  db/db_iterator_test.cc                                                \
   db/db_log_iter_test.cc                                                \
+  db/db_memtable_test.cc                                                \
+  db/db_merge_operator_test.cc                                          \
   db/db_options_test.cc                                                 \
-  db/db_range_del_test.cc						\
+  db/db_range_del_test.cc                                               \
   db/db_sst_test.cc                                                     \
   db/external_sst_file_test.cc                                          \
   db/db_tailing_iter_test.cc                                            \
@@ -314,7 +316,7 @@ MAIN_SOURCES =                                                    \
   utilities/write_batch_with_index/write_batch_with_index_test.cc       \
   utilities/column_aware_encoding_test.cc                               \
   utilities/lua/rocks_lua_test.cc                                       \
-	util/iostats_context_test.cc																					\
+  util/iostats_context_test.cc                                          \
   util/log_write_bench.cc                                               \
   util/mock_env_test.cc                                                 \
   util/options_test.cc                                                  \

table/sst_file_writer.cc

@@ -21,11 +21,12 @@ const std::string ExternalSstFilePropertyNames::kGlobalSeqno =
 
 struct SstFileWriter::Rep {
   Rep(const EnvOptions& _env_options, const Options& options,
-      const Comparator* _user_comparator)
+      const Comparator* _user_comparator, ColumnFamilyHandle* _cfh)
       : env_options(_env_options),
         ioptions(options),
         mutable_cf_options(options),
-        internal_comparator(_user_comparator) {}
+        internal_comparator(_user_comparator),
+        cfh(_cfh) {}
 
   std::unique_ptr<WritableFileWriter> file_writer;
   std::unique_ptr<TableBuilder> builder;
@@ -34,14 +35,16 @@ struct SstFileWriter::Rep {
   MutableCFOptions mutable_cf_options;
   InternalKeyComparator internal_comparator;
   ExternalSstFileInfo file_info;
-  std::string column_family_name;
   InternalKey ikey;
+  std::string column_family_name;
+  ColumnFamilyHandle* cfh;
 };
 
 SstFileWriter::SstFileWriter(const EnvOptions& env_options,
                              const Options& options,
-                             const Comparator* user_comparator)
-    : rep_(new Rep(env_options, options, user_comparator)) {}
+                             const Comparator* user_comparator,
+                             ColumnFamilyHandle* column_family)
+    : rep_(new Rep(env_options, options, user_comparator, column_family)) {}
 
 SstFileWriter::~SstFileWriter() {
   if (rep_->builder) {
@@ -89,6 +92,18 @@ Status SstFileWriter::Open(const std::string& file_path) {
             user_collector_factories[i]));
   }
   int unknown_level = -1;
+  uint32_t cf_id;
+
+  if (r->cfh != nullptr) {
+    // user explicitly specified that this file will be ingested into cfh,
+    // we can persist this information in the file.
+    cf_id = r->cfh->GetID();
+    r->column_family_name = r->cfh->GetName();
+  } else {
+    r->column_family_name = "";
+    cf_id = TablePropertiesCollectorFactory::Context::kUnknownColumnFamily;
+  }
+
   TableBuilderOptions table_builder_options(
       r->ioptions, r->internal_comparator, &int_tbl_prop_collector_factories,
       compression_type, r->ioptions.compression_opts,
@@ -100,9 +115,7 @@ Status SstFileWriter::Open(const std::string& file_path) {
   // TODO(tec) : If table_factory is using compressed block cache, we will
   // be adding the external sst file blocks into it, which is wasteful.
   r->builder.reset(r->ioptions.table_factory->NewTableBuilder(
-      table_builder_options,
-      TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
-      r->file_writer.get()));
+      table_builder_options, cf_id, r->file_writer.get()));
 
   r->file_info.file_path = file_path;
   r->file_info.file_size = 0;