Track newer snapshot versions in RocksDB metadata reads

Summary:
Add experimental `DB::GetWithMetadata()` and `DB::MultiGetWithMetadata()` APIs with opt-in timestamp and `newer_version_present` outputs. Append the new virtuals to `DB` to preserve existing vtable slot order, and forward them through `StackableDB`, BlobDB, TTL, compacted, read-only, secondary, C, and Java/JNI surfaces.

For explicit-snapshot reads, sample a published sequence upper bound and scan through it while a read callback preserves the snapshot-visible result. Point writes and covering range tombstones in memtables and SSTs update per-key metadata. Row-cache lookups are bypassed only while sequence inspection is still required and replay uses the original snapshot sequence.

Keep ordinary `Get()` / `MultiGet()` paths free of metadata discovery scans and context allocation. Reject newer-version tracking for `kPersistedTier` and transaction reads that require custom visibility callbacks. Latest reads without an explicit snapshot return false metadata.

Integrate both metadata APIs into batched `db_stress`, including consistency validation for atomic batches, and add a `db_bench --read_with_metadata` mode for repeatable performance comparisons. Add the public API release note.

Rebased onto latest remote master `b79bb51e6349` and resolved conflicts in the transaction and TTL wrappers/tests. This version also keeps the prior review-feedback fixes around metadata context mutation, compacted-DB output initialization, TTL MultiGet allocation, and MultiGet status preservation.

This version applies clang-format fixes for the `check-format-and-targets` workflow without changing behavior.

Differential Revision: D104182397

Author: xingbowang

db/c.cc

@@ -2623,6 +2623,59 @@ char* rocksdb_get(rocksdb_t* db, const rocksdb_readoptions_t* options,
   return result;
 }
 
