Let table factories declare that they cannot build tables

Summary:
This adds `TableFactory::IsTableBuilderSupported()` and `ExternalTableFactory::IsTableBuilderSupported()`, so a table factory that only reads files produced elsewhere can say so, and makes RocksDB keep every compaction away from a column family that uses one.

Some external table factories read a format that RocksDB cannot write: their files are produced by an outside pipeline and only ever ingested, and `NewTableBuilder()` returns a builder that fails. Today nothing tells RocksDB this, so any flush or compaction that reaches such a factory fails in the background, and the resulting background error stops writes to the whole DB, not just to that column family. Flushes can already be ruled out with `disallow_memtable_writes`, but a manual compaction cannot: with a single ingested file on the last level, a default `CompactRange()` is a no-op, while `CompactRange()` with `BottommostLevelCompaction::kForce` rewrites the file, fails with the builder's error, and leaves the next `Put()` to the default column family failing with the same status. Automatic compactions have the same problem unless the user remembers `disable_auto_compactions`.

With this change, a factory that returns false gets four guarantees. Opening or creating a column family with it fails with `InvalidArgument` unless `disallow_memtable_writes` is set, so nothing can reach a flush. `ColumnFamilyData::NeedsCompaction()` returns false, so automatic compactions are never scheduled, whatever the compaction style and `disable_auto_compactions`. `CompactRange()` and `CompactFiles()` return `NotSupported` before doing any work, so a caller gets a foreground error and the DB stays writable. And its L0 files and pending compaction bytes never stall or stop writes, the same as for a column family with `disable_auto_compactions`: since nothing can compact it, counting that backlog would let one such column family stop writes to the whole DB for good, for example after an overlapping ingestion lands in L0 of a column family whose L0 stop trigger it reaches. This follows `IsDeleteRangeSupported()`: the column family caches the capability when it is created, and the external table adapter caches the wrapped factory's answer so that RocksDB never calls external code while holding the DB mutex.

The default is true, so existing table factories behave as before. Blocking automatic compactions also blocks compactions that need no builder, such as FIFO deletions and trivial moves; a factory that cannot build tables should not rely on them. `OpenAndTrimHistory()` calls `CompactRangeInternal()` directly, but only for column families with user-defined timestamps, which external tables reject.

Changes:
1. `TableFactory::IsTableBuilderSupported()` and `ExternalTableFactoryBase::IsTableBuilderSupported()` default to true; `ExternalTableFactoryAdapter` caches and forwards the wrapped factory's value, including after `PrepareOptions()`.
2. `ColumnFamilyData` caches the capability, `NeedsCompaction()` returns false without it, and `ColumnFamilyData::ValidateOptions()` requires `disallow_memtable_writes` without it.
3. `GetWriteStallConditionAndCause()` takes the capability and, like `disable_auto_compactions`, ignores L0 file counts and pending compaction bytes without it; its three callers pass the column family's cached value.
4. `DBImpl::CompactRange()` and `DBImpl::CompactFiles()` return `NotSupported` for such a column family.
5. `ExternalTableTest.FactoryWithoutTableBuilderIsNeverCompacted` compares two FIFO column families holding the same ingested file, one with a factory that can build tables and one without, and checks creation-time validation, that only the first is compacted automatically, that `CompactRange()` with and without `kForce` and `CompactFiles()` return `NotSupported` for the second, and that the DB stays writable and the file readable.
6. `ExternalTableTest.FactoryWithoutTableBuilderDoesNotStallWrites` ingests two overlapping files into a two-level column family without a builder, so one lands in L0 and reaches every L0 trigger, and checks that writes to the default column family are neither stopped nor delayed.

Reviewed By: xingbowang

Differential Revision: D128259405

db/column_family.cc

