Support all operation types in User Defined Index (UDI) interface (#14399)

Summary:
Remove the restriction that limited UDI to ingest-only, Puts-only use cases. This enables UDI plugins (including the trie index from https://github.com/facebook/rocksdb/issues/14310) to work with all operation types: Put, Delete, Merge, SingleDelete, PutEntity, etc.

Part of https://github.com/facebook/rocksdb/issues/12396

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

Reviewed By: pdillinger

Differential Revision: D96392636

Pulled By: xingbowang

fbshipit-source-id: 0f1e6c38531fa72539a0e2c6a3dffff333392b4c

BUCK

@@ -5628,6 +5628,12 @@ cpp_unittest_wrapper(name="transaction_test",
             extra_compiler_flags=[])
 
 
+cpp_unittest_wrapper(name="trie_index_db_test",
+            srcs=["utilities/trie_index/trie_index_db_test.cc"],
+            deps=[":rocksdb_test_lib"],
+            extra_compiler_flags=[])
+
+
 cpp_unittest_wrapper(name="trie_index_test",
             srcs=["utilities/trie_index/trie_index_test.cc"],
             deps=[":rocksdb_test_lib"],

CMakeLists.txt

@@ -1557,6 +1557,7 @@ if(WITH_TESTS)
         utilities/transactions/write_unprepared_transaction_test.cc
         utilities/transactions/lock/range/range_locking_test.cc
         utilities/transactions/timestamped_snapshot_test.cc
+        utilities/trie_index/trie_index_db_test.cc
         utilities/trie_index/trie_index_test.cc
         utilities/ttl/ttl_test.cc
         utilities/types_util_test.cc

Makefile

@@ -1608,6 +1608,9 @@ object_registry_test: $(OBJ_DIR)/utilities/object_registry_test.o $(TEST_LIBRARY
 ttl_test: $(OBJ_DIR)/utilities/ttl/ttl_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 
+trie_index_db_test: $(OBJ_DIR)/utilities/trie_index/trie_index_db_test.o $(TEST_LIBRARY) $(LIBRARY)
+	$(AM_LINK)
+
 trie_index_test: $(OBJ_DIR)/utilities/trie_index/trie_index_test.o $(TEST_LIBRARY) $(LIBRARY)
 	$(AM_LINK)
 

db_stress_tool/db_stress_gflags.cc

@@ -627,11 +627,10 @@ DEFINE_double(uniform_cv_threshold,
               "CV threshold for marking index blocks as uniform. Set to -1 to "
               "disable. (see `uniform_cv_threshold` in table.h)");
 
-DEFINE_bool(
-    use_trie_index, false,
-    "Use trie-based user defined index (UDI) for SST files. "
-    "Only compatible with Put operations (no Merge/Delete/SingleDelete). "
-    "When enabled, incompatible flags are automatically adjusted.");
+DEFINE_bool(use_trie_index, false,
+            "Use trie-based user defined index (UDI) for SST files. "
+            "Compatible with all operation types (Put, Delete, Merge, etc.). "
+            "Backward scan is disabled when this is enabled.");
 
 DEFINE_bool(test_backward_scan, true,
             "Test backward iteration (Prev, SeekForPrev) in stress tests. "

db_stress_tool/db_stress_test_base.cc

@@ -1999,13 +1999,17 @@ Status StressTest::TestIterateImpl(ThreadState* thread,
 
     Slice key(key_str);
 
-    // UserDefinedIndexIterator only supports Seek(target) + Next() - it
-    // requires a target key for seeks. SeekToFirst/SeekToLast have no target
-    // key, and SeekForPrev/Prev are not supported. Check if UDI is being used
-    // either via ReadOptions or CF-level configuration.
+    // UserDefinedIndexIterator supports Seek(target), Next(), and
+    // SeekToFirst(). However, SeekToLast, SeekForPrev, and Prev are not
+    // supported. Check if UDI is being used either via ReadOptions or
+    // CF-level configuration.
     const bool using_udi =
         (ro.table_index_factory != nullptr) || (udi_factory_ != nullptr);
-    const bool support_seek_first_or_last =
+    // SeekToFirst is supported by UDI, so only total_order is required.
+    const bool support_seek_to_first =
+        expect_total_order && (FLAGS_test_backward_scan || using_udi);
+    // SeekToLast requires backward scan support which UDI does not provide.
+    const bool support_seek_to_last =
         expect_total_order && FLAGS_test_backward_scan && !using_udi;
     const bool support_seek_for_prev = FLAGS_test_backward_scan && !using_udi;
 
@@ -2022,12 +2026,12 @@ Status StressTest::TestIterateImpl(ThreadState* thread,
     }
 
     LastIterateOp last_op;
-    if (support_seek_first_or_last && thread->rand.OneIn(100)) {
+    if (support_seek_to_first && thread->rand.OneIn(100)) {
       iter->SeekToFirst();
       cmp_iter->SeekToFirst();
       last_op = kLastOpSeekToFirst;
       op_logs += "STF ";
-    } else if (support_seek_first_or_last && thread->rand.OneIn(100)) {
+    } else if (support_seek_to_last && thread->rand.OneIn(100)) {
       iter->SeekToLast();
       cmp_iter->SeekToLast();
       last_op = kLastOpSeekToLast;

db_stress_tool/db_stress_tool.cc

@@ -229,65 +229,20 @@ int db_stress_tool(int argc, char** argv) {
     FLAGS_atomic_flush = true;
   }
 
-  // Trie UDI only supports Seek + Next. Disable backward scan testing so
-  // that other features lacking backward scan support can reuse this flag.
+  // Trie UDI supports Seek, Next, and SeekToFirst, but not SeekToLast,
+  // SeekForPrev, or Prev. Disable backward scan testing.
   if (FLAGS_use_trie_index) {
     FLAGS_test_backward_scan = false;
   }
 
-  // Trie UDI only supports kTypeValue (Put) entries. Reject incompatible
-  // operations that would produce non-Put types during flush/compaction.
-  if (FLAGS_use_trie_index) {
-    if (FLAGS_delpercent > 0) {
-      fprintf(stderr,
-              "Error: use_trie_index is incompatible with delpercent > 0\n");
-      exit(1);
-    }
-    if (FLAGS_delrangepercent > 0) {
-      fprintf(stderr,
-              "Error: use_trie_index is incompatible with "
-              "delrangepercent > 0\n");
-      exit(1);
-    }
-    if (FLAGS_use_merge) {
-      fprintf(stderr, "Error: use_trie_index is incompatible with use_merge\n");
-      exit(1);
-    }
-    if (FLAGS_use_put_entity_one_in > 0) {
-      fprintf(stderr,
-              "Error: use_trie_index is incompatible with "
-              "use_put_entity_one_in > 0\n");
-      exit(1);
-    }
-    if (FLAGS_use_timed_put_one_in > 0) {
-      fprintf(stderr,
-              "Error: use_trie_index is incompatible with "
-              "use_timed_put_one_in > 0\n");
-      exit(1);
-    }
-    if (FLAGS_use_txn) {
-      fprintf(stderr,
-              "Error: use_trie_index is incompatible with use_txn. "
-              "TransactionDB rollback is executed as delete operation during "
-              "crash recovery, which are non-Put types, not supported by "
-              "user-defined index.\n");
-      exit(1);
-    }
-    if (FLAGS_mmap_read) {
-      fprintf(stderr,
-              "Error: use_trie_index is incompatible with mmap_read. "
-              "The trie index uses zero-copy pointers into block data "
-              "which is unsafe with mmap'd reads.\n");
-      exit(1);
-    }
-    if (FLAGS_enable_blob_files ||
-        FLAGS_allow_setting_blob_options_dynamically) {
-      fprintf(stderr,
-              "Error: use_trie_index is incompatible with BlobDB. "
-              "BlobDB writes kTypeBlobIndex entries in SSTs which are "
-              "non-Put types, not supported by user-defined index.\n");
-      exit(1);
-    }
+  // Trie UDI uses zero-copy pointers into block data, which is
+  // incompatible with mmap_read.
+  if (FLAGS_use_trie_index && FLAGS_mmap_read) {
+    fprintf(stderr,
+            "Error: use_trie_index is incompatible with mmap_read. "
+            "The trie index uses zero-copy pointers into block data "
+            "which is unsafe with mmap'd reads.\n");
+    exit(1);
   }
 
   if (FLAGS_read_only) {

include/rocksdb/options.h

@@ -2275,9 +2275,10 @@ struct ReadOptions {
   // block based table index. The table_factory used for the column family
   // must support building/reading this index.
   //
-  // Currently, only forward scans are supported. For forward scans, only Seek()
-  // is supported. SeekToFirst() is not supported. If the caller wishes to scan
-  // from start to end, the native index must be used.
+  // Forward scans (SeekToFirst, Seek, Next) and point lookups (Get) are
+  // supported. Reverse operations (SeekToLast, SeekForPrev, Prev) are not
+  // yet supported and will return NotSupported when this is set. Leave this
+  // null to use the native index for reverse operations.
   const UserDefinedIndexFactory* table_index_factory = nullptr;
 
   // *** END options only relevant to iterators or scans ***

include/rocksdb/user_defined_index.h

@@ -21,26 +21,29 @@
 namespace ROCKSDB_NAMESPACE {
 
 // Prefix for user-defined index block names
-inline const std::string kUserDefinedIndexPrefix =
+inline constexpr const char* kUserDefinedIndexPrefix =
     "rocksdb.user_defined_index.";
 
 // This is a public API for user-defined index builders.
 // It allows users to define their own index format and build custom
 // indexes during table building. Currently, only a monolithic index
 // block is supported (no partitioned index).
-//
-// This is currently supported only for a restricted set of use cases. The
-// CF must be ingest only, and only files containing Puts generated by
-// SstFileWriter are supported.
 
 // The interface for building user-defined index.
 class UserDefinedIndexBuilder {
  public:
-  // Right now, we only support Puts. In the future, we may support merges,
-  // deletions etc.
-  enum ValueType {
-    kValue,
-    kTypeMax,
+  // Indicates the type of key-value entry being added via OnKeyAdded().
+  // UDI builders that only use AddIndexEntry() (e.g., trie-based indexes)
+  // can safely ignore this.
+  enum ValueType : uint8_t {
+    kValue = 0,   // Put: the value is the full user value.
+    kDelete = 1,  // Deletion (Delete, SingleDelete, or DeleteWithTimestamp):
+                  // the value is typically empty.
+    kMerge = 2,   // Merge operand: the value is a partial update.
+    kOther = 3,   // Other types (e.g., blob reference, wide-column entity).
+                  // The value format is type-specific and may not be the
+                  // actual user data.
+    kTypeMax,     // Sentinel — must be last. Value may change across releases.
   };
 
   // File offset and size of the data block
@@ -92,13 +95,25 @@ class UserDefinedIndexBuilder {
                               std::string* separator_scratch,
                               const IndexEntryContext& context) = 0;
 
-  // This method will be called whenever a key is added. The subclasses may
-  // override OnKeyAdded() if they need to collect additional information.
-  // The type argument indicates whether the value is a full value or partial.
-  // At the moment, only full values are supported.
+  // Called for every key-value pair added to the SST file. UDI builders may
+  // override this to collect per-key information (e.g., for secondary
+  // indexes). Builders that only use separator keys from AddIndexEntry()
+  // (e.g., trie-based indexes) can leave this as a no-op.
   //
-  // The key will be a user key. RocksDB guarantees that there will only be
-  // one entry for each key in the file/index.
+  // @key:   The user key (without sequence number or type suffix).
+  // @type:  The entry type — kValue (Put), kDelete, kMerge, or kOther.
+  //         For kDelete entries, the value may be empty. For kOther, the
+  //         value format is type-specific and may not be actual user data.
+  // @value: The associated value (may be empty for deletions).
+  //
+  // NOTE: In SST files produced by flush or compaction, there may be multiple
+  // entries for the same user key with different sequence numbers (e.g., when
+  // snapshots are active). UDI builders that use OnKeyAdded() should be
+  // prepared for this.
+  //
+  // Thread safety: For a given builder instance, OnKeyAdded() and
+  // AddIndexEntry() are always called from a single thread. Builders do
+  // not need internal synchronization.
   virtual void OnKeyAdded(const Slice& /*key*/, ValueType /*type*/,
                           const Slice& /*value*/) {}
 
@@ -129,6 +144,17 @@ class UserDefinedIndexIterator {
     SequenceNumber target_seq = 0;
   };
 
+  // Position the index iterator at the very first index entry. The result
+  // must be populated the same way as SeekAndGetResult.
+  //
+  // The default implementation calls SeekAndGetResult with an empty key,
+  // which works for BytewiseComparator (empty string is the smallest key).
+  // Implementations should override this if they can reach the first entry
+  // more efficiently or if they use a comparator where empty is not smallest.
+  virtual Status SeekToFirstAndGetResult(IterateResult* result) {
+    return SeekAndGetResult(Slice(), result, SeekContext{});
+  }
+
   // Given the target key, position the index iterator at the index entry
   // for the data block that may contain the target.
   //
@@ -186,7 +212,7 @@ struct UserDefinedIndexOption {
 // Factory for creating user-defined index builders.
 class UserDefinedIndexFactory : public Customizable {
  public:
-  virtual ~UserDefinedIndexFactory() = default;
+  ~UserDefinedIndexFactory() override = default;
 
   static const char* Type() { return "UserDefinedIndexFactory"; }
 

src.mk

@@ -671,6 +671,7 @@ TEST_MAIN_SOURCES =                                                     \
   utilities/transactions/write_committed_transaction_ts_test.cc         \
   utilities/transactions/timestamped_snapshot_test.cc                   \
   utilities/ttl/ttl_test.cc                                             \
+  utilities/trie_index/trie_index_db_test.cc                            \
   utilities/trie_index/trie_index_test.cc                               \
   utilities/types_util_test.cc                                          \
   utilities/util_merge_operators_test.cc                                \

table/block_based/user_defined_index_wrapper.h

@@ -9,6 +9,7 @@
 #include <string>
 #include <unordered_map>
 
+#include "db/seqno_to_time_mapping.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/status.h"
 #include "rocksdb/user_defined_index.h"
@@ -43,7 +44,7 @@ class UserDefinedIndexBuilderWrapper : public IndexBuilder {
                       const BlockHandle& block_handle,
                       std::string* separator_scratch,
                       bool skip_delta_encoding) override {
-    UserDefinedIndexBuilder::BlockHandle handle;
+    UserDefinedIndexBuilder::BlockHandle handle{};
     handle.offset = block_handle.offset();
     handle.size = block_handle.size();
     // Forward the call to both index builders.
@@ -80,7 +81,11 @@ class UserDefinedIndexBuilderWrapper : public IndexBuilder {
         separator_scratch, skip_delta_encoding);
   }
 
-  // Not supported with parallel compression
+  // Parallel compression splits AddIndexEntry() into PrepareIndexEntry() (emit
+  // thread) and FinishIndexEntry() (worker thread). This wrapper does not
+  // implement that split yet, so parallel compression is rejected at option
+  // validation time (see BlockBasedTableFactory::ValidateOptions and the Rep
+  // constructor). These stubs exist only to satisfy the interface.
   std::unique_ptr<PreparedIndexEntry> CreatePreparedIndexEntry() override {
     return nullptr;
   }
@@ -111,39 +116,35 @@ class UserDefinedIndexBuilderWrapper : public IndexBuilder {
 
     ParsedInternalKey pkey;
     if (status_.ok()) {
+      // Defensive: value should always be present since OnKeyAdded() is called
+      // on the main thread in Add() with the original value Slice. No current
+      // code path passes std::nullopt here.
       if (!value.has_value()) {
+        assert(false);
         status_ = Status::InvalidArgument(
-            "user_defined_index_factory not supported with parallel "
-            "compression");
+            "OnKeyAdded called without a value; UDI requires the value to "
+            "forward to the plugin builder");
       } else {
         status_ = ParseInternalKey(key, &pkey, /*log_err_key*/ false);
-        // UDI only supports kTypeValue (Put) entries. Non-Put types include:
-        // - kTypeDeletion, kTypeSingleDeletion, kTypeRangeDeletion (deletes)
-        // - kTypeMerge (merge operands)
-        // - kTypeWideColumnEntity (PutEntity)
-        // - kTypeBlobIndex (BlobDB stores values in blob files)
-        // This makes UDI incompatible with:
-        // - Delete/Merge/SingleDelete/DeleteRange operations
-        // - TransactionDB (ROLLBACK writes DELETE entries to undo changes)
-        // - BlobDB (writes kTypeBlobIndex entries during flush)
-        // See T257683723 for analysis of TransactionDB incompatibility.
-        // See T258398372 for analysis of BlobDB incompatibility.
-        if (status_.ok() && pkey.type != ValueType::kTypeValue) {
-          status_ = Status::InvalidArgument(
-              "user_defined_index_factory only supported with Puts");
-        }
       }
     }
     if (!status_.ok()) {
       return;
     }
 
-    // Pass the user key to the UDI. UDI is designed for ingest-only use cases
-    // where files contain only Put entries with unique keys. We don't expect
-    // multiple entries with different sequence numbers for the same key.
+    // Pass the user key to the UDI with the mapped value type. In SST files
+    // produced by flush or compaction, there may be multiple entries for the
+    // same user key with different sequence numbers (e.g., when snapshots are
+    // active). UDI builders that use OnKeyAdded() should handle this; builders
+    // that only use AddIndexEntry() separator keys (e.g., trie) are unaffected.
+    Slice udi_value = value.value();
+    if (pkey.type == kTypeValuePreferredSeqno) {
+      // Strip the packed preferred seqno suffix so the UDI plugin receives
+      // only the user value, consistent with the kValue contract.
+      udi_value = ParsePackedValueForValue(udi_value);
+    }
     user_defined_index_builder_->OnKeyAdded(
-        pkey.user_key, UserDefinedIndexBuilder::ValueType::kValue,
-        value.value());
+        pkey.user_key, MapToUDIValueType(pkey.type), udi_value);
   }
 
   Status Finish(IndexBlocks* index_blocks,
@@ -189,6 +190,30 @@ class UserDefinedIndexBuilderWrapper : public IndexBuilder {
   }
 
  private:
+  static UserDefinedIndexBuilder::ValueType MapToUDIValueType(
+      ROCKSDB_NAMESPACE::ValueType t) {
+    switch (t) {
+      case kTypeValue:
+      case kTypeValuePreferredSeqno:
+        return UserDefinedIndexBuilder::kValue;
+      case kTypeDeletion:
+      case kTypeSingleDeletion:
+      case kTypeDeletionWithTimestamp:
+        return UserDefinedIndexBuilder::kDelete;
+      case kTypeMerge:
+        return UserDefinedIndexBuilder::kMerge;
+      case kTypeBlobIndex:
+      case kTypeWideColumnEntity:
+        return UserDefinedIndexBuilder::kOther;
+      default:
+        // Any new type that reaches OnKeyAdded() should be explicitly mapped
+        // above. Falling through to kOther is a safe default but indicates a
+        // missing case that should be added.
+        assert(false);
+        return UserDefinedIndexBuilder::kOther;
+    }
+  }
+
   const std::string name_;
   std::unique_ptr<IndexBuilder> internal_index_builder_;
   std::unique_ptr<UserDefinedIndexBuilder> user_defined_index_builder_;
@@ -208,8 +233,15 @@ class UserDefinedIndexIteratorWrapper
   bool Valid() const override { return valid_; }
 
   void SeekToFirst() override {
-    valid_ = false;
-    status_ = Status::NotSupported("SeekToFirst not supported");
+    status_ = udi_iter_->SeekToFirstAndGetResult(&result_);
+    if (status_.ok()) {
+      valid_ = result_.bound_check_result == IterBoundCheck::kInbound;
+      if (valid_) {
+        SetInternalKeyFromUDIResult();
+      }
+    } else {
+      valid_ = false;
+    }
   }
 
   void SeekToLast() override {
@@ -233,12 +265,7 @@ class UserDefinedIndexIteratorWrapper
     if (status_.ok()) {
       valid_ = result_.bound_check_result == IterBoundCheck::kInbound;
       if (valid_) {
-        // Use seq=0 for the internal key because this is a separator key
-        // (upper bound on block contents), not a real data key. seq=0 makes
-        // the key compare as "greater" in internal key order (since lower
-        // seqno = greater internal key for the same user key), which is the
-        // correct behavior for a separator used as an index entry.
-        ikey_.Set(result_.key, 0, ValueType::kTypeValue);
+        SetInternalKeyFromUDIResult();
       }
     } else {
       valid_ = false;
@@ -250,7 +277,7 @@ class UserDefinedIndexIteratorWrapper
     if (status_.ok()) {
       valid_ = result_.bound_check_result == IterBoundCheck::kInbound;
       if (valid_) {
-        ikey_.Set(result_.key, 0, ValueType::kTypeValue);
+        SetInternalKeyFromUDIResult();
       }
     } else {
       valid_ = false;
@@ -262,9 +289,11 @@ class UserDefinedIndexIteratorWrapper
     if (status_.ok()) {
       valid_ = result_.bound_check_result == IterBoundCheck::kInbound;
       if (valid_) {
-        ikey_.Set(result_.key, 0, ValueType::kTypeValue);
+        SetInternalKeyFromUDIResult();
+        result->key = key();
       }
-      *result = result_;
+      result->bound_check_result = result_.bound_check_result;
+      result->value_prepared = result_.value_prepared;
     } else {
       valid_ = false;
     }
@@ -303,6 +332,17 @@ class UserDefinedIndexIteratorWrapper
   }
 
  private:
+  // Convert the UDI result's user key into an internal key for the index
+  // iterator contract. UDI separators are user keys, but
+  // InternalIteratorBase<IndexValue> must expose internal keys (user key +
+  // 8-byte trailer). We use seq=0 / kTypeValue so that the resulting
+  // internal key compares as "greater than or equal to" any real data key
+  // with the same user key (lower seqno = later in internal key order),
+  // which is the correct upper-bound semantics for an index separator.
+  void SetInternalKeyFromUDIResult() {
+    ikey_.Set(result_.key, 0, ValueType::kTypeValue);
+  }
+
   std::unique_ptr<UserDefinedIndexIterator> udi_iter_;
   IterateResult result_;
   InternalKey ikey_;
@@ -320,7 +360,7 @@ class UserDefinedIndexReaderWrapper : public BlockBasedTable::IndexReader {
         reader_(std::move(reader)),
         udi_reader_(std::move(udi_reader)) {}
 
-  virtual InternalIteratorBase<IndexValue>* NewIterator(
+  InternalIteratorBase<IndexValue>* NewIterator(
       const ReadOptions& read_options, bool disable_prefix_seek,
       IndexBlockIter* iter, GetContext* get_context,
       BlockCacheLookupContext* lookup_context) override {
@@ -337,17 +377,14 @@ class UserDefinedIndexReaderWrapper : public BlockBasedTable::IndexReader {
     std::unique_ptr<UserDefinedIndexIterator> udi_iter =
         udi_reader_->NewIterator(read_options);
     if (udi_iter) {
-      InternalIteratorBase<IndexValue>* wrap_iter =
-          new UserDefinedIndexIteratorWrapper(std::move(udi_iter));
-      return wrap_iter;
+      return new UserDefinedIndexIteratorWrapper(std::move(udi_iter));
     }
     return NewErrorInternalIterator<IndexValue>(
         Status::NotFound("Could not create UDI iterator"));
   }
 
-  virtual Status CacheDependencies(
-      const ReadOptions& ro, bool pin,
-      FilePrefetchBuffer* tail_prefetch_buffer) override {
+  Status CacheDependencies(const ReadOptions& ro, bool pin,
+                           FilePrefetchBuffer* tail_prefetch_buffer) override {
     return reader_->CacheDependencies(ro, pin, tail_prefetch_buffer);
   }
 
@@ -356,7 +393,7 @@ class UserDefinedIndexReaderWrapper : public BlockBasedTable::IndexReader {
            udi_reader_->ApproximateMemoryUsage();
   }
 
-  virtual void EraseFromCacheBeforeDestruction(
+  void EraseFromCacheBeforeDestruction(
       uint32_t uncache_aggressiveness) override {
     reader_->EraseFromCacheBeforeDestruction(uncache_aggressiveness);
   }

table/table_test.cc

@@ -7617,10 +7617,23 @@ class UserDefinedIndexTestBase : public BlockBasedTableTestBase {
     Status NewBuilder(
         const UserDefinedIndexOption& /*option*/,
         std::unique_ptr<UserDefinedIndexBuilder>& builder) const override {
-      builder = std::make_unique<TestUserDefinedIndexBuilder>();
+      auto b = std::make_unique<TestUserDefinedIndexBuilder>();
+      b->skip_key_size_check_ = skip_key_size_check_;
+      // Share the factory's key_type_log so tests can inspect after flush.
+      b->shared_key_type_log_ = &key_type_log_;
+      builder = std::move(b);
       return Status::OK();
     }
 
+    // When true, builders skip key-size assertions (for variable-length keys).
+    bool skip_key_size_check_ = false;
+
+    // Accumulated log of (key, ValueType) pairs from all builders created
+    // by this factory. Tests can inspect this after flush/compaction.
+    mutable std::vector<
+        std::pair<std::string, UserDefinedIndexBuilder::ValueType>>
+        key_type_log_;
+
     struct CustomizedMapComparator {
       CustomizedMapComparator(const Comparator* _comparator)
           : comparator(_comparator) {}
@@ -7662,9 +7675,11 @@ class UserDefinedIndexTestBase : public BlockBasedTableTestBase {
         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);
+        if (!skip_key_size_check_) {
+          EXPECT_EQ(last_key_in_current_block.size(), 5);
+          if (first_key_in_next_block) {
+            EXPECT_EQ(first_key_in_next_block->size(), 5);
+          }
         }
         // Unused parameters
         (void)separator_scratch;
@@ -7681,18 +7696,27 @@ class UserDefinedIndexTestBase : public BlockBasedTableTestBase {
         return last_key_in_current_block;
       }
 
-      void OnKeyAdded(const Slice& key, ValueType /*value*/,
+      void OnKeyAdded(const Slice& key, ValueType type,
                       const Slice& /*value*/) override {
         if (key.starts_with("dummy")) {
           return;
         }
-        EXPECT_EQ(key.size(), 5);
-        // Track keys added to the index
+        if (!skip_key_size_check_) {
+          EXPECT_EQ(key.size(), 5);
+        }
+        // Record the ValueType for each key so tests can verify the mapping.
+        if (shared_key_type_log_) {
+          shared_key_type_log_->emplace_back(key.ToString(), type);
+        }
+        // Track keys added to the current block (used by AddIndexEntry).
         keys_added_++;
-        // Add dummy entry
-        PutFixed64(&index_data_[key.ToString()], 0);
-        PutFixed64(&index_data_[key.ToString()], 0);
-        PutFixed32(&index_data_[key.ToString()], 0);
+        if (!skip_key_size_check_) {
+          // For fixed-size key tests, add a dummy per-key entry that the
+          // TestUserDefinedIndexReader can parse alongside block-level entries.
+          PutFixed64(&index_data_[key.ToString()], 0);
+          PutFixed64(&index_data_[key.ToString()], 0);
+          PutFixed32(&index_data_[key.ToString()], 0);
+        }
       }
 
       Status Finish(Slice* index_contents) override {
@@ -7713,6 +7737,14 @@ class UserDefinedIndexTestBase : public BlockBasedTableTestBase {
 
       int GetEntriesAdded() const { return entries_added_; }
 
+      // When true, skip the EXPECT_EQ(key.size(), 5) checks, allowing
+      // variable-length keys (e.g., from DB flush/compaction).
+      bool skip_key_size_check_ = false;
+
+      // Points to the factory's shared log vector. Set by the factory.
+      mutable std::vector<std::pair<std::string, ValueType>>*
+          shared_key_type_log_ = nullptr;
+
      private:
       int entries_added_;
       std::map<std::string, std::string> index_data_;
@@ -7799,11 +7831,16 @@ class UserDefinedIndexTestBase : public BlockBasedTableTestBase {
           iter_ = index_.lower_bound(key.ToString());
           if ((iter_ != index_.end()) && IsInbound()) {
             AdvanceToNextIndexEntry();
-            result->bound_check_result = IterBoundCheck::kInbound;
-            result->key = Slice(iter_->first);
-            if (scan_opts_ && target_num_keys_ > 0 &&
-                comparator_->Compare(key, iter_->first) == 0) {
-              target_num_keys_--;
+            if (iter_ != index_.end()) {
+              result->bound_check_result = IterBoundCheck::kInbound;
+              result->key = Slice(iter_->first);
+              if (scan_opts_ && target_num_keys_ > 0 &&
+                  comparator_->Compare(key, iter_->first) == 0) {
+                target_num_keys_--;
+              }
+            } else {
+              result->bound_check_result = IterBoundCheck::kUnknown;
+              result->key = Slice();
             }
           } else {
             result->bound_check_result = IterBoundCheck::kOutOfBound;
@@ -7838,10 +7875,16 @@ class UserDefinedIndexTestBase : public BlockBasedTableTestBase {
           iter_++;
           if ((iter_ != index_.end()) && IsInbound()) {
             AdvanceToNextIndexEntry();
-            result->bound_check_result = IterBoundCheck::kInbound;
-            result->key = Slice(iter_->first);
-            target_num_keys_ -=
-                std::min(target_num_keys_, iter_->second.second);
+            if (iter_ != index_.end()) {
+              result->bound_check_result = IterBoundCheck::kInbound;
+              result->key = Slice(iter_->first);
+              target_num_keys_ -=
+                  std::min(target_num_keys_, iter_->second.second);
+            } else {
+              // AdvanceToNextIndexEntry reached end of map.
+              result->bound_check_result = IterBoundCheck::kUnknown;
+              result->key = Slice();
+            }
           } else {
             // EOF
             result->bound_check_result = IterBoundCheck::kUnknown;
@@ -7851,7 +7894,7 @@ class UserDefinedIndexTestBase : public BlockBasedTableTestBase {
         }
 
         void AdvanceToNextIndexEntry() {
-          while (iter_->second.second == 0) {
+          while (iter_ != index_.end() && iter_->second.second == 0) {
             iter_++;
           }
         }
@@ -8071,7 +8114,8 @@ void UserDefinedIndexTestBase::BasicTest(bool use_partitioned_index) {
       /* unique_id */ {}, /* largest_seqno */ 0,
       /* tail_size */ 0, ioptions.persist_user_defined_timestamps);
   // Verify that the user-defined index was created
-  std::string meta_block_name = kUserDefinedIndexPrefix + "test_index";
+  std::string meta_block_name =
+      std::string(kUserDefinedIndexPrefix) + "test_index";
   BlockHandle block_handle;
   uint64_t file_size = 0;
   std::unique_ptr<FSRandomAccessFile> file;
@@ -8220,31 +8264,273 @@ TEST_P(UserDefinedIndexTest, InvalidArgumentTest1) {
   writer.reset();
 }
 
-TEST_P(UserDefinedIndexTest, InvalidArgumentTest2) {
+TEST_P(UserDefinedIndexTest, MergeWithUDI) {
+  // Verify that Merge operations work correctly with user-defined index.
   BlockBasedTableOptions table_options;
   std::string dbname = test::PerThreadDBPath("user_defined_index_test");
   std::string ingest_file = dbname + "test.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));
+  options_.merge_operator = MergeOperators::CreateStringAppendOperator();
 
   std::unique_ptr<SstFileWriter> writer;
   writer.reset(new SstFileWriter(EnvOptions(), options_));
   ASSERT_OK(writer->Open(ingest_file));
 
-  std::string key = "foo";
-  std::string value = "bar";
-  ASSERT_OK(writer->Merge(key, value));
-  ASSERT_EQ(writer->Finish(), Status::InvalidArgument());
+  // Use 5-byte keys to match TestUserDefinedIndexBuilder expectations.
+  ASSERT_OK(writer->Merge("key_a", "val_a"));
+  ASSERT_OK(writer->Finish());
   writer.reset();
+
+  // Read back and verify the merge entry is present in the SST.
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(ingest_file));
+  ReadOptions ro;
+  std::unique_ptr<Iterator> iter(reader.NewIterator(ro));
+  iter->SeekToFirst();
+  ASSERT_TRUE(iter->Valid());
+  ASSERT_EQ(iter->key().ToString(), "key_a");
+  ASSERT_EQ(iter->value().ToString(), "val_a");
+  iter->Next();
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_OK(iter->status());
+}
+
+TEST_P(UserDefinedIndexTest, DBFlushWithMixedOpsAndUDI) {
+  // Verify that Put, Delete, Merge, and SingleDelete all flow correctly
+  // through the UDI builder when flushed via DB::Flush.
+  std::string dbname = test::PerThreadDBPath("udi_db_flush_test");
+  ASSERT_OK(DestroyDB(dbname, options_));
+
+  BlockBasedTableOptions table_options;
+  auto user_defined_index_factory =
+      std::make_shared<TestUserDefinedIndexFactory>();
+  user_defined_index_factory->skip_key_size_check_ = true;
+  table_options.user_defined_index_factory = user_defined_index_factory;
+  table_options.flush_block_policy_factory =
+      std::make_shared<CustomFlushBlockPolicyFactory>();
+  options_.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  options_.merge_operator = MergeOperators::CreateStringAppendOperator();
+  options_.create_if_missing = true;
+
+  std::unique_ptr<DB> db;
+  ASSERT_OK(DB::Open(options_, dbname, &db));
+
+  // Write mixed operations.
+  ASSERT_OK(db->Put(WriteOptions(), "key_aa", "val_put"));
+  ASSERT_OK(db->Merge(WriteOptions(), "key_bb", "val_merge"));
+  ASSERT_OK(db->Delete(WriteOptions(), "key_cc"));
+  ASSERT_OK(db->Put(WriteOptions(), "key_dd", "val_put2"));
+  ASSERT_OK(db->SingleDelete(WriteOptions(), "key_dd"));
+  ASSERT_OK(db->Put(WriteOptions(), "key_ee", "val_put3"));
+
+  // Flush to produce an SST with UDI.
+  ASSERT_OK(db->Flush(FlushOptions()));
+
+  // Verify data is readable via the native index (which always works with
+  // SeekToFirst). key_aa (put), key_bb (merge), key_ee (put) should be
+  // visible. key_cc was deleted, key_dd was single-deleted.
+  {
+    ReadOptions ro;
+    std::unique_ptr<Iterator> iter(db->NewIterator(ro));
+    iter->SeekToFirst();
+    std::vector<std::string> visible;
+    for (; iter->Valid(); iter->Next()) {
+      visible.push_back(iter->key().ToString());
+    }
+    ASSERT_OK(iter->status());
+    ASSERT_EQ(visible.size(), 3u);
+    // With reverse comparator, keys are in reverse order.
+    if (is_reverse_comparator_) {
+      std::vector<std::string> expected = {"key_ee", "key_bb", "key_aa"};
+      ASSERT_EQ(visible, expected);
+    } else {
+      std::vector<std::string> expected = {"key_aa", "key_bb", "key_ee"};
+      ASSERT_EQ(visible, expected);
+    }
+  }
+
+  ASSERT_OK(db->Close());
+  ASSERT_OK(DestroyDB(dbname, options_));
+}
+
+TEST_P(UserDefinedIndexTest, ValueTypeMappingViaDBFlush) {
+  // Verify that MapToUDIValueType correctly maps internal ValueTypes to UDI
+  // ValueTypes by writing various operation types via the DB API, flushing,
+  // and inspecting what the TestUserDefinedIndexBuilder received.
+  if (is_reverse_comparator_) {
+    // Skip for reverse comparator — the key ordering makes this test
+    // unnecessarily complex and the mapping logic is comparator-independent.
+    ROCKSDB_GTEST_SKIP("Skipped for reverse comparator");
+    return;
+  }
+  std::string dbname = test::PerThreadDBPath("udi_valuetype_mapping_test");
+  ASSERT_OK(DestroyDB(dbname, options_));
+
+  BlockBasedTableOptions table_options;
+  auto user_defined_index_factory =
+      std::make_shared<TestUserDefinedIndexFactory>();
+  user_defined_index_factory->skip_key_size_check_ = true;
+  table_options.user_defined_index_factory = user_defined_index_factory;
+  options_.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  options_.merge_operator = MergeOperators::CreateStringAppendOperator();
+  options_.create_if_missing = true;
+  options_.disable_auto_compactions = true;
+
+  std::unique_ptr<DB> db;
+  ASSERT_OK(DB::Open(options_, dbname, &db));
+
+  // Write one entry of each type that goes through the flush path.
+  // kTypeValue:
+  ASSERT_OK(db->Put(WriteOptions(), "key_01_put", "v1"));
+  // kTypeMerge:
+  ASSERT_OK(db->Merge(WriteOptions(), "key_02_merge", "m1"));
+  // kTypeDeletion:
+  ASSERT_OK(db->Delete(WriteOptions(), "key_03_del"));
+  // kTypeSingleDeletion:
+  ASSERT_OK(db->SingleDelete(WriteOptions(), "key_04_sdel"));
+  // kTypeWideColumnEntity:
+  ASSERT_OK(db->PutEntity(WriteOptions(), db->DefaultColumnFamily(),
+                          "key_05_entity", WideColumns{{"col1", "val1"}}));
+
+  ASSERT_OK(db->Flush(FlushOptions()));
+
+  // The builder recorded all (key, ValueType) pairs via the shared log.
+  const auto& log = user_defined_index_factory->key_type_log_;
+  ASSERT_FALSE(log.empty());
+
+  // Build a map from key to the ValueType received by OnKeyAdded.
+  std::map<std::string, UserDefinedIndexBuilder::ValueType> type_map;
+  for (const auto& entry : log) {
+    type_map[entry.first] = entry.second;
+  }
+
+  // Verify each mapping.
+  ASSERT_EQ(type_map.count("key_01_put"), 1u);
+  EXPECT_EQ(type_map["key_01_put"], UserDefinedIndexBuilder::kValue);
+
+  ASSERT_EQ(type_map.count("key_02_merge"), 1u);
+  EXPECT_EQ(type_map["key_02_merge"], UserDefinedIndexBuilder::kMerge);
+
+  ASSERT_EQ(type_map.count("key_03_del"), 1u);
+  EXPECT_EQ(type_map["key_03_del"], UserDefinedIndexBuilder::kDelete);
+
+  ASSERT_EQ(type_map.count("key_04_sdel"), 1u);
+  EXPECT_EQ(type_map["key_04_sdel"], UserDefinedIndexBuilder::kDelete);
+
+  ASSERT_EQ(type_map.count("key_05_entity"), 1u);
+  EXPECT_EQ(type_map["key_05_entity"], UserDefinedIndexBuilder::kOther);
+
+  ASSERT_OK(db->Close());
+  ASSERT_OK(DestroyDB(dbname, options_));
+}
+
+TEST_P(UserDefinedIndexTest, CompactionWithSnapshotsAndUDI) {
+  // Verify that compaction with snapshots (producing multiple versions of the
+  // same user key) works correctly with UDI.
+  if (is_reverse_comparator_) {
+    ROCKSDB_GTEST_SKIP("Skipped for reverse comparator");
+    return;
+  }
+  std::string dbname = test::PerThreadDBPath("udi_compaction_snapshot_test");
+  ASSERT_OK(DestroyDB(dbname, options_));
+
+  BlockBasedTableOptions table_options;
+  auto user_defined_index_factory =
+      std::make_shared<TestUserDefinedIndexFactory>();
+  user_defined_index_factory->skip_key_size_check_ = true;
+  table_options.user_defined_index_factory = user_defined_index_factory;
+  options_.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  options_.create_if_missing = true;
+  // Disable auto-compaction so we control when compaction runs.
+  options_.disable_auto_compactions = true;
+
+  std::unique_ptr<DB> db;
+  ASSERT_OK(DB::Open(options_, dbname, &db));
+
+  // Write version 1 and flush.
+  ASSERT_OK(db->Put(WriteOptions(), "key_aa", "v1"));
+  ASSERT_OK(db->Put(WriteOptions(), "key_bb", "v1"));
+  ASSERT_OK(db->Flush(FlushOptions()));
+
+  // Take a snapshot to force compaction to keep both versions.
+  const Snapshot* snap = db->GetSnapshot();
+
+  // Write version 2 and flush (creates a second L0 file).
+  ASSERT_OK(db->Put(WriteOptions(), "key_aa", "v2"));
+  ASSERT_OK(db->Delete(WriteOptions(), "key_bb"));
+  ASSERT_OK(db->Flush(FlushOptions()));
+
+  // Compact L0 → L1. With the snapshot held, both versions of key_aa
+  // and the delete tombstone for key_bb must be preserved in the compaction
+  // output. The UDI builder receives multiple entries for key_aa.
+  ASSERT_OK(db->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+
+  // Verify the UDI builder saw entries during compaction. The key_type_log
+  // accumulates from all builders (two flushes + one compaction). The
+  // compaction output must contain multiple versions of key_aa (v2 and v1,
+  // due to the snapshot) and both the delete tombstone and old value of key_bb.
+  const auto& log = user_defined_index_factory->key_type_log_;
+  ASSERT_FALSE(log.empty());
+
+  // Count total occurrences of key_aa across all builders — at least 4:
+  // flush1 (v1) + flush2 (v2) + compaction (v2, v1).
+  int key_aa_count = 0;
+  int key_bb_count = 0;
+  for (const auto& entry : log) {
+    if (entry.first == "key_aa") {
+      key_aa_count++;
+    } else if (entry.first == "key_bb") {
+      key_bb_count++;
+    }
+  }
+  // flush1 (1) + flush2 (1) + compaction (2 versions due to snapshot) = 4.
+  ASSERT_GE(key_aa_count, 4) << "Expected key_aa from flush1 + flush2 + "
+                                "compaction (2 versions due to snapshot)";
+  // flush1 (1) + flush2 (1) + compaction (tombstone + old value) = 4.
+  ASSERT_GE(key_bb_count, 4) << "Expected key_bb from flush1 + flush2 + "
+                                "compaction (tombstone + old value)";
+
+  // Verify current view via native index: key_aa=v2, key_bb deleted.
+  {
+    ReadOptions ro;
+    std::unique_ptr<Iterator> iter(db->NewIterator(ro));
+    iter->SeekToFirst();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_EQ(iter->key().ToString(), "key_aa");
+    ASSERT_EQ(iter->value().ToString(), "v2");
+    iter->Next();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+  }
+
+  // Verify snapshot view via native index: key_aa=v1, key_bb=v1.
+  {
+    ReadOptions ro;
+    ro.snapshot = snap;
+    std::unique_ptr<Iterator> iter(db->NewIterator(ro));
+    iter->SeekToFirst();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_EQ(iter->key().ToString(), "key_aa");
+    ASSERT_EQ(iter->value().ToString(), "v1");
+    iter->Next();
+    ASSERT_TRUE(iter->Valid());
+    ASSERT_EQ(iter->key().ToString(), "key_bb");
+    ASSERT_EQ(iter->value().ToString(), "v1");
+    iter->Next();
+    ASSERT_FALSE(iter->Valid());
+    ASSERT_OK(iter->status());
+  }
+
+  db->ReleaseSnapshot(snap);
+  ASSERT_OK(db->Close());
+  ASSERT_OK(DestroyDB(dbname, options_));
 }
 
 TEST_P(UserDefinedIndexTest, IngestTest) {

tools/db_crashtest.py

@@ -242,8 +242,8 @@ default_params = {
     "uncache_aggressiveness": lambda: int(math.pow(10, 4.0 * random.random()) - 1.0),
     "use_full_merge_v1": lambda: random.randint(0, 1),
     "use_merge": lambda: random.randint(0, 1),
-    # use_trie_index must be the same across invocations because it restricts
-    # operations (no deletes/merges) and existing SSTs may contain non-Put types
+    # use_trie_index must be the same across invocations so that all SSTs
+    # in a DB are opened with matching table options
     "use_trie_index": random.choice([0, 0, 0, 0, 0, 0, 0, 1]),
     # use_put_entity_one_in has to be the same across invocations for verification to work, hence no lambda
     "use_put_entity_one_in": random.choice([0] * 7 + [1, 5, 10]),
@@ -903,44 +903,14 @@ def finalize_and_sanitize(src_params):
     else:
         dest_params["allow_resumption_one_in"] = 0
 
-    # Trie UDI only supports Put (kTypeValue). Disable incompatible operations.
+    # UDI now supports all operation types (Put, Delete, Merge, etc.).
+    # Only parallel compression and mmap_read remain incompatible.
     if dest_params.get("use_trie_index") == 1:
-        dest_params["use_merge"] = 0
-        dest_params["use_put_entity_one_in"] = 0
-        dest_params["use_timed_put_one_in"] = 0
-        dest_params["use_get_entity"] = 0
-        dest_params["use_multi_get_entity"] = 0
-        # TransactionDB ROLLBACK writes DELETE entries to WAL to undo
-        # uncommitted changes. These DELETEs violate UDI's Put-only restriction.
-        dest_params["use_txn"] = 0
-        dest_params["use_optimistic_txn"] = 0
-        dest_params["test_multi_ops_txns"] = 0
         # Trie UDI uses zero-copy pointers into block data, which is
         # incompatible with mmap_read.
         dest_params["mmap_read"] = 0
-        # Redistribute delete/delrange percents to write percent
-        dest_params["writepercent"] += dest_params["delpercent"]
-        dest_params["writepercent"] += dest_params["delrangepercent"]
-        dest_params["delpercent"] = 0
-        dest_params["delrangepercent"] = 0
-        # Ingestion with standalone range deletions is incompatible
-        dest_params["test_ingest_standalone_range_deletion_one_in"] = 0
         # Parallel compression is incompatible with UDI
         dest_params["compression_parallel_threads"] = 1
-        # Trie UDI does not support SeekToFirst/SeekToLast. Prefix scanning
-        # calls SeekToFirst internally, so disable it. Additionally,
-        # LevelIterator::SkipEmptyFileForward() calls SeekToFirst() when
-        # Next() crosses file boundaries, so general iteration (iterpercent)
-        # also fails with trie UDI. Redistribute both to reads.
-        dest_params["readpercent"] += dest_params.get("prefixpercent", 0)
-        dest_params["prefixpercent"] = 0
-        dest_params["readpercent"] += dest_params.get("iterpercent", 0)
-        dest_params["iterpercent"] = 0
-        # BlobDB writes kTypeBlobIndex entries in SSTs instead of kTypeValue,
-        # which violates UDI's Put-only restriction. Also disable dynamic
-        # blob options to prevent SetOptions from re-enabling blob files.
-        dest_params["enable_blob_files"] = 0
-        dest_params["allow_setting_blob_options_dynamically"] = 0
 
     # Multi-key operations are not currently compatible with transactions or
     # timestamp.

utilities/trie_index/bitvector.h

@@ -176,7 +176,9 @@ class BitvectorBuilder {
   // when possible, which is significantly faster than the bit-by-bit loop for
   // large counts (e.g., appending 256 zeros for an empty dense node).
   void AppendMultiple(bool bit, uint64_t count) {
-    if (count == 0) return;
+    if (count == 0) {
+      return;
+    }
 
     // Fill partial word at the end of the current buffer.
     uint64_t partial = num_bits_ % 64;
@@ -267,6 +269,8 @@ class Bitvector {
   // std::string's move constructor preserves the buffer address for
   // SSO-exceeding strings. For the InitFromData case, the pointers reference
   // external memory and are unaffected by moving owned_data_ (which is empty).
+  ~Bitvector() = default;
+
   Bitvector(const Bitvector&) = delete;
   Bitvector& operator=(const Bitvector&) = delete;
   // Move constructor delegates to default ctor + move assignment to avoid
@@ -363,7 +367,8 @@ class Bitvector {
       return num_bits_;
     }
     // Use select hints to narrow the search range.
-    uint64_t lo, hi;
+    uint64_t lo;
+    uint64_t hi;
     if (num_select1_hints_ > 0) {
       uint64_t hint_idx = i / kOnesPerSelectHint;
       lo = select1_hints_[hint_idx];
@@ -501,6 +506,8 @@ class EliasFano {
         low_words_(nullptr),
         num_low_words_(0) {}
 
+  ~EliasFano() = default;
+
   EliasFano(const EliasFano&) = delete;
   EliasFano& operator=(const EliasFano&) = delete;
 

utilities/trie_index/louds_trie.cc

@@ -51,10 +51,12 @@ void LoudsTrieBuilder::AddKeyWithSeqno(const Slice& key,
 
 void LoudsTrieBuilder::AddOverflowBlock(const TrieBlockHandle& handle,
                                         uint64_t seqno) {
-  // Overflow blocks always represent actual key versions within a same-key
-  // run, so seqno must be > 0. Seqno 0 is reserved as the sentinel meaning
-  // "never advance past this leaf" in the reader's post-seek correction.
-  assert(seqno != 0);
+  // Seqno may be 0 when bottommost compaction zeroes all sequence numbers.
+  // In that case, every block in the same-key run has seqno=0. The reader's
+  // post-seek correction handles this correctly: the primary leaf's seqno=0
+  // triggers the "never advance" guard (leaf_seqno != 0 check), so the seek
+  // returns the primary block. Next() iterates overflow blocks by index, not
+  // seqno, so all blocks are still visited in order.
   overflow_handles_.push_back(handle);
   overflow_seqnos_.push_back(seqno);
 }
@@ -328,11 +330,49 @@ void LoudsTrieBuilder::Finish() {
   // Only used when has_seqno_encoding_ is true.
   std::vector<uint64_t> bfs_ordered_seqnos;
   std::vector<uint32_t> bfs_ordered_block_counts;
+  // Overflow arrays must also be BFS-reordered. Without this, the
+  // overflow_base_ prefix sum (computed from BFS-ordered block_counts)
+  // would index into key-sorted overflow arrays, mapping overflow blocks
+  // to the wrong leaves when BFS order differs from key-sorted order
+  // (which happens whenever separator keys have different lengths).
+  std::vector<uint32_t> key_sorted_overflow_base;
+  std::vector<TrieBlockHandle> bfs_ordered_overflow_handles;
+  std::vector<uint64_t> bfs_ordered_overflow_seqnos;
   if (has_seqno_encoding_) {
     bfs_ordered_seqnos.reserve(keys_.size());
     bfs_ordered_block_counts.reserve(keys_.size());
+    if (!overflow_handles_.empty()) {
+      key_sorted_overflow_base.resize(keys_.size());
+      uint32_t sum = 0;
+      for (size_t i = 0; i < keys_.size(); i++) {
+        key_sorted_overflow_base[i] = sum;
+        assert(block_counts_[i] >= 1);
+        sum += block_counts_[i] - 1;
+      }
+      assert(sum == static_cast<uint32_t>(overflow_handles_.size()));
+      bfs_ordered_overflow_handles.reserve(overflow_handles_.size());
+      bfs_ordered_overflow_seqnos.reserve(overflow_seqnos_.size());
+    }
   }
 
+  // Emit BFS-ordered data for a leaf with key index ki: primary handle,
+  // seqno side-table fields, and overflow blocks (if any).
+  auto emit_leaf = [&](size_t ki) {
+    bfs_ordered_handles.push_back(handles_[ki]);
+    if (has_seqno_encoding_) {
+      bfs_ordered_seqnos.push_back(seqnos_[ki]);
+      bfs_ordered_block_counts.push_back(block_counts_[ki]);
+      if (!key_sorted_overflow_base.empty() && block_counts_[ki] > 1) {
+        uint32_t base = key_sorted_overflow_base[ki];
+        uint32_t count = block_counts_[ki] - 1;
+        for (uint32_t j = 0; j < count; j++) {
+          bfs_ordered_overflow_handles.push_back(overflow_handles_[base + j]);
+          bfs_ordered_overflow_seqnos.push_back(overflow_seqnos_[base + j]);
+        }
+      }
+    }
+  };
+
   for (uint32_t level = 0; level <= max_depth_; level++) {
     const auto& ld = levels[level];
     if (ld.node_count() == 0) {
@@ -350,12 +390,7 @@ void LoudsTrieBuilder::Finish() {
 
       // ---- Handle reordering: emit prefix key handle ----
       if (ld.is_prefix[ni] && ld.prefix_handle[ni] >= 0) {
-        size_t ki = static_cast<size_t>(ld.prefix_handle[ni]);
-        bfs_ordered_handles.push_back(handles_[ki]);
-        if (has_seqno_encoding_) {
-          bfs_ordered_seqnos.push_back(seqnos_[ki]);
-          bfs_ordered_block_counts.push_back(block_counts_[ki]);
-        }
+        emit_leaf(static_cast<size_t>(ld.prefix_handle[ni]));
       }
 
       // Skip pure leaf nodes (no labels) — they are accounted for by
@@ -394,12 +429,7 @@ void LoudsTrieBuilder::Finish() {
 
           if (!is_internal) {
             if (ld.leaf_handle[li] >= 0) {
-              size_t ki = static_cast<size_t>(ld.leaf_handle[li]);
-              bfs_ordered_handles.push_back(handles_[ki]);
-              if (has_seqno_encoding_) {
-                bfs_ordered_seqnos.push_back(seqnos_[ki]);
-                bfs_ordered_block_counts.push_back(block_counts_[ki]);
-              }
+              emit_leaf(static_cast<size_t>(ld.leaf_handle[li]));
             }
             dense_leaf_count_++;
           } else if (level == cutoff_level_ - 1) {
@@ -424,12 +454,7 @@ void LoudsTrieBuilder::Finish() {
 
           if (!is_internal) {
             if (ld.leaf_handle[li] >= 0) {
-              size_t ki = static_cast<size_t>(ld.leaf_handle[li]);
-              bfs_ordered_handles.push_back(handles_[ki]);
-              if (has_seqno_encoding_) {
-                bfs_ordered_seqnos.push_back(seqnos_[ki]);
-                bfs_ordered_block_counts.push_back(block_counts_[ki]);
-              }
+              emit_leaf(static_cast<size_t>(ld.leaf_handle[li]));
             }
           }
         }
@@ -450,6 +475,12 @@ void LoudsTrieBuilder::Finish() {
     assert(bfs_ordered_block_counts.size() == keys_.size());
     seqnos_ = std::move(bfs_ordered_seqnos);
     block_counts_ = std::move(bfs_ordered_block_counts);
+    if (!overflow_handles_.empty()) {
+      assert(bfs_ordered_overflow_handles.size() == overflow_handles_.size());
+      assert(bfs_ordered_overflow_seqnos.size() == overflow_seqnos_.size());
+      overflow_handles_ = std::move(bfs_ordered_overflow_handles);
+      overflow_seqnos_ = std::move(bfs_ordered_overflow_seqnos);
+    }
   }
 
   SerializeAll();
@@ -1773,6 +1804,26 @@ bool LoudsTrieIterator::DescendToLeftmostLeaf(bool in_dense,
   }
 }
 
+bool LoudsTrieIterator::SeekToFirst() {
+  valid_ = false;
+  leaf_index_ = 0;
+  key_len_ = 0;
+  path_.clear();
+  is_at_prefix_key_ = false;
+
+  if (trie_->NumKeys() == 0) {
+    return false;
+  }
+
+  // Descend directly from root to the leftmost leaf.
+  // Compared to Seek(""), this skips the SeekImpl target-consumption loop
+  // (a no-op for empty target) and the redundant prefix key check that
+  // SeekImpl performs at the root before calling DescendToLeftmostLeaf.
+  // DescendToLeftmostLeaf itself handles prefix keys at every node.
+  bool in_dense = (trie_->cutoff_level_ > 0);
+  return DescendToLeftmostLeaf(in_dense, /*node_num=*/0);
+}
+
 // Main Seek implementation.
 // Uses SuRF-style Select-free traversal for sparse regions: instead of
 // tracking node_num and calling FindNthOneBit to find node boundaries, we

utilities/trie_index/louds_trie.h

@@ -244,6 +244,8 @@ class LoudsTrie {
   // forbidden since C++11). Trie data always exceeds the SSO threshold
   // (hundreds to thousands of bytes), so aligned_copy_ is always
   // heap-allocated, and move always preserves the buffer address.
+  ~LoudsTrie() = default;
+
   LoudsTrie(const LoudsTrie&) = delete;
   LoudsTrie& operator=(const LoudsTrie&) = delete;
   LoudsTrie(LoudsTrie&&) = default;
@@ -470,6 +472,13 @@ class LoudsTrieIterator {
  public:
   explicit LoudsTrieIterator(const LoudsTrie* trie);
 
+  // Position on the very first leaf (smallest key) by descending from the
+  // root to the leftmost leaf. More efficient than Seek(Slice()) because it
+  // skips SeekImpl's target-consumption loop and its redundant prefix key
+  // check at root (DescendToLeftmostLeaf handles prefix keys at every node).
+  // Returns true if positioned on a valid leaf.
+  bool SeekToFirst();
+
   // Seek to the first leaf whose key is >= `target`.
   // Returns true if positioned on a valid leaf.
   //

utilities/trie_index/trie_index_factory.cc

@@ -77,20 +77,29 @@ Slice TrieIndexBuilder::AddIndexEntry(const Slice& last_key_in_current_block,
       }
     }
   } else {
-    // Last block: use a short successor of the last key.
-    *separator_scratch = last_key_in_current_block.ToString();
-    comparator_->FindShortSuccessor(separator_scratch);
-    separator = Slice(*separator_scratch);
+    // Last block: use the last key itself as the separator, NOT a shortened
+    // successor. This matches the standard ShortenedIndexBuilder behavior
+    // (see index_builder.h GetSeparatorWithSeq lines 278-286): it only calls
+    // FindShortInternalKeySuccessor when shortening_mode is
+    // kShortenSeparatorsAndSuccessor, which is not the default. With the
+    // default kShortenSeparators, the last block's separator is simply
+    // last_key_in_current_block.
+    //
+    // Why this matters: FindShortSuccessor can widen the key range. For
+    // example, if the actual last key is "9\xff\xff", FindShortSuccessor
+    // produces ":" (0x3A). The trie would then claim to cover keys up to
+    // ":", but the data block only contains keys up to "9\xff\xff". A seek
+    // targeting a key in that gap (e.g., "9\xff\xff\x01") would find a
+    // block via the trie that contains no matching data, causing iterator
+    // desynchronization — the trie index returns a valid block while the
+    // standard index correctly reports no match.
+    separator = last_key_in_current_block;
 
-    // Edge case: FindShortSuccessor may fail to shorten the key (e.g.,
-    // all-0xFF keys like "\xff\xff" — no byte can be incremented). When
-    // this happens AND the previous entry has the same separator, the last
-    // block is actually part of a same-user-key run. Without this check,
-    // the last block would get seqno=kMaxSequenceNumber (sentinel), but
-    // Finish() would group it into the run and hit the assert that overflow
-    // blocks must have real seqnos.
+    // Edge case: if this last block's separator matches the previous entry's
+    // separator, they share the same user key (same-user-key run boundary).
     if (!buffered_entries_.empty() &&
-        buffered_entries_.back().separator_key == *separator_scratch) {
+        comparator_->Compare(buffered_entries_.back().separator_key,
+                             separator) == 0) {
       same_user_key = true;
       if (!must_use_separator_with_seq_) {
         must_use_separator_with_seq_ = true;
@@ -172,8 +181,10 @@ Status TrieIndexBuilder::Finish(Slice* index_contents) {
                                     seqno, block_count);
 
       // Add overflow blocks (2nd, 3rd, ... in the run).
-      // Overflow blocks only exist within same-key runs, which always
-      // receive real seqnos from AddIndexEntry, never kMaxSequenceNumber.
+      // Overflow blocks only exist within same-key runs, so their seqnos
+      // come from last_key_seq in AddIndexEntry (never kMaxSequenceNumber).
+      // The seqno may be 0 when bottommost compaction zeroes all sequence
+      // numbers — this is valid; see AddOverflowBlock comment.
       for (size_t j = run_start + 1; j < run_end; j++) {
         assert(buffered_entries_[j].seqno != kMaxSequenceNumber);
         trie_builder_.AddOverflowBlock(buffered_entries_[j].handle,
@@ -231,6 +242,37 @@ void TrieIndexIterator::Prepare(const ScanOptions scan_opts[],
   prepared_ = true;
 }
 
+Status TrieIndexIterator::SeekToFirstAndGetResult(IterateResult* result) {
+  // Reset overflow state — SeekToFirst always lands on the primary block
+  // of the first trie leaf.
+  overflow_run_index_ = 0;
+  overflow_run_size_ = 1;
+  overflow_base_idx_ = 0;
+
+  if (!iter_.SeekToFirst()) {
+    result->bound_check_result = IterBoundCheck::kUnknown;
+    result->key = Slice();
+    return Status::OK();
+  }
+
+  result->key = iter_.Key();
+  current_key_scratch_ = result->key.ToString();
+  result->key = Slice(current_key_scratch_);
+
+  // Set up overflow state for the first leaf if seqno encoding is active.
+  if (has_seqno_encoding_) {
+    uint64_t leaf_idx = iter_.LeafIndex();
+    uint32_t block_count = trie_->GetLeafBlockCount(leaf_idx);
+    overflow_run_size_ = block_count;
+    overflow_base_idx_ = trie_->GetOverflowBase(leaf_idx);
+  }
+
+  // The very first entry is always in bounds (no target to compare against
+  // the limit, and the first block cannot precede any scan range).
+  result->bound_check_result = IterBoundCheck::kInbound;
+  return Status::OK();
+}
+
 Status TrieIndexIterator::SeekAndGetResult(const Slice& target,
                                            IterateResult* result,
                                            const SeekContext& context) {
@@ -506,13 +548,13 @@ Status TrieIndexFactory::NewBuilder(
 Status TrieIndexFactory::NewReader(
     const UserDefinedIndexOption& option, Slice& index_block,
     std::unique_ptr<UserDefinedIndexReader>& reader) const {
-  if (option.comparator != nullptr &&
-      option.comparator != BytewiseComparator()) {
+  const Comparator* cmp =
+      option.comparator ? option.comparator : BytewiseComparator();
+  if (cmp != BytewiseComparator()) {
     return Status::NotSupported(
-        "TrieIndexFactory requires BytewiseComparator; got: ",
-        option.comparator->Name());
+        "TrieIndexFactory requires BytewiseComparator; got: ", cmp->Name());
   }
-  auto trie_reader = std::make_unique<TrieIndexReader>(option.comparator);
+  auto trie_reader = std::make_unique<TrieIndexReader>(cmp);
   Status s = trie_reader->InitFromSlice(index_block);
   if (!s.ok()) {
     return s;

utilities/trie_index/trie_index_factory.h

@@ -111,7 +111,7 @@ class TrieIndexBuilder : public UserDefinedIndexBuilder {
   //     at Finish() time as a sentinel meaning "never advance")
   struct BufferedEntry {
     std::string separator_key;
-    SequenceNumber seqno;
+    SequenceNumber seqno{};
     TrieBlockHandle handle;
   };
   std::vector<BufferedEntry> buffered_entries_;
@@ -136,6 +136,10 @@ class TrieIndexIterator : public UserDefinedIndexIterator {
   // Prepare for a batch of scans. Stores scan bounds for later use.
   void Prepare(const ScanOptions scan_opts[], size_t num_opts) override;
 
+  // Position at the very first index entry. Descends directly to the
+  // leftmost leaf without a full seek traversal.
+  Status SeekToFirstAndGetResult(IterateResult* result) override;
+
   // Seek to the first index entry >= target. When has_seqno_encoding_ is
   // true, the trie is searched with user_key only, then post-seek correction
   // uses target_seq from context to advance through overflow blocks as needed.