+static char* rocksdb_get_with_metadata_impl(
+    rocksdb_t* db, const rocksdb_readoptions_t* options,
+    ColumnFamilyHandle* column_family, const char* key, size_t keylen,
+    size_t* vallen, char** timestamp, size_t* timestamp_len,
+    unsigned char* newer_version_present, char** errptr) {
+  ROCKSDB_NAMESPACE::OutputMetadata output_metadata;
+  if (timestamp != nullptr && timestamp_len != nullptr) {
+    output_metadata.WantTimestamp();
+  }
+  if (newer_version_present != nullptr) {
+    output_metadata.WantNewerVersionPresent();
+  }
+
+  PinnableSlice pinnable_val;
+  const Status s =
+      db->rep->GetWithMetadata(options->rep, column_family, Slice(key, keylen),
+                               &pinnable_val, &output_metadata);
+  if (newer_version_present != nullptr) {
+    *newer_version_present = *output_metadata.newer_version_present ? 1 : 0;
+  }
+
+  char* result = nullptr;
+  if (s.ok()) {
+    *vallen = pinnable_val.size();
+    result = CopyString(pinnable_val);
+    if (output_metadata.timestamp.has_value()) {
+      *timestamp_len = output_metadata.timestamp->size();
+      *timestamp = CopyString(*output_metadata.timestamp);
+    }
+  } else {
+    *vallen = 0;
+    if (output_metadata.timestamp.has_value()) {
+      *timestamp = nullptr;
+      *timestamp_len = 0;
+    }
+    if (!s.IsNotFound()) {
+      SaveError(errptr, s);
+    }
+  }
+  return result;
+}
+
+char* rocksdb_get_with_metadata(rocksdb_t* db,
+                                const rocksdb_readoptions_t* options,
+                                const char* key, size_t keylen, size_t* vallen,
+                                char** timestamp, size_t* timestamp_len,
+                                unsigned char* newer_version_present,
+                                char** errptr) {
+  return rocksdb_get_with_metadata_impl(
+      db, options, db->rep->DefaultColumnFamily(), key, keylen, vallen,
+      timestamp, timestamp_len, newer_version_present, errptr);
+}
+
 char* rocksdb_get_cf(rocksdb_t* db, const rocksdb_readoptions_t* options,
                      rocksdb_column_family_handle_t* column_family,
                      const char* key, size_t keylen, size_t* vallen,
@@ -2645,6 +2698,16 @@ char* rocksdb_get_cf(rocksdb_t* db, const rocksdb_readoptions_t* options,
   return result;
 }
 
+char* rocksdb_get_cf_with_metadata(
+    rocksdb_t* db, const rocksdb_readoptions_t* options,
+    rocksdb_column_family_handle_t* column_family, const char* key,
+    size_t keylen, size_t* vallen, char** timestamp, size_t* timestamp_len,
+    unsigned char* newer_version_present, char** errptr) {
+  return rocksdb_get_with_metadata_impl(
+      db, options, column_family->rep, key, keylen, vallen, timestamp,
+      timestamp_len, newer_version_present, errptr);
+}
+
 char* rocksdb_get_with_ts(rocksdb_t* db, const rocksdb_readoptions_t* options,
                           const char* key, size_t keylen, size_t* vallen,
                           char** ts, size_t* tslen, char** errptr) {
@@ -2736,6 +2799,72 @@ void rocksdb_multi_get(rocksdb_t* db, const rocksdb_readoptions_t* options,
   }
 }
 
+static void rocksdb_multi_get_with_metadata_impl(
+    rocksdb_t* db, const rocksdb_readoptions_t* options,
+    ColumnFamilyHandle* const* column_families, size_t num_keys,
+    const char* const* keys_list, const size_t* keys_list_sizes,
+    char** values_list, size_t* values_list_sizes, char** timestamp_list,
+    size_t* timestamp_list_sizes, unsigned char* newer_version_present,
+    char** errs) {
+  std::unique_ptr<Slice[]> keys(new Slice[num_keys]);
+  std::vector<PinnableSlice> values(num_keys);
+  std::vector<Status> statuses(num_keys);
+  for (size_t i = 0; i < num_keys; ++i) {
+    keys[i] = Slice(keys_list[i], keys_list_sizes[i]);
+  }
+
+  ROCKSDB_NAMESPACE::MultiGetOutputMetadata output_metadata;
+  if (timestamp_list != nullptr && timestamp_list_sizes != nullptr) {
+    output_metadata.WantTimestamps();
+  }
+  if (newer_version_present != nullptr) {
+    output_metadata.WantNewerVersionPresent();
+  }
+  db->rep->MultiGetWithMetadata(options->rep, num_keys, column_families,
+                                keys.get(), values.data(), statuses.data(),
+                                &output_metadata);
+
+  for (size_t i = 0; i < num_keys; ++i) {
+    if (newer_version_present != nullptr) {
+      newer_version_present[i] =
+          (*output_metadata.newer_version_present)[i] ? 1 : 0;
+    }
+    if (statuses[i].ok()) {
+      values_list[i] = CopyString(values[i]);
+      values_list_sizes[i] = values[i].size();
+      if (output_metadata.timestamps.has_value()) {
+        timestamp_list[i] = CopyString((*output_metadata.timestamps)[i]);
+        timestamp_list_sizes[i] = (*output_metadata.timestamps)[i].size();
+      }
+      errs[i] = nullptr;
+    } else {
+      values_list[i] = nullptr;
+      values_list_sizes[i] = 0;
+      if (output_metadata.timestamps.has_value()) {
+        timestamp_list[i] = nullptr;
+        timestamp_list_sizes[i] = 0;
+      }
+      errs[i] = statuses[i].IsNotFound()
+                    ? nullptr
+                    : strdup(statuses[i].ToString().c_str());
+    }
+  }
+}
+
+void rocksdb_multi_get_with_metadata(
+    rocksdb_t* db, const rocksdb_readoptions_t* options, size_t num_keys,
+    const char* const* keys_list, const size_t* keys_list_sizes,
+    char** values_list, size_t* values_list_sizes, char** timestamp_list,
+    size_t* timestamp_list_sizes, unsigned char* newer_version_present,
+    char** errs) {
+  std::vector<ColumnFamilyHandle*> column_families(
+      num_keys, db->rep->DefaultColumnFamily());
+  rocksdb_multi_get_with_metadata_impl(
+      db, options, column_families.data(), num_keys, keys_list, keys_list_sizes,
+      values_list, values_list_sizes, timestamp_list, timestamp_list_sizes,
+      newer_version_present, errs);
+}
+
 void rocksdb_multi_get_with_ts(rocksdb_t* db,
                                const rocksdb_readoptions_t* options,
                                size_t num_keys, const char* const* keys_list,
@@ -2811,6 +2940,24 @@ void rocksdb_multi_get_cf(
   }
 }
 
+void rocksdb_multi_get_cf_with_metadata(
+    rocksdb_t* db, const rocksdb_readoptions_t* options,
+    const rocksdb_column_family_handle_t* const* column_families,
+    size_t num_keys, const char* const* keys_list,
+    const size_t* keys_list_sizes, char** values_list,
+    size_t* values_list_sizes, char** timestamp_list,
+    size_t* timestamp_list_sizes, unsigned char* newer_version_present,
+    char** errs) {
+  std::vector<ColumnFamilyHandle*> cfs(num_keys);
+  for (size_t i = 0; i < num_keys; ++i) {
+    cfs[i] = column_families[i]->rep;
+  }
+  rocksdb_multi_get_with_metadata_impl(
+      db, options, cfs.data(), num_keys, keys_list, keys_list_sizes,
+      values_list, values_list_sizes, timestamp_list, timestamp_list_sizes,
+      newer_version_present, errs);
+}
+
 void rocksdb_multi_get_cf_with_ts(
     rocksdb_t* db, const rocksdb_readoptions_t* options,
     const rocksdb_column_family_handle_t* const* column_families,

db/c_test.c

@@ -2474,6 +2474,35 @@ int main(int argc, char** argv) {
     CheckNoError(err);
     rocksdb_readoptions_set_snapshot(roptions, snap);
     CheckGet(db, roptions, "foo", "hello");
+
+    size_t metadata_value_len = 0;
+    unsigned char newer_version_present = 0;
+    char* metadata_value =
+        rocksdb_get_with_metadata(db, roptions, "foo", 3, &metadata_value_len,
+                                  NULL, NULL, &newer_version_present, &err);
+    // NOLINTNEXTLINE(clang-analyzer-security.insecureAPI.DeprecatedOrUnsafeBufferHandling)
+    CheckNoError(err);
+    CheckEqual("hello", metadata_value, metadata_value_len);
+    // NOLINTNEXTLINE(clang-analyzer-security.insecureAPI.DeprecatedOrUnsafeBufferHandling)
+    CheckCondition(newer_version_present == 1);
+    Free(&metadata_value);
+
+    const char* metadata_keys[] = {"foo"};
+    const size_t metadata_key_sizes[] = {3};
+    char* metadata_values[1];
+    size_t metadata_value_sizes[1];
+    unsigned char newer_versions[1];
+    char* metadata_errors[1];
+    rocksdb_multi_get_with_metadata(
+        db, roptions, 1, metadata_keys, metadata_key_sizes, metadata_values,
+        metadata_value_sizes, NULL, NULL, newer_versions, metadata_errors);
+    // NOLINTNEXTLINE(clang-analyzer-security.insecureAPI.DeprecatedOrUnsafeBufferHandling)
+    CheckCondition(metadata_errors[0] == NULL);
+    CheckEqual("hello", metadata_values[0], metadata_value_sizes[0]);
+    // NOLINTNEXTLINE(clang-analyzer-security.insecureAPI.DeprecatedOrUnsafeBufferHandling)
+    CheckCondition(newer_versions[0] == 1);
+    Free(&metadata_values[0]);
+
     rocksdb_readoptions_set_snapshot(roptions, NULL);
     CheckGet(db, roptions, "foo", NULL);
     rocksdb_release_snapshot(db, snap);