@@ -635,6 +635,8 @@ ColumnFamilyData::ColumnFamilyData(
       mutable_cf_options_(initial_cf_options_),
       is_delete_range_supported_(
           cf_options.table_factory->IsDeleteRangeSupported()),
+      is_table_builder_supported_(
+          cf_options.table_factory->IsTableBuilderSupported()),
       write_buffer_manager_(write_buffer_manager),
       mem_(nullptr),
       imm_(ioptions_.min_write_buffer_number_to_merge,
@@ -1025,13 +1027,16 @@ ColumnFamilyData::GetWriteStallConditionAndCause(
     int num_unflushed_memtables, int num_l0_files,
     uint64_t num_compaction_needed_bytes,
     const MutableCFOptions& mutable_cf_options,
-    const ImmutableCFOptions& immutable_cf_options) {
+    const ImmutableCFOptions& immutable_cf_options,
+    bool table_builder_supported) {
+  const bool auto_compactions =
+      table_builder_supported && !mutable_cf_options.disable_auto_compactions;
   if (num_unflushed_memtables >= mutable_cf_options.max_write_buffer_number) {
     return {WriteStallCondition::kStopped, WriteStallCause::kMemtableLimit};
-  } else if (!mutable_cf_options.disable_auto_compactions &&
+  } else if (auto_compactions &&
              num_l0_files >= mutable_cf_options.level0_stop_writes_trigger) {
     return {WriteStallCondition::kStopped, WriteStallCause::kL0FileCountLimit};
-  } else if (!mutable_cf_options.disable_auto_compactions &&
+  } else if (auto_compactions &&
              mutable_cf_options.hard_pending_compaction_bytes_limit > 0 &&
              num_compaction_needed_bytes >=
                  mutable_cf_options.hard_pending_compaction_bytes_limit) {
@@ -1043,12 +1048,12 @@ ColumnFamilyData::GetWriteStallConditionAndCause(
              num_unflushed_memtables - 1 >=
                  immutable_cf_options.min_write_buffer_number_to_merge) {
     return {WriteStallCondition::kDelayed, WriteStallCause::kMemtableLimit};
-  } else if (!mutable_cf_options.disable_auto_compactions &&
+  } else if (auto_compactions &&
              mutable_cf_options.level0_slowdown_writes_trigger >= 0 &&
              num_l0_files >=
                  mutable_cf_options.level0_slowdown_writes_trigger) {
     return {WriteStallCondition::kDelayed, WriteStallCause::kL0FileCountLimit};
-  } else if (!mutable_cf_options.disable_auto_compactions &&
+  } else if (auto_compactions &&
              mutable_cf_options.soft_pending_compaction_bytes_limit > 0 &&
              num_compaction_needed_bytes >=
                  mutable_cf_options.soft_pending_compaction_bytes_limit) {
@@ -1070,7 +1075,7 @@ WriteStallCondition ColumnFamilyData::RecalculateWriteStallConditions(
     auto write_stall_condition_and_cause = GetWriteStallConditionAndCause(
         imm()->NumNotFlushed(), vstorage->l0_delay_trigger_count(),
         vstorage->estimated_compaction_needed_bytes(), mutable_cf_options,
-        ioptions());
+        ioptions(), is_table_builder_supported_);
     write_stall_condition = write_stall_condition_and_cause.first;
     auto write_stall_cause = write_stall_condition_and_cause.second;
 
@@ -1276,7 +1281,8 @@ void ColumnFamilyData::CreateNewMemtable(SequenceNumber earliest_seq) {
 }
 
 bool ColumnFamilyData::NeedsCompaction() const {
-  return !mutable_cf_options_.disable_auto_compactions &&
+  return is_table_builder_supported_ &&
+         !mutable_cf_options_.disable_auto_compactions &&
          compaction_picker_->NeedsCompaction(current_->storage_info());
 }
 
@@ -1541,6 +1547,13 @@ Status ColumnFamilyData::ValidateOptions(
     return s;
   }
 
+  if (!cf_options.table_factory->IsTableBuilderSupported() &&
+      !cf_options.disallow_memtable_writes) {
+    return Status::InvalidArgument(
+        "A table factory that cannot build tables requires "
+        "disallow_memtable_writes");
+  }
+
   if (cf_options.ttl > 0 && cf_options.ttl != kDefaultTtl) {
     if (!cf_options.table_factory->IsInstanceOf(
             TableFactory::kBlockBasedTableName())) {

db/column_family.h

@@ -368,6 +368,12 @@ class ColumnFamilyData {
 
   bool is_delete_range_supported() { return is_delete_range_supported_; }
 
+  // False when the table factory cannot write tables, so the column family is
+  // never compacted.
+  bool is_table_builder_supported() const {
+    return is_table_builder_supported_;
+  }
+
   // Validate CF options against DB options
   static Status ValidateOptions(const DBOptions& db_options,
                                 const ColumnFamilyOptions& cf_options);
@@ -562,12 +568,15 @@ class ColumnFamilyData {
     --num_in_flight_compactions_;
   }
 
+  // L0 files and pending compaction bytes stall writes only when the column
+  // family can be compacted automatically, which requires
+  // 'table_builder_supported' and no disable_auto_compactions.
   static std::pair<WriteStallCondition, WriteStallCause>
-  GetWriteStallConditionAndCause(
-      int num_unflushed_memtables, int num_l0_files,
-      uint64_t num_compaction_needed_bytes,
-      const MutableCFOptions& mutable_cf_options,
-      const ImmutableCFOptions& immutable_cf_options);
+  GetWriteStallConditionAndCause(int num_unflushed_memtables, int num_l0_files,
+                                 uint64_t num_compaction_needed_bytes,
+                                 const MutableCFOptions& mutable_cf_options,
+                                 const ImmutableCFOptions& immutable_cf_options,
+                                 bool table_builder_supported);
 
   // Recalculate some stall conditions, which are changed only during
   // compaction, adding new memtable and/or recalculation of compaction score.
@@ -705,6 +714,7 @@ class ColumnFamilyData {
   MutableCFOptions mutable_cf_options_;
 
   const bool is_delete_range_supported_;
+  const bool is_table_builder_supported_;
 
   // True from a successful SetOptions() until MarkOptionsInstalled(). A
   // DB::SetOptions() call that fails on a later column family leaves it set

db/db_impl/db_impl_compaction_flush.cc

@@ -62,6 +62,11 @@ void RecordAtomicFlushRequestReason(Statistics* stats,
   }
 }
 
+Status TableBuilderNotSupported(const ColumnFamilyData& cfd) {
+  return Status::NotSupported(
+      "Column family's table factory cannot build tables", cfd.GetName());
+}
+
 std::vector<uint64_t> MemTableIds(const autovector<ReadOnlyMemTable*>& mems) {
   std::vector<uint64_t> ids;
   ids.reserve(mems.size());
@@ -158,7 +163,7 @@ bool DBImpl::ShouldRescheduleFlushRequestToRetainUDT(
           cfd->GetUnflushedMemTableCountForWriteStallCheck(),
           /*num_l0_files=*/0,
           /*num_compaction_needed_bytes=*/0, mutable_cf_options,
-          cfd->ioptions())
+          cfd->ioptions(), cfd->is_table_builder_supported())
           .first;
   if (write_stall != WriteStallCondition::kNormal) {
     return false;
@@ -1193,6 +1198,10 @@ Status DBImpl::CompactRange(const CompactRangeOptions& options,
       static_cast_with_check<ColumnFamilyHandleImpl>(column_family)->cfd();
   assert(cfd);
 
+  if (!cfd->is_table_builder_supported()) {
+    return TableBuilderNotSupported(*cfd);
+  }
+
   if (manual_compaction_paused_.load(std::memory_order_acquire) > 0) {
     return Status::Incomplete(Status::SubCode::kManualCompactionPaused);
   }
@@ -1619,6 +1628,10 @@ Status DBImpl::CompactFiles(const CompactionOptions& compact_options,
       static_cast_with_check<ColumnFamilyHandleImpl>(column_family)->cfd();
   assert(cfd);
 
+  if (!cfd->is_table_builder_supported()) {
+    return TableBuilderNotSupported(*cfd);
+  }
+
   {
     InstrumentedMutexLock l(&mutex_);
     BeginInFlightCompaction(cfd);
@@ -3362,7 +3375,7 @@ Status DBImpl::WaitUntilFlushWouldNotStallWrites(ColumnFamilyData* cfd,
               cfd->GetUnflushedMemTableCountForWriteStallCheck(),
               vstorage->l0_delay_trigger_count() + 1,
               vstorage->estimated_compaction_needed_bytes(), mutable_cf_options,
-              cfd->ioptions())
+              cfd->ioptions(), cfd->is_table_builder_supported())
               .first;
     } while (write_stall_condition != WriteStallCondition::kNormal);
   }

include/rocksdb/external_table.h

@@ -576,6 +576,11 @@ class ExternalTableFactoryBase : public Customizable {
   // requires the corresponding builder and reader to implement
   // PutRangeDeletionBlock() and GetRangeDeletionBlock(), respectively.
   virtual bool IsDeleteRangeSupported() const { return false; }
+
+  // Returns whether NewTableBuilder() can write tables. A factory that only
+  // reads files written elsewhere returns false; see
+  // TableFactory::IsTableBuilderSupported() for what RocksDB then requires.
+  virtual bool IsTableBuilderSupported() const { return true; }
 };
 
 using ExternalTableFactory =

include/rocksdb/table.h

@@ -1302,6 +1302,12 @@ class TableFactory : public Customizable {
 
   // Return is delete range supported
   virtual bool IsDeleteRangeSupported() const { return false; }
+
+  // Returns whether NewTableBuilder() can write tables. A column family whose
+  // table factory returns false holds only ingested or imported files: it must
+  // set disallow_memtable_writes, automatic compactions are not scheduled for
+  // it, and CompactRange() and CompactFiles() on it return NotSupported.
+  virtual bool IsTableBuilderSupported() const { return true; }
 };
 
 // Create a special table factory that can open either of the supported

table/external_table.cc

@@ -1100,7 +1100,8 @@ class ExternalTableFactoryAdapter : public TableFactory {
   explicit ExternalTableFactoryAdapter(
       std::shared_ptr<ExternalTableFactoryBase<Mode>> inner)
       : inner_(std::move(inner)),
-        supports_range_deletions_(inner_->IsDeleteRangeSupported()) {
+        supports_range_deletions_(inner_->IsDeleteRangeSupported()),
+        supports_table_builder_(inner_->IsTableBuilderSupported()) {
     RegisterOptions(&options_,
                     &GetExternalTableFactoryAdapterOptionsTypeInfo());
   }
@@ -1113,6 +1114,10 @@ class ExternalTableFactoryAdapter : public TableFactory {
     return Mode == ExternalTableMode::kFull && supports_range_deletions_;
   }
 
+  bool IsTableBuilderSupported() const override {
+    return supports_table_builder_;
+  }
+
   bool IsInstanceOf(const std::string& name) const override {
     return name == kClassName() || TableFactory::IsInstanceOf(name);
   }
@@ -1124,6 +1129,7 @@ class ExternalTableFactoryAdapter : public TableFactory {
     }
     if (status.ok()) {
       supports_range_deletions_ = inner_->IsDeleteRangeSupported();
+      supports_table_builder_ = inner_->IsTableBuilderSupported();
     }
     return status;
   }
@@ -1229,9 +1235,10 @@ class ExternalTableFactoryAdapter : public TableFactory {
  private:
   std::shared_ptr<ExternalTableFactoryBase<Mode>> inner_;
   ExternalTableFactoryAdapterOptions options_;
-  // Avoid invoking external code when RocksDB queries this capability while
+  // Avoid invoking external code when RocksDB queries these capabilities while
   // holding the DB mutex.
   bool supports_range_deletions_;
+  bool supports_table_builder_;
 };
 
 template <ExternalTableMode Mode>

table/table_test.cc

@@ -8384,6 +8384,12 @@ class ExternalTableTest : public DBTestBase {
       : DBTestBase("external_table_test", /*env_do_fsync=*/false) {}
 
  protected:
+  // Reads the files of SimpleExternalTableFactory but cannot write any.
+  class ReadOnlyExternalTableFactory : public SimpleExternalTableFactory {
+   public:
+    bool IsTableBuilderSupported() const override { return false; }
+  };
+
   // A reader that pins values from its internal buffer, exercising the
   // zero-copy path in ExternalTableReaderAdapter::Get().
   class PinnedSimpleExternalTableReader
@@ -8618,6 +8624,140 @@ TEST_F(ExternalTableTest, WrappedFactoryValidatesColumnFamilyOptions) {
       column_families, std::vector<Options>{options, ingest_only}));
 }
 
+TEST_F(ExternalTableTest, FactoryWithoutTableBuilderIsNeverCompacted) {
+  if (encrypted_env_) {
+    ROCKSDB_GTEST_SKIP("Test requires non-encrypted environment");
+    return;
+  }
+  Options options = GetDefaultOptions();
+  // FIFO compaction, used below, requires it.
+  options.max_open_files = -1;
+  Reopen(options);
+  Options writable = options;
+  writable.table_factory =
+      NewExternalTableFactory(std::make_shared<SimpleExternalTableFactory>());
+  writable.disallow_memtable_writes = true;
+  // Any ingested file exceeds the FIFO size limit, so the picker always wants
+  // an automatic compaction.
+  writable.compaction_style = kCompactionStyleFIFO;
+  writable.compaction_options_fifo.max_table_files_size = 1;
+  Options read_only = writable;
+  read_only.table_factory =
+      NewExternalTableFactory(std::make_shared<ReadOnlyExternalTableFactory>());
+  ASSERT_TRUE(writable.table_factory->IsTableBuilderSupported());
+  ASSERT_FALSE(read_only.table_factory->IsTableBuilderSupported());
+
+  // Memtable writes would reach the missing builder at the next flush.
+  Options accepts_writes = read_only;
+  accepts_writes.disallow_memtable_writes = false;
+  ColumnFamilyHandle* handle = nullptr;
+  Status s = db_->CreateColumnFamily(accepts_writes, "accepts_writes", &handle);
+  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+  ASSERT_EQ(handle, nullptr);
+
+  // handles_[1] can build tables and handles_[2] cannot; both get the same
+  // ingested file.
+  CreateColumnFamilies({"writable", "read_only"}, writable);
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "writable", "read_only"},
+                           {options, writable, read_only});
+  const std::string file = dbname_ + "/external.immutable";
+  SstFileWriter writer(EnvOptions(writable), writable);
+  ASSERT_OK(writer.Open(file));
+  ASSERT_OK(writer.Put("a", "va"));
+  ASSERT_OK(writer.Put("b", "vb"));
+  ASSERT_OK(writer.Finish());
+  IngestExternalFileOptions ingest_options;
+  ingest_options.allow_global_seqno = true;
+  const auto live_files = [&](int cf) {
+    ColumnFamilyMetaData metadata;
+    db_->GetColumnFamilyMetaData(handles_[cf], &metadata);
+    std::vector<std::string> names;
+    for (const auto& level : metadata.levels) {
+      for (const auto& file_metadata : level.files) {
+        names.push_back(file_metadata.name);
+      }
+    }
+    return names;
+  };
+  std::vector<std::string> ingested[3];
+  for (int cf : {1, 2}) {
+    ASSERT_OK(db_->IngestExternalFile(handles_[cf], {file}, ingest_options));
+    ingested[cf] = live_files(cf);
+    ASSERT_EQ(ingested[cf].size(), 1);
+  }
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+  ASSERT_TRUE(live_files(1).empty());
+  ASSERT_EQ(live_files(2), ingested[2]);
+  ASSERT_FALSE(static_cast_with_check<ColumnFamilyHandleImpl>(handles_[2])
+                   ->cfd()
+                   ->NeedsCompaction());
+
+  CompactRangeOptions forced;
+  forced.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+  s = db_->CompactRange(CompactRangeOptions(), handles_[2], nullptr, nullptr);
+  ASSERT_TRUE(s.IsNotSupported()) << s.ToString();
+  s = db_->CompactRange(forced, handles_[2], nullptr, nullptr);
+  ASSERT_TRUE(s.IsNotSupported()) << s.ToString();
+  // FIFO compaction keeps every file on level 0.
+  s = db_->CompactFiles(CompactionOptions(), handles_[2], ingested[2],
+                        /*output_level=*/0);
+  ASSERT_TRUE(s.IsNotSupported()) << s.ToString();
+  ASSERT_EQ(live_files(2), ingested[2]);
+
+  // The rejected compactions leave the DB writable and the file readable.
+  ASSERT_OK(Put("key", "value"));
+  ASSERT_OK(Flush());
+  ASSERT_EQ(Get(2, "a"), "va");
+  ASSERT_EQ(Get(2, "b"), "vb");
+}
+
+TEST_F(ExternalTableTest, FactoryWithoutTableBuilderDoesNotStallWrites) {
+  if (encrypted_env_) {
+    ROCKSDB_GTEST_SKIP("Test requires non-encrypted environment");
+    return;
+  }
+  Options options = GetDefaultOptions();
+  Reopen(options);
+  Options writable = options;
+  writable.table_factory =
+      NewExternalTableFactory(std::make_shared<SimpleExternalTableFactory>());
+  Options read_only = writable;
+  read_only.table_factory =
+      NewExternalTableFactory(std::make_shared<ReadOnlyExternalTableFactory>());
+  read_only.disallow_memtable_writes = true;
+  // A second overlapping file can only go to L0, which alone reaches every L0
+  // trigger.
+  read_only.num_levels = 2;
+  read_only.level0_file_num_compaction_trigger = 1;
+  read_only.level0_slowdown_writes_trigger = 1;
+  read_only.level0_stop_writes_trigger = 1;
+  CreateColumnFamilies({"read_only"}, read_only);
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "read_only"},
+                           std::vector<Options>{options, read_only});
+
+  IngestExternalFileOptions ingest_options;
+  ingest_options.allow_global_seqno = true;
+  for (const std::string value : {"v1", "v2"}) {
+    const std::string file = dbname_ + "/external_" + value + ".immutable";
+    SstFileWriter writer(EnvOptions(writable), writable);
+    ASSERT_OK(writer.Open(file));
+    ASSERT_OK(writer.Put("a", value));
+    ASSERT_OK(writer.Finish());
+    ASSERT_OK(db_->IngestExternalFile(handles_[1], {file}, ingest_options));
+  }
+  ASSERT_EQ(NumTableFilesAtLevel(0, 1), 1);
+  ASSERT_EQ(NumTableFilesAtLevel(1, 1), 1);
+  ASSERT_EQ(Get(1, "a"), "v2");
+
+  // No compaction can clear that L0 file, so it must not stall other column
+  // families.
+  ASSERT_FALSE(dbfull()->TEST_write_controler().IsStopped());
+  ASSERT_FALSE(dbfull()->TEST_write_controler().NeedsDelay());
+  WriteOptions no_slowdown;
+  no_slowdown.no_slowdown = true;
+  ASSERT_OK(db_->Put(no_slowdown, "key", "value"));
+}
+
 TEST_F(ExternalTableTest, GetWrappedExternalTableFactory) {
   std::shared_ptr<ExternalTableFactory> inner =
       std::make_shared<SimpleExternalTableFactory>();

unreleased_history/new_features/table_factory_without_builder.md

@@ -0,0 +1 @@
+Added `TableFactory::IsTableBuilderSupported()` and `ExternalTableFactory::IsTableBuilderSupported()` for table factories that only read ingested or imported files. A column family whose factory returns false must set `disallow_memtable_writes`, is never scheduled for automatic compaction, does not stall writes for L0 files or pending compaction bytes, and returns `NotSupported` from `CompactRange()` and `CompactFiles()` instead of failing a background compaction.