set ignore_unknown_options when parsing options (#13443)

Summary:
In case the primary host has a new option added which isn't available in the remote worker yet, the remote compaction currently fails. In most cases, these new options are not relevant to the remote compaction and the worker should be able to move on by ignoring it.

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

Test Plan: Verified internally in Meta Infra.

Reviewed By: anand1976

Differential Revision: D70744359

Pulled By: jaykorean

fbshipit-source-id: eb6a388c2358a7f8089f2e35a378b7017b9e03f3

BUCK

@@ -214,7 +214,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[
         "table/cuckoo/cuckoo_table_builder.cc",
         "table/cuckoo/cuckoo_table_factory.cc",
         "table/cuckoo/cuckoo_table_reader.cc",
-        "table/external_table_reader.cc",
+        "table/external_table.cc",
         "table/format.cc",
         "table/get_context.cc",
         "table/iterator.cc",

CMakeLists.txt

@@ -835,7 +835,7 @@ set(SOURCES
         table/cuckoo/cuckoo_table_builder.cc
         table/cuckoo/cuckoo_table_factory.cc
         table/cuckoo/cuckoo_table_reader.cc
-        table/external_table_reader.cc
+        table/external_table.cc
         table/format.cc
         table/get_context.cc
         table/iterator.cc

HISTORY.md

@@ -1,6 +1,35 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 10.0.1 (03/05/2025)
+### Public API Changes
+* Add an unordered map of name/value pairs, ReadOptions::property_bag, to pass opaque options through to an external table when creating an Iterator.
+* Introduced CompactionServiceJobStatus::kAborted to allow handling aborted scenario in Schedule(), Wait() or OnInstallation() APIs in Remote Compactions.
+* Added a column family option disallow_memtable_writes to safely fail any attempts to write to a non-default column family. This can be used for column families that are ingest only.
+
+## 10.0.0 (02/21/2025)
+### New Features
+* Introduced new `auto_refresh_iterator_with_snapshot` opt-in knob that (when enabled) will periodically release obsolete memory and storage resources for as long as the iterator is making progress and its supplied `read_options.snapshot` was initialized with non-nullptr value.
+* Added the ability to plug-in a custom table reader implementation. See include/rocksdb/external_table_reader.h for more details.
+* Experimental feature: RocksDB now supports FAISS inverted file based indices via the secondary indexing framework. Applications can use FAISS secondary indices to automatically quantize embeddings and perform K-nearest-neighbors similarity searches. See `FaissIVFIndex` and `SecondaryIndex` for more details. Note: the FAISS integration currently requires using the BUCK build.
+* Add new DB property `num_running_compaction_sorted_runs` that tracks the number of sorted runs being processed by currently running compactions
+* Experimental feature: added support for simple secondary indices that index the specified column as-is. See `SimpleSecondaryIndex` and `SecondaryIndex` for more details.
+* Added new `TransactionDBOptions::txn_commit_bypass_memtable_threshold`, which enables optimized transaction commit (see `TransactionOptions::commit_bypass_memtable`) when the transaction size exceeds a configured threshold.
+
+### Public API Changes
+* Updated the query API of the experimental secondary indexing feature by removing the earlier `SecondaryIndex::NewIterator` virtual and adding a `SecondaryIndexIterator` class that can be utilized by applications to find the primary keys for a given search target.
+* Added back the ability to leverage the primary key when building secondary index entries. This involved changes to the signatures of `SecondaryIndex::GetSecondary{KeyPrefix,Value}` as well as the addition of a new method `SecondaryIndex::FinalizeSecondaryKeyPrefix`. See the API comments for more details.
+* Minimum supported version of ZSTD is now 1.4.0, for code simplification. Obsolete `CompressionType` `kZSTDNotFinalCompression` is also removed.
+
+### Behavior Changes
+* `VerifyBackup` in `verify_with_checksum`=`true` mode will now evaluate checksums in parallel. As a result, unlike in case of original implementation, the API won't bail out on a very first corruption / mismatch and instead will iterate over all the backup files logging success / _degree_of_failure_ for each.
+* Reversed the order of updates to the same key in WriteBatchWithIndex. This means if there are multiple updates to the same key, the most recent update is ordered first. This affects the output of WBWIIterator. When WriteBatchWithIndex is created with `overwrite_key=true`, this affects the output only if Merge is used (#13387).
+* Added support for Merge operations in transactions using option `TransactionOptions::commit_bypass_memtable`.
+
+### Bug Fixes
+* Fixed GetMergeOperands() API in ReadOnlyDB and SecondaryDB
+* Fix a bug in `GetMergeOperands()` that can return incorrect status (MergeInProgress) and incorrect number of merge operands. This can happen when `GetMergeOperandsOptions::continue_cb` is set, both active and immutable memtables have merge operands and the callback stops the look up at the immutable memtable.
+
 ## 9.11.0 (01/17/2025)
 ### New Features
 * Introduce CancelAwaitingJobs() in CompactionService interface which will allow users to implement cancellation of running remote compactions from the primary instance

db/column_family.cc

@@ -168,7 +168,8 @@ Status CheckConcurrentWritesSupported(const ColumnFamilyOptions& cf_options) {
   }
   if (!cf_options.memtable_factory->IsInsertConcurrentlySupported()) {
     return Status::InvalidArgument(
-        "Memtable doesn't allow concurrent writes (allow_concurrent_memtable_write)");
+        "Memtable doesn't allow concurrent writes "
+        "(allow_concurrent_memtable_write)");
   }
   return Status::OK();
 }
@@ -239,6 +240,10 @@ ColumnFamilyOptions SanitizeOptions(const ImmutableDBOptions& db_options,
 
     result.min_write_buffer_number_to_merge = 1;
   }
+  if (result.disallow_memtable_writes) {
+    // A simple memtable that enforces MarkReadOnly (unlike skip list)
+    result.memtable_factory = std::make_shared<VectorRepFactory>();
+  }
 
   if (result.num_levels < 1) {
     result.num_levels = 1;

db/column_family.h

@@ -392,6 +392,9 @@ class ColumnFamilyData {
   void SetMemtable(MemTable* new_mem) {
     AssignMemtableID(new_mem);
     mem_ = new_mem;
+    if (ioptions_.disallow_memtable_writes) {
+      mem_->MarkImmutable();
+    }
   }
 
   void AssignMemtableID(ReadOnlyMemTable* new_imm) {

db/compaction/compaction_job.cc

@@ -1130,8 +1130,7 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
   if (db_options_.compaction_service) {
     CompactionServiceJobStatus comp_status =
         ProcessKeyValueCompactionWithCompactionService(sub_compact);
-    if (comp_status == CompactionServiceJobStatus::kSuccess ||
-        comp_status == CompactionServiceJobStatus::kFailure) {
+    if (comp_status != CompactionServiceJobStatus::kUseLocal) {
       return;
     }
     // fallback to local compaction

db/compaction/compaction_service_job.cc

@@ -83,6 +83,14 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
   switch (response.status) {
     case CompactionServiceJobStatus::kSuccess:
       break;
+    case CompactionServiceJobStatus::kAborted:
+      sub_compact->status =
+          Status::Aborted("Scheduling a remote compaction job was aborted");
+      ROCKS_LOG_WARN(
+          db_options_.info_log,
+          "[%s] [JOB %d] Remote compaction was aborted at Schedule()",
+          compaction->column_family_data()->GetName().c_str(), job_id_);
+      return response.status;
     case CompactionServiceJobStatus::kFailure:
       sub_compact->status = Status::Incomplete(
           "CompactionService failed to schedule a remote compaction job.");
@@ -118,6 +126,16 @@ CompactionJob::ProcessKeyValueCompactionWithCompactionService(
     return compaction_status;
   }
 
+  if (compaction_status == CompactionServiceJobStatus::kAborted) {
+    sub_compact->status =
+        Status::Aborted("Waiting a remote compaction job was aborted");
+    ROCKS_LOG_INFO(db_options_.info_log,
+                   "[%s] [JOB %d] Remote compaction was aborted during Wait()",
+                   compaction->column_family_data()->GetName().c_str(),
+                   job_id_);
+    return compaction_status;
+  }
+
   CompactionServiceResult compaction_result;
   s = CompactionServiceResult::Read(compaction_result_binary,
                                     &compaction_result);

db/compaction/compaction_service_test.cc

@@ -1471,6 +1471,40 @@ TEST_F(CompactionServiceTest, FallbackLocalManual) {
   VerifyTestData();
 }
 
+TEST_F(CompactionServiceTest, AbortedWhileWait) {
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  ReopenWithCompactionService(&options);
+
+  GenerateTestData();
+  VerifyTestData();
+
+  auto my_cs = GetCompactionService();
+  Statistics* compactor_statistics = GetCompactorStatistics();
+  Statistics* primary_statistics = GetPrimaryStatistics();
+
+  my_cs->ResetOverride();
+  std::string start_str = Key(15);
+  std::string end_str = Key(45);
+  Slice start(start_str);
+  Slice end(end_str);
+
+  // Override Wait() result with kAborted
+  my_cs->OverrideWaitStatus(CompactionServiceJobStatus::kAborted);
+  start_str = Key(120);
+  start = start_str;
+
+  Status s = db_->CompactRange(CompactRangeOptions(), &start, nullptr);
+  ASSERT_NOK(s);
+  ASSERT_TRUE(s.IsAborted());
+  // no remote compaction is run
+  ASSERT_EQ(my_cs->GetCompactionNum(), 0);
+  // make sure the compaction statistics is not recorded any side
+  ASSERT_EQ(primary_statistics->getTickerCount(COMPACT_WRITE_BYTES), 0);
+  ASSERT_EQ(primary_statistics->getTickerCount(REMOTE_COMPACT_WRITE_BYTES), 0);
+  ASSERT_EQ(compactor_statistics->getTickerCount(COMPACT_WRITE_BYTES), 0);
+}
+
 TEST_F(CompactionServiceTest, RemoteEventListener) {
   class RemoteEventListenerTest : public EventListener {
    public:

db/db_basic_test.cc

@@ -4994,6 +4994,97 @@ TEST_F(DBBasicTest, VerifyFileChecksumsReadahead) {
             (sst_size + alignment - 1) / (alignment));
 }
 
+TEST_F(DBBasicTest, DisallowMemtableWrite) {
+  // This test is mostly about what you can't do with memtable writes
+  // disallowed. For what you can do, see
+  // ExternalSSTFileBasicTest.FailIfNotBottommostLevelAndDisallowMemtable
+  Options options_allow = GetDefaultOptions();
+  options_allow.create_if_missing = true;
+  Options options_disallow = options_allow;
+  options_disallow.disallow_memtable_writes = true;
+
+  DestroyAndReopen(options_allow);
+  // CFs allowing and disallowing memtable write
+  CreateColumnFamilies({"cf1", "cf2"}, options_allow);
+  CreateColumnFamilies({"cf3"}, options_disallow);
+  // XXX: needed to get consistent handles_ mappings
+  ReopenWithColumnFamilies(
+      {"default", "cf1", "cf2", "cf3"},
+      {options_allow, options_allow, options_allow, options_disallow});
+
+  EXPECT_EQ(Put(0, "a0", "1").code(), Status::Code::kOk);
+  EXPECT_EQ(Put(1, "a1", "1").code(), Status::Code::kOk);
+  EXPECT_EQ(Put(2, "a2", "1").code(), Status::Code::kOk);
+  EXPECT_EQ(Put(3, "a3", "1").code(), Status::Code::kInvalidArgument);
+
+  EXPECT_EQ(Get(0, "a0"), "1");
+  EXPECT_EQ(Get(1, "a1"), "1");
+  EXPECT_EQ(Get(2, "a2"), "1");
+  EXPECT_EQ(Get(3, "a3"), "NOT_FOUND");
+
+  EXPECT_EQ(Delete(0, "z0").code(), Status::Code::kOk);
+  EXPECT_EQ(Delete(1, "z1").code(), Status::Code::kOk);
+  EXPECT_EQ(Delete(2, "z2").code(), Status::Code::kOk);
+  EXPECT_EQ(Delete(3, "z3").code(), Status::Code::kInvalidArgument);
+
+  WriteBatch wb;
+  EXPECT_EQ(wb.Put(handles_[0], "b0", "2").code(), Status::Code::kOk);
+  EXPECT_EQ(wb.Put(handles_[1], "b1", "2").code(), Status::Code::kOk);
+  EXPECT_EQ(wb.Put(handles_[2], "b2", "2").code(), Status::Code::kOk);
+  EXPECT_EQ(wb.Put(handles_[3], "b3", "2").code(),
+            Status::Code::kInvalidArgument);
+  ASSERT_OK(db_->Write({}, &wb));
+  wb.Clear();
+
+  EXPECT_EQ(Get(0, "b0"), "2");
+  EXPECT_EQ(Get(1, "b1"), "2");
+  EXPECT_EQ(Get(2, "b2"), "2");
+  EXPECT_EQ(Get(3, "b3"), "NOT_FOUND");
+
+  // When the DB is re-opened with WAL entries for a CF that is newly setting
+  // disallow_memtable_writes, we detect that and fail the open gracefully.
+  ASSERT_EQ(TryReopenWithColumnFamilies(
+                {"default", "cf1", "cf2", "cf3"},
+                {options_allow, options_allow, options_disallow, options_allow})
+                .code(),
+            Status::Code::kInvalidArgument);
+
+  // Successfully opening with allow creates L0 files from the WAL
+  ReopenWithColumnFamilies({"default", "cf1", "cf2", "cf3"}, options_allow);
+
+  EXPECT_EQ(Get(0, "a0"), "1");
+  EXPECT_EQ(Get(1, "a1"), "1");
+  EXPECT_EQ(Get(2, "a2"), "1");
+  EXPECT_EQ(Get(3, "a3"), "NOT_FOUND");
+
+  // Now able to disallow on CF2 because no relevant WAL entries
+  ReopenWithColumnFamilies(
+      {"default", "cf1", "cf2", "cf3"},
+      {options_allow, options_allow, options_disallow, options_allow});
+
+  EXPECT_EQ(Get(0, "a0"), "1");
+  EXPECT_EQ(Get(1, "a1"), "1");
+  EXPECT_EQ(Get(2, "a2"), "1");
+  EXPECT_EQ(Get(3, "a3"), "NOT_FOUND");
+
+  // Now able to write to CF 3 but not CF 2
+  EXPECT_EQ(Put(0, "c0", "3").code(), Status::Code::kOk);
+  EXPECT_EQ(Put(1, "c1", "3").code(), Status::Code::kOk);
+  EXPECT_EQ(Put(2, "c2", "3").code(), Status::Code::kInvalidArgument);
+  EXPECT_EQ(Put(3, "c3", "3").code(), Status::Code::kOk);
+
+  EXPECT_EQ(Get(0, "c0"), "3");
+  EXPECT_EQ(Get(1, "c1"), "3");
+  EXPECT_EQ(Get(2, "c2"), "NOT_FOUND");
+  EXPECT_EQ(Get(3, "c3"), "3");
+
+  // disallow_memtable_writes not supported on default column family.
+  // (Would be complicated to make a WriteBatch aware of the setting in order
+  // to reject the write before entering the write path.)
+  Destroy(options_allow);
+  EXPECT_EQ(TryReopen(options_disallow).code(), Status::Code::kInvalidArgument);
+}
+
 // TODO: re-enable after we provide finer-grained control for WAL tracking to
 // meet the needs of different use cases, durability levels and recovery modes.
 TEST_F(DBBasicTest, DISABLED_ManualWalSync) {

db/db_impl/db_impl_open.cc

@@ -224,6 +224,12 @@ Status DBImpl::ValidateOptions(
     if (!s.ok()) {
       return s;
     }
+    if (cfd.name == kDefaultColumnFamilyName) {
+      if (cfd.options.disallow_memtable_writes) {
+        return Status::InvalidArgument(
+            "Default column family cannot use disallow_memtable_writes=true");
+      }
+    }
   }
   s = ValidateOptions(db_options);
   return s;

db/db_impl/db_impl_secondary.cc

@@ -913,6 +913,10 @@ Status DBImplSecondary::CompactWithoutInstallation(
   Status s = cfd->compaction_picker()->GetCompactionInputsFromFileNumbers(
       &input_files, &input_set, vstorage, comp_options);
   if (!s.ok()) {
+    ROCKS_LOG_ERROR(
+        immutable_db_options_.info_log,
+        "GetCompactionInputsFromFileNumbers() failed - %s.\n DebugString: %s",
+        s.ToString().c_str(), version->DebugString().c_str());
     return s;
   }
 
@@ -990,6 +994,7 @@ Status DB::OpenAndCompact(
   DBOptions db_options;
   ConfigOptions config_options;
   config_options.env = override_options.env;
+  config_options.ignore_unknown_options = true;
   std::vector<ColumnFamilyDescriptor> all_column_families;
 
   TEST_SYNC_POINT_CALLBACK(

db/external_sst_file_basic_test.cc

@@ -2669,51 +2669,83 @@ TEST_F(ExternalSSTFileBasicTest, IngestWithTemperature) {
   }
 }
 
-TEST_F(ExternalSSTFileBasicTest, FailIfNotBottommostLevel) {
-  Options options = GetDefaultOptions();
+TEST_F(ExternalSSTFileBasicTest, FailIfNotBottommostLevelAndDisallowMemtable) {
+  for (bool disallow_memtable : {false, true}) {
+    Options options = GetDefaultOptions();
 
-  std::string file_path = sst_files_dir_ + std::to_string(1);
-  SstFileWriter sfw(EnvOptions(), options);
+    // First test with universal compaction
+    options.create_if_missing = true;
+    options.compaction_style = CompactionStyle::kCompactionStyleUniversal;
+    DestroyAndReopen(options);
 
-  ASSERT_OK(sfw.Open(file_path));
-  ASSERT_OK(sfw.Put("b", "dontcare"));
-  ASSERT_OK(sfw.Finish());
+    // And a CF potentially disallowing memtable write
+    options.disallow_memtable_writes = disallow_memtable;
+    CreateColumnFamilies({"cf0"}, options);
+    ASSERT_EQ(db_->GetOptions(handles_[0]).disallow_memtable_writes,
+              disallow_memtable);
 
-  // Test universal compaction + ingest with snapshot consistency
-  options.create_if_missing = true;
-  options.compaction_style = CompactionStyle::kCompactionStyleUniversal;
-  DestroyAndReopen(options);
-  {
-    const Snapshot* snapshot = db_->GetSnapshot();
-    ManagedSnapshot snapshot_guard(db_, snapshot);
-    IngestExternalFileOptions ifo;
-    ifo.fail_if_not_bottommost_level = true;
-    ifo.snapshot_consistency = true;
-    const Status s = db_->IngestExternalFile({file_path}, ifo);
-    ASSERT_TRUE(s.ok());
-  }
+    // Ingest with snapshot consistency
+    std::string file_path = sst_files_dir_ + std::to_string(1);
+    SstFileWriter sfw(EnvOptions(), options);
 
-  // Test level compaction
-  options.compaction_style = CompactionStyle::kCompactionStyleLevel;
-  options.num_levels = 2;
-  DestroyAndReopen(options);
-  ASSERT_OK(db_->Put(WriteOptions(), "a", "dontcare"));
-  ASSERT_OK(db_->Put(WriteOptions(), "c", "dontcare"));
-  ASSERT_OK(db_->Flush(FlushOptions()));
+    ASSERT_OK(sfw.Open(file_path));
+    ASSERT_OK(sfw.Put("b", "dontcare"));
+    ASSERT_OK(sfw.Finish());
 
-  ASSERT_OK(db_->Put(WriteOptions(), "b", "dontcare"));
-  ASSERT_OK(db_->Put(WriteOptions(), "d", "dontcare"));
-  ASSERT_OK(db_->Flush(FlushOptions()));
+    {
+      const Snapshot* snapshot = db_->GetSnapshot();
+      ManagedSnapshot snapshot_guard(db_, snapshot);
+      IngestExternalFileOptions ifo;
+      ifo.fail_if_not_bottommost_level = true;
+      ifo.snapshot_consistency = true;
+      ASSERT_OK(db_->IngestExternalFile(handles_[0], {file_path}, ifo));
+    }
 
-  {
-    CompactRangeOptions cro;
-    cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
-    ASSERT_OK(db_->CompactRange(cro, nullptr, nullptr));
+    // Test level compaction
+    options.compaction_style = CompactionStyle::kCompactionStyleLevel;
+    options.num_levels = 2;
+    CreateColumnFamilies({"cf1"}, options);
+    ASSERT_EQ(db_->GetOptions(handles_[1]).disallow_memtable_writes,
+              disallow_memtable);
 
-    IngestExternalFileOptions ifo;
-    ifo.fail_if_not_bottommost_level = true;
-    const Status s = db_->IngestExternalFile({file_path}, ifo);
-    ASSERT_TRUE(s.IsTryAgain());
+    if (!disallow_memtable) {
+      ASSERT_OK(Put(1, "a", "1"));
+      ASSERT_OK(Put(1, "c", "3"));
+      ASSERT_OK(Flush(1));
+
+      ASSERT_OK(Put(1, "b", "2"));
+      ASSERT_OK(Put(1, "d", "4"));
+      ASSERT_OK(Flush(1));
+    } else {
+      // Memtable write disallowed
+      EXPECT_EQ(Put(1, "a", "1").code(), Status::Code::kInvalidArgument);
+
+      // Use ingestion to get to the same state as above
+      std::string file_path2 = sst_files_dir_ + std::to_string(2);
+
+      ASSERT_OK(sfw.Open(file_path2));
+      ASSERT_OK(sfw.Put("a", "1"));
+      ASSERT_OK(sfw.Put("c", "3"));
+      ASSERT_OK(sfw.Finish());
+      ASSERT_OK(db_->IngestExternalFile(handles_[1], {file_path2}, {}));
+
+      ASSERT_OK(sfw.Open(file_path2));
+      ASSERT_OK(sfw.Put("b", "2"));
+      ASSERT_OK(sfw.Put("d", "4"));
+      ASSERT_OK(sfw.Finish());
+      ASSERT_OK(db_->IngestExternalFile(handles_[1], {file_path2}, {}));
+    }
+
+    {
+      CompactRangeOptions cro;
+      cro.bottommost_level_compaction = BottommostLevelCompaction::kForce;
+      ASSERT_OK(db_->CompactRange(cro, handles_[1], nullptr, nullptr));
+
+      IngestExternalFileOptions ifo;
+      ifo.fail_if_not_bottommost_level = true;
+      const Status s = db_->IngestExternalFile(handles_[1], {file_path}, ifo);
+      ASSERT_TRUE(s.IsTryAgain());
+    }
   }
 }
 

db/write_batch.cc

@@ -815,6 +815,12 @@ WriteBatchInternal::GetColumnFamilyIdAndTimestampSize(
         s = Status::InvalidArgument("Default cf timestamp size mismatch");
       }
     }
+    auto* cfd =
+        static_cast_with_check<ColumnFamilyHandleImpl>(column_family)->cfd();
+    if (cfd && cfd->ioptions().disallow_memtable_writes) {
+      s = Status::InvalidArgument(
+          "This column family has disallow_memtable_writes=true");
+    }
   } else if (b->default_cf_ts_sz_ > 0) {
     ts_sz = b->default_cf_ts_sz_;
   }
@@ -836,6 +842,12 @@ Status CheckColumnFamilyTimestampSize(ColumnFamilyHandle* column_family,
   if (cf_ts_sz != ts.size()) {
     return Status::InvalidArgument("timestamp size mismatch");
   }
+  auto* cfd =
+      static_cast_with_check<ColumnFamilyHandleImpl>(column_family)->cfd();
+  if (cfd && cfd->ioptions().disallow_memtable_writes) {
+    return Status::InvalidArgument(
+        "This column family has disallow_memtable_writes=true");
+  }
   return Status::OK();
 }
 }  // anonymous namespace
@@ -2185,6 +2197,13 @@ class MemTableInserter : public WriteBatch::Handler {
       }
       return false;
     }
+    auto* current = cf_mems_->current();
+    if (current && current->ioptions().disallow_memtable_writes) {
+      *s = Status::InvalidArgument(
+          "This column family has disallow_memtable_writes=true");
+      return false;
+    }
+
     if (recovering_log_number_ != 0 &&
         recovering_log_number_ < cf_mems_->GetLogNumber()) {
       // This is true only in recovery environment (recovering_log_number_ is

include/rocksdb/advanced_options.h

@@ -719,6 +719,17 @@ struct AdvancedColumnFamilyOptions {
   // Dynamically changeable through SetOptions() API
   bool report_bg_io_stats = false;
 
+  // Setting this option to true disallows ordinary writes to the column family
+  // and it can only be populated through import and ingestion. It is intended
+  // to protect "ingestion only" column families. This option is not currently
+  // supported on the default column family because of error handling challenges
+  // analogous to https://github.com/facebook/rocksdb/issues/13429
+  //
+  // This option is not mutable with SetOptions(). It can be changed between
+  // DB::Open() calls, but open will fail if recovering WAL writes to a CF with
+  // this option set.
+  bool disallow_memtable_writes = false;
+
   // This option has different meanings for different compaction styles:
   //
   // Leveled: Non-bottom-level files with all keys older than TTL will go

include/rocksdb/external_table.h

@@ -6,6 +6,7 @@
 #pragma once
 
 #include "rocksdb/customizable.h"
+#include "rocksdb/file_checksum.h"
 #include "rocksdb/iterator.h"
 #include "rocksdb/options.h"
 #include "rocksdb/status.h"
@@ -18,22 +19,21 @@ class ExternalTableFactory;
 // The interface defined in this file is subject to change at any time without
 // warning!!
 
-// This file defines an interface for plugging in an external table reader
+// This file defines an interface for plugging in an external table
 // into RocksDB. The external table reader will be used instead of the
-// BlockBasedTable to load and query sst files. As of now, creating the
-// external table files using RocksDB is not supported, but will be added in
-// the near future. The external table files can be created outside and
-// RocksDB and ingested into a RocksDB instance using the IngestExternalFIle()
-// API.
+// BlockBasedTable to load and query sst files.
+// The external table files can be created using an SstFileWriter. Eventually
+// external tables will be allowed to be ingested into a RocksDB instance
+// using the IngestExternalFIle() API.
 //
-// Initial support is for loading and querying the files using an
-// SstFileReader. We will add support for ingestion of an external table
-// into a limited RocksDB instance that only supports ingestion and not live
-// writes in the near future. It'll be followed by support for replacing the
-// column family by ingesting a new set of files. In all cases, the external
-// table files will only be allowed in the bottommost level.
+// Initial support is for writing and querying the files using an
+// SstFileWriter and SstFileReader. We will add support for ingestion of an
+// external table into a limited RocksDB instance that only supports ingestion
+// and not live writes in the near future. It'll be followed by support for
+// replacing the column family by ingesting a new set of files. In all cases,
+// the external table files will only be allowed in the bottommost level.
 //
-// The external table reader can support one or both of the following layouts -
+// The external table can support one or both of the following layouts -
 // 1. Total order seek - All the keys in the files are in sorted order, and a
 //    user can seek to the first, last, or any key in between and iterate
 //    forwards or backwards till the end of the range. To support this mode,
@@ -54,8 +54,8 @@ class ExternalTableFactory;
 //    true to seek to the first non-empty prefix (as determined by the key
 //    order) if the seek prefix is empty.
 //
-// Many of the options in ReadOptions may not be relevant to the external
-// table implementation.
+// Many of the options in ReadOptions and WriteOptions may not be relevant to
+// the external table implementation.
 // TODO: Specify which options are relevant
 
 class ExternalTableReader {
@@ -94,6 +94,62 @@ class ExternalTableReader {
   }
 };
 
+// A table builder interface that can be used by SstFileWriter to allow
+// RocksDB users to write external table files. The sequence of operations
+// to write an external table is as follows -
+// 1. Add() is called one or more times to write all key-values to the table.
+//    Its called in increasing key order, as determined by the comparator.
+//    The input key is a user key, i.e sequence number and value type are
+//    stripped out.
+// 2. After every Add() operation, status() is called to check the current
+//    status.
+// 3. After the last key is added, Finish() is called to do whatever is
+//    necessary to ensure the data is persisted in the table file.
+// 4. If there is a failure midway for some reason, Abandon() is called
+//    instead of Finish().
+// 5. At the end, FileSize(), GetTableProperties(), and status() are called to
+//    get the final size of the file, the table properties, and the final
+//    status. GetFileChecksum() and GetFileChecksumFuncName() may also be
+//    called to get checksum information about the whole file, but their
+//    implementation is optional.
+class ExternalTableBuilder {
+ public:
+  virtual ~ExternalTableBuilder() {}
+
+  // Write a single KV to the table file. This is guaranteed to be called
+  // in key order, and the write may be buffered and flushed at a later time.
+  virtual void Add(const Slice& key, const Slice& value) = 0;
+
+  // Return the current Status. This could return non-ok, for example, if
+  // Add() fails for some reason.
+  virtual Status status() const = 0;
+
+  // Flush and close the table file
+  virtual Status Finish() = 0;
+
+  // Delete the partial file and release any allocated resources. Either this
+  // or Finish() will be called, but not both.
+  virtual void Abandon() = 0;
+
+  // Return the size of the table file. Will be called at the end, after
+  // Finish().
+  virtual uint64_t FileSize() const = 0;
+
+  //  As mentioned in earlier comments, the following table properties must be
+  //  returned at a minimum -
+  //  comparator_name
+  //  num_entries
+  //  raw_key_size
+  //  raw_value_size
+  virtual TableProperties GetTableProperties() const = 0;
+
+  virtual std::string GetFileChecksum() const { return kUnknownFileChecksum; }
+
+  virtual const char* GetFileChecksumFuncName() const {
+    return kUnknownFileChecksumFuncName;
+  }
+};
+
 struct ExternalTableOptions {
   const std::shared_ptr<const SliceTransform>& prefix_extractor;
   const Comparator* comparator;
@@ -104,6 +160,29 @@ struct ExternalTableOptions {
       : prefix_extractor(_prefix_extractor), comparator(_comparator) {}
 };
 
+struct ExternalTableBuilderOptions {
+  const ReadOptions& read_options;
+  const WriteOptions& write_options;
+  const std::shared_ptr<const SliceTransform>& prefix_extractor;
+  const Comparator* comparator;
+  const std::string& column_family_name;
+  const std::string db_id;
+  const std::string db_session_id;
+  const TableFileCreationReason reason;
+
+  ExternalTableBuilderOptions(
+      const ReadOptions& _read_options, const WriteOptions& _write_options,
+      const std::shared_ptr<const SliceTransform>& _prefix_extractor,
+      const Comparator* _comparator, const std::string& _column_family_name,
+      const TableFileCreationReason _reason)
+      : read_options(_read_options),
+        write_options(_write_options),
+        prefix_extractor(_prefix_extractor),
+        comparator(_comparator),
+        column_family_name(_column_family_name),
+        reason(_reason) {}
+};
+
 class ExternalTableFactory : public Customizable {
  public:
   ~ExternalTableFactory() override {}
@@ -113,7 +192,11 @@ class ExternalTableFactory : public Customizable {
   virtual Status NewTableReader(
       const ReadOptions& read_options, const std::string& file_path,
       const ExternalTableOptions& table_options,
-      std::unique_ptr<ExternalTableReader>* table_reader) = 0;
+      std::unique_ptr<ExternalTableReader>* table_reader) const = 0;
+
+  virtual ExternalTableBuilder* NewTableBuilder(
+      const ExternalTableBuilderOptions& builder_options,
+      const std::string& file_path) const = 0;
 };
 
 // Allocate a TableFactory that wraps around an ExternalTableFactory. Use this

include/rocksdb/options.h

@@ -454,6 +454,7 @@ extern const char* kHostnameForDbHostId;
 enum class CompactionServiceJobStatus : char {
   kSuccess,
   kFailure,
+  kAborted,
   kUseLocal,
 };
 
@@ -1985,6 +1986,12 @@ struct ReadOptions {
   // reader implementation.
   uint64_t weight = 0;
 
+  // A map of name,value pairs that can be passed by the user to an
+  // external table reader. This is completely opaque to RocksDB and is
+  // ignored by the natively supported table readers like block based and plain
+  // table. This is only useful for Iterator.
+  std::optional<std::unordered_map<std::string, std::string>> property_bag;
+
   // *** END options for RocksDB internal use only ***
 
   ReadOptions() {}

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 10
 #define ROCKSDB_MINOR 0
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 1
 
 // 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

options/cf_options.cc

@@ -736,6 +736,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct ImmutableCFOptions, force_consistency_checks),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone}},
+        {"disallow_memtable_writes",
+         {offsetof(struct ImmutableCFOptions, disallow_memtable_writes),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
         {"default_temperature",
          {offsetof(struct ImmutableCFOptions, default_temperature),
           OptionType::kTemperature, OptionVerificationType::kNormal,
@@ -998,6 +1002,7 @@ ImmutableCFOptions::ImmutableCFOptions(const ColumnFamilyOptions& cf_options)
       num_levels(cf_options.num_levels),
       optimize_filters_for_hits(cf_options.optimize_filters_for_hits),
       force_consistency_checks(cf_options.force_consistency_checks),
+      disallow_memtable_writes(cf_options.disallow_memtable_writes),
       default_temperature(cf_options.default_temperature),
       memtable_insert_with_hint_prefix_extractor(
           cf_options.memtable_insert_with_hint_prefix_extractor),

options/cf_options.h

@@ -68,6 +68,8 @@ struct ImmutableCFOptions {
 
   bool force_consistency_checks;
 
+  bool disallow_memtable_writes;
+
   Temperature default_temperature;
 
   std::shared_ptr<const SliceTransform>

options/options.cc

@@ -90,6 +90,7 @@ AdvancedColumnFamilyOptions::AdvancedColumnFamilyOptions(const Options& options)
       paranoid_file_checks(options.paranoid_file_checks),
       force_consistency_checks(options.force_consistency_checks),
       report_bg_io_stats(options.report_bg_io_stats),
+      disallow_memtable_writes(options.disallow_memtable_writes),
       ttl(options.ttl),
       periodic_compaction_seconds(options.periodic_compaction_seconds),
       sample_for_compression(options.sample_for_compression),
@@ -395,6 +396,8 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
                    force_consistency_checks);
   ROCKS_LOG_HEADER(log, "               Options.report_bg_io_stats: %d",
                    report_bg_io_stats);
+  ROCKS_LOG_HEADER(log, "               Options.disallow_memtable_writes: %d",
+                   disallow_memtable_writes);
   ROCKS_LOG_HEADER(log, "                              Options.ttl: %" PRIu64,
                    ttl);
   ROCKS_LOG_HEADER(log,

options/options_helper.cc

@@ -326,6 +326,7 @@ void UpdateColumnFamilyOptions(const ImmutableCFOptions& ioptions,
   cf_opts->num_levels = ioptions.num_levels;
   cf_opts->optimize_filters_for_hits = ioptions.optimize_filters_for_hits;
   cf_opts->force_consistency_checks = ioptions.force_consistency_checks;
+  cf_opts->disallow_memtable_writes = ioptions.disallow_memtable_writes;
   cf_opts->memtable_insert_with_hint_prefix_extractor =
       ioptions.memtable_insert_with_hint_prefix_extractor;
   cf_opts->cf_paths = ioptions.cf_paths;

options/options_settable_test.cc

@@ -644,6 +644,7 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "hard_pending_compaction_bytes_limit=0;"
       "disable_auto_compactions=false;"
       "report_bg_io_stats=true;"
+      "disallow_memtable_writes=true;"
       "ttl=60;"
       "periodic_compaction_seconds=3600;"
       "sample_for_compression=0;"

src.mk

@@ -205,7 +205,7 @@ LIB_SOURCES =                                                   \
   table/cuckoo/cuckoo_table_builder.cc                          \
   table/cuckoo/cuckoo_table_factory.cc                          \
   table/cuckoo/cuckoo_table_reader.cc                           \
-  table/external_table_reader.cc				\
+  table/external_table.cc					\
   table/format.cc                                               \
   table/get_context.cc                                          \
   table/iterator.cc                                             \

table/external_table.cc

@@ -3,7 +3,7 @@
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 
-#include "rocksdb/external_table_reader.h"
+#include "rocksdb/external_table.h"
 
 #include "rocksdb/table.h"
 #include "table/internal_iterator.h"
@@ -16,7 +16,8 @@ namespace {
 
 class ExternalTableIterator : public InternalIterator {
  public:
-  explicit ExternalTableIterator(Iterator* iterator) : iterator_(iterator) {}
+  explicit ExternalTableIterator(Iterator* iterator)
+      : iterator_(iterator), valid_(false) {}
 
   // No copying allowed
   ExternalTableIterator(const ExternalTableIterator&) = delete;
@@ -24,7 +25,7 @@ class ExternalTableIterator : public InternalIterator {
 
   ~ExternalTableIterator() override {}
 
-  bool Valid() const override { return iterator_ && iterator_->Valid(); }
+  bool Valid() const override { return valid_; }
 
   void SeekToFirst() override {
     status_ = Status::OK();
@@ -94,23 +95,29 @@ class ExternalTableIterator : public InternalIterator {
     return Slice();
   }
 
-  Status status() const override {
-    return !status_.ok() ? status_
-                         : (iterator_ ? iterator_->status() : Status::OK());
-  }
+  Status status() const override { return status_; }
 
  private:
   std::unique_ptr<Iterator> iterator_;
   InternalKey key_;
+  bool valid_;
   Status status_;
 
-  void UpdateKey() { key_.Set(iterator_->key(), 0, ValueType::kTypeValue); }
+  void UpdateKey() {
+    if (iterator_) {
+      valid_ = iterator_->Valid();
+      status_ = iterator_->status();
+      if (valid_ && status_.ok()) {
+        key_.Set(iterator_->key(), 0, ValueType::kTypeValue);
+      }
+    }
+  }
 };
 
 class ExternalTableReaderAdapter : public TableReader {
  public:
   explicit ExternalTableReaderAdapter(
-      std::unique_ptr<ExternalTableReader> reader)
+      std::unique_ptr<ExternalTableReader>&& reader)
       : reader_(std::move(reader)) {}
 
   ~ExternalTableReaderAdapter() override {}
@@ -170,6 +177,62 @@ class ExternalTableReaderAdapter : public TableReader {
   std::unique_ptr<ExternalTableReader> reader_;
 };
 
+class ExternalTableBuilderAdapter : public TableBuilder {
+ public:
+  explicit ExternalTableBuilderAdapter(
+      std::unique_ptr<ExternalTableBuilder>&& builder)
+      : builder_(std::move(builder)), num_entries_(0) {}
+
+  void Add(const Slice& key, const Slice& value) override {
+    ParsedInternalKey pkey;
+    status_ = ParseInternalKey(key, &pkey, /*log_err_key=*/false);
+    if (status_.ok()) {
+      if (pkey.type != ValueType::kTypeValue) {
+        status_ = Status::NotSupported(
+            "Value type " + std::to_string(pkey.type) + "not supported");
+      } else {
+        builder_->Add(pkey.user_key, value);
+        num_entries_++;
+      }
+    }
+  }
+
+  Status status() const override {
+    if (status_.ok()) {
+      return builder_->status();
+    } else {
+      return status_;
+    }
+  }
+
+  IOStatus io_status() const override { return status_to_io_status(status()); }
+
+  Status Finish() override { return builder_->Finish(); }
+
+  void Abandon() override { builder_->Abandon(); }
+
+  uint64_t FileSize() const override { return builder_->FileSize(); }
+
+  uint64_t NumEntries() const override { return num_entries_; }
+
+  TableProperties GetTableProperties() const override {
+    return builder_->GetTableProperties();
+  }
+
+  std::string GetFileChecksum() const override {
+    return builder_->GetFileChecksum();
+  }
+
+  const char* GetFileChecksumFuncName() const override {
+    return builder_->GetFileChecksumFuncName();
+  }
+
+ private:
+  Status status_;
+  std::unique_ptr<ExternalTableBuilder> builder_;
+  uint64_t num_entries_;
+};
+
 class ExternalTableFactoryAdapter : public TableFactory {
  public:
   explicit ExternalTableFactoryAdapter(
@@ -197,8 +260,18 @@ class ExternalTableFactoryAdapter : public TableFactory {
     return Status::OK();
   }
 
-  TableBuilder* NewTableBuilder(const TableBuilderOptions&,
-                                WritableFileWriter*) const override {
+  using TableFactory::NewTableBuilder;
+  TableBuilder* NewTableBuilder(const TableBuilderOptions& topts,
+                                WritableFileWriter* file) const override {
+    std::unique_ptr<ExternalTableBuilder> builder;
+    ExternalTableBuilderOptions ext_topts(
+        topts.read_options, topts.write_options,
+        topts.moptions.prefix_extractor, topts.ioptions.user_comparator,
+        topts.column_family_name, topts.reason);
+    builder.reset(inner_->NewTableBuilder(ext_topts, file->file_name()));
+    if (builder) {
+      return new ExternalTableBuilderAdapter(std::move(builder));
+    }
     return nullptr;
   }
 

table/table_test.cc

@@ -36,7 +36,7 @@
 #include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
-#include "rocksdb/external_table_reader.h"
+#include "rocksdb/external_table.h"
 #include "rocksdb/file_checksum.h"
 #include "rocksdb/file_system.h"
 #include "rocksdb/filter_policy.h"
@@ -6531,28 +6531,112 @@ class ExternalTableReaderTest : public DBTestBase {
       : DBTestBase("external_table_reader_test", /*env_do_fsync=*/false) {}
 
  protected:
+  class DummyExternalTableFile {
+   public:
+    explicit DummyExternalTableFile(const std::string& file_path)
+        : file_path_(file_path), file_size_(0) {
+      props_.comparator_name = BytewiseComparator()->Name();
+    }
+
+    Status Serialize(
+        const std::vector<std::pair<std::string, std::string>>& kv_vec) {
+      for (auto& kv : kv_vec) {
+        SerializeOne(kv.first, kv.second);
+        props_.raw_key_size += kv.first.length();
+        props_.raw_value_size += kv.second.length();
+      }
+      props_.num_entries = kv_vec.size();
+      file_size_ = buf_.length();
+      return WriteStringToFile(Env::Default(), buf_, file_path_);
+    }
+
+    Status Deserialize(std::map<std::string, std::string>& kv_map) {
+      Status s = ReadFileToString(Env::Default(), file_path_, &buf_);
+      if (!s.ok()) {
+        return s;
+      }
+
+      while (buf_.length() > 0) {
+        std::pair<std::string, std::string> kv;
+        s = DeserializeOne(kv);
+        if (!s.ok()) {
+          break;
+        }
+        size_t key_size = kv.first.length();
+        size_t value_size = kv.second.length();
+        kv_map.emplace(std::move(kv));
+        props_.raw_key_size += key_size;
+        props_.raw_value_size += value_size;
+      }
+      props_.num_entries = kv_map.size();
+      return s;
+    }
+
+    TableProperties GetTableProperties() const { return props_; }
+
+    uint64_t FileSize() const { return file_size_; }
+
+   private:
+    struct ItemHeader {
+      uint32_t key_size;
+      uint32_t value_size;
+    };
+
+    void SerializeOne(const Slice& key, const Slice& value) {
+      ItemHeader hdr;
+      hdr.key_size = static_cast<uint32_t>(key.size());
+      hdr.value_size = static_cast<uint32_t>(value.size());
+      buf_.append(static_cast<char*>(static_cast<void*>(&hdr)), sizeof(hdr));
+      buf_.append(key.data(), key.size());
+      buf_.append(value.data(), value.size());
+    }
+
+    Status DeserializeOne(std::pair<std::string, std::string>& kv) {
+      ItemHeader hdr;
+      size_t copied =
+          buf_.copy(static_cast<char*>(static_cast<void*>(&hdr)), sizeof(hdr));
+      if (copied < sizeof(hdr)) {
+        return Status::Corruption();
+      }
+      buf_.erase(0, sizeof(hdr));
+      if (buf_.length() < hdr.key_size + hdr.value_size) {
+        return Status::Corruption();
+      }
+      kv.first.assign(std::string_view(buf_.data(), hdr.key_size));
+      buf_.erase(0, hdr.key_size);
+      kv.second.assign(std::string_view(buf_.data(), hdr.value_size));
+      buf_.erase(0, hdr.value_size);
+      return Status::OK();
+    }
+
+    std::string file_path_;
+    std::string buf_;
+    TableProperties props_;
+    uint64_t file_size_;
+  };
+
   class DummyExternalTableIterator : public Iterator {
    public:
-    explicit DummyExternalTableIterator(bool empty) : empty_(empty) {}
+    explicit DummyExternalTableIterator(
+        const ReadOptions& ro, const std::map<std::string, std::string>& kv_map)
+        : weight_(ro.weight), kv_map_(kv_map), valid_(false) {}
 
-    bool Valid() const override { return empty_ ? !empty_ : valid_; }
+    bool Valid() const override { return valid_; }
 
     void SeekToFirst() override {
-      valid_ = true;
+      iter_ = kv_map_.begin();
+      valid_ = iter_ != kv_map_.end();
       status_ = Status::OK();
     }
 
     void SeekToLast() override {
-      valid_ = true;
-      status_ = Status::OK();
+      valid_ = false;
+      status_ = Status::NotSupported();
     }
 
     void Seek(const Slice& target) override {
-      if (target.compare(key_str) <= 0) {
-        valid_ = true;
-      } else {
-        valid_ = false;
-      }
+      iter_ = kv_map_.find(target.ToString());
+      valid_ = iter_ != kv_map_.end();
       status_ = Status::OK();
     }
 
@@ -6562,7 +6646,9 @@ class ExternalTableReaderTest : public DBTestBase {
     }
 
     void Next() override {
-      valid_ = false;
+      iter_++;
+      weight_--;
+      valid_ = iter_ != kv_map_.end() && weight_ > 0;
       // status_ is still ok. valid_ indicates end of scan
     }
 
@@ -6573,7 +6659,7 @@ class ExternalTableReaderTest : public DBTestBase {
 
     Slice key() const override {
       // If valid_ is false or status_ is non-ok, behavior is indeterminate
-      return Slice(key_str);
+      return Slice(iter_->first);
     }
 
     Status status() const override {
@@ -6583,31 +6669,36 @@ class ExternalTableReaderTest : public DBTestBase {
 
     Slice value() const override {
       // If valid_ is false or status_ is non-ok, behavior is indeterminate
-      return Slice(value_str);
+      return Slice(iter_->second);
     }
 
    private:
-    static const std::string key_str;
-    static const std::string value_str;
-
+    uint64_t weight_;
+    std::map<std::string, std::string> kv_map_;
     bool valid_ = false;
-    bool empty_;
     Status status_ = Status::OK();
+    std::map<std::string, std::string>::iterator iter_;
   };
 
   class DummyExternalTableReader : public ExternalTableReader {
    public:
+    explicit DummyExternalTableReader(const std::string& file_path)
+        : file_(file_path) {
+      Status s = file_.Deserialize(kv_map_);
+      EXPECT_OK(s);
+    }
+
     Iterator* NewIterator(const ReadOptions& read_options,
                           const SliceTransform* /*prefix_extractor*/) override {
-      return new DummyExternalTableIterator((read_options.weight == 0) ? true
-                                                                       : false);
+      return new DummyExternalTableIterator(read_options, kv_map_);
     }
 
     Status Get(const ReadOptions& /*read_options*/, const Slice& key,
                const SliceTransform* /*prefix_extractor*/,
                std::string* value) override {
-      if (!key.compare("foo")) {
-        value->assign("bar");
+      auto iter = kv_map_.find(key.ToString());
+      if (iter != kv_map_.end()) {
+        value->assign(iter->second);
         return Status::OK();
       }
       return Status::NotFound();
@@ -6635,6 +6726,43 @@ class ExternalTableReaderTest : public DBTestBase {
       props->raw_value_size = 3;
       return props;
     }
+
+   private:
+    std::map<std::string, std::string> kv_map_;
+    DummyExternalTableFile file_;
+  };
+
+  class DummyExternalTableBuilder : public ExternalTableBuilder {
+   public:
+    explicit DummyExternalTableBuilder(const std::string& file_path)
+        : file_(file_path) {}
+
+    void Add(const Slice& key, const Slice& value) override {
+      if (!kv_vec_.empty()) {
+        ASSERT_LT(BytewiseComparator()->Compare(kv_vec_.back().first, key), 0);
+      }
+      kv_vec_.emplace_back(key.ToString(), value.ToString());
+    }
+
+    Status Finish() override {
+      status_ = file_.Serialize(kv_vec_);
+      return status_;
+    }
+
+    void Abandon() override { kv_vec_.clear(); }
+
+    uint64_t FileSize() const override { return file_.FileSize(); }
+
+    TableProperties GetTableProperties() const override {
+      return file_.GetTableProperties();
+    }
+
+    Status status() const override { return status_; }
+
+   private:
+    std::vector<std::pair<std::string, std::string>> kv_vec_;
+    DummyExternalTableFile file_;
+    Status status_;
   };
 
   class DummyExternalTableFactory : public ExternalTableFactory {
@@ -6642,28 +6770,42 @@ class ExternalTableReaderTest : public DBTestBase {
     const char* Name() const override { return "DummyExternalTableFactory"; }
 
     Status NewTableReader(
-        const ReadOptions& /*read_options*/, const std::string& /*file_path*/,
+        const ReadOptions& /*read_options*/, const std::string& file_path,
         const ExternalTableOptions& /*topts*/,
-        std::unique_ptr<ExternalTableReader>* table_reader) override {
-      table_reader->reset(new DummyExternalTableReader());
+        std::unique_ptr<ExternalTableReader>* table_reader) const override {
+      table_reader->reset(new DummyExternalTableReader(file_path));
       return Status::OK();
     }
+
+    ExternalTableBuilder* NewTableBuilder(
+        const ExternalTableBuilderOptions& /*opts*/,
+        const std::string& file_path) const override {
+      return new DummyExternalTableBuilder(file_path);
+    }
   };
 };
 
-const std::string ExternalTableReaderTest::DummyExternalTableIterator::key_str =
-    "foo";
-const std::string
-    ExternalTableReaderTest::DummyExternalTableIterator::value_str = "bar";
-
 TEST_F(ExternalTableReaderTest, BasicTest) {
   std::shared_ptr<ExternalTableFactory> factory =
       std::make_shared<DummyExternalTableFactory>();
 
+  std::string file_path = test::PerThreadDBPath("external_table");
+  {
+    std::unique_ptr<ExternalTableBuilder> builder;
+    builder.reset(factory->NewTableBuilder(
+        ExternalTableBuilderOptions(ReadOptions(), WriteOptions(),
+                                    std::shared_ptr<const SliceTransform>(),
+                                    BytewiseComparator(), "default",
+                                    TableFileCreationReason::kMisc),
+        file_path));
+    builder->Add("foo", "bar");
+    ASSERT_OK(builder->Finish());
+  }
+
   std::unique_ptr<ExternalTableReader> reader;
   std::shared_ptr<SliceTransform> prefix_extractor;
   ASSERT_OK(factory->NewTableReader(
-      {}, "", ExternalTableOptions(prefix_extractor, nullptr), &reader));
+      {}, file_path, ExternalTableOptions(prefix_extractor, nullptr), &reader));
 
   ReadOptions ro;
   ro.weight = 1;
@@ -6694,14 +6836,19 @@ TEST_F(ExternalTableReaderTest, SstReaderTest) {
   std::string dbname = test::PerThreadDBPath("external_table_reader_test");
   std::string ingest_file = dbname + "test.immutabledb";
   dbname += "_db";
+  // This test doesn't work with some custom Envs, like EncryptedEnv
+  options.env = Env::Default();
 
   std::shared_ptr<ExternalTableFactory> factory =
       std::make_shared<DummyExternalTableFactory>();
   options.table_factory = NewExternalTableFactory(factory);
 
-  // Create a file
-  ASSERT_OK(WriteStringToFile(options.env, "Hello World", ingest_file,
-                              /*should_sync=*/true));
+  std::unique_ptr<SstFileWriter> writer;
+  writer.reset(new SstFileWriter(EnvOptions(), options));
+  ASSERT_OK(writer->Open(ingest_file));
+  ASSERT_OK(writer->Put("foo", "bar"));
+  ASSERT_OK(writer->Finish());
+  writer.reset();
 
   std::unique_ptr<SstFileReader> reader(new SstFileReader(options));
   ASSERT_OK(reader->Open(ingest_file));

unreleased_history/behavior_changes/parallelize_backup_verification.md

@@ -1 +0,0 @@
-`VerifyBackup` in `verify_with_checksum`=`true` mode will now evaluate checksums in parallel. As a result, unlike in case of original implementation, the API won't bail out on a very first corruption / mismatch and instead will iterate over all the backup files logging success / _degree_of_failure_ for each.

unreleased_history/behavior_changes/wbwi-ordering.md

@@ -1 +0,0 @@
-* Reversed the order of updates to the same key in WriteBatchWithIndex. This means if there are multiple updates to the same key, the most recent update is ordered first. This affects the output of WBWIIterator. When WriteBatchWithIndex is created with `overwrite_key=true`, this affects the output only if Merge is used (#13387).

unreleased_history/bug_fixes/get_merge_operand_fix.md

@@ -1 +0,0 @@
-Fixed GetMergeOperands() API in ReadOnlyDB and SecondaryDB

unreleased_history/bug_fixes/get_merge_operand_imm.md

@@ -1 +0,0 @@
-* Fix a bug in `GetMergeOperands()` that can return incorrect status (MergeInProgress) and incorrect number of merge operands. This can happen when `GetMergeOperandsOptions::continue_cb` is set, both active and immutable memtables have merge operands and the callback stops the look up at the immutable memtable.

unreleased_history/new_features/auto_refresh_iterator.md

@@ -1 +0,0 @@
-Introduced new `auto_refresh_iterator_with_snapshot` opt-in knob that (when enabled) will periodically release obsolete memory and storage resources for as long as the iterator is making progress and its supplied `read_options.snapshot` was initialized with non-nullptr value.

unreleased_history/new_features/external_table_reader.md

@@ -1 +0,0 @@
-Added the ability to plug-in a custom table reader implementation. See include/rocksdb/external_table_reader.h for more details.

unreleased_history/new_features/faiss_ivf_index.md

@@ -1 +0,0 @@
-Experimental feature: RocksDB now supports FAISS inverted file based indices via the secondary indexing framework. Applications can use FAISS secondary indices to automatically quantize embeddings and perform K-nearest-neighbors similarity searches. See `FaissIVFIndex` and `SecondaryIndex` for more details. Note: the FAISS integration currently requires using the BUCK build.

unreleased_history/new_features/num_running_compaction_sorted_runs.md

@@ -1 +0,0 @@
-Add new DB property `num_running_compaction_sorted_runs` that tracks the number of sorted runs being processed by currently running compactions

unreleased_history/new_features/simple_secondary_index.md

@@ -1 +0,0 @@
-Experimental feature: added support for simple secondary indices that index the specified column as-is. See `SimpleSecondaryIndex` and `SecondaryIndex` for more details.

unreleased_history/public_api_changes/secondary_index_iterator.md

@@ -1 +0,0 @@
-Updated the query API of the experimental secondary indexing feature by removing the earlier `SecondaryIndex::NewIterator` virtual and adding a `SecondaryIndexIterator` class that can be utilized by applications to find the primary keys for a given search target.

unreleased_history/public_api_changes/secondary_index_primary_key.md

@@ -1 +0,0 @@
-Added back the ability to leverage the primary key when building secondary index entries. This involved changes to the signatures of `SecondaryIndex::GetSecondary{KeyPrefix,Value}` as well as the addition of a new method `SecondaryIndex::FinalizeSecondaryKeyPrefix`. See the API comments for more details.

unreleased_history/public_api_changes/zstd.md

@@ -1 +0,0 @@
-* Minimum supported version of ZSTD is now 1.4.0, for code simplification. Obsolete `CompressionType` `kZSTDNotFinalCompression` is also removed.