Introduce column family option cf_allow_ingest_behind (#13810)

Summary:
this option has the same functionality as DBOptions::allow_ingest_behind but allows the feature at per CF level. `DBOptions::allow_ingest_behind` is deprecated after this PR and users should use `cf_allow_ingest_behind` instead.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/13810

Test Plan: updated some existing tests to use the new option.

Reviewed By: xingbowang

Differential Revision: D79191969

Pulled By: cbi42

fbshipit-source-id: 0da45f6be472ace6754ad15df93d45ac86313837

db/column_family.cc

@@ -280,7 +280,8 @@ ColumnFamilyOptions SanitizeCfOptions(const ImmutableDBOptions& db_options,
   }
 
   if (result.compaction_style == kCompactionStyleUniversal &&
-      db_options.allow_ingest_behind && result.num_levels < 3) {
+      (db_options.allow_ingest_behind || result.cf_allow_ingest_behind) &&
+      result.num_levels < 3) {
     result.num_levels = 3;
   }
 

db/column_family.h

@@ -600,6 +600,11 @@ class ColumnFamilyData {
     return (mem_->IsEmpty() ? 0 : 1) + imm_.NumNotFlushed();
   }
 
+  // thread-safe, DB mutex not needed.
+  bool AllowIngestBehind() const {
+    return ioptions_.cf_allow_ingest_behind || ioptions_.allow_ingest_behind;
+  }
+
  private:
   friend class ColumnFamilySet;
   ColumnFamilyData(

db/compaction/compaction_iterator.h

@@ -145,7 +145,8 @@ class CompactionIterator {
     }
 
     bool allow_ingest_behind() const override {
-      return compaction_->immutable_options().allow_ingest_behind;
+      return compaction_->immutable_options().cf_allow_ingest_behind ||
+             compaction_->immutable_options().allow_ingest_behind;
     }
 
     bool allow_mmap_reads() const override {

db/compaction/compaction_picker.cc

@@ -619,8 +619,8 @@ Compaction* CompactionPicker::PickCompactionForCompactRange(
     // Universal compaction with more than one level always compacts all the
     // files together to the last level.
     assert(vstorage->num_levels() > 1);
-    int max_output_level =
-        vstorage->MaxOutputLevel(ioptions_.allow_ingest_behind);
+    int max_output_level = vstorage->MaxOutputLevel(
+        ioptions_.cf_allow_ingest_behind || ioptions_.allow_ingest_behind);
     // DBImpl::CompactRange() set output level to be the last level
     assert(output_level == max_output_level);
     // DBImpl::RunManualCompaction will make full range for universal compaction

db/compaction/compaction_picker_test.cc

@@ -544,9 +544,15 @@ TEST_F(CompactionPickerTest, NeedsCompactionUniversal) {
 }
 
 TEST_F(CompactionPickerTest, CompactionUniversalIngestBehindReservedLevel) {
+  for (bool cf_option : {false, true}) {
+    SCOPED_TRACE("cf_option = " + std::to_string(cf_option));
     const uint64_t kFileSize = 100000;
     NewVersionStorage(3 /* num_levels */, kCompactionStyleUniversal);
+    if (cf_option) {
+      ioptions_.cf_allow_ingest_behind = true;
+    } else {
       ioptions_.allow_ingest_behind = true;
+    }
     ioptions_.num_levels = 3;
     UniversalCompactionPicker universal_compaction_picker(ioptions_, &icmp_);
     UpdateVersionStorageInfo();
@@ -581,6 +587,7 @@ TEST_F(CompactionPickerTest, CompactionUniversalIngestBehindReservedLevel) {
         ASSERT_LT(compaction_input.level, 2);
       }
     }
+  }
 }
 // Tests if the files can be trivially moved in multi level
 // universal compaction when allow_trivial_move option is set

db/compaction/compaction_picker_universal.cc

@@ -48,7 +48,9 @@ class UniversalCompactionBuilder {
         vstorage_(vstorage),
         picker_(picker),
         log_buffer_(log_buffer),
-        require_max_output_level_(require_max_output_level) {
+        require_max_output_level_(require_max_output_level),
+        allow_ingest_behind_(ioptions.cf_allow_ingest_behind ||
+                             ioptions.allow_ingest_behind) {
     assert(icmp_);
     const auto* ucmp = icmp_->user_comparator();
     assert(ucmp);
@@ -422,8 +424,7 @@ class UniversalCompactionBuilder {
   bool MeetsOutputLevelRequirements(int output_level) const {
     return !require_max_output_level_ ||
            Compaction::OutputToNonZeroMaxOutputLevel(
-               output_level,
-               vstorage_->MaxOutputLevel(ioptions_.allow_ingest_behind));
+               output_level, vstorage_->MaxOutputLevel(allow_ingest_behind_));
   }
 
   const ImmutableOptions& ioptions_;
@@ -437,7 +438,6 @@ class UniversalCompactionBuilder {
   VersionStorageInfo* vstorage_;
   UniversalCompactionPicker* picker_;
   LogBuffer* log_buffer_;
-  bool require_max_output_level_;
   // Optional earliest snapshot at time of compaction picking. This is only
   // provided if the column family doesn't enable user-defined timestamps.
   // And this information is only passed to `Compaction` picked by deletion
@@ -448,6 +448,8 @@ class UniversalCompactionBuilder {
   // marked for compaction. This is only populated when snapshot info is
   // populated.
   std::map<uint64_t, size_t> file_marked_for_compaction_to_sorted_run_index_;
+  bool require_max_output_level_;
+  bool allow_ingest_behind_;
 
   std::vector<UniversalCompactionBuilder::SortedRun> CalculateSortedRuns(
       const VersionStorageInfo& vstorage, int last_level,
@@ -733,8 +735,7 @@ bool UniversalCompactionBuilder::ShouldSkipMarkedFile(
 Compaction* UniversalCompactionBuilder::PickCompaction() {
   const int kLevel0 = 0;
   score_ = vstorage_->CompactionScore(kLevel0);
-  const int max_output_level =
-      vstorage_->MaxOutputLevel(ioptions_.allow_ingest_behind);
+  const int max_output_level = vstorage_->MaxOutputLevel(allow_ingest_behind_);
   const int file_num_compaction_trigger =
       mutable_cf_options_.level0_file_num_compaction_trigger;
   const unsigned int ratio =
@@ -781,8 +782,7 @@ Compaction* UniversalCompactionBuilder::PickCompaction() {
         "UniversalCompactionBuilder::PickCompaction:Return", nullptr);
     return nullptr;
   }
-  assert(c->output_level() <=
-         vstorage_->MaxOutputLevel(ioptions_.allow_ingest_behind));
+  assert(c->output_level() <= vstorage_->MaxOutputLevel(allow_ingest_behind_));
   assert(MeetsOutputLevelRequirements(c->output_level()));
 
   if (mutable_cf_options_.compaction_options_universal.allow_trivial_move ==
@@ -1024,8 +1024,7 @@ Compaction* UniversalCompactionBuilder::PickCompactionToReduceSortedRuns(
   int start_level = sorted_runs_[start_index].level;
   int output_level;
   // last level is reserved for the files ingested behind
-  int max_output_level =
-      vstorage_->MaxOutputLevel(ioptions_.allow_ingest_behind);
+  int max_output_level = vstorage_->MaxOutputLevel(allow_ingest_behind_);
   if (first_index_after == sorted_runs_.size()) {
     output_level = max_output_level;
   } else if (sorted_runs_[first_index_after].level == 0) {
@@ -1517,8 +1516,7 @@ Compaction* UniversalCompactionBuilder::PickDeleteTriggeredCompaction() {
       return nullptr;
     }
 
-    int max_output_level =
-        vstorage_->MaxOutputLevel(ioptions_.allow_ingest_behind);
+    int max_output_level = vstorage_->MaxOutputLevel(allow_ingest_behind_);
     // Pick the first non-empty level after the start_level
     for (output_level = start_level + 1; output_level <= max_output_level;
          output_level++) {
@@ -1621,8 +1619,7 @@ Compaction* UniversalCompactionBuilder::PickCompactionWithSortedRunRange(
   uint32_t path_id =
       GetPathId(ioptions_, mutable_cf_options_, estimated_total_size);
   int start_level = sorted_runs_[start_index].level;
-  int max_output_level =
-      vstorage_->MaxOutputLevel(ioptions_.allow_ingest_behind);
+  int max_output_level = vstorage_->MaxOutputLevel(allow_ingest_behind_);
   std::vector<CompactionInputFiles> inputs(max_output_level + 1);
   for (size_t i = 0; i < inputs.size(); ++i) {
     inputs[i].level = start_level + static_cast<int>(i);

db/db_impl/db_impl.cc

@@ -4345,7 +4345,7 @@ void DBImpl::ReleaseSnapshot(const Snapshot* s) {
     CfdList cf_scheduled;
     if (oldest_snapshot > bottommost_files_mark_threshold_) {
       for (auto* cfd : *versions_->GetColumnFamilySet()) {
-        if (!cfd->ioptions().allow_ingest_behind) {
+        if (!cfd->AllowIngestBehind()) {
           cfd->current()->storage_info()->UpdateOldestSnapshot(
               oldest_snapshot, /*allow_ingest_behind=*/false);
           if (!cfd->current()
@@ -4365,8 +4365,7 @@ void DBImpl::ReleaseSnapshot(const Snapshot* s) {
       // inaccurate.
       SequenceNumber new_bottommost_files_mark_threshold = kMaxSequenceNumber;
       for (auto* cfd : *versions_->GetColumnFamilySet()) {
-        if (CfdListContains(cf_scheduled, cfd) ||
-            cfd->ioptions().allow_ingest_behind) {
+        if (CfdListContains(cf_scheduled, cfd) || cfd->AllowIngestBehind()) {
           continue;
         }
         new_bottommost_files_mark_threshold = std::min(
@@ -5761,10 +5760,6 @@ Status DBImpl::IngestExternalFiles(
   for (const auto& arg : args) {
     const IngestExternalFileOptions& ingest_opts = arg.options;
     if (ingest_opts.ingest_behind) {
-      if (!immutable_db_options_.allow_ingest_behind) {
-        return Status::InvalidArgument(
-            "can't ingest_behind file in DB with allow_ingest_behind=false");
-      }
       auto ucmp = arg.column_family->GetComparator();
       assert(ucmp);
       if (ucmp->timestamp_size() > 0) {
@@ -5772,6 +5767,14 @@ Status DBImpl::IngestExternalFiles(
             "Column family with user-defined "
             "timestamps enabled doesn't support ingest behind.");
       }
+
+      if (!static_cast<ColumnFamilyHandleImpl*>(arg.column_family)
+               ->cfd()
+               ->AllowIngestBehind()) {
+        return Status::InvalidArgument(
+            "Can't ingest_behind file in ColumnFamily %s with "
+            "cf_allow_ingest_behind=false");
+      }
     }
     if (arg.atomic_replace_range.has_value()) {
       if (ingest_opts.ingest_behind) {

db/db_impl/db_impl.h

@@ -1388,6 +1388,9 @@ class DBImpl : public DB {
   // logs_, cur_wal_number_. Refer to the definition of each variable below for
   // more description.
   //
+  // Protects access to most ColumnFamilyData methods, see more in comment for
+  // each method.
+  //
   // `mutex_` can be a hot lock in some workloads, so it deserves dedicated
   // cachelines.
   mutable CacheAlignedInstrumentedMutex mutex_;

db/db_impl/db_impl_compaction_flush.cc

@@ -1111,8 +1111,7 @@ Status DBImpl::CompactRangeInternal(const CompactRangeOptions& options,
       cfd->NumberLevels() > 1) {
     // Always compact all files together.
     final_output_level = cfd->NumberLevels() - 1;
-    // if bottom most level is reserved
-    if (immutable_db_options_.allow_ingest_behind) {
+    if (cfd->AllowIngestBehind()) {
       final_output_level--;
     }
     s = RunManualCompaction(cfd, ColumnFamilyData::kCompactAllLevels,
@@ -1460,7 +1459,7 @@ Status DBImpl::CompactFilesImpl(
     }
   }
 
-  if (cfd->ioptions().allow_ingest_behind &&
+  if (cfd->AllowIngestBehind() &&
       output_level >= cfd->ioptions().num_levels - 1) {
     return Status::InvalidArgument(
         "Exceed the maximum output level defined by "
@@ -4155,6 +4154,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
                      ->current()
                      ->storage_info()
                      ->MaxOutputLevel(
+                         c->immutable_options().cf_allow_ingest_behind ||
                          immutable_db_options_.allow_ingest_behind)) &&
              env_->GetBackgroundThreads(Env::Priority::BOTTOM) > 0) {
     assert(thread_pri == Env::Priority::LOW);
@@ -4660,7 +4660,7 @@ void DBImpl::InstallSuperVersionAndScheduleWork(
   bottommost_files_mark_threshold_ = kMaxSequenceNumber;
   standalone_range_deletion_files_mark_threshold_ = kMaxSequenceNumber;
   for (auto* my_cfd : *versions_->GetColumnFamilySet()) {
-    if (!my_cfd->ioptions().allow_ingest_behind) {
+    if (!my_cfd->AllowIngestBehind()) {
       bottommost_files_mark_threshold_ = std::min(
           bottommost_files_mark_threshold_,
           my_cfd->current()->storage_info()->bottommost_files_mark_threshold());

db/db_impl/db_impl_open.cc

@@ -599,7 +599,7 @@ Status DBImpl::Recover(
         // allow_ingest_behind does not support Level Compaction,
         // and per_key_placement can have infinite compaction loop for Level
         // Compaction. Adjust to_level here just to be safe.
-        if (cfd->ioptions().allow_ingest_behind ||
+        if (cfd->AllowIngestBehind() ||
             moptions.preclude_last_level_data_seconds > 0) {
           to_level -= 1;
         }

db/db_universal_compaction_test.cc

@@ -2106,6 +2106,8 @@ TEST_F(DBTestUniversalCompaction2, OverlappingL0) {
 }
 
 TEST_F(DBTestUniversalCompaction2, IngestBehind) {
+  for (bool cf_option : {false, true}) {
+    SCOPED_TRACE("cf_option = " + std::to_string(cf_option));
     const int kNumKeys = 3000;
     const int kWindowSize = 100;
     const int kNumDelsTrigger = 90;
@@ -2116,7 +2118,11 @@ TEST_F(DBTestUniversalCompaction2, IngestBehind) {
     opts.compaction_style = kCompactionStyleUniversal;
     opts.level0_file_num_compaction_trigger = 2;
     opts.compression = kNoCompression;
+    if (cf_option) {
+      opts.cf_allow_ingest_behind = true;
+    } else {
       opts.allow_ingest_behind = true;
+    }
     opts.compaction_options_universal.size_ratio = 10;
     opts.compaction_options_universal.min_merge_width = 2;
     opts.compaction_options_universal.max_size_amplification_percent = 200;
@@ -2146,6 +2152,33 @@ TEST_F(DBTestUniversalCompaction2, IngestBehind) {
     ASSERT_EQ(0, NumTableFilesAtLevel(0));
     ASSERT_EQ(0, NumTableFilesAtLevel(6));
     ASSERT_GT(NumTableFilesAtLevel(5), 0);
+
+    if (cf_option) {
+      // Test that another CF does not allow ingest behind
+      ColumnFamilyHandle* new_cfh;
+      Options new_cf_option;
+      new_cf_option.compaction_style = kCompactionStyleUniversal;
+      new_cf_option.num_levels = 7;
+      // CreateColumnFamilies({"new_cf"}, new_cf_option);
+      ASSERT_OK(db_->CreateColumnFamily(new_cf_option, "new_cf", &new_cfh));
+      // handles_.push_back(new_cfh);
+      for (i = 0; i < 10; ++i) {
+        // ASSERT_OK(Put(1, Key(i), "val"));
+        ASSERT_OK(db_->Put(WriteOptions(), new_cfh, Key(i), "val"));
+      }
+      ASSERT_OK(
+          db_->CompactRange(CompactRangeOptions(), new_cfh, nullptr, nullptr));
+      // This CF can use the last leve.
+      std::string property;
+      EXPECT_TRUE(db_->GetProperty(
+          new_cfh, "rocksdb.num-files-at-level" + std::to_string(6),
+          &property));
+      ASSERT_EQ(1, atoi(property.c_str()));
+
+      ASSERT_OK(db_->DropColumnFamily(new_cfh));
+      ASSERT_OK(db_->DestroyColumnFamilyHandle(new_cfh));
+    }
+  }
 }
 
 TEST_F(DBTestUniversalCompaction2, PeriodicCompactionDefault) {

db/external_sst_file_basic_test.cc

@@ -2567,7 +2567,14 @@ TEST_F(ExternalSSTFileBasicTest, IngestWithTemperature) {
     options.default_write_temperature = Temperature::kHot;
     SstFileWriter sst_file_writer(EnvOptions(), options);
     options.level0_file_num_compaction_trigger = 2;
+    bool cf_option = Random::GetTLSInstance()->OneIn(2);
+    SCOPED_TRACE(std::string("Use ") + (cf_option ? "CF" : "DB") +
+                 " option for ingest behind");
+    if (cf_option) {
+      options.cf_allow_ingest_behind = (mode == "ingest_behind");
+    } else {
       options.allow_ingest_behind = (mode == "ingest_behind");
+    }
     Reopen(options);
     Defer destroyer([&]() { Destroy(options); });
 

db/external_sst_file_ingestion_job.cc

@@ -1277,13 +1277,13 @@ Status ExternalSstFileIngestionJob::CheckLevelForIngestedBehindFile(
         "at the last level!");
   }
 
-  // Second, check if despite allow_ingest_behind=true we still have 0 seqnums
-  // at some upper level
+  // Second, check if despite cf_allow_ingest_behind=true we still have 0
+  // seqnums at some upper level
   for (int lvl = 0; lvl < cfd_->NumberLevels() - 1; lvl++) {
     for (auto file : vstorage->LevelFiles(lvl)) {
       if (file->fd.smallest_seqno == 0) {
         return Status::InvalidArgument(
-            "Can't ingest_behind file as despite allow_ingest_behind=true "
+            "Can't ingest_behind file as despite cf_allow_ingest_behind=true "
             "there are files with 0 seqno in database at upper levels!");
       }
     }

db/external_sst_file_ingestion_job.h

@@ -349,7 +349,7 @@ class ExternalSstFileIngestionJob {
       std::optional<int> prev_batch_uppermost_level);
 
   // File that we want to ingest behind always goes to the lowest level;
-  // we just check that it fits in the level, that DB allows ingest_behind,
+  // we just check that it fits in the level, that the CF allows ingest_behind,
   // and that we don't have 0 seqnums at the upper levels.
   // REQUIRES: Mutex held
   Status CheckLevelForIngestedBehindFile(IngestedFileInfo* file_to_ingest);

db/external_sst_file_test.cc

@@ -2417,6 +2417,8 @@ TEST_F(ExternalSSTFileTest, SnapshotInconsistencyBug) {
 }
 
 TEST_P(ExternalSSTFileTest, IngestBehind) {
+  for (bool cf_option : {false, true}) {
+    SCOPED_TRACE("cf_option = " + std::to_string(cf_option));
     Options options = CurrentOptions();
     options.compaction_style = kCompactionStyleUniversal;
     options.num_levels = 3;
@@ -2430,7 +2432,7 @@ TEST_P(ExternalSSTFileTest, IngestBehind) {
       ASSERT_OK(Put(Key(i), "memtable"));
     }
 
-  // Insert 0 -> 20 using IngestExternalFile
+    // Insert 100 -> 200 using IngestExternalFile
     file_data.clear();
     for (int i = 0; i <= 20; i++) {
       file_data.emplace_back(Key(i), "ingest_behind");
@@ -2448,7 +2450,11 @@ TEST_P(ExternalSSTFileTest, IngestBehind) {
         verify_checksums_before_ingest, ingest_behind, false /*sort_data*/,
         &true_data));
 
+    if (cf_option) {
+      options.cf_allow_ingest_behind = true;
+    } else {
       options.allow_ingest_behind = true;
+    }
     // check that we still can open the DB, as num_levels should be
     // sanitized to 3
     options.num_levels = 2;
@@ -2462,41 +2468,13 @@ TEST_P(ExternalSSTFileTest, IngestBehind) {
       ASSERT_OK(Put(Key(i), "memtable"));
       true_data[Key(i)] = "memtable";
     }
-
-  // Test that tombstones for Key(7) and Key(8) are not dropped during
-  // compaction. Will verify below that after ingesting Puts for Key(7) and
-  // Key(8), they are covered by these two tombstones.
-  ASSERT_OK(Delete(Key(7)));
-  ASSERT_OK(SingleDelete(Key(8)));
     ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
     // Universal picker should go at second from the bottom level
     ASSERT_EQ("0,1", FilesPerLevel());
-
-  // Test that SingleDelte overwritten by Put is not dropped.
-  // From old to new, we issue SD, PUT, CompactRange, SD, CompactRange. The
-  // first CompactRange() should not drop the overwritten SD. The second
-  // CompactRange() will drop the new SD with PUT. If the older SD was dropped,
-  // the ingested behind data will be incorrectly visible below.
-  ASSERT_OK(SingleDelete(Key(1)));
-  ASSERT_OK(Put(Key(1), "overwrite_sd"));
-  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
-  ASSERT_OK(SingleDelete(Key(1)));
-  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
-
     ASSERT_OK(GenerateAndAddExternalFile(
         options, file_data, -1, allow_global_seqno, write_global_seqno,
         verify_checksums_before_ingest, true /*ingest_behind*/,
         false /*sort_data*/, &true_data));
-  // adjust expected data for tombtones
-  true_data.erase(Key(7));
-  true_data.erase(Key(8));
-  true_data.erase(Key(1));
-  std::unordered_set<std::string> not_found_set;
-  // Tombstones will be verified in VerifyDBFromMap() below.
-  not_found_set.insert(Key(7));
-  not_found_set.insert(Key(8));
-  not_found_set.insert(Key(1));
-
     ASSERT_EQ("0,1,1", FilesPerLevel());
     // this time ingest should fail as the file doesn't fit to the bottom level
     ASSERT_NOK(GenerateAndAddExternalFile(
@@ -2505,15 +2483,17 @@ TEST_P(ExternalSSTFileTest, IngestBehind) {
         false /*sort_data*/, &true_data));
     ASSERT_EQ("0,1,1", FilesPerLevel());
     std::vector<std::vector<FileMetaData>> level_to_files;
-  dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(), &level_to_files);
+    dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(),
+                                    &level_to_files);
     uint64_t ingested_file_number = level_to_files[2][0].fd.GetNumber();
     ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
     // Last level should not be compacted
     ASSERT_EQ("0,1,1", FilesPerLevel());
-  dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(), &level_to_files);
+    dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(),
+                                    &level_to_files);
     ASSERT_EQ(ingested_file_number, level_to_files[2][0].fd.GetNumber());
     size_t kcnt = 0;
-  VerifyDBFromMap(true_data, &kcnt, false, nullptr, nullptr, &not_found_set);
+    VerifyDBFromMap(true_data, &kcnt, false);
 
     // Auto-compaction should not include the last level.
     // Trigger compaction if size amplification exceeds 110%.
@@ -2529,18 +2509,38 @@ TEST_P(ExternalSSTFileTest, IngestBehind) {
       ASSERT_OK(Flush());
     }
     ASSERT_OK(dbfull()->TEST_WaitForCompact());
-  dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(), &level_to_files);
+    dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(),
+                                    &level_to_files);
     ASSERT_EQ(1, level_to_files[2].size());
     ASSERT_EQ(ingested_file_number, level_to_files[2][0].fd.GetNumber());
 
     // Turning off the option allows DB to compact ingested files.
+    if (cf_option) {
+      // Test that another CF does not allow ingest behind
+      ColumnFamilyHandle* new_cfh;
+      Options new_cf_option;
+      ASSERT_OK(db_->CreateColumnFamily(new_cf_option, "new_cf", &new_cfh));
+      ASSERT_TRUE(GenerateAndAddExternalFile(
+                      new_cf_option, file_data, -1, allow_global_seqno,
+                      write_global_seqno, verify_checksums_before_ingest,
+                      true /*ingest_behind*/, false /*sort_data*/, nullptr,
+                      /*cfh=*/new_cfh)
+                      .IsInvalidArgument());
+      ASSERT_OK(db_->DropColumnFamily(new_cfh));
+      ASSERT_OK(db_->DestroyColumnFamilyHandle(new_cfh));
+
+      options.cf_allow_ingest_behind = false;
+    } else {
       options.allow_ingest_behind = false;
+    }
     ASSERT_OK(TryReopen(options));
     ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
-  dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(), &level_to_files);
+    dbfull()->TEST_GetFilesMetaData(db_->DefaultColumnFamily(),
+                                    &level_to_files);
     ASSERT_EQ(1, level_to_files[2].size());
     ASSERT_NE(ingested_file_number, level_to_files[2][0].fd.GetNumber());
     VerifyDBFromMap(true_data, &kcnt, false);
+  }
 }
 
 TEST_F(ExternalSSTFileTest, SkipBloomFilter) {
@@ -3542,19 +3542,26 @@ TEST_F(ExternalSSTFileWithTimestampTest, SanityCheck) {
   // overlapping key ranges.
   ASSERT_TRUE(IngestExternalUDTFile({file1, file2}).IsNotSupported());
 
+  for (bool cf_option : {false, true}) {
+    SCOPED_TRACE("cf_option = " + std::to_string(cf_option));
+    if (cf_option) {
+      options.cf_allow_ingest_behind = true;
+    } else {
       options.allow_ingest_behind = true;
+    }
     DestroyAndReopen(options);
     IngestExternalFileOptions opts;
 
     // TODO(yuzhangyu): support ingestion behind for user-defined timestamps?
     // Ingesting external files with user-defined timestamps requires searching
-  // through the whole lsm tree to make sure there is no key range overlap with
-  // the db. Ingestion behind currently is doing a simply placing it at the
-  // bottom level step without a search, so we don't allow it either.
+    // through the whole lsm tree to make sure there is no key range overlap
+    // with the db. Ingestion behind currently is doing a simply placing it at
+    // the bottom level step without a search, so we don't allow it either.
     opts.ingest_behind = true;
     ASSERT_TRUE(db_->IngestExternalFile({file1}, opts).IsNotSupported());
 
     DestroyAndRecreateExternalSSTFilesDir();
+  }
 }
 
 TEST_F(ExternalSSTFileWithTimestampTest, UDTSettingsCompatibilityCheck) {

db/version_edit.h

@@ -110,7 +110,7 @@ constexpr uint64_t kUnknownOldestAncesterTime = 0;
 constexpr uint64_t kUnknownNewestKeyTime = 0;
 constexpr uint64_t kUnknownFileCreationTime = 0;
 constexpr uint64_t kUnknownEpochNumber = 0;
-// If `Options::allow_ingest_behind` is true, this epoch number
+// If `Options::cf_allow_ingest_behind` is true, this epoch number
 // will be dedicated to files ingested behind.
 constexpr uint64_t kReservedEpochNumberForFileIngestedBehind = 1;
 

db/version_set.cc

@@ -3528,7 +3528,9 @@ void VersionStorageInfo::ComputeCompactionScore(
   // maintaining it to be over 1.0, we scale the original score by 10x
   // if it is larger than 1.0.
   const double kScoreScale = 10.0;
-  int max_output_level = MaxOutputLevel(immutable_options.allow_ingest_behind);
+  int max_output_level =
+      MaxOutputLevel(immutable_options.cf_allow_ingest_behind ||
+                     immutable_options.allow_ingest_behind);
   for (int level = 0; level <= MaxInputLevel(); level++) {
     double score;
     if (level == 0) {
@@ -3713,6 +3715,7 @@ void VersionStorageInfo::ComputeCompactionScore(
   }
   ComputeFilesMarkedForCompaction(max_output_level);
   ComputeBottommostFilesMarkedForCompaction(
+      immutable_options.cf_allow_ingest_behind ||
       immutable_options.allow_ingest_behind);
   ComputeExpiredTtlFiles(immutable_options, mutable_cf_options.ttl);
   ComputeFilesMarkedForPeriodicCompaction(
@@ -4710,8 +4713,7 @@ void VersionStorageInfo::RecoverEpochNumbers(ColumnFamilyData* cfd,
   if (restart_epoch) {
     cfd->ResetNextEpochNumber();
 
-    bool reserve_epoch_num_for_file_ingested_behind =
-        cfd->ioptions().allow_ingest_behind;
+    bool reserve_epoch_num_for_file_ingested_behind = cfd->AllowIngestBehind();
     if (reserve_epoch_num_for_file_ingested_behind) {
       uint64_t reserved_epoch_number = cfd->NewEpochNumber();
       assert(reserved_epoch_number ==
@@ -4719,7 +4721,8 @@ void VersionStorageInfo::RecoverEpochNumbers(ColumnFamilyData* cfd,
       ROCKS_LOG_INFO(cfd->ioptions().info_log.get(),
                      "[%s]CF has reserved epoch number %" PRIu64
                      " for files ingested "
-                     "behind since `Options::allow_ingest_behind` is true",
+                     "behind since `Options::allow_ingest_behind` or "
+                     "`Options::cf_allow_ingest_behind` is true",
                      cfd->GetName().c_str(), reserved_epoch_number);
     }
   }

include/rocksdb/advanced_options.h

@@ -526,7 +526,7 @@ struct AdvancedColumnFamilyOptions {
   // By doing it, we give max_bytes_for_level_multiplier a priority against
   // max_bytes_for_level_base, for a more predictable LSM tree shape. It is
   // useful to limit worse case space amplification.
-  // If `allow_ingest_behind=true` or `preclude_last_level_data_seconds > 0`,
+  // If `cf_allow_ingest_behind=true` or `preclude_last_level_data_seconds > 0`,
   // then the last level is reserved, and we will start filling LSM from the
   // second last level.
   //
@@ -1146,6 +1146,33 @@ struct AdvancedColumnFamilyOptions {
   // Dynamically changeable through the SetOptions() API.
   uint32_t memtable_avg_op_scan_flush_trigger = 0;
 
+  // If either DBOptions::allow_ingest_behind or this option is set to true,
+  // this column family will prepare for ingesting files to the last level
+  // (IngestExternalFiles() with ingest_behind=true). Users should set only
+  // this option since DBOptions::allow_ingest_behind is deprecated.
+  //
+  // Specifically, preparing a column family for ingesting files to the last
+  // level has the following effects:
+  // 1) Disables some internal optimizations around SST file compression.
+  // 2) Reserves the last level for ingested files only.
+  // 3) Compaction will not include any file from the last level.
+  // 4) Compaction will preserve necessary tombstones that can apply on
+  // top of ingested files.
+  //
+  // Note that only Universal Compaction supports cf_allow_ingest_behind.
+  // `num_levels` should be >= 3 if this option is turned on.
+  //
+  // Note that this option needs to be set to true before any write to the CF.
+  // It's recommended to set the option to true since CF creation. Otherwise,
+  // ingestion with ingest_behind = true might fail. Once file ingestions are
+  // done, the option should be flipped to false. Flipping this option to false
+  // allows the CF to disable the behavior changes detailed above and resume
+  // more efficient operation.
+  //
+  // Default: false
+  // Immutable.
+  bool cf_allow_ingest_behind = false;
+
   // Create ColumnFamilyOptions with default values for all fields
   AdvancedColumnFamilyOptions();
   // Create ColumnFamilyOptions from Options

include/rocksdb/db.h

@@ -1984,7 +1984,7 @@ class DB {
   //     even if the file compression doesn't match the level compression
   // (3) If IngestExternalFileOptions->ingest_behind is set to true,
   //     we always ingest at the bottommost level, which should be reserved
-  //     for this purpose (see DBOPtions::allow_ingest_behind flag).
+  //     for this purpose (see ColumnFamilyOptions::cf_allow_ingest_behind).
   // (4) If IngestExternalFileOptions->fail_if_not_bottommost_level is set to
   //     true, then this method can return Status:TryAgain() indicating that
   //     the files cannot be ingested to the bottommost level, and it is the

include/rocksdb/options.h

@@ -1361,19 +1361,11 @@ struct DBOptions {
   // Dynamically changeable through SetDBOptions() API.
   bool avoid_flush_during_shutdown = false;
 
-  // Set this option to true during creation of database if you want
-  // to be able to ingest behind (call IngestExternalFile() skipping keys
-  // that already exist, rather than overwriting matching keys).
-  // Setting this option to true has the following effects:
-  // 1) Disable some internal optimizations around SST file compression.
-  // 2) Reserve the last level for ingested files only.
-  // 3) Compaction will not include any file from the last level.
-  // 4) Compaction will preserve necessary tombstones that can apply on
-  // top of ingested files.
-  // Note that only Universal Compaction supports allow_ingest_behind.
-  // `num_levels` should be >= 3 if this option is turned on.
-  // Note that if TimedPut was issued to a CF, ingest behind into that
-  // CF may fail.
+  // DEPRECATED: use ColumnFamilyOptions::cf_allow_ingest_behind instead.
+  // This option might be removed in a future release.
+  //
+  // See comment for `ColumnFamilyOptions::cf_allow_ingest_behind` for
+  // detail about the option's functionality and use cases.
   //
   // DEFAULT: false
   // Immutable.
@@ -2380,8 +2372,8 @@ struct IngestExternalFileOptions {
   // to be skipped rather than overwriting existing data under that key.
   // Use case: back-fill of some historical data in the database without
   // over-writing existing newer version of data.
-  // This option could only be used if the DB has been running
-  // with allow_ingest_behind=true since the dawn of time.
+  // This option could only be used if the CF has been running
+  // with cf_allow_ingest_behind=true since CF creation (or before any write).
   // All files will be ingested at the bottommost level with seqno=0.
   bool ingest_behind = false;
   // DEPRECATED - Set to true if you would like to write global_seqno to

options/cf_options.cc

@@ -893,6 +893,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct ImmutableCFOptions, persist_user_defined_timestamps),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kCompareLoose}},
+        {"cf_allow_ingest_behind",
+         {offsetof(struct ImmutableCFOptions, cf_allow_ingest_behind),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
 };
 
 const std::string OptionsHelper::kCFOptionsName = "ColumnFamilyOptions";
@@ -1032,7 +1036,8 @@ ImmutableCFOptions::ImmutableCFOptions(const ColumnFamilyOptions& cf_options)
       sst_partitioner_factory(cf_options.sst_partitioner_factory),
       blob_cache(cf_options.blob_cache),
       persist_user_defined_timestamps(
-          cf_options.persist_user_defined_timestamps) {}
+          cf_options.persist_user_defined_timestamps),
+      cf_allow_ingest_behind(cf_options.cf_allow_ingest_behind) {}
 
 ImmutableOptions::ImmutableOptions() : ImmutableOptions(Options()) {}
 

options/cf_options.h

@@ -82,6 +82,8 @@ struct ImmutableCFOptions {
   std::shared_ptr<Cache> blob_cache;
 
   bool persist_user_defined_timestamps;
+
+  bool cf_allow_ingest_behind;
 };
 
 struct ImmutableOptions : public ImmutableDBOptions, public ImmutableCFOptions {

options/options.cc

@@ -462,6 +462,8 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
                    experimental_mempurge_threshold);
   ROCKS_LOG_HEADER(log, "           Options.memtable_max_range_deletions: %d",
                    memtable_max_range_deletions);
+  ROCKS_LOG_HEADER(log, "                 Options.cf_allow_ingest_behind: %s",
+                   cf_allow_ingest_behind ? "true" : "false");
 }  // ColumnFamilyOptions::Dump
 
 void Options::Dump(Logger* log) const {

options/options_helper.cc

@@ -339,6 +339,7 @@ void UpdateColumnFamilyOptions(const ImmutableCFOptions& ioptions,
   cf_opts->persist_user_defined_timestamps =
       ioptions.persist_user_defined_timestamps;
   cf_opts->default_temperature = ioptions.default_temperature;
+  cf_opts->cf_allow_ingest_behind = ioptions.cf_allow_ingest_behind;
 
   // TODO(yhchiang): find some way to handle the following derived options
   // * max_file_size

options/options_settable_test.cc

@@ -681,7 +681,8 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "uncache_aggressiveness=1234;"
       "paranoid_memory_checks=1;"
       "memtable_op_scan_flush_trigger=123;"
-      "memtable_avg_op_scan_flush_trigger=12;",
+      "memtable_avg_op_scan_flush_trigger=12;"
+      "cf_allow_ingest_behind=1;",
       new_options));
 
   ASSERT_NE(new_options->blob_cache.get(), nullptr);

unreleased_history/new_features/cf-ingest-behind.md

@@ -0,0 +1 @@
+* Introduce column family option `cf_allow_ingest_behind`. This option aims to replace `DBOptions::allow_ingest_behind` to enable ingest behind at the per-CF level. `DBOptions::allow_ingest_behind` is deprecated.