Update HISTORY and version

Summary:
**Context/Summary:**
Fix a race condition (illustrated below) in FIFO size-based compaction where concurrent threads could select the same non-L0 file, causing assertion failures in debug builds or "Cannot delete table file from LSM tree" errors in release builds.
```
Thread 1                           Thread 2
--------                           --------
FIFO size-based compaction
   ↓
Pick L2 file
   ↓
Mark: file.being_compacted = true (file.being_compacted was false)
   ↓
WriteManifestStart (unlock mutex) ─→ FIFO size-based compaction starts
   ↓                                   ↓
Continue manifest write...          Pick SAME L2 file
   ↓                                Mark: file.being_compacted = true  (file.being_compacted was true) ❌
   ↓                                   ↓
   ↓                                Unlock mutex, wait for manifest
   ↓                                   ↓
Lock mutex ←─────────────────────────────────┘
   ↓
Delete L2 file ✅
   ↓
Complete ─────────────────────────────→ Try delete same file ❌
                                        ↓
                                     ERROR: "file not in LSM tree"

🐛 BUG: Both threads pick the same file!
    Thread 2 doesn't properly check file.being_compacted flag
```

**Test**
New test that fails before the fix and passes after

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

Reviewed By: xingbowang

Differential Revision: D82279731

Pulled By: hx235

fbshipit-source-id: b426517f2d1b23dd7d4951157822a2d322fe1435

HISTORY.md

@@ -1,6 +1,43 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 10.6.2 (09/15/2025)
+### Bug Fixes
+* Fix a race condition in FIFO size-based compaction where concurrent threads could select the same non-L0 file, causing assertion failures in debug builds or "Cannot delete table file from LSM tree" errors in release builds.
+
+## 10.6.1 (09/05/2025)
+### New Features
+* Add the fail_if_no_udi_on_open flag in BlockBasedTableOption to control whether a missing user defined index block in a SST is a hard error or not.
+* Add new option `MultiScanArgs::max_prefetch_size` that limits the memory usage of per file pinning of prefetched blocks.
+
+## 10.6.0 (08/22/2025)
+### New Features
+* 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.
+* Introduce `MultiScanArgs::io_coalesce_threshold` to allow a configurable IO coalescing threshold.
+
+### Public API Changes
+* `IngestExternalFileOptions::allow_db_generated_files` now allows files ingestion of any DB generated SST file, instead of only the ones with all keys having sequence number 0.
+* `decouple_partitioned_filters = true` is now the default in BlockBasedTableOptions.
+* GetTtl() API is now available in TTL DB
+* Minimum supported version of LZ4 library is now 1.7.0 (r129 from 2015)
+* Some changes to experimental Compressor and CompressionManager APIs
+* A new Filesystem::SyncFile function is added for syncing a file that was already written, such as on file ingestion. The default implementation matches previous RocksDB behavior: re-open the file for read-write, sync it, and close it. We recommend overriding for FileSystems that do not require syncing for crash recovery or do not handle (well) re-opening for writes.
+
+### Behavior Changes
+* When `allow_ingest_behind` is enabled, compaction will no longer drop tombstones based on the absence of underlying data. Tombstones will be preserved to apply to ingested files.
+
+### Bug Fixes
+* Files in dropped column family won't be returned to the caller upon successful, offline MANIFEST iteration in `GetFileChecksumsFromCurrentManifest`.
+* Fix a bug in MultiScan that causes it to fall back to a normal scan when dictionary compression is enabled.
+* Fix a crash in iterator Prepare() when fill_cache=false
+* Fix a bug in MultiScan where incorrect results can be returned when a Scan's range is across multiple files.
+* Fixed a bug in remote compaction that may mistakenly delete live SST file(s) during the cleanup phase when no keys survive the compaction (all expired)
+* Allow a user defined index to be configured from a string.
+* Make the User Defined Index interface consistently use the user key format, fixing the previous mixed usage of internal and user key.
+
+### Performance Improvements
+* Small improvement to CPU efficiency of compression using built-in algorithms, and a dramatic efficiency improvement for LZ4HC, based on reusing data structures between invocations.
+
 ## 10.5.0 (07/18/2025)
 ### Public API Changes
 * DB option skip_checking_sst_file_sizes_on_db_open is deprecated, in favor of validating file size in parallel in a thread pool, when db is opened. When DB is opened, with paranoid check enabled, a file with the wrong size would fail the DB open. With paranoid check disabled, the DB open would succeed, the column family with the corrupted file would not be read or write, while the other healthy column families could be read and write normally. When max_open_files option is not set to -1, only a subset of the files will be opened and checked. The rest of the files will be opened and checked when they are accessed.

db/compaction/compaction_picker_fifo.cc

