Track each SST's timestamp information as user properties

db/db_with_timestamp_basic_test.cc

@@ -595,6 +595,41 @@ TEST_F(DBBasicTestWithTimestamp, SimpleIterate) {
   Close();
 }
 
+#ifndef ROCKSDB_LITE
+TEST_F(DBBasicTestWithTimestamp, GetTimestampTableProperties) {
+  Options options = CurrentOptions();
+  const size_t kTimestampSize = Timestamp(0, 0).size();
+  TestComparator test_cmp(kTimestampSize);
+  options.comparator = &test_cmp;
+  DestroyAndReopen(options);
+  // Create 2 tables
+  for (int table = 0; table < 2; ++table) {
+    for (int i = 0; i < 10; i++) {
+      WriteOptions write_opts;
+      std::string ts_str = Timestamp(i, 0);
+      Slice ts = ts_str;
+      write_opts.timestamp = &ts;
+      ASSERT_OK(db_->Put(write_opts, "key", Key(i)));
+    }
+    ASSERT_OK(Flush());
+  }
+
+  TablePropertiesCollection props;
+  ASSERT_OK(db_->GetPropertiesOfAllTables(&props));
+  ASSERT_EQ(2U, props.size());
+  for (const auto& item : props) {
+    auto& user_collected = item.second->user_collected_properties;
+    ASSERT_TRUE(user_collected.find("rocksdb.min_timestamp") !=
+                user_collected.end());
+    ASSERT_TRUE(user_collected.find("rocksdb.max_timestamp") !=
+                user_collected.end());
+    ASSERT_EQ(user_collected.at("rocksdb.min_timestamp"), Timestamp(0, 0));
+    ASSERT_EQ(user_collected.at("rocksdb.max_timestamp"), Timestamp(9, 0));
+  }
+  Close();
+}
+#endif  // !ROCKSDB_LITE
+
 class DBBasicTestWithTimestampTableOptions
     : public DBBasicTestWithTimestampBase,
       public testing::WithParamInterface<BlockBasedTableOptions::IndexType> {

db/table_properties_collector.h

@@ -6,12 +6,14 @@
 // This file defines a collection of statistics collectors.
 #pragma once
 
-#include "rocksdb/table_properties.h"
-
 #include <memory>
 #include <string>
 #include <vector>
 
+#include "db/dbformat.h"
+#include "rocksdb/comparator.h"
+#include "rocksdb/table_properties.h"
+
 namespace ROCKSDB_NAMESPACE {
 
 // Base class for internal table properties collector.
@@ -108,4 +110,65 @@ class UserKeyTablePropertiesCollectorFactory
   std::shared_ptr<TablePropertiesCollectorFactory> user_collector_factory_;
 };
 
+// When rocksdb creates a newtable, it will encode all "user keys" into
+// "internal keys". This class collects min/max timestamp from the encoded
+// internal key when Add() is invoked.
+//
+// @param cmp the Comparator to compare timestamp of key by the
+//     Comparator::CompareTimestamp function.
+class TimestampTablePropertiesCollector : public IntTblPropCollector {
+ public:
+  explicit TimestampTablePropertiesCollector(const Comparator* cmp)
+      : cmp_(cmp) {}
+
+  Status InternalAdd(const Slice& key, const Slice& /* value */,
+                     uint64_t /* file_size */) override {
+    auto user_key = ExtractUserKey(key);
+    assert(cmp_ && cmp_->timestamp_size() > 0);
+    if (user_key.size() < cmp_->timestamp_size()) {
+      return Status::Corruption(
+          "User key size mismatch when comparing to timestamp size.");
+    }
+    auto timestamp_in_key =
+        ExtractTimestampFromUserKey(user_key, cmp_->timestamp_size());
+    if (max_timestamp_ == kDisableUserTimestamp ||
+        cmp_->CompareTimestamp(timestamp_in_key, max_timestamp_) > 0) {
+      max_timestamp_ = timestamp_in_key.ToString();
+    }
+    if (min_timestamp_ == kDisableUserTimestamp ||
+        cmp_->CompareTimestamp(min_timestamp_, timestamp_in_key) > 0) {
+      min_timestamp_ = timestamp_in_key.ToString();
+    }
+    return Status::OK();
+  }
+
+  void BlockAdd(uint64_t /* block_raw_bytes */,
+                uint64_t /* block_compressed_bytes_fast */,
+                uint64_t /* block_compressed_bytes_slow */) override {
+    return;
+  }
+
+  Status Finish(UserCollectedProperties* properties) override {
+    assert(min_timestamp_.size() == max_timestamp_.size() &&
+           max_timestamp_.size() == cmp_->timestamp_size());
+    properties->insert({"rocksdb.min_timestamp", min_timestamp_});
+    properties->insert({"rocksdb.max_timestamp", max_timestamp_});
+    return Status::OK();
+  }
+
+  const char* Name() const override {
+    return "TimestampTablePropertiesCollector";
+  }
+
+  UserCollectedProperties GetReadableProperties() const override {
+    return {{"rocksdb.min_timestamp", Slice(min_timestamp_).ToString(true)},
+            {"rocksdb.max_timestamp", Slice(max_timestamp_).ToString(true)}};
+  }
+
+ protected:
+  const Comparator* const cmp_;
+  std::string max_timestamp_;
+  std::string min_timestamp_;
+};
+
 }  // namespace ROCKSDB_NAMESPACE

table/block_based/block_based_table_builder.cc

@@ -514,6 +514,12 @@ struct BlockBasedTableBuilder::Rep {
         new BlockBasedTablePropertiesCollector(
             table_options.index_type, table_options.whole_key_filtering,
             moptions.prefix_extractor != nullptr));
+    const Comparator* ucmp = tbo.internal_comparator.user_comparator();
+    assert(ucmp);
+    if (ucmp->timestamp_size() > 0) {
+      table_properties_collectors.emplace_back(
+          new TimestampTablePropertiesCollector(ucmp));
+    }
     if (table_options.verify_compression) {
       for (uint32_t i = 0; i < compression_opts.parallel_threads; i++) {
         verify_ctxs[i].reset(new UncompressionContext(compression_type));