@@ -258,6 +258,9 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
     // better serves a major type of FIFO use cases where smaller keys are
     // associated with older data.
     for (const auto& f : last_level_files) {
+      if (f->being_compacted) {
+        continue;
+      }
       total_size -= f->fd.file_size;
       inputs[0].files.push_back(f);
       char tmp_fsize[16];

db/db_compaction_test.cc

@@ -7128,6 +7128,70 @@ TEST_F(DBCompactionTest, PartialManualCompaction) {
   ASSERT_OK(dbfull()->CompactRange(cro, nullptr, nullptr));
 }
 
+TEST_F(DBCompactionTest, ConcurrentFIFOPickingSameFileBug) {
+  Options opts = CurrentOptions();
+  opts.compaction_style = CompactionStyle::kCompactionStyleLevel;
+  opts.num_levels = 3;
+  opts.disable_auto_compactions = true;
+  opts.max_background_jobs = 3;
+
+  DestroyAndReopen(opts);
+
+  ASSERT_OK(Put("k1", "v1"));
+  ASSERT_OK(Flush());
+
+  // Create a non-L0 SST file for multi-level FIFO size-based compaction later
+  MoveFilesToLevel(2);
+
+  Options opts_new(opts);
+  opts_new.compaction_style = CompactionStyle::kCompactionStyleFIFO;
+  opts_new.max_open_files = -1;
+  // Set a low threshold to trigger multi-level size-based compaction
+  opts_new.compaction_options_fifo.max_table_files_size = 1;
+
+  Reopen(opts_new);
+
+  const CompactRangeOptions cro;
+  const Slice begin_key("k1");
+  const Slice end_key("k2");
+
+  std::unique_ptr<port::Thread> concurrent_compaction;
+
+  bool within_first_compaction = true;
+  SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::LogAndApply:WriteManifestStart", [&](void* /*arg*/) {
+        if (!within_first_compaction) {
+          return;
+        }
+        within_first_compaction = false;
+
+        // To allow the second/concurrent compaction to still see the non-L0
+        // SST file and coerce the bug of picking that file
+        SyncPoint::GetInstance()->LoadDependency({
+            {"DBImpl::BackgroundCompaction:BeforeCompaction",
+             "VersionSet::LogAndApply:WriteManifest"},
+        });
+
+        concurrent_compaction.reset(new port::Thread([&]() {
+          // Before the fix, the second CompactRange() will either fail the
+          // assertion of double file picking `being_compacted !=
+          // inputs_[i][j]->being_compacted` in debug mode or cause LSM shape
+          // corruption "Cannot delete table file XXX from level 2 since it is
+          // not in the LSM tree" in release mode
+          Status s = db_->CompactRange(cro, &begin_key, &end_key);
+          ASSERT_OK(s);
+        }));
+      });
+
+  SyncPoint::GetInstance()->EnableProcessing();
+  Status s = db_->CompactRange(cro, &begin_key, &end_key);
+  SyncPoint::GetInstance()->DisableProcessing();
+
+  ASSERT_OK(s);
+
+  concurrent_compaction->join();
+}
+
 TEST_F(DBCompactionTest, ManualCompactionFailsInReadOnlyMode) {
   // Regression test for bug where manual compaction hangs forever when the DB
   // is in read-only mode. Verify it now at least returns, despite failing.

db/external_sst_file_basic_test.cc

@@ -1311,7 +1311,7 @@ TEST_F(ExternalSSTFileBasicTest, SyncFailure) {
     });
     if (i == 0) {
       SyncPoint::GetInstance()->SetCallBack(
-          "ExternalSstFileIngestionJob::CheckSyncReturnCode", [&](void* s) {
+          "ExternalSstFileIngestionJob::Prepare:Reopen", [&](void* s) {
             Status* status = static_cast<Status*>(s);
             if (status->IsNotSupported()) {
               no_sync = true;
@@ -1372,11 +1372,11 @@ TEST_F(ExternalSSTFileBasicTest, ReopenNotSupported) {
   options.create_if_missing = true;
   options.env = env_;
 
-  SyncPoint::GetInstance()->SetCallBack("FileSystem::SyncFile:Open",
-                                        [&](void* arg) {
-                                          Status* s = static_cast<Status*>(arg);
-                                          *s = Status::NotSupported();
-                                        });
+  SyncPoint::GetInstance()->SetCallBack(
+      "ExternalSstFileIngestionJob::Prepare:Reopen", [&](void* arg) {
+        Status* s = static_cast<Status*>(arg);
+        *s = Status::NotSupported();
+      });
   SyncPoint::GetInstance()->EnableProcessing();
 
   DestroyAndReopen(options);

db/external_sst_file_ingestion_job.cc

@@ -160,26 +160,35 @@ Status ExternalSstFileIngestionJob::Prepare(
         // It is unsafe to assume application had sync the file and file
         // directory before ingest the file. For integrity of RocksDB we need
         // to sync the file.
-        TEST_SYNC_POINT("ExternalSstFileIngestionJob::BeforeSyncIngestedFile");
-        auto s = fs_->SyncFile(path_inside_db, env_options_, IOOptions(),
-                               db_options_.use_fsync, nullptr);
-        TEST_SYNC_POINT("ExternalSstFileIngestionJob::AfterSyncIngestedFile");
-        TEST_SYNC_POINT_CALLBACK(
-            "ExternalSstFileIngestionJob::CheckSyncReturnCode", &s);
-        if (!s.ok()) {
-          if (s.IsNotSupported()) {
-            // Some file systems (especially remote/distributed) don't support
-            // SyncFile API. Ignore the NotSupported error in that case.
-            ROCKS_LOG_WARN(db_options_.info_log,
-                           "After link the file, SyncFile API is not supported "
-                           "for file %s: %s",
-                           path_inside_db.c_str(), status.ToString().c_str());
-          } else {
-            // for other errors, propagate the error
-            status = s;
-            ROCKS_LOG_WARN(db_options_.info_log,
-                           "Failed to sync ingested file %s: %s",
-                           path_inside_db.c_str(), status.ToString().c_str());
+
+        // TODO(xingbo), We should in general be moving away from production
+        // uses of ReuseWritableFile (except explicitly for WAL recycling),
+        // ReopenWritableFile, and NewRandomRWFile. We should create a
+        // FileSystem::SyncFile/FsyncFile API that by default does the
+        // re-open+sync+close combo but can (a) be reused easily, and (b) be
+        // overridden to do that more cleanly, e.g. in EncryptedEnv.
+        // https://github.com/facebook/rocksdb/issues/13741
+        std::unique_ptr<FSWritableFile> file_to_sync;
+        Status s = fs_->ReopenWritableFile(path_inside_db, env_options_,
+                                           &file_to_sync, nullptr);
+        TEST_SYNC_POINT_CALLBACK("ExternalSstFileIngestionJob::Prepare:Reopen",
+                                 &s);
+        // Some file systems (especially remote/distributed) don't support
+        // reopening a file for writing and don't require reopening and
+        // syncing the file. Ignore the NotSupported error in that case.
+        if (!s.IsNotSupported()) {
+          status = s;
+          if (status.ok()) {
+            TEST_SYNC_POINT(
+                "ExternalSstFileIngestionJob::BeforeSyncIngestedFile");
+            status = SyncIngestedFile(file_to_sync.get());
+            TEST_SYNC_POINT(
+                "ExternalSstFileIngestionJob::AfterSyncIngestedFile");
+            if (!status.ok()) {
+              ROCKS_LOG_WARN(db_options_.info_log,
+                             "Failed to sync ingested file %s: %s",
+                             path_inside_db.c_str(), status.ToString().c_str());
+            }
           }
         }
       } else if (status.IsNotSupported() &&

env/composite_env_wrapper.h

@@ -142,12 +142,6 @@ class CompositeEnv : public Env {
     return file_system_->LinkFile(s, t, io_opts, &dbg);
   }
 
-  Status SyncFile(const std::string& fname, const EnvOptions& env_options,
-                  bool use_fsync) override {
-    return file_system_->SyncFile(fname, env_options, IOOptions(), use_fsync,
-                                  nullptr);
-  }
-
   Status NumFileLinks(const std::string& fname, uint64_t* count) override {
     IOOptions io_opts;
     IODebugContext dbg;

env/env.cc

@@ -528,13 +528,6 @@ class LegacyFileSystemWrapper : public FileSystem {
     return status_to_io_status(target_->LinkFile(s, t));
   }
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& /*io_options*/, bool use_fsync,
-                    IODebugContext* /*dbg*/) override {
-    return status_to_io_status(
-        target_->SyncFile(fname, file_options, use_fsync));
-  }
-
   IOStatus NumFileLinks(const std::string& fname, const IOOptions& /*options*/,
                         uint64_t* count, IODebugContext* /*dbg*/) override {
     return status_to_io_status(target_->NumFileLinks(fname, count));
@@ -866,43 +859,6 @@ std::string Env::GenerateUniqueId() {
   return result;
 }
 
-// This API Env::SyncFile is used for testing for 2 reasons:
-//
-// 1. The default implementation of SyncFile API is essentially a wrapper of
-// other FileSystem APIs. FaultInjectionTestEnv uses this default
-// implementation to call other FileSystem APIs defined at
-// FaultInjectionTestEnv class to inject failurses. See
-// FaultInjectionTestEnv::SyncFile for more details
-//
-// 2. Some of old tests are using LegacyFileSystemWrapper.
-// LegacyFileSystemWrapper forwards the API call to EnvWrapper, which forwards
-// to CompositeEnv, and then forwards to the actual FileSystem implemention.
-// Without this API in Env, LegacyFileSystemWrapper will not be able to
-// forward the API call to EnvWrapper, causing the default FileSystem API to
-// be called.
-//
-// Due to the above reason, adding a new API in FileSystem, would very likely
-// require the same API to be added to Env.
-//
-// TODO xingbo. Getting rid of FileSystem functions from Env.
-// We need to simplify the relationship between Env and FileSystem. At least
-// for internal test, we should stop using Env and switch to FileSystem, if
-// possible. Related github issue #9274
-Status Env::SyncFile(const std::string& fname, const EnvOptions& env_options,
-                     bool use_fsync) {
-  std::unique_ptr<WritableFile> file_to_sync;
-  auto status = ReopenWritableFile(fname, &file_to_sync, env_options);
-  TEST_SYNC_POINT_CALLBACK("FileSystem::SyncFile:Open", &status);
-  if (status.ok()) {
-    if (use_fsync) {
-      status = file_to_sync->Fsync();
-    } else {
-      status = file_to_sync->Sync();
-    }
-  }
-  return status;
-}
-
 SequentialFile::~SequentialFile() = default;
 
 RandomAccessFile::~RandomAccessFile() = default;

env/env_encryption.cc

@@ -664,8 +664,6 @@ class EncryptedFileSystemImpl : public EncryptedFileSystem {
                               const FileOptions& options,
                               std::unique_ptr<FSWritableFile>* result,
                               IODebugContext* dbg) override {
-    // TODO xingbo Add unit test for the new implementation of
-    // EncryptedFileSysmteImpl::ReopenWritableFile.
     result->reset();
     if (options.use_mmap_reads || options.use_mmap_writes) {
       return IOStatus::InvalidArgument();
@@ -816,15 +814,6 @@ class EncryptedFileSystemImpl : public EncryptedFileSystem {
     return status;
   }
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& io_options, bool use_fsync,
-                    IODebugContext* dbg) override {
-    // Use the underlying file system to sync the file, as we don't need to
-    // read/write the file.
-    return FileSystemWrapper::SyncFile(fname, file_options, io_options,
-                                       use_fsync, dbg);
-  }
-
  private:
   std::shared_ptr<EncryptionProvider> provider_;
 };

env/file_system.cc

@@ -107,23 +107,6 @@ IOStatus FileSystem::ReuseWritableFile(const std::string& fname,
   return NewWritableFile(fname, opts, result, dbg);
 }
 
-IOStatus FileSystem::SyncFile(const std::string& fname,
-                              const FileOptions& file_options,
-                              const IOOptions& io_options, bool use_fsync,
-                              IODebugContext* dbg) {
-  std::unique_ptr<FSWritableFile> file_to_sync;
-  auto status = ReopenWritableFile(fname, file_options, &file_to_sync, dbg);
-  TEST_SYNC_POINT_CALLBACK("FileSystem::SyncFile:Open", &status);
-  if (status.ok()) {
-    if (use_fsync) {
-      status = file_to_sync->Fsync(io_options, dbg);
-    } else {
-      status = file_to_sync->Sync(io_options, dbg);
-    }
-  }
-  return status;
-}
-
 IOStatus FileSystem::NewLogger(const std::string& fname,
                                const IOOptions& io_opts,
                                std::shared_ptr<Logger>* result,

env/mock_env.cc

@@ -957,14 +957,6 @@ IOStatus MockFileSystem::LinkFile(const std::string& src,
   return IOStatus::OK();
 }
 
-IOStatus MockFileSystem::SyncFile(const std::string& /*fname*/,
-                                  const FileOptions& /*file_options*/,
-                                  const IOOptions& /*io_options*/,
-                                  bool /*use_fsync*/, IODebugContext* /*dbg*/) {
-  // Noop
-  return IOStatus::OK();
-}
-
 IOStatus MockFileSystem::NewLogger(const std::string& fname,
                                    const IOOptions& io_opts,
                                    std::shared_ptr<Logger>* result,

env/mock_env.h

@@ -86,10 +86,6 @@ class MockFileSystem : public FileSystem {
   IOStatus LinkFile(const std::string& /*src*/, const std::string& /*target*/,
                     const IOOptions& /*options*/,
                     IODebugContext* /*dbg*/) override;
-  IOStatus SyncFile(const std::string& /*fname*/,
-                    const FileOptions& /*file_options*/,
-                    const IOOptions& /*io_options*/, bool /*use_fsync*/,
-                    IODebugContext* /*dbg*/) override;
   IOStatus LockFile(const std::string& fname, const IOOptions& options,
                     FileLock** lock, IODebugContext* dbg) override;
   IOStatus UnlockFile(FileLock* lock, const IOOptions& options,

include/rocksdb/env.h

@@ -385,13 +385,6 @@ class Env : public Customizable {
     return Status::NotSupported("LinkFile is not supported for this Env");
   }
 
-  // Sync the file content to file system.
-  // This API is only used for testing.
-  // See FileSystem::SyncFile comment for details
-  virtual Status SyncFile(const std::string& /*fname*/,
-                          const EnvOptions& /*env_options*/,
-                          bool /*use_fsync*/);
-
   virtual Status NumFileLinks(const std::string& /*fname*/,
                               uint64_t* /*count*/) {
     return Status::NotSupported(
@@ -1550,11 +1543,6 @@ class EnvWrapper : public Env {
     return target_.env->LinkFile(s, t);
   }
 
-  Status SyncFile(const std::string& fname, const EnvOptions& env_options,
-                  bool use_fsync) override {
-    return target_.env->SyncFile(fname, env_options, use_fsync);
-  }
-
   Status NumFileLinks(const std::string& fname, uint64_t* count) override {
     return target_.env->NumFileLinks(fname, count);
   }

include/rocksdb/file_system.h

@@ -606,18 +606,6 @@ class FileSystem : public Customizable {
         "LinkFile is not supported for this FileSystem");
   }
 
-  // Sync the file content to file system.
-  // The default implementation would open, sync and close the file.
-  // This function could be overridden with no-op, if the file system
-  // automatically sync the data when file is closed.
-  // This is used when a user-provided file, probably unsynced, is pulled into a
-  // context where power-outage-proof persistence is required (e.g.
-  // IngestExternalFile without copy).
-  virtual IOStatus SyncFile(const std::string& fname,
-                            const FileOptions& file_options,
-                            const IOOptions& io_options, bool use_fsync,
-                            IODebugContext* dbg);
-
   virtual IOStatus NumFileLinks(const std::string& /*fname*/,
                                 const IOOptions& /*options*/,
                                 uint64_t* /*count*/, IODebugContext* /*dbg*/) {
@@ -1604,12 +1592,6 @@ class FileSystemWrapper : public FileSystem {
     return target_->LinkFile(s, t, options, dbg);
   }
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& io_options, bool use_fsync,
-                    IODebugContext* dbg) override {
-    return target_->SyncFile(fname, file_options, io_options, use_fsync, dbg);
-  }
-
   IOStatus NumFileLinks(const std::string& fname, const IOOptions& options,
                         uint64_t* count, IODebugContext* dbg) override {
     return target_->NumFileLinks(fname, options, count, dbg);

include/rocksdb/options.h

@@ -1790,9 +1790,11 @@ class MultiScanArgs {
     comp_ = other.comp_;
     original_ranges_ = other.original_ranges_;
     io_coalesce_threshold = other.io_coalesce_threshold;
+    max_prefetch_size = other.max_prefetch_size;
   }
   MultiScanArgs(MultiScanArgs&& other) noexcept
       : io_coalesce_threshold(other.io_coalesce_threshold),
+        max_prefetch_size(other.max_prefetch_size),
         comp_(other.comp_),
         original_ranges_(std::move(other.original_ranges_)) {}
 
@@ -1800,6 +1802,7 @@ class MultiScanArgs {
     comp_ = other.comp_;
     original_ranges_ = other.original_ranges_;
     io_coalesce_threshold = other.io_coalesce_threshold;
+    max_prefetch_size = other.max_prefetch_size;
     return *this;
   }
 
@@ -1808,6 +1811,7 @@ class MultiScanArgs {
       comp_ = other.comp_;
       original_ranges_ = std::move(other.original_ranges_);
       io_coalesce_threshold = other.io_coalesce_threshold;
+      max_prefetch_size = other.max_prefetch_size;
     }
     return *this;
   }
@@ -1849,6 +1853,18 @@ class MultiScanArgs {
 
   uint64_t io_coalesce_threshold = 16 << 10;  // 16KB by default
 
+  // Maximum size (in bytes) for the data blocks loaded by a MultiScan.
+  // This limits the amount of I/O and memory usage by pinned data blocks.
+  //
+  // When set to 0 (the default), there is no limit. When the limit is reached,
+  // the iterator will start returning Status::PrefetchLimitReached().
+  //
+  // Note that prefetching happens only once in Prepare(), which is different
+  // from ReadOptions::readahead_size, which applies any time the iterator does
+  // I/O.
+  // Note that this limit is per file and applies to compressed block size.
+  uint64_t max_prefetch_size = 0;
+
  private:
   // The comparator used for ordering ranges
   const Comparator* comp_;

include/rocksdb/statistics.h

@@ -542,6 +542,9 @@ enum Tickers : uint32_t {
   // TransactionOptions::large_txn_commit_optimize_threshold.
   NUMBER_WBWI_INGEST,
 
+  // Failure to load the UDI during SST table open
+  SST_USER_DEFINED_INDEX_LOAD_FAIL_COUNT,
+
   TICKER_ENUM_MAX
 };
 

include/rocksdb/status.h

@@ -115,6 +115,7 @@ class Status {
     kIOFenced = 14,
     kMergeOperatorFailed = 15,
     kMergeOperandThresholdExceeded = 16,
+    kPrefetchLimitReached = 17,
     kMaxSubCode
   };
 
@@ -318,6 +319,10 @@ class Status {
 
   static Status LockLimit() { return Status(kAborted, kLockLimit); }
 
+  static Status PrefetchLimitReached() {
+    return Status(kIncomplete, kPrefetchLimitReached);
+  }
+
   // Returns true iff the status indicates success.
   bool ok() const {
     MarkChecked();
@@ -486,6 +491,13 @@ class Status {
     return (code() == kIOError) && (subcode() == kIOFenced);
   }
 
+  // Returns true iff the status indicates prefetch limit reached during
+  // MultiScan.
+  bool IsPrefetchLimitReached() const {
+    MarkChecked();
+    return (code() == kIncomplete) && (subcode() == kPrefetchLimitReached);
+  }
+
   // Return a string representation of this status suitable for printing.
   // Returns the string "OK" for success.
   std::string ToString() const;

include/rocksdb/table.h

@@ -502,6 +502,12 @@ struct BlockBasedTableOptions {
   // during table building.
   std::shared_ptr<UserDefinedIndexFactory> user_defined_index_factory = nullptr;
 
+  // EXPERIMENTAL
+  //
+  // Return an error Status if a user_defined_index_factory is configured,
+  // but there's no corresponding UDI block in the SST file being opened.
+  bool fail_if_no_udi_on_open = false;
+
   // If true, place whole keys in the filter (not just prefixes).
   // This must generally be true for gets to be efficient.
   bool whole_key_filtering = true;

include/rocksdb/version.h

@@ -13,7 +13,7 @@
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 10
 #define ROCKSDB_MINOR 6
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_PATCH 2
 
 // Do not use these. We made the mistake of declaring macros starting with
 // double underscore. Now we have to live with our choice. We'll deprecate these

monitoring/statistics.cc

@@ -273,6 +273,8 @@ const std::vector<std::pair<Tickers, std::string>> TickersNameMap = {
     {FILE_READ_CORRUPTION_RETRY_SUCCESS_COUNT,
      "rocksdb.file.read.corruption.retry.success.count"},
     {NUMBER_WBWI_INGEST, "rocksdb.number.wbwi.ingest"},
+    {SST_USER_DEFINED_INDEX_LOAD_FAIL_COUNT,
+     "rocksdb.sst.user.defined.index.load.fail.count"},
 };
 
 const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {

options/options_settable_test.cc

@@ -203,7 +203,8 @@ TEST_F(OptionsSettableTest, BlockBasedTableOptionsAllFieldsSettable) {
       "max_auto_readahead_size=0;"
       "prepopulate_block_cache=kDisable;"
       "initial_auto_readahead_size=0;"
-      "num_file_reads_for_auto_readahead=0",
+      "num_file_reads_for_auto_readahead=0;"
+      "fail_if_no_udi_on_open=true",
       new_bbto));
 
   ASSERT_EQ(unset_bytes_base,

table/block_based/block_based_table_factory.cc

@@ -399,6 +399,9 @@ static struct BlockBasedTableTypeInfo {
          {offsetof(struct BlockBasedTableOptions,
                    num_file_reads_for_auto_readahead),
           OptionType::kUInt64T, OptionVerificationType::kNormal}},
+        {"fail_if_no_udi_on_open",
+         {offsetof(struct BlockBasedTableOptions, fail_if_no_udi_on_open),
+          OptionType::kBoolean, OptionVerificationType::kNormal}},
     };
   }
 } block_based_table_type_info;
@@ -874,6 +877,14 @@ std::string BlockBasedTableFactory::GetPrintableOptions() const {
                ? "nullptr"
                : table_options_.filter_policy->Name());
   ret.append(buffer);
+  snprintf(buffer, kBufferSize, "  user_defined_index_factory: %s\n",
+           table_options_.user_defined_index_factory == nullptr
+               ? "nullptr"
+               : table_options_.user_defined_index_factory->Name());
+  ret.append(buffer);
+  snprintf(buffer, kBufferSize, "  fail_if_no_udi_on_open: %d\n",
+           table_options_.fail_if_no_udi_on_open);
+  ret.append(buffer);
   snprintf(buffer, kBufferSize, "  whole_key_filtering: %d\n",
            table_options_.whole_key_filtering);
   ret.append(buffer);

table/block_based/block_based_table_iterator.cc

@@ -982,7 +982,6 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
 
   // Gather all relevant data block handles
   std::vector<BlockHandle> blocks_to_prepare;
-  Status s;
   std::vector<std::tuple<size_t, size_t>> block_ranges_per_scan;
   for (const auto& scan_opt : *scan_opts) {
     size_t num_blocks = 0;
@@ -1042,11 +1041,26 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
   // Look up entries in cache and pin if exist.
   // Store indices of blocks to read.
   std::vector<size_t> blocks_to_read;
-  std::vector<CachableEntry<Block>> pinned_data_blocks_guard;
-  pinned_data_blocks_guard.resize(blocks_to_prepare.size());
+  std::vector<CachableEntry<Block>> pinned_data_blocks_guard(
+      blocks_to_prepare.size());
+  uint64_t total_prefetch_size = 0;
+
   for (size_t i = 0; i < blocks_to_prepare.size(); ++i) {
     const auto& data_block_handle = blocks_to_prepare[i];
-    s = table_->LookupAndPinBlocksInCache<Block_kData>(
+
+    // Check if we would exceed the prefetch size limit with this block
+    total_prefetch_size +=
+        BlockBasedTable::BlockSizeWithTrailer(data_block_handle);
+    if (multiscan_opts->max_prefetch_size > 0 &&
+        total_prefetch_size > multiscan_opts->max_prefetch_size) {
+      // All remaining blocks are by default empty.
+      for (size_t j = i; j < blocks_to_prepare.size(); ++j) {
+        assert(pinned_data_blocks_guard[j].IsEmpty());
+      }
+      break;
+    }
+
+    Status s = table_->LookupAndPinBlocksInCache<Block_kData>(
         read_options_, data_block_handle,
         &pinned_data_blocks_guard[i].As<Block_kData>());
 
@@ -1088,10 +1102,13 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
 
     // do IO
     IOOptions io_opts;
-    s = table_->get_rep()->file->PrepareIOOptions(read_options_, io_opts);
-    if (!s.ok()) {
-      // Abort: PrepareIOOptions failed
-      return;
+    {
+      Status s =
+          table_->get_rep()->file->PrepareIOOptions(read_options_, io_opts);
+      if (!s.ok()) {
+        // Abort: PrepareIOOptions failed
+        return;
+      }
     }
 
     // Init read requests for Multi-Read
@@ -1163,11 +1180,13 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
     }
 
     AlignedBuf aligned_buf;
-    s = table_->get_rep()->file.get()->MultiRead(
-        io_opts, read_reqs.data(), read_reqs.size(),
-        direct_io ? &aligned_buf : nullptr);
-    if (!s.ok()) {
-      return;
+    {
+      Status s = table_->get_rep()->file.get()->MultiRead(
+          io_opts, read_reqs.data(), read_reqs.size(),
+          direct_io ? &aligned_buf : nullptr);
+      if (!s.ok()) {
+        return;
+      }
     }
     for (auto& req : read_reqs) {
       if (!req.status.ok()) {
@@ -1181,7 +1200,8 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
         table_->get_rep()->decompressor.get();
     CachableEntry<DecompressorDict> cached_dict;
     if (table_->get_rep()->uncompression_dict_reader) {
-      s = table_->get_rep()
+      Status s =
+          table_->get_rep()
               ->uncompression_dict_reader->GetOrReadUncompressionDictionary(
                   /* prefetch_buffer= */ nullptr, read_options_,
                   /* get_context= */ nullptr, /* lookup_context= */ nullptr,
@@ -1226,7 +1246,7 @@ void BlockBasedTableIterator::Prepare(const MultiScanArgs* multiscan_opts) {
             table_->get_rep()->footer.GetBlockTrailerSize() > 0;
 #endif
         assert(pinned_data_blocks_guard[block_idx].IsEmpty());
-        s = table_->CreateAndPinBlockInCache<Block_kData>(
+        Status s = table_->CreateAndPinBlockInCache<Block_kData>(
             read_options_, block, decompressor, &tmp_contents,
             &(pinned_data_blocks_guard[block_idx].As<Block_kData>()));
         if (!s.ok()) {
@@ -1290,6 +1310,16 @@ bool BlockBasedTableIterator::SeekMultiScan(const Slice* target) {
       }
 
       ResetDataIter();
+
+      // Check if we've hit an empty entry indicating prefetch limit reached
+      if (multi_scan_->pinned_data_blocks[cur_scan_start_idx].IsEmpty()) {
+        multi_scan_->cur_data_block_idx = cur_scan_start_idx;
+        multi_scan_->prefetch_limit_reached = true;
+        assert(!Valid());
+        assert(status().IsPrefetchLimitReached());
+        return true;
+      }
+
       // Note that the block_iter_ takes ownership of the pinned data block
       // TODO: we can delegate the clean up like with pinned_iters_mgr_ if
       // need to pin blocks longer.
@@ -1346,6 +1376,16 @@ void BlockBasedTableIterator::FindBlockForwardInMultiScan() {
     // Move to the next pinned data block
     ResetDataIter();
     ++multi_scan_->cur_data_block_idx;
+
+    // Check if we've hit an empty entry indicating prefetch limit reached
+    if (multi_scan_->pinned_data_blocks[multi_scan_->cur_data_block_idx]
+            .IsEmpty()) {
+      multi_scan_->prefetch_limit_reached = true;
+      assert(!Valid());
+      assert(status().IsPrefetchLimitReached());
+      return;
+    }
+
     table_->NewDataBlockIterator<DataBlockIter>(
         read_options_,
         multi_scan_->pinned_data_blocks[multi_scan_->cur_data_block_idx],

table/block_based/block_based_table_iterator.h

@@ -145,10 +145,14 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
       assert(!multi_scan_);
       return index_iter_->status();
     } else if (block_iter_points_to_real_block_) {
+      // This is the common case.
       return block_iter_.status();
     } else if (async_read_in_progress_) {
       assert(!multi_scan_);
       return Status::TryAgain("Async read in progress");
+    } else if (multi_scan_ && multi_scan_->prefetch_limit_reached) {
+      assert(!Valid());
+      return Status::PrefetchLimitReached();
     } else {
       return Status::OK();
     }
@@ -385,6 +389,10 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
     size_t next_scan_idx;
     size_t cur_data_block_idx;
 
+    // When true, the iterator will return
+    // Status::Incomplete(Status::kPrefetchLimitReached).
+    bool prefetch_limit_reached;
+
     MultiScanState(
         const MultiScanArgs* _scan_opts,
         std::vector<CachableEntry<Block>>&& _pinned_data_blocks,
@@ -393,7 +401,8 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
           pinned_data_blocks(std::move(_pinned_data_blocks)),
           block_ranges_per_scan(std::move(_block_ranges_per_scan)),
           next_scan_idx(0),
-          cur_data_block_idx(0) {}
+          cur_data_block_idx(0),
+          prefetch_limit_reached(false) {}
   };
 
   std::unique_ptr<MultiScanState> multi_scan_;

table/block_based/block_based_table_reader.cc

@@ -1333,25 +1333,54 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks(
     s = FindMetaBlock(meta_iter, kUserDefinedIndexPrefix + udi_name,
                       &udi_block_handle);
     if (!s.ok()) {
-      return s;
+      RecordTick(rep_->ioptions.statistics.get(),
+                 SST_USER_DEFINED_INDEX_LOAD_FAIL_COUNT);
+      if (table_options.fail_if_no_udi_on_open) {
+        ROCKS_LOG_ERROR(rep_->ioptions.logger,
+                        "Failed to find the the UDI block %s in file %s; %s",
+                        udi_name.c_str(), rep_->file->file_name().c_str(),
+                        s.ToString().c_str());
+        // MAke the status more informative
+        s = Status::Corruption(s.ToString(), rep_->file->file_name());
+        return s;
+      } else {
+        // Emit a warning, but ignore the error status
+        ROCKS_LOG_WARN(rep_->ioptions.logger,
+                       "Failed to find the the UDI block %s in file %s; %s",
+                       udi_name.c_str(), rep_->file->file_name().c_str(),
+                       s.ToString().c_str());
+        s = Status::OK();
+      }
     }
-    // Read the block, and allocate on heap or pin in cache. The UDI block is
-    // not compressed. RetrieveBlock will verify the checksum.
-    s = RetrieveBlock(prefetch_buffer, ro, udi_block_handle,
-                      rep_->decompressor.get(), &rep_->udi_block,
-                      /*get_context=*/nullptr, lookup_context,
-                      /*for_compaction=*/false, use_cache, /*async_read=*/false,
-                      /*use_block_cache_for_lookup=*/false);
-    if (!s.ok()) {
-      return s;
-    }
-    assert(!rep_->udi_block.IsEmpty());
 
-    std::unique_ptr<UserDefinedIndexReader> udi_reader =
-        table_options.user_defined_index_factory->NewReader(
-            rep_->udi_block.GetValue()->data);
-    index_reader = std::make_unique<UserDefinedIndexReaderWrapper>(
-        udi_name, std::move(index_reader), std::move(udi_reader));
+    // If the UDI block size is 0, that means there's effectively no user
+    // defined index. In that case, skip setting up the reader.
+    if (udi_block_handle.size() > 0) {
+      // Read the block, and allocate on heap or pin in cache. The UDI block is
+      // not compressed. RetrieveBlock will verify the checksum.
+      if (s.ok()) {
+        s = RetrieveBlock(prefetch_buffer, ro, udi_block_handle,
+                          rep_->decompressor.get(), &rep_->udi_block,
+                          /*get_context=*/nullptr, lookup_context,
+                          /*for_compaction=*/false, use_cache,
+                          /*async_read=*/false,
+                          /*use_block_cache_for_lookup=*/false);
+      }
+      if (s.ok()) {
+        assert(!rep_->udi_block.IsEmpty());
+
+        std::unique_ptr<UserDefinedIndexReader> udi_reader =
+            table_options.user_defined_index_factory->NewReader(
+                rep_->udi_block.GetValue()->data);
+        if (udi_reader) {
+          index_reader = std::make_unique<UserDefinedIndexReaderWrapper>(
+              udi_name, std::move(index_reader), std::move(udi_reader));
+        } else {
+          s = Status::Corruption("Failed to create UDI reader for " + udi_name +
+                                 " in file " + rep_->file->file_name());
+        }
+      }
+    }
   }
 
   rep_->index_reader = std::move(index_reader);
@@ -1359,7 +1388,7 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks(
   // The partitions of partitioned index are always stored in cache. They
   // are hence follow the configuration for pin and prefetch regardless of
   // the value of cache_index_and_filter_blocks
-  if (prefetch_all || pin_partition) {
+  if (s.ok() && (prefetch_all || pin_partition)) {
     s = rep_->index_reader->CacheDependencies(ro, pin_partition,
                                               prefetch_buffer);
   }

table/block_based/block_based_table_reader_test.cc

@@ -1176,6 +1176,205 @@ TEST_P(BlockBasedTableReaderTest, MultiScanPrepare) {
   ASSERT_OK(iter->status());
 }
 
+TEST_P(BlockBasedTableReaderTest, MultiScanPrefetchSizeLimit) {
+  if (compression_type_ != kNoCompression) {
+    // This test relies on block sizes to be close to what's set in option.
+    ROCKSDB_GTEST_BYPASS("This test assumes no compression.");
+    return;
+  }
+  Options options;
+  ReadOptions read_opts;
+  size_t ts_sz = options.comparator->timestamp_size();
+
+  // Generate data that spans multiple blocks
+  std::vector<std::pair<std::string, std::string>> kv =
+      BlockBasedTableReaderBaseTest::GenerateKVMap(
+          20 /* num_block */, true /* mixed_with_human_readable_string_value */,
+          ts_sz);
+
+  std::string table_name = "BlockBasedTableReaderTest_PrefetchSizeLimit" +
+                           CompressionTypeToString(compression_type_);
+
+  ImmutableOptions ioptions(options);
+  CreateTable(table_name, ioptions, compression_type_, kv,
+              compression_parallel_threads_, compression_dict_bytes_);
+
+  std::unique_ptr<BlockBasedTable> table;
+  FileOptions foptions;
+  foptions.use_direct_reads = use_direct_reads_;
+  InternalKeyComparator comparator(options.comparator);
+  NewBlockBasedTableReader(foptions, ioptions, comparator, table_name, &table,
+                           true /* bool prefetch_index_and_filter_in_cache */,
+                           nullptr /* status */, persist_udt_);
+
+  // Default block size is 4KB
+  //
+  // Tests when no block is loaded
+  {
+    std::unique_ptr<InternalIterator> iter;
+    iter.reset(table->NewIterator(
+        read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
+        /*skip_filters=*/false, TableReaderCaller::kUncategorized));
+
+    MultiScanArgs scan_options(BytewiseComparator());
+    scan_options.max_prefetch_size = 1024;  // less than block size
+    scan_options.insert(ExtractUserKey(kv[0].first),
+                        ExtractUserKey(kv[5].first));
+
+    iter->Prepare(&scan_options);
+
+    // Should be able to scan the first block, but not more
+    iter->Seek(kv[0].first);
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_TRUE(iter->status().IsPrefetchLimitReached());
+  }
+
+  // Some blocks are loaded
+  {
+    std::unique_ptr<InternalIterator> iter;
+    iter.reset(table->NewIterator(
+        read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
+        /*skip_filters=*/false, TableReaderCaller::kUncategorized));
+
+    MultiScanArgs scan_options(BytewiseComparator());
+    scan_options.max_prefetch_size = 9 * 1024;  // 9KB - 2 blocks with buffer
+    scan_options.insert(ExtractUserKey(kv[1 * kEntriesPerBlock].first),
+                        ExtractUserKey(kv[8 * kEntriesPerBlock].first));
+
+    iter->Prepare(&scan_options);
+    iter->Seek(kv[1 * kEntriesPerBlock].first);
+    size_t scanned_keys = 0;
+
+    // Should be able to scan up to 2 blocks worth of data
+    while (iter->Valid()) {
+      ASSERT_EQ(iter->key().ToString(),
+                kv[scanned_keys + 1 * kEntriesPerBlock].first);
+      iter->Next();
+      scanned_keys++;
+    }
+
+    ASSERT_TRUE(iter->status().IsPrefetchLimitReached());
+    ASSERT_EQ(scanned_keys, 2 * kEntriesPerBlock);
+  }
+
+  // Tests with some block loaded in cache already:
+  // Blocks 1 and 2 are already in cache by the above test.
+  // Here we try blocks 0 - 5, with prefetch limit to 3 blocks, and expect to
+  // read 3 blocks.
+  {
+    std::unique_ptr<InternalIterator> iter;
+    iter.reset(table->NewIterator(
+        read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
+        /*skip_filters=*/false, TableReaderCaller::kUncategorized));
+
+    MultiScanArgs scan_options(BytewiseComparator());
+    scan_options.max_prefetch_size = 3 * 4 * 1024 + 1024;  // 3 blocks + 1KB
+    scan_options.insert(ExtractUserKey(kv[0].first),
+                        ExtractUserKey(kv[5 * kEntriesPerBlock].first));
+
+    iter->Prepare(&scan_options);
+    iter->Seek(kv[0].first);
+    size_t scanned_keys = 0;
+    // Should only read 3 blocks (blocks 0, 1, 2)
+    // already cached.
+    while (iter->Valid()) {
+      ASSERT_EQ(iter->key().ToString(), kv[scanned_keys].first);
+      iter->Next();
+      scanned_keys++;
+    }
+    ASSERT_TRUE(iter->status().IsPrefetchLimitReached());
+    ASSERT_EQ(scanned_keys, 3 * kEntriesPerBlock);
+  }
+
+  // Multiple scan ranges with prefetch limit
+  {
+    std::unique_ptr<InternalIterator> iter;
+    iter.reset(table->NewIterator(
+        read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
+        /*skip_filters=*/false, TableReaderCaller::kUncategorized));
+
+    MultiScanArgs scan_options(BytewiseComparator());
+    scan_options.max_prefetch_size = 5 * 4 * 1024 + 1024;  // 5 blocks + 1KB
+    // Will read 5 entries from first scan range, and 4 blocks from the second
+    // scan range
+    scan_options.insert(ExtractUserKey(kv[0].first),
+                        ExtractUserKey(kv[5].first));
+    scan_options.insert(ExtractUserKey(kv[12 * kEntriesPerBlock].first),
+                        ExtractUserKey(kv[17 * kEntriesPerBlock].first));
+    scan_options.insert(ExtractUserKey(kv[18 * kEntriesPerBlock].first),
+                        ExtractUserKey(kv[19 * kEntriesPerBlock].first));
+
+    iter->Prepare(&scan_options);
+
+    iter->Seek(kv[0].first);
+    size_t scanned_keys = 0;
+    size_t key_idx = 0;
+    while (iter->Valid()) {
+      ASSERT_EQ(iter->key().ToString(), kv[key_idx].first);
+      iter->Next();
+      scanned_keys++;
+      key_idx++;
+      if (key_idx == 5) {
+        iter->Seek(kv[12 * kEntriesPerBlock].first);
+        key_idx = 12 * kEntriesPerBlock;
+      }
+    }
+    ASSERT_EQ(scanned_keys, 5 + 4 * kEntriesPerBlock);
+    ASSERT_TRUE(iter->status().IsPrefetchLimitReached());
+  }
+
+  // Prefetch limit is big enough for all scan ranges.
+  {
+    std::unique_ptr<InternalIterator> iter;
+    iter.reset(table->NewIterator(
+        read_opts, options_.prefix_extractor.get(), /*arena=*/nullptr,
+        /*skip_filters=*/false, TableReaderCaller::kUncategorized));
+
+    MultiScanArgs scan_options(BytewiseComparator());
+    scan_options.max_prefetch_size = 10 * 1024 * 1024;  // 10MB
+    scan_options.insert(ExtractUserKey(kv[0].first),
+                        ExtractUserKey(kv[5].first));
+    scan_options.insert(ExtractUserKey(kv[8 * kEntriesPerBlock].first),
+                        ExtractUserKey(kv[12 * kEntriesPerBlock].first));
+    scan_options.insert(ExtractUserKey(kv[18 * kEntriesPerBlock].first),
+                        ExtractUserKey(kv[19 * kEntriesPerBlock].first));
+
+    iter->Prepare(&scan_options);
+
+    iter->Seek(kv[0].first);
+    size_t scanned_keys = 0;
+    size_t key_idx = 0;
+    // Scan first range
+    while (iter->Valid() && key_idx < 5) {
+      ASSERT_EQ(iter->key().ToString(), kv[key_idx].first);
+      iter->Next();
+      scanned_keys++;
+      key_idx++;
+    }
+    // Move to second range
+    iter->Seek(kv[8 * kEntriesPerBlock].first);
+    key_idx = 8 * kEntriesPerBlock;
+    while (iter->Valid() && key_idx < 12 * kEntriesPerBlock) {
+      ASSERT_EQ(iter->key().ToString(), kv[key_idx].first);
+      iter->Next();
+      scanned_keys++;
+      key_idx++;
+    }
+    // Move to third range
+    iter->Seek(kv[18 * kEntriesPerBlock].first);
+    key_idx = 18 * kEntriesPerBlock;
+    while (iter->Valid() && key_idx < 19 * kEntriesPerBlock) {
+      ASSERT_EQ(iter->key().ToString(), kv[key_idx].first);
+      iter->Next();
+      scanned_keys++;
+      key_idx++;
+    }
+    // Should not hit prefetch limit
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(scanned_keys, 5 + 4 * kEntriesPerBlock + 1 * kEntriesPerBlock);
+  }
+}
+
 // Param 1: compression type
 // Param 2: whether to use direct reads
 // Param 3: Block Based Table Index type, partitioned filters are also enabled

table/table_test.cc

@@ -7480,6 +7480,9 @@ class UserDefinedIndexTest : public BlockBasedTableTestBase {
                           const Slice* first_key_in_next_block,
                           const BlockHandle& block_handle,
                           std::string* separator_scratch) override {
+        if (keys_added_ == 0) {
+          return last_key_in_current_block;
+        }
         EXPECT_EQ(last_key_in_current_block.size(), 5);
         if (first_key_in_next_block) {
           EXPECT_EQ(first_key_in_next_block->size(), 5);
@@ -7500,12 +7503,19 @@ class UserDefinedIndexTest : public BlockBasedTableTestBase {
 
       void OnKeyAdded(const Slice& key, ValueType /*value*/,
                       const Slice& /*value*/) override {
+        if (key.starts_with("dummy")) {
+          return;
+        }
         EXPECT_EQ(key.size(), 5);
         // Track keys added to the index
         keys_added_++;
       }
 
       Status Finish(Slice* index_contents) override {
+        if (entries_added_ == 0) {
+          *index_contents = Slice();
+          return Status::OK();
+        }
         // Serialize the index data
         std::string result;
         for (const auto& entry : index_data_) {
@@ -8020,6 +8030,7 @@ TEST_F(UserDefinedIndexTest, IngestTest) {
 
 // Verify that external file ingestion fails if we try to ingest an SST file
 // without the UDI and a UDI factory is configured in BlockBasedTableOptions
+// and fail_if_no_udi_on_open is true in BlockBasedTableOptions.
 TEST_F(UserDefinedIndexTest, IngestFailTest) {
   Options options;
   BlockBasedTableOptions table_options;
@@ -8051,6 +8062,7 @@ TEST_F(UserDefinedIndexTest, IngestFailTest) {
   auto user_defined_index_factory =
       std::make_shared<TestUserDefinedIndexFactory>();
   table_options.user_defined_index_factory = user_defined_index_factory;
+  table_options.fail_if_no_udi_on_open = true;
   options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
   std::unique_ptr<DB> db;
@@ -8065,6 +8077,72 @@ TEST_F(UserDefinedIndexTest, IngestFailTest) {
   s = db->IngestExternalFile(cfh, {ingest_file}, ifo);
   ASSERT_NOK(s);
 
+  ASSERT_OK(db->SetOptions(
+      cfh, {{"block_based_table_factory", "{fail_if_no_udi_on_open=false;}"}}));
+  s = db->IngestExternalFile(cfh, {ingest_file}, ifo);
+  ASSERT_OK(s);
+
+  ASSERT_OK(db->DestroyColumnFamilyHandle(cfh));
+  ASSERT_OK(db->Close());
+  ASSERT_OK(DestroyDB(dbname, options));
+}
+
+TEST_F(UserDefinedIndexTest, IngestEmptyUDI) {
+  Options options;
+  BlockBasedTableOptions table_options;
+  std::string dbname = test::PerThreadDBPath("user_defined_index_test");
+  std::string ingest_file = dbname + "test.sst";
+  std::string ingest_file2 = dbname + "dummy.sst";
+
+  // Set up the user-defined index factory
+  auto user_defined_index_factory =
+      std::make_shared<TestUserDefinedIndexFactory>();
+  table_options.user_defined_index_factory = user_defined_index_factory;
+  // Set up custom flush block policy that flushes every 3 keys
+  table_options.flush_block_policy_factory =
+      std::make_shared<CustomFlushBlockPolicyFactory>();
+
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+  std::unique_ptr<SstFileWriter> writer;
+  writer.reset(new SstFileWriter(EnvOptions(), options));
+  ASSERT_OK(writer->Open(ingest_file));
+
+  // Add 100 keys instead of just 5
+  for (int i = 0; i < 100; i++) {
+    std::stringstream ss;
+    ss << std::setw(2) << std::setfill('0') << i;
+    std::string key = "key" + ss.str();
+    std::string value = "value" + ss.str();
+    ASSERT_OK(writer->Put(key, value));
+  }
+  ASSERT_OK(writer->Finish());
+  writer.reset();
+  writer.reset(new SstFileWriter(EnvOptions(), options));
+  ASSERT_OK(writer->Open(ingest_file2));
+  ASSERT_OK(writer->Put("dummy", "val"));
+  ASSERT_OK(writer->Finish());
+  writer.reset();
+
+  table_options.fail_if_no_udi_on_open = true;
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+
+  std::unique_ptr<DB> db;
+  options.create_if_missing = true;
+  Status s = DB::Open(options, dbname, &db);
+  ASSERT_OK(s);
+  ASSERT_TRUE(db != nullptr);
+  ColumnFamilyHandle* cfh = nullptr;
+  ASSERT_OK(db->CreateColumnFamily(options, "new_cf", &cfh));
+
+  std::vector<IngestExternalFileArg> ifa;
+  ifa.emplace_back();
+  ifa[0].column_family = cfh;
+  ifa[0].external_files.emplace_back(ingest_file);
+  ifa[0].external_files.emplace_back(ingest_file2);
+  s = db->IngestExternalFiles(ifa);
+  ASSERT_OK(s);
+
   ASSERT_OK(db->DestroyColumnFamilyHandle(cfh));
   ASSERT_OK(db->Close());
   ASSERT_OK(DestroyDB(dbname, options));

unreleased_history/behavior_changes/ingest_behind_tombstone.md

@@ -1 +0,0 @@
-* When `allow_ingest_behind` is enabled, compaction will no longer drop tombstones based on the absence of underlying data. Tombstones will be preserved to apply to ingested files.

unreleased_history/bug_fixes/handle_drop_column_family_edit_version_in_file_checksum_retriever.md

@@ -1 +0,0 @@
-* Files in dropped column family won't be returned to the caller upon successful, offline MANIFEST iteration in `GetFileChecksumsFromCurrentManifest`.

unreleased_history/bug_fixes/multi-scan-dict-compression.md

@@ -1 +0,0 @@
-* Fix a bug in MultiScan that causes it to fall back to a normal scan when dictionary compression is enabled.

unreleased_history/bug_fixes/multi-scan.md

@@ -1 +0,0 @@
-* Fix a bug in MultiScan where incorrect results can be returned when a Scan's range is across multiple files.

unreleased_history/bug_fixes/multiscan_fill_cache.md

@@ -1 +0,0 @@
-Fix a crash in iterator Prepare() when fill_cache=false

unreleased_history/bug_fixes/remote_compaction_empty_result.md

@@ -1 +0,0 @@
-Fixed a bug in remote compaction that may mistakenly delete live SST file(s) during the cleanup phase when no keys survive the compaction (all expired)

unreleased_history/bug_fixes/udi_config.md

@@ -1 +0,0 @@
-Allow a user defined index to be configured from a string.

unreleased_history/bug_fixes/udi_index_key_format.md

@@ -1 +0,0 @@
-Make the User Defined Index interface consistently use the user key format, fixing the previous mixed usage of internal and user key.

unreleased_history/new_features/cf-ingest-behind.md

@@ -1 +0,0 @@
-* 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.

unreleased_history/new_features/multiscan-io-coalesce.md

@@ -1,2 +0,0 @@
-* Introduce `MultiScanArgs::io_coalesce_threshold` to allow a configurable IO coalescing threshold.
-

unreleased_history/performance_improvements/compression_perf.md

@@ -1 +0,0 @@
-* Small improvement to CPU efficiency of compression using built-in algorithms, and a dramatic efficiency improvement for LZ4HC, based on reusing data structures between invocations.

unreleased_history/public_api_changes/db-gen-file-ingestion.md

@@ -1 +0,0 @@
-* `IngestExternalFileOptions::allow_db_generated_files` now allows files ingestion of any DB generated SST file, instead of only the ones with all keys having sequence number 0.

unreleased_history/public_api_changes/decouple.md

@@ -1 +0,0 @@
-* `decouple_partitioned_filters = true` is now the default in BlockBasedTableOptions.

unreleased_history/public_api_changes/get_ttl_in_ttl_db.md

@@ -1 +0,0 @@
-GetTtl() API is now available in TTL DB

unreleased_history/public_api_changes/lz4_etc.md

@@ -1,2 +0,0 @@
-* Minimum supported version of LZ4 library is now 1.7.0 (r129 from 2015)
-* Some changes to experimental Compressor and CompressionManager APIs

unreleased_history/public_api_changes/new_SyncFile_api_at_FileSystem_interface

@@ -1 +0,0 @@
-A new Filesystem::SyncFile function is added for syncing a file that was already written, such as on file ingestion. The default implementation matches previous RocksDB behavior: re-open the file for read-write, sync it, and close it. We recommend overriding for FileSystems that do not require syncing for crash recovery or do not handle (well) re-opening for writes.

util/status.cc

@@ -46,6 +46,7 @@ static const char* msgs[static_cast<int>(Status::kMaxSubCode)] = {
     "IO fenced off",          // kIOFenced
     "Merge operator failed",  // kMergeOperatorFailed
     "Number of operands merged exceeded threshold",  // kMergeOperandThresholdExceeded
+    "MultiScan reached file prefetch limit",         // kMultiScanPrefetchLimit
 };
 
 Status::Status(Code _code, SubCode _subcode, const Slice& msg,

utilities/fault_injection_env.cc

@@ -464,17 +464,6 @@ Status FaultInjectionTestEnv::LinkFile(const std::string& s,
   return ret;
 }
 
-Status FaultInjectionTestEnv::SyncFile(const std::string& fname,
-                                       const EnvOptions& env_options,
-                                       bool use_fsync) {
-  // Call the default implement of SyncFile API in Env, so that it would call
-  // other FileSystem API at FaultInjectionTestEnv layer for failure injection.
-  // Otherwise, the default behavior is WrapperEnv::SyncFile, which forward the
-  // call to the underlying FileSystem, instead of the ones in
-  // FaultInjectionTestEnv.
-  return Env::SyncFile(fname, env_options, use_fsync);
-}
-
 void FaultInjectionTestEnv::WritableFileClosed(const FileState& state) {
   MutexLock l(&mutex_);
   if (open_managed_files_.find(state.filename_) != open_managed_files_.end()) {

utilities/fault_injection_env.h

@@ -177,9 +177,6 @@ class FaultInjectionTestEnv : public EnvWrapper {
 
   Status LinkFile(const std::string& s, const std::string& t) override;
 
-  Status SyncFile(const std::string& fname, const EnvOptions& env_options,
-                  bool use_fsync) override;
-
 // Undef to eliminate clash on Windows
 #undef GetFreeSpace
   Status GetFreeSpace(const std::string& path, uint64_t* disk_free) override {

utilities/fault_injection_fs.cc

@@ -1200,17 +1200,6 @@ IOStatus FaultInjectionTestFS::LinkFile(const std::string& s,
   }
   return io_s;
 }
-IOStatus FaultInjectionTestFS::SyncFile(const std::string& fname,
-                                        const FileOptions& file_options,
-                                        const IOOptions& io_options,
-                                        bool use_fsync, IODebugContext* dbg) {
-  // Call the default implement of SyncFile API in FileSystem, so that it would
-  // call other FileSystem API at FaultInjectionTestFS layer for failure
-  // injection. Otherwise, the default behavior is calling target()->SyncFile,
-  // which forward the call to the underlying FileSystem, instead of the ones in
-  // FaultInjectionTestFS.
-  return FileSystem::SyncFile(fname, file_options, io_options, use_fsync, dbg);
-}
 
 IOStatus FaultInjectionTestFS::NumFileLinks(const std::string& fname,
                                             const IOOptions& options,

utilities/fault_injection_fs.h

@@ -301,10 +301,6 @@ class FaultInjectionTestFS : public FileSystemWrapper {
   IOStatus LinkFile(const std::string& src, const std::string& target,
                     const IOOptions& options, IODebugContext* dbg) override;
 
-  IOStatus SyncFile(const std::string& fname, const FileOptions& file_options,
-                    const IOOptions& io_options, bool use_fsync,
-                    IODebugContext* dbg) override;
-
   IOStatus NumFileLinks(const std::string& fname, const IOOptions& options,
                         uint64_t* count, IODebugContext* dbg) override;