Merge pull request #13756 from mszeszko-meta/backport_13609_to_10.4

10.4.2 Patch Release

.github/workflows/pr-jobs.yml

@@ -484,7 +484,7 @@ jobs:
     runs-on:
       labels: 4-core-ubuntu
     container:
-      image: evolvedbinary/rocksjava:centos6_x64-be
+      image: evolvedbinary/rocksjava:centos7_x64-be
       options: --shm-size=16gb
     steps:
     # The docker image is intentionally based on an OS that has an older GLIBC version.
@@ -512,7 +512,7 @@ jobs:
     runs-on:
       labels: 4-core-ubuntu
     container:
-      image: evolvedbinary/rocksjava:centos6_x64-be
+      image: evolvedbinary/rocksjava:centos7_x64-be
       options: --shm-size=16gb
     steps:
     # The docker image is intentionally based on an OS that has an older GLIBC version.
@@ -607,7 +607,7 @@ jobs:
     runs-on:
       labels: 4-core-ubuntu
     container:
-      image: evolvedbinary/rocksjava:rockylinux8_x64-be
+      image: evolvedbinary/rocksjava:alpine3_x64-be
       options: --shm-size=16gb
     steps:
     - uses: actions/checkout@v4.1.0

HISTORY.md

@@ -1,12 +1,23 @@
 # Rocksdb Change Log
 > NOTE: Entries for next release do not go here. Follow instructions in `unreleased_history/README.txt`
 
+## 10.4.2 (07/09/2025)
+### Bug Fixes
+* Fix a race condition between concurrent DB::Open sharing the same SstFileManager instance.
+
+## 10.4.1 (07/01/2025)
+### Behavior Changes
+* RocksDB now triggers eligible compactions every 12 hours when periodic compaction is configured. This solves a limitation of the compaction trigger mechanism, which would only trigger compaction after specific events like flush, compaction, or SetOptions.
+### Bug Fixes
+* Fix a bug in BackupEngine that can crash backup due to a null FSWritableFile passed to WritableFileWriter.
+
 ## 10.4.0 (06/20/2025)
 ### New Features
 * Add a new CF option `memtable_avg_op_scan_flush_trigger` that supports triggering memtable flush when an iterator scans through an expensive range of keys, with the average number of skipped keys from the active memtable exceeding the threshold.
 * Vector based memtable now supports concurrent writers (DBOptions::allow_concurrent_memtable_write) #13675.
 * Add new experimental `TransactionOptions::large_txn_commit_optimize_byte_threshold` to enable optimizations for large transaction commit by transaction batch data size.
 * Add a new option `CompactionOptionsUniversal::reduce_file_locking` and if it's true, auto universal compaction picking will adjust to minimize locking of input files when bottom priority compactions are waiting to run. This can increase the likelihood of existing L0s being selected for compaction, thereby improving write stall and reducing read regression.
+* Add new `format_version=7` to aid experimental support of custom compression algorithms with CompressionManager and block-based table. This format version includes changing the format of `TableProperties::compression_name`.
 
 ### Public API Changes
 * Change NewExternalTableFactory to return a unique_ptr instead of shared_ptr.

Makefile

@@ -2147,14 +2147,14 @@ ZLIB_DOWNLOAD_BASE ?= http://zlib.net
 BZIP2_VER ?= 1.0.8
 BZIP2_SHA256 ?= ab5a03176ee106d3f0fa90e381da478ddae405918153cca248e682cd0c4a2269
 BZIP2_DOWNLOAD_BASE ?= http://sourceware.org/pub/bzip2
-SNAPPY_VER ?= 1.2.1
-SNAPPY_SHA256 ?= 736aeb64d86566d2236ddffa2865ee5d7a82d26c9016b36218fcc27ea4f09f86
+SNAPPY_VER ?= 1.2.2
+SNAPPY_SHA256 ?= 90f74bc1fbf78a6c56b3c4a082a05103b3a56bb17bca1a27e052ea11723292dc
 SNAPPY_DOWNLOAD_BASE ?= https://github.com/google/snappy/archive
-LZ4_VER ?= 1.9.4
-LZ4_SHA256 ?= 0b0e3aa07c8c063ddf40b082bdf7e37a1562bda40a0ff5272957f3e987e0e54b
+LZ4_VER ?= 1.10.0
+LZ4_SHA256 ?= 537512904744b35e232912055ccf8ec66d768639ff3abe5788d90d792ec5f48b
 LZ4_DOWNLOAD_BASE ?= https://github.com/lz4/lz4/archive
-ZSTD_VER ?= 1.5.5
-ZSTD_SHA256 ?= 98e9c3d949d1b924e28e01eccb7deed865eefebf25c2f21c702e5cd5b63b85e1
+ZSTD_VER ?= 1.5.7
+ZSTD_SHA256 ?= 37d7284556b20954e56e1ca85b80226768902e2edabd3b649e9e72c0c9012ee3
 ZSTD_DOWNLOAD_BASE ?= https://github.com/facebook/zstd/archive
 CURL_SSL_OPTS ?= --tlsv1
 
@@ -2375,27 +2375,27 @@ rocksdbjavastaticreleasedocker: rocksdbjavastaticosx rocksdbjavastaticdockerx86
 
 rocksdbjavastaticdockerx86:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_x86-be --platform linux/386 --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos6_x86-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_x86-be --platform linux/386 --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos7_x86-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockerx86_64:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_x64-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos6_x64-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_x64-be --platform linux/amd64 --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos7_x64-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockerppc64le:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_ppc64le-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos7_ppc64le-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_ppc64le-be --platform linux/ppc64le --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos7_ppc64le-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockerarm64v8:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_arm64v8-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos7_arm64v8-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_arm64v8-be --platform linux/aarch64 --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:centos7_arm64v8-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockers390x:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_s390x-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:ubuntu18_s390x-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_s390x-be --platform linux/s390x --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:ubuntu18_s390x-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockerriscv64:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_riscv64-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:ubuntu20_riscv64-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_riscv64-be --platform linux/riscv64 --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:ubuntu20_riscv64-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockerx86musl:
 	mkdir -p java/target
@@ -2403,19 +2403,19 @@ rocksdbjavastaticdockerx86musl:
 
 rocksdbjavastaticdockerx86_64musl:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_x64-musl-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_x64-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_x64-musl-be --platform linux/amd64 --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_x64-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockerppc64lemusl:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_ppc64le-musl-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_ppc64le-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_ppc64le-musl-be --platform linux/ppc64le --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_ppc64le-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockerarm64v8musl:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_arm64v8-musl-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_arm64v8-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_arm64v8-musl-be --platform linux/aarch64 --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_arm64v8-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticdockers390xmusl:
 	mkdir -p java/target
-	docker run --rm --name rocksdb_linux_s390x-musl-be --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_s390x-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
+	docker run --rm --name rocksdb_linux_s390x-musl-be --platform linux/s390x --attach stdin --attach stdout --attach stderr --volume $(HOME)/.m2:/root/.m2:ro --volume `pwd`:/rocksdb-host:ro --volume /rocksdb-local-build --volume `pwd`/java/target:/rocksdb-java-target --env DEBUG_LEVEL=$(DEBUG_LEVEL) --env J=$(J) evolvedbinary/rocksjava:alpine3_s390x-be /rocksdb-host/java/crossbuild/docker-build-linux.sh
 
 rocksdbjavastaticpublish: rocksdbjavastaticrelease rocksdbjavastaticpublishcentral
 
@@ -2470,8 +2470,8 @@ jtest_run:
 jtest: rocksdbjava
 	cd java;$(MAKE) sample test
 
-jpmd: rocksdbjava rocksdbjavageneratepom
-	cd java;$(MAKE) pmd
+jpmd: rocksdbjavageneratepom
+	cd java;$(MAKE) java java_test pmd
 
 jdb_bench:
 	cd java;$(MAKE) db_bench;

cache/compressed_secondary_cache.cc

@@ -80,15 +80,16 @@ std::unique_ptr<SecondaryCacheResultHandle> CompressedSecondaryCache::Lookup(
     ptr = reinterpret_cast<CacheAllocationPtr*>(handle_value);
     handle_value_charge = cache_->GetCharge(lru_handle);
     data_ptr = ptr->get();
-    data_ptr = GetVarint32Ptr(data_ptr, data_ptr + 1,
-                              static_cast<uint32_t*>(&type_32));
+    const char* limit = ptr->get() + handle_value_charge;
+    data_ptr =
+        GetVarint32Ptr(data_ptr, limit, static_cast<uint32_t*>(&type_32));
     type = static_cast<CompressionType>(type_32);
-    data_ptr = GetVarint32Ptr(data_ptr, data_ptr + 1,
-                              static_cast<uint32_t*>(&source_32));
+    data_ptr =
+        GetVarint32Ptr(data_ptr, limit, static_cast<uint32_t*>(&source_32));
     source = static_cast<CacheTier>(source_32);
     uint64_t data_size = 0;
-    data_ptr = GetVarint64Ptr(data_ptr, ptr->get() + handle_value_charge,
-                              static_cast<uint64_t*>(&data_size));
+    data_ptr =
+        GetVarint64Ptr(data_ptr, limit, static_cast<uint64_t*>(&data_size));
     assert(handle_value_charge > data_size);
     handle_value_charge = data_size;
   }

db/column_family.cc

@@ -110,32 +110,48 @@ void GetInternalTblPropCollFactory(
   }
 }
 
-bool CompressionSupportedWithManager(CompressionType type,
-                                     UnownedPtr<CompressionManager> mgr) {
-  return mgr ? mgr->SupportsCompressionType(type)
-             : CompressionTypeSupported(type);
+Status CheckCompressionSupportedWithManager(
+    CompressionType type, UnownedPtr<CompressionManager> mgr) {
+  if (mgr) {
+    if (!mgr->SupportsCompressionType(type)) {
+      return Status::NotSupported("Compression type " +
+                                  CompressionTypeToString(type) +
+                                  " is not recognized/supported by this "
+                                  "version of CompressionManager " +
+                                  mgr->GetId());
+    }
+  } else {
+    if (!CompressionTypeSupported(type)) {
+      if (type <= kLastBuiltinCompression) {
+        return Status::InvalidArgument("Compression type " +
+                                       CompressionTypeToString(type) +
+                                       " is not linked with the binary.");
+      } else {
+        return Status::NotSupported(
+            "Compression type " + CompressionTypeToString(type) +
+            " is not recognized/supported by built-in CompressionManager.");
+      }
+    }
+  }
+  return Status::OK();
 }
 
 Status CheckCompressionSupported(const ColumnFamilyOptions& cf_options) {
   if (!cf_options.compression_per_level.empty()) {
     for (size_t level = 0; level < cf_options.compression_per_level.size();
          ++level) {
-      if (!CompressionSupportedWithManager(
-              cf_options.compression_per_level[level],
-              cf_options.compression_manager.get())) {
-        return Status::InvalidArgument(
-            "Compression type " +
-            CompressionTypeToString(cf_options.compression_per_level[level]) +
-            " is not linked with the binary.");
+      Status s = CheckCompressionSupportedWithManager(
+          cf_options.compression_per_level[level],
+          cf_options.compression_manager.get());
+      if (!s.ok()) {
+        return s;
       }
     }
   } else {
-    if (!CompressionSupportedWithManager(
-            cf_options.compression, cf_options.compression_manager.get())) {
-      return Status::InvalidArgument(
-          "Compression type " +
-          CompressionTypeToString(cf_options.compression) +
-          " is not linked with the binary.");
+    Status s = CheckCompressionSupportedWithManager(
+        cf_options.compression, cf_options.compression_manager.get());
+    if (!s.ok()) {
+      return s;
     }
   }
   if (cf_options.compression_opts.zstd_max_train_bytes > 0) {
@@ -1617,6 +1633,8 @@ Status ColumnFamilyData::SetOptions(
   Status s = GetColumnFamilyOptionsFromMap(config_opts, cf_opts, options_map,
                                            &cf_opts);
   if (s.ok()) {
+    // FIXME: we should call SanitizeOptions() too or consolidate it with
+    // ValidateOptions().
     s = ValidateOptions(db_opts, cf_opts);
   }
   if (s.ok()) {

db/db_block_cache_test.cc

@@ -833,7 +833,6 @@ TEST_F(DBBlockCacheTest, CacheCompressionDict) {
       GetSupportedDictCompressions();
   Random rnd(301);
   // Format version before and after compression handling changes
-  TEST_AllowUnsupportedFormatVersion() = true;
   for (int format_version : {6, 7}) {
     // Test all supported compression types because (at least historically)
     // dictionary compression could be enabled and a dictionary block saved

db/db_compaction_test.cc

@@ -11093,6 +11093,59 @@ TEST_F(DBCompactionTest, RecordNewestKeyTimeForTtlCompaction) {
   ASSERT_OK(dbfull()->TEST_WaitForCompact());
   ASSERT_EQ(NumTableFilesAtLevel(0), 0);
 }
+
+class PeriodicCompactionListener : public EventListener {
+ public:
+  explicit PeriodicCompactionListener() {}
+  void OnCompactionBegin(DB* /*db*/, const CompactionJobInfo& ci) override {
+    if (ci.compaction_reason == CompactionReason::kPeriodicCompaction) {
+      ++num_periodic_compactions;
+    }
+  }
+
+  std::atomic<int> num_periodic_compactions = 0;
+};
+
+TEST_F(DBCompactionTest, PeriodicTask) {
+  // Tests that when no trigger event is fired (flush/compaction/setoptions),
+  // periodic compaction is still triggered by a scheduled periodic function.
+  auto mock_clock = std::make_shared<MockSystemClock>(env_->GetSystemClock());
+  mock_clock->SetCurrentTime(100);
+  mock_clock->InstallTimedWaitFixCallback();
+  auto mock_env = std::make_unique<CompositeEnvWrapper>(env_, mock_clock);
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::StartPeriodicTaskScheduler:Init", [&](void* arg) {
+        auto periodic_task_scheduler_ptr =
+            static_cast<PeriodicTaskScheduler*>(arg);
+        periodic_task_scheduler_ptr->TEST_OverrideTimer(mock_clock.get());
+      });
+
+  Options options;
+  options.env = mock_env.get();
+  options.compaction_style = kCompactionStyleUniversal;
+  options.statistics = CreateDBStatistics();
+  int kPeriodicCompactionSeconds = 7 * 24 * 60 * 60;  // 1 week
+  options.periodic_compaction_seconds = kPeriodicCompactionSeconds;
+  options.num_levels = 50;
+  auto listener = std::make_shared<PeriodicCompactionListener>();
+  options.listeners.push_back(listener);
+  ASSERT_OK(TryReopen(options));
+
+  Random* rnd = Random::GetTLSInstance();
+  for (int k = 0; k < 10; ++k) {
+    ASSERT_OK(Put(Key(k), rnd->RandomString(100)));
+  }
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->CompactRange({}, nullptr, nullptr));
+  ASSERT_EQ(1, NumTableFilesAtLevel(49));
+
+  dbfull()->TEST_WaitForPeriodicTaskRun(
+      [&] { mock_clock->MockSleepForSeconds(kPeriodicCompactionSeconds + 1); });
+  ASSERT_OK(db_->WaitForCompact({}));
+
+  ASSERT_EQ(listener->num_periodic_compactions, 1);
+  Close();
+}
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

db/db_impl/db_impl.cc

@@ -253,6 +253,9 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
   periodic_task_functions_.emplace(
       PeriodicTaskType::kRecordSeqnoTime,
       [this]() { this->RecordSeqnoToTimeMapping(); });
+  periodic_task_functions_.emplace(
+      PeriodicTaskType::kTriggerCompaction,
+      [this]() { this->TriggerPeriodicCompaction(); });
 
   versions_.reset(new VersionSet(
       dbname_, &immutable_db_options_, file_options_, table_cache_.get(),
@@ -787,7 +790,8 @@ Status DBImpl::StartPeriodicTaskScheduler() {
     Status s = periodic_task_scheduler_.Register(
         PeriodicTaskType::kDumpStats,
         periodic_task_functions_.at(PeriodicTaskType::kDumpStats),
-        mutable_db_options_.stats_dump_period_sec);
+        mutable_db_options_.stats_dump_period_sec,
+        /*run_immediately=*/true);
     if (!s.ok()) {
       return s;
     }
@@ -796,7 +800,8 @@ Status DBImpl::StartPeriodicTaskScheduler() {
     Status s = periodic_task_scheduler_.Register(
         PeriodicTaskType::kPersistStats,
         periodic_task_functions_.at(PeriodicTaskType::kPersistStats),
-        mutable_db_options_.stats_persist_period_sec);
+        mutable_db_options_.stats_persist_period_sec,
+        /*run_immediately=*/true);
     if (!s.ok()) {
       return s;
     }
@@ -804,7 +809,15 @@ Status DBImpl::StartPeriodicTaskScheduler() {
 
   Status s = periodic_task_scheduler_.Register(
       PeriodicTaskType::kFlushInfoLog,
-      periodic_task_functions_.at(PeriodicTaskType::kFlushInfoLog));
+      periodic_task_functions_.at(PeriodicTaskType::kFlushInfoLog),
+      /*run_immediately=*/true);
+
+  if (s.ok()) {
+    s = periodic_task_scheduler_.Register(
+        PeriodicTaskType::kTriggerCompaction,
+        periodic_task_functions_.at(PeriodicTaskType::kTriggerCompaction),
+        /*run_immediately=*/false);
+  }
 
   return s;
 }
@@ -855,7 +868,7 @@ Status DBImpl::RegisterRecordSeqnoTimeWorker() {
     s = periodic_task_scheduler_.Register(
         PeriodicTaskType::kRecordSeqnoTime,
         periodic_task_functions_.at(PeriodicTaskType::kRecordSeqnoTime),
-        seqno_time_cadence);
+        seqno_time_cadence, /*run_immediately=*/true);
   }
 
   return s;
@@ -1365,7 +1378,7 @@ Status DBImpl::SetDBOptions(
         s = periodic_task_scheduler_.Register(
             PeriodicTaskType::kDumpStats,
             periodic_task_functions_.at(PeriodicTaskType::kDumpStats),
-            new_options.stats_dump_period_sec);
+            new_options.stats_dump_period_sec, /*run_immediately=*/true);
       }
       if (new_options.max_total_wal_size !=
           mutable_db_options_.max_total_wal_size) {
@@ -1380,7 +1393,7 @@ Status DBImpl::SetDBOptions(
           s = periodic_task_scheduler_.Register(
               PeriodicTaskType::kPersistStats,
               periodic_task_functions_.at(PeriodicTaskType::kPersistStats),
-              new_options.stats_persist_period_sec);
+              new_options.stats_persist_period_sec, /*run_immediately=*/true);
         }
       }
       mutex_.Lock();
@@ -6884,6 +6897,35 @@ void DBImpl::RecordSeqnoToTimeMapping() {
   sv_context.Clean();
 }
 
+void DBImpl::TriggerPeriodicCompaction() {
+  TEST_SYNC_POINT("DBImpl::TriggerPeriodicCompaction:StartRunning");
+  {
+    InstrumentedMutexLock l(&mutex_);
+    ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                   "Running the periodic task to trigger compactions.");
+
+    for (ColumnFamilyData* cfd : *versions_->GetColumnFamilySet()) {
+      if (cfd->IsDropped()) {
+        continue;
+      }
+      if (cfd->GetLatestCFOptions().periodic_compaction_seconds &&
+          !cfd->queued_for_compaction()) {
+        cfd->current()->storage_info()->ComputeCompactionScore(
+            cfd->ioptions(), cfd->GetLatestMutableCFOptions());
+        EnqueuePendingCompaction(cfd);
+        if (cfd->queued_for_compaction()) {
+          ROCKS_LOG_INFO(immutable_db_options_.info_log,
+                         "Periodic task to trigger compaction queued Column "
+                         "family [%s] for compaction.",
+                         cfd->GetName().c_str());
+        }
+      }
+    }
+    MaybeScheduleFlushOrCompaction();
+    bg_cv_.SignalAll();
+  }
+}
+
 void DBImpl::TrackOrUntrackFiles(
     const std::vector<std::string>& existing_data_files, bool track) {
   auto sfm = static_cast_with_check<SstFileManagerImpl>(

db/db_impl/db_impl.h

@@ -1286,6 +1286,12 @@ class DBImpl : public DB {
   // For the background timer job
   void RecordSeqnoToTimeMapping();
 
+  // Compactions rely on an event triggers like flush/compaction/SetOptions.
+  // We need to trigger periodic compactions even when there is no such trigger.
+  // This function checks and schedules available compactions and will run
+  // periodically.
+  void TriggerPeriodicCompaction();
+
   // REQUIRES: DB mutex held
   std::pair<SequenceNumber, uint64_t> GetSeqnoToTimeSample() const;
 

db/db_iter.cc

@@ -511,7 +511,6 @@ bool DBIter::FindNextUserEntryInternal(bool skipping_saved_key,
 
             valid_ = true;
             return true;
-            break;
           case kTypeMerge:
             if (!PrepareValueInternal()) {
               return false;
@@ -523,7 +522,6 @@ bool DBIter::FindNextUserEntryInternal(bool skipping_saved_key,
             current_entry_is_merged_ = true;
             valid_ = true;
             return MergeValuesNewToOld();  // Go to a different state machine
-            break;
           default:
             valid_ = false;
             status_ = Status::Corruption(
@@ -1124,7 +1122,6 @@ bool DBIter::FindValueForCurrentKey() {
         }
         return true;
       }
-      break;
     case kTypeValue:
     case kTypeValuePreferredSeqno:
       SetValueAndColumnsFromPlain(pinned_value_);

db/db_test.cc

@@ -6209,6 +6209,7 @@ TEST_F(DBTest, L0L1L2AndUpHitCounter) {
 }
 
 TEST_F(DBTest, EncodeDecompressedBlockSizeTest) {
+  // Allow testing format_version=1
   bool& allow_unsupported_fv = TEST_AllowUnsupportedFormatVersion();
   SaveAndRestore guard(&allow_unsupported_fv);
   ASSERT_FALSE(allow_unsupported_fv);

db/db_test2.cc

@@ -24,6 +24,7 @@
 #include "rocksdb/persistent_cache.h"
 #include "rocksdb/trace_record.h"
 #include "rocksdb/trace_record_result.h"
+#include "rocksdb/utilities/object_registry.h"
 #include "rocksdb/utilities/replayer.h"
 #include "rocksdb/wal_filter.h"
 #include "test_util/testutil.h"
@@ -1885,8 +1886,8 @@ TEST_F(DBTest2, CompressionOptions) {
 
 TEST_F(DBTest2, RoundRobinManager) {
   if (ZSTD_Supported()) {
-    auto mgr = std::make_shared<RoundRobinManager>(
-        GetDefaultBuiltinCompressionManager());
+    auto mgr =
+        std::make_shared<RoundRobinManager>(GetBuiltinV2CompressionManager());
 
     std::vector<std::string> values;
     for (bool use_wrapper : {true}) {
@@ -1934,7 +1935,7 @@ TEST_F(DBTest2, RoundRobinManager) {
 TEST_F(DBTest2, RandomMixedCompressionManager) {
   if (ZSTD_Supported()) {
     auto mgr = std::make_shared<RandomMixedCompressionManager>(
-        GetDefaultBuiltinCompressionManager());
+        GetBuiltinV2CompressionManager());
     std::vector<std::string> values;
     for (bool use_wrapper : {true}) {
       SCOPED_TRACE((use_wrapper ? "With " : "No ") + std::string("wrapper"));
@@ -2026,7 +2027,7 @@ TEST_F(DBTest2, CompressionManagerWrapper) {
           wrapped_->GetCompressorForSST(context, opts, preferred));
     }
   };
-  auto mgr = std::make_shared<MyManager>(GetDefaultBuiltinCompressionManager());
+  auto mgr = std::make_shared<MyManager>(GetBuiltinV2CompressionManager());
 
   for (CompressionType type : GetSupportedCompressions()) {
     for (bool use_wrapper : {false, true}) {
@@ -2117,8 +2118,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     bool SupportsCompressionType(CompressionType type) const override {
       return type == kCustomCompression8A || type == kCustomCompression8B ||
              type == kCustomCompression8C ||
-             GetDefaultBuiltinCompressionManager()->SupportsCompressionType(
-                 type);
+             GetBuiltinV2CompressionManager()->SupportsCompressionType(type);
     }
 
     int used_compressor8A_count_ = 0;
@@ -2139,8 +2139,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
           return std::make_unique<Compressor8C>();
         // Also support built-in compression algorithms
         default:
-          return GetDefaultBuiltinCompressionManager()->GetCompressor(opts,
-                                                                      type);
+          return GetBuiltinV2CompressionManager()->GetCompressor(opts, type);
       }
     }
 
@@ -2148,13 +2147,14 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
       return std::make_shared<test::DecompressorCustomAlg>();
     }
 
-    CompressionType last_specific_decompressor_type_ = kNoCompression;
+    RelaxedAtomic<CompressionType> last_specific_decompressor_type_{
+        kNoCompression};
 
     std::shared_ptr<Decompressor> GetDecompressorForTypes(
         const CompressionType* types_begin,
         const CompressionType* types_end) override {
       assert(types_end > types_begin);
-      last_specific_decompressor_type_ = *types_begin;
+      last_specific_decompressor_type_.StoreRelaxed(*types_begin);
       auto decomp = std::make_shared<test::DecompressorCustomAlg>();
       decomp->SetAllowedTypes(types_begin, types_end);
       return decomp;
@@ -2191,9 +2191,11 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     // respect their distinct compatibility names and treat them as incompatible
     // (or else risk processing data incorrectly)
     // NOTE: these are not registered in ObjectRegistry to test what happens
-    // when the original CompressionManager might not be available.
+    // when the original CompressionManager might not be available, but
+    // mgr_bar will be registered during the test, with different names to
+    // prevent interference between iterations.
     auto mgr_foo = std::make_shared<MyManager>("Foo");
-    auto mgr_bar = std::make_shared<MyManager>("Bar");
+    auto mgr_bar = std::make_shared<MyManager>(use_dict ? "Bar1" : "Bar2");
 
     // And this one claims to be fully compatible with the built-in compression
     // manager when it's not fully compatible (for custom CompressionTypes)
@@ -2256,10 +2258,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     options.compression = kLZ4Compression;
     ASSERT_EQ(TryReopen(options).code(), Status::Code::kInvalidArgument);
 
-    // TODO: eliminate this hack when format_version=7 is published
-    SaveAndRestore guard(&TEST_AllowUnsupportedFormatVersion(), true);
-
-    // Set new format version
+    // Set format version supporting custom compression
     bbto.format_version = 7;
     options.table_factory.reset(NewBlockBasedTableFactory(bbto));
 
@@ -2269,8 +2268,13 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     options.compression = kCustomCompression8B;
     ASSERT_EQ(TryReopen(options).code(), Status::Code::kInvalidArgument);
 
-    // Using a built-in compression type with fv=7 but named custom schema
+    // Custom compression schema, but specifying a custom compression type it
+    // doesn't support.
     options.compression_manager = mgr_foo;
+    options.compression = kCustomCompressionF0;
+    ASSERT_EQ(TryReopen(options).code(), Status::Code::kNotSupported);
+
+    // Using a built-in compression type with fv=7 but named custom schema
     options.compression = kLZ4Compression;
     Reopen(options);
     ASSERT_OK(
@@ -2279,7 +2283,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(NumTableFilesAtLevel(0), 2);
     ASSERT_EQ(Get("b"), value);
 
-    // Verify it was compressed with snappy
+    // Verify it was compressed with LZ4
     r = {"b", "b0"};
     tables_properties.clear();
     ASSERT_OK(db_->GetPropertiesOfTablesInRange(db_->DefaultColumnFamily(), &r,
@@ -2288,7 +2292,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     // Uses new format for "compression_name" property
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;04;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kLZ4Compression);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kLZ4Compression);
 
     // Custom compression type
     options.compression = kCustomCompression8A;
@@ -2309,7 +2314,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(tables_properties.size(), 1U);
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;8A;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kCustomCompression8A);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kCustomCompression8A);
 
     // Also dynamically changeable, because the compression manager will respect
     // the current setting as reported under the legacy logic
@@ -2320,7 +2326,7 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(NumTableFilesAtLevel(0), 4);
     ASSERT_EQ(Get("d"), value);
 
-    // Verify it was compressed with snappy
+    // Verify it was compressed with LZ4
     r = {"d", "d0"};
     tables_properties.clear();
     ASSERT_OK(db_->GetPropertiesOfTablesInRange(db_->DefaultColumnFamily(), &r,
@@ -2328,7 +2334,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(tables_properties.size(), 1U);
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;04;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kLZ4Compression);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kLZ4Compression);
 
     // Dynamically changeable to custom compressions also
     ASSERT_OK(dbfull()->SetOptions({{"compression", "kCustomCompression8B"}}));
@@ -2346,7 +2353,8 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(tables_properties.size(), 1U);
     EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
     EXPECT_EQ(tables_properties.begin()->second->compression_name, "Foo;8B;");
-    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_, kCustomCompression8B);
+    EXPECT_EQ(mgr_foo->last_specific_decompressor_type_.LoadRelaxed(),
+              kCustomCompression8B);
 
     // Fails to re-open with incompatible compression manager (can't find
     // compression manager Foo because it's not registered nor known by Bar)
@@ -2365,10 +2373,50 @@ TEST_F(DBTest2, CompressionManagerCustomCompression) {
     ASSERT_EQ(Get("d").size(), kValueSize);
     ASSERT_EQ(Get("e").size(), kValueSize);
 
-    // TODO: mix of compatibility names in same DB
+    // Add a file using mgr_bar
+    ASSERT_OK(
+        Put("f", test::CompressibleString(&rnd, 0.1, kValueSize, &value)));
+    ASSERT_OK(Flush());
+    ASSERT_EQ(NumTableFilesAtLevel(0), 6);
+    ASSERT_EQ(Get("f"), value);
+
+    // Verify it was compressed appropriately
+    r = {"f", "f0"};
+    tables_properties.clear();
+    ASSERT_OK(db_->GetPropertiesOfTablesInRange(db_->DefaultColumnFamily(), &r,
+                                                1, &tables_properties));
+    ASSERT_EQ(tables_properties.size(), 1U);
+    EXPECT_LT(tables_properties.begin()->second->data_size, kValueSize / 2);
+    EXPECT_EQ(mgr_bar->last_specific_decompressor_type_.LoadRelaxed(),
+              kLZ4Compression);
+
+    // Fails to re-open with incompatible compression manager (can't find
+    // compression manager Bar because it's not registered nor known by Foo)
+    options.compression_manager = mgr_foo;
+    ASSERT_EQ(TryReopen(options).code(), Status::Code::kNotSupported);
+
+    // Register and re-open
+    auto& library = *ObjectLibrary::Default();
+    library.AddFactory<CompressionManager>(
+        mgr_bar->CompatibilityName(),
+        [mgr_bar](const std::string& /*uri*/,
+                  std::unique_ptr<CompressionManager>* guard,
+                  std::string* /*errmsg*/) {
+          *guard = std::make_unique<MyManager>(mgr_bar->CompatibilityName());
+          return guard->get();
+        });
+    Reopen(options);
+
+    // Can still read everything
+    ASSERT_EQ(Get("a").size(), kValueSize);
+    ASSERT_EQ(Get("b").size(), kValueSize);
+    ASSERT_EQ(Get("c").size(), kValueSize);
+    ASSERT_EQ(Get("d").size(), kValueSize);
+    ASSERT_EQ(Get("e").size(), kValueSize);
+    ASSERT_EQ(Get("f").size(), kValueSize);
+
     // TODO: test old version of a compression manager unable to read a
     // compression type
-    // TODO: test getting compression manager from object registry
   }
 }
 

db/periodic_task_scheduler.cc

@@ -26,6 +26,7 @@ static const std::map<PeriodicTaskType, uint64_t> kDefaultPeriodSeconds = {
     {PeriodicTaskType::kPersistStats, kInvalidPeriodSec},
     {PeriodicTaskType::kFlushInfoLog, 10},
     {PeriodicTaskType::kRecordSeqnoTime, kInvalidPeriodSec},
+    {PeriodicTaskType::kTriggerCompaction, 12 * 60 * 60}  // 12 hours
 };
 
 static const std::map<PeriodicTaskType, std::string> kPeriodicTaskTypeNames = {
@@ -33,16 +34,20 @@ static const std::map<PeriodicTaskType, std::string> kPeriodicTaskTypeNames = {
     {PeriodicTaskType::kPersistStats, "pst_st"},
     {PeriodicTaskType::kFlushInfoLog, "flush_info_log"},
     {PeriodicTaskType::kRecordSeqnoTime, "record_seq_time"},
+    {PeriodicTaskType::kTriggerCompaction, "trigger_compaction"},
 };
 
 Status PeriodicTaskScheduler::Register(PeriodicTaskType task_type,
-                                       const PeriodicTaskFunc& fn) {
-  return Register(task_type, fn, kDefaultPeriodSeconds.at(task_type));
+                                       const PeriodicTaskFunc& fn,
+                                       bool run_immediately) {
+  return Register(task_type, fn, kDefaultPeriodSeconds.at(task_type),
+                  run_immediately);
 }
 
 Status PeriodicTaskScheduler::Register(PeriodicTaskType task_type,
                                        const PeriodicTaskFunc& fn,
-                                       uint64_t repeat_period_seconds) {
+                                       uint64_t repeat_period_seconds,
+                                       bool run_immediately) {
   MutexLock l(&timer_mutex);
   static std::atomic<uint64_t> initial_delay(0);
 
@@ -65,10 +70,13 @@ Status PeriodicTaskScheduler::Register(PeriodicTaskType task_type,
   std::string unique_id =
       kPeriodicTaskTypeNames.at(task_type) + std::to_string(id_++);
 
-  bool succeeded = timer_->Add(
-      fn, unique_id,
-      (initial_delay.fetch_add(1) % repeat_period_seconds) * kMicrosInSecond,
-      repeat_period_seconds * kMicrosInSecond);
+  uint64_t initial_delay_micros =
+      (initial_delay.fetch_add(1) % repeat_period_seconds) * kMicrosInSecond;
+  if (!run_immediately) {
+    initial_delay_micros += repeat_period_seconds * kMicrosInSecond;
+  }
+  bool succeeded = timer_->Add(fn, unique_id, initial_delay_micros,
+                               repeat_period_seconds * kMicrosInSecond);
   if (!succeeded) {
     return Status::Aborted("Failed to register periodic task");
   }

db/periodic_task_scheduler.h

@@ -21,6 +21,7 @@ enum class PeriodicTaskType : uint8_t {
   kPersistStats,
   kFlushInfoLog,
   kRecordSeqnoTime,
+  kTriggerCompaction,
   kMax,
 };
 
@@ -42,13 +43,16 @@ class PeriodicTaskScheduler {
   PeriodicTaskScheduler& operator=(PeriodicTaskScheduler&&) = delete;
 
   // Register a task with its default repeat period. Thread safe call.
-  Status Register(PeriodicTaskType task_type, const PeriodicTaskFunc& fn);
+  // @param run_immediately If true, the task will run soon after it's
+  // scheduled, instead of waiting for the repeat period.
+  Status Register(PeriodicTaskType task_type, const PeriodicTaskFunc& fn,
+                  bool run_immediately);
 
   // Register a task with specified repeat period. 0 is an invalid argument
   // (kInvalidPeriodSec). To stop the task, please use Unregister().
   // Thread safe call.
   Status Register(PeriodicTaskType task_type, const PeriodicTaskFunc& fn,
-                  uint64_t repeat_period_seconds);
+                  uint64_t repeat_period_seconds, bool run_immediately);
 
   // Unregister the task. Thread safe call.
   Status Unregister(PeriodicTaskType task_type);

db/periodic_task_scheduler_test.cc

@@ -56,6 +56,12 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) {
   SyncPoint::GetInstance()->SetCallBack(
       "DBImpl::FlushInfoLog:StartRunning",
       [&](void*) { flush_info_log_counter++; });
+
+  int trigger_compaction_counter = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::TriggerPeriodicCompaction:StartRunning",
+      [&](void*) { trigger_compaction_counter++; });
+
   SyncPoint::GetInstance()->EnableProcessing();
 
   Reopen(options);
@@ -70,7 +76,7 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) {
 
   const PeriodicTaskScheduler& scheduler =
       dbfull()->TEST_GetPeriodicTaskScheduler();
-  ASSERT_EQ(3, scheduler.TEST_GetValidTaskNum());
+  ASSERT_EQ((int)PeriodicTaskType::kMax - 1, scheduler.TEST_GetValidTaskNum());
 
   ASSERT_EQ(1, dump_st_counter);
   ASSERT_EQ(1, pst_st_counter);
@@ -103,14 +109,14 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) {
   ASSERT_EQ(3, pst_st_counter);
   ASSERT_EQ(4, flush_info_log_counter);
 
-  ASSERT_EQ(1u, scheduler.TEST_GetValidTaskNum());
+  ASSERT_EQ(2u, scheduler.TEST_GetValidTaskNum());
 
   // Re-enable one task
   ASSERT_OK(dbfull()->SetDBOptions({{"stats_dump_period_sec", "5"}}));
   ASSERT_EQ(5u, dbfull()->GetDBOptions().stats_dump_period_sec);
   ASSERT_EQ(0u, dbfull()->GetDBOptions().stats_persist_period_sec);
 
-  ASSERT_EQ(2, scheduler.TEST_GetValidTaskNum());
+  ASSERT_EQ(3, scheduler.TEST_GetValidTaskNum());
 
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kPeriodSec)); });
@@ -118,6 +124,16 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) {
   ASSERT_EQ(3, pst_st_counter);
   ASSERT_EQ(5, flush_info_log_counter);
 
+  ASSERT_EQ(0, trigger_compaction_counter);
+  dbfull()->TEST_WaitForPeriodicTaskRun([&] {
+    mock_clock_->MockSleepForSeconds(static_cast<int>(12 * 60 * 60));
+  });
+  ASSERT_EQ(1, trigger_compaction_counter);
+  dbfull()->TEST_WaitForPeriodicTaskRun([&] {
+    mock_clock_->MockSleepForSeconds(static_cast<int>(12 * 60 * 60));
+  });
+  ASSERT_EQ(2, trigger_compaction_counter);
+
   Close();
 }
 
@@ -150,7 +166,9 @@ TEST_F(PeriodicTaskSchedulerTest, MultiInstances) {
   auto dbi = static_cast_with_check<DBImpl>(dbs[kInstanceNum - 1]);
 
   const PeriodicTaskScheduler& scheduler = dbi->TEST_GetPeriodicTaskScheduler();
-  ASSERT_EQ(kInstanceNum * 3, scheduler.TEST_GetValidTaskNum());
+  // kRecordSeqnoTime is not registered since the feature is not enabled
+  ASSERT_EQ(kInstanceNum * ((int)PeriodicTaskType::kMax - 1),
+            scheduler.TEST_GetValidTaskNum());
 
   int expected_run = kInstanceNum;
   dbi->TEST_WaitForPeriodicTaskRun(

db_stress_tool/db_stress_compression_manager.h

@@ -41,7 +41,7 @@ class DbStressCustomCompressionManager : public CompressionManager {
             test::CompressorCustomAlg<kCustomCompressionAC>>();
       // Also support built-in compression algorithms
       default:
-        return GetDefaultBuiltinCompressionManager()->GetCompressor(opts, type);
+        return GetBuiltinV2CompressionManager()->GetCompressor(opts, type);
     }
   }
 
@@ -59,7 +59,7 @@ class DbStressCustomCompressionManager : public CompressionManager {
 
  protected:
   std::shared_ptr<CompressionManager> default_ =
-      GetDefaultBuiltinCompressionManager();
+      GetBuiltinV2CompressionManager();
 };
 
 }  // namespace ROCKSDB_NAMESPACE

db_stress_tool/db_stress_test_base.cc

@@ -3435,15 +3435,15 @@ void StressTest::Open(SharedState* shared, bool reopen) {
     options_.compression_manager =
         std::make_shared<DbStressCustomCompressionManager>();
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "mixed")) {
-    options_.compression_manager = std::make_shared<RoundRobinManager>(
-        GetDefaultBuiltinCompressionManager());
+    options_.compression_manager =
+        std::make_shared<RoundRobinManager>(GetBuiltinV2CompressionManager());
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "randommixed")) {
     options_.compression_manager =
         std::make_shared<RandomMixedCompressionManager>(
-            GetDefaultBuiltinCompressionManager());
+            GetBuiltinV2CompressionManager());
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "autoskip")) {
     options_.compression_manager =
-        CreateAutoSkipCompressionManager(GetDefaultBuiltinCompressionManager());
+        CreateAutoSkipCompressionManager(GetBuiltinV2CompressionManager());
   } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "none")) {
     // Nothing to do using default compression manager
   } else {

env/env_posix.cc

@@ -169,8 +169,9 @@ class PosixClock : public SystemClock {
     struct timespec ts;
     clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts);
     return (static_cast<uint64_t>(ts.tv_sec) * 1000000000 + ts.tv_nsec) / 1000;
-#endif
+#else
     return 0;
+#endif
   }
 
   uint64_t CPUNanos() override {
@@ -179,8 +180,9 @@ class PosixClock : public SystemClock {
     struct timespec ts;
     clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts);
     return static_cast<uint64_t>(ts.tv_sec) * 1000000000 + ts.tv_nsec;
-#endif
+#else
     return 0;
+#endif
   }
 
   void SleepForMicroseconds(int micros) override { usleep(micros); }

env/fs_posix.cc

@@ -868,7 +868,6 @@ class PosixFileSystem : public FileSystem {
       IOOptions opts;
       return CreateDirIfMissing(*result, opts, nullptr);
     }
-    return IOStatus::OK();
   }
 
   IOStatus GetFreeSpace(const std::string& fname, const IOOptions& /*opts*/,

file/sst_file_manager_impl.h

@@ -162,7 +162,6 @@ class SstFileManagerImpl : public SstFileManager {
   void Close();
 
   void SetStatisticsPtr(const std::shared_ptr<Statistics>& stats) override {
-    stats_ = stats;
     delete_scheduler_.SetStatisticsPtr(stats);
   }
 
@@ -216,7 +215,6 @@ class SstFileManagerImpl : public SstFileManager {
   std::list<ErrorHandler*> error_handler_list_;
   // Pointer to ErrorHandler instance that is currently processing recovery
   ErrorHandler* cur_instance_;
-  std::shared_ptr<Statistics> stats_;
 };
 
 }  // namespace ROCKSDB_NAMESPACE

include/rocksdb/advanced_compression.h

@@ -371,8 +371,9 @@ class CompressionManager
                                  const std::string& id,
                                  std::shared_ptr<CompressionManager>* result);
 
-  // Will this compression type be used if requested in calling
-  // GetCompressor/GetCompressorForSST?
+  // Returns false iff a configuration that would pass the given compression
+  // type to GetCompressor/GetCompressorForSST should be rejected (not
+  // supported)
   virtual bool SupportsCompressionType(CompressionType type) const = 0;
 
   // TODO: function to check compatibility with or sanitize CompressionOptions
@@ -577,11 +578,17 @@ class CompressionManagerWrapper : public CompressionManager {
   std::shared_ptr<CompressionManager> wrapped_;
 };
 
-// Compression manager that implements built-in compression strategy. The
-// behavior of compression_manager=nullptr is essentially equivalent to
-// using this compression manager.
-const std::shared_ptr<CompressionManager>&
-GetDefaultBuiltinCompressionManager();
+// Compression manager that implements the second schema for RocksDB built-in
+// compression support. (The first schema is intentionally not provided here.)
+// *** CURRENT STATE ***
+// This is currently the latest schema for built-in compression, and the
+// compression manager used when compression_manager=nullptr.
+const std::shared_ptr<CompressionManager>& GetBuiltinV2CompressionManager();
+
+// NOTE: No GetLatestBuiltinCompressionManager() is provided because that could
+// lead to unexpected schema changes for user CompressionManagers building on
+// the built-in schema, in the unlikely/rare case of a new built-in schema.
+
 // Gets CompressionManager designed for the automated compression strategy.
 // This may include deciding to compress or not.
 // In future should be able to select compression algorithm based on the CPU

include/rocksdb/compression_type.h

@@ -32,6 +32,7 @@ enum CompressionType : unsigned char {
 
   // For use by user custom CompressionManagers
   kCustomCompression80 = 0x80,
+  kFirstCustomCompression = kCustomCompression80,
   kCustomCompression81 = 0x81,
   kCustomCompression82 = 0x82,
   kCustomCompression83 = 0x83,
@@ -158,6 +159,7 @@ enum CompressionType : unsigned char {
   kCustomCompressionFC = 0xFC,
   kCustomCompressionFD = 0xFD,
   kCustomCompressionFE = 0xFE,
+  kLastCustomCompression = kCustomCompressionFE,
 
   // kDisableCompressionOption is used to disable some compression options.
   kDisableCompressionOption = 0xff,

include/rocksdb/table.h

@@ -561,6 +561,10 @@ struct BlockBasedTableOptions {
   // misplaced within or between files is as likely to fail checksum
   // verification as random corruption. Also checksum-protects SST footer.
   // Can be read by RocksDB versions >= 8.6.0.
+  // 7 -- Support for custom compression algorithms with a CompressionManager
+  // using a non-built-in CompatibilityName(). See `compression_manager` in
+  // ColumnFamilyOptions. Also changes the format of TableProperties field
+  // `compression_name`. Can be read by RocksDB versions >= 10.4.0.
   //
   // Using the default setting of format_version is strongly recommended, so
   // that available enhancements are adopted eventually and automatically. The

include/rocksdb/table_properties.h

@@ -355,7 +355,13 @@ struct TableProperties {
   // behave, this must be set to "ZSTD" if any blocks are compressed
   // with zstd and must NOT be set to "NoCompression" if any blocks are
   // compressed.
-  // * For format_version >= 7, it is ...
+  // * For format_version >= 7, the format is
+  //   <compatibility_name>;<hex-coded compression types>;<future use>
+  // where <compatibility_name> is the CompatibilityName() of the
+  // CompressionManager used for the file, or empty if compression was
+  // disabled; <hex-coded compression types> represents a sorted set of
+  // CompressionType values used in the file other than kNoCompression, each
+  // as 2-digit hex, e.g. 04 for LZ$, 07 for ZSTD, etc.
   std::string compression_name;
 
   // Compression options used to compress the SST files.

include/rocksdb/version.h

@@ -12,8 +12,8 @@
 // NOTE: in 'main' development branch, this should be the *next*
 // minor or major version number planned for release.
 #define ROCKSDB_MAJOR 10
-#define ROCKSDB_MINOR 5
-#define ROCKSDB_PATCH 0
+#define ROCKSDB_MINOR 4
+#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

java/src/main/java/org/rocksdb/NativeLibraryLoader.java

@@ -30,6 +30,14 @@ public class NativeLibraryLoader {
   private static final String tempFilePrefix = "librocksdbjni";
   private static final String tempFileSuffix = Environment.getJniLibraryExtension();
 
+  /**
+   * If you set the System Property ROCKS_JAVA_DEBUG_NLL can be to true
+   * messages about attempts to load the native library will be printed
+   * to std out.
+   */
+  private static boolean DEBUG_LOADING =
+      "true".equals(System.getProperty("ROCKS_JAVA_DEBUG_NLL", "false"));
+
   /**
    * Get a reference to the NativeLibraryLoader
    *
@@ -55,7 +63,7 @@ public class NativeLibraryLoader {
    *
    * @throws java.io.IOException if a filesystem operation fails.
    */
-  @SuppressWarnings("PMD.EmptyCatchBlock")
+  @SuppressWarnings({"PMD.EmptyCatchBlock", "PMD.SystemPrintln"})
   public synchronized void loadLibrary(final String tmpDir) throws IOException {
     try {
       // try dynamic library
@@ -63,6 +71,9 @@ public class NativeLibraryLoader {
       return;
     } catch (final UnsatisfiedLinkError ule) {
       // ignore - try from static library
+      if (DEBUG_LOADING) {
+        System.out.println("Unable to load shared dynamic library: " + sharedLibraryName);
+      }
     }
 
     try {
@@ -71,6 +82,9 @@ public class NativeLibraryLoader {
       return;
     } catch (final UnsatisfiedLinkError ule) {
       // ignore - then try static library fallback or from jar
+      if (DEBUG_LOADING) {
+        System.out.println("Unable to load shared static library: " + jniLibraryName);
+      }
     }
 
     if (fallbackJniLibraryName != null) {
@@ -80,6 +94,10 @@ public class NativeLibraryLoader {
         return;
       } catch (final UnsatisfiedLinkError ule) {
         // ignore - then try from jar
+        if (DEBUG_LOADING) {
+          System.out.println(
+              "Unable to load shared static fallback library: " + fallbackJniLibraryName);
+        }
       }
     }
 
@@ -137,18 +155,23 @@ public class NativeLibraryLoader {
     }
   }
 
-  @SuppressWarnings({"PMD.UseProperClassLoader", "PMD.UseTryWithResources"})
+  @SuppressWarnings({"PMD.UseProperClassLoader", "PMD.UseTryWithResources", "PMD.SystemPrintln"})
   File loadLibraryFromJarToTemp(final String tmpDir) throws IOException {
     try (InputStream is = getClass().getClassLoader().getResourceAsStream(jniLibraryFileName)) {
       if (is != null) {
         final File temp = createTemp(tmpDir, jniLibraryFileName);
         Files.copy(is, temp.toPath(), StandardCopyOption.REPLACE_EXISTING);
         return temp;
+      } else {
+        if (DEBUG_LOADING) {
+          System.out.println("Unable to find: " + jniLibraryFileName + " on the classpath");
+        }
       }
     }
 
     if (fallbackJniLibraryFileName == null) {
-      throw new RuntimeException(fallbackJniLibraryFileName + " was not found inside JAR.");
+      throw new RuntimeException(
+          jniLibraryFileName + " was not found inside JAR, and there is no fallback.");
     }
 
     try (InputStream is =
@@ -157,10 +180,16 @@ public class NativeLibraryLoader {
         final File temp = createTemp(tmpDir, fallbackJniLibraryFileName);
         Files.copy(is, temp.toPath(), StandardCopyOption.REPLACE_EXISTING);
         return temp;
+      } else {
+        if (DEBUG_LOADING) {
+          System.out.println(
+              "Unable to find fallback: " + fallbackJniLibraryFileName + " on the classpath");
+        }
       }
     }
 
-    throw new RuntimeException(jniLibraryFileName + " was not found inside JAR.");
+    throw new RuntimeException("Neither " + jniLibraryFileName + " or " + fallbackJniLibraryFileName
+        + " were found inside the JAR, and there is no fallback.");
   }
 
   /**

java/src/main/java/org/rocksdb/RocksDB.java

@@ -84,13 +84,7 @@ public class RocksDB extends RocksObject {
       return;
     }
 
-    while (libraryLoaded.get() == LibraryState.LOADING) {
-      try {
-        Thread.sleep(10);
-      } catch(final InterruptedException e) {
-        //ignore
-      }
-    }
+    waitForLibraryToBeLoaded();
   }
 
   /**
@@ -146,12 +140,28 @@ public class RocksDB extends RocksObject {
       return;
     }
 
-    while (libraryLoaded.get() == LibraryState.LOADING) {
-      try {
-        Thread.sleep(10);
-      } catch(final InterruptedException e) {
-        //ignore
+    waitForLibraryToBeLoaded();
+  }
+
+  private static void waitForLibraryToBeLoaded() {
+    final long wait = 10; // Time to wait before re-checking if another thread loaded the library
+    final long timeout =
+        10 * 1000; // Maximum time to wait for another thread to load the library (10 seconds)
+    long waited = 0;
+    try {
+      while (libraryLoaded.get() == LibraryState.LOADING) {
+        Thread.sleep(wait);
+        waited += wait;
+
+        if (waited >= timeout) {
+          throw new RuntimeException(
+              "Exceeded timeout whilst trying to load the RocksDB shared library");
+        }
       }
+    } catch (final InterruptedException e) {
+      // restore interrupted status
+      Thread.currentThread().interrupt();
+      throw new RuntimeException("Interrupted whilst trying to load the RocksDB shared library", e);
     }
   }
 

options/options_settable_test.cc

@@ -705,10 +705,9 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       12345);
   // TODO: try to enhance ObjectLibrary to support singletons
   // ASSERT_EQ(new_options->compression_manager,
-  //           GetBuiltinCompressionManager(/*compression_format_version*/ 2));
-  ASSERT_STREQ(
-      new_options->compression_manager->Name(),
-      GetBuiltinCompressionManager(/*compression_format_version*/ 2)->Name());
+  //           GetBuiltinV2CompressionManager());
+  ASSERT_STREQ(new_options->compression_manager->Name(),
+               GetBuiltinV2CompressionManager()->Name());
 
   ColumnFamilyOptions rnd_filled_options = *new_options;
 

port/port_posix.cc

@@ -220,8 +220,9 @@ int GetMaxOpenFiles() {
     return std::numeric_limits<int>::max();
   }
   return static_cast<int>(no_files_limit.rlim_cur);
-#endif
+#else
   return -1;
+#endif
 }
 
 void* cacheline_aligned_alloc(size_t size) {

table/block_based/block_based_table_builder.cc

@@ -977,7 +977,10 @@ struct BlockBasedTableBuilder::Rep {
       assert(mgr);
       // Use newer compression_name property
       props.compression_name.reserve(32);
-      props.compression_name.append(mgr->CompatibilityName());
+      // If compression is disabled, use empty manager name
+      if (basic_compressor) {
+        props.compression_name.append(mgr->CompatibilityName());
+      }
       props.compression_name.push_back(';');
       // Rest of property to be filled out at the end of building the file
     } else {

table/block_based/block_based_table_reader.cc

@@ -606,7 +606,8 @@ Status GetDecompressor(const std::string& compression_name,
       if (!s.ok()) {
         return s;
       }
-      assert(mgr_to_use);
+      assert(mgr_to_use || compatibility_name == kNullptrString ||
+             compatibility_name.empty());
     }
 
     // Second field is set of compression types actually used in the file
@@ -632,9 +633,15 @@ Status GetDecompressor(const std::string& compression_name,
       }
       ctypes[i] = static_cast<CompressionType>(val);
     }
-    *out_decompressor =
-        mgr_to_use->GetDecompressorForTypes(ctypes.get(), ctypes.get() + count);
-    assert(*out_decompressor || count == 0);
+    if (mgr_to_use) {
+      *out_decompressor = mgr_to_use->GetDecompressorForTypes(
+          ctypes.get(), ctypes.get() + count);
+      assert(*out_decompressor || count == 0);
+    } else {
+      // Compression/decompression disabled
+      *out_decompressor = nullptr;
+      assert(count == 0);
+    }
     // Can ignore possible additional future fields
   } else {
     // No explicit CompressionManager, e.g. legacy file support where

table/format.h

@@ -157,10 +157,15 @@ inline uint32_t GetCompressFormatForVersion(uint32_t format_version) {
   // As of format_version 2, we encode compressed block with
   // compress_format_version == 2. Before that, the version is 1.
   // DO NOT CHANGE THIS FUNCTION, it affects disk format
+  // As of format_version 7 and opening up to custom compression, the
+  // compression format version is essentially independent of the block-based
+  // table format version, and encoded in the compression_name table property.
+  // Thus, this function can go away once we remove support for reading
+  // format_version=1.
   return format_version >= 2 ? 2 : 1;
 }
 
-constexpr uint32_t kLatestFormatVersion = 6;
+constexpr uint32_t kLatestFormatVersion = 7;
 
 inline bool IsSupportedFormatVersion(uint32_t version) {
   return version <= kLatestFormatVersion;

table/table_test.cc

@@ -1795,18 +1795,23 @@ TEST_P(BlockBasedTableTest, IndexUncompressed) {
 #endif  // SNAPPY
 
 TEST_P(BlockBasedTableTest, BlockBasedTableProperties2) {
-  TableConstructor c(&reverse_key_comparator);
+  TableConstructor c(&reverse_key_comparator,
+                     true /* convert_to_internal_key_ */);
   std::vector<std::string> keys;
   stl_wrappers::KVMap kvmap;
 
-  {
+  for (CompressionType ct : {kNoCompression, kSnappyCompression}) {
+    if (!Snappy_Supported() && ct == kSnappyCompression) {
+      continue;
+    }
     Options options;
-    options.compression = CompressionType::kNoCompression;
+    options.compression = ct;
     BlockBasedTableOptions table_options = GetBlockBasedTableOptions();
     options.table_factory.reset(NewBlockBasedTableFactory(table_options));
 
     const ImmutableOptions ioptions(options);
     const MutableCFOptions moptions(options);
+    c.Add("blah", std::string(200, 'x'));  // something to compress
     c.Finish(options, ioptions, moptions, table_options,
              GetPlainInternalComparator(options.comparator), &keys, &kvmap);
 
@@ -1823,7 +1828,13 @@ TEST_P(BlockBasedTableTest, BlockBasedTableProperties2) {
     // No filter policy is used
     ASSERT_EQ("", props.filter_policy_name);
     // Compression type == that set:
-    ASSERT_EQ("NoCompression", props.compression_name);
+    if (FormatVersionUsesCompressionManagerName(table_options.format_version)) {
+      ASSERT_EQ(ct == kNoCompression ? ";;" : "BuiltinV2;01;",
+                props.compression_name);
+    } else {
+      ASSERT_EQ(ct == kNoCompression ? "NoCompression" : "Snappy",
+                props.compression_name);
+    }
     c.ResetTableReader();
   }
 

test_util/testutil.h

@@ -740,9 +740,9 @@ template <CompressionType kCompression>
 struct CompressorCustomAlg : public CompressorWrapper {
   static bool Supported() { return LZ4_Supported(); }
 
-  explicit CompressorCustomAlg(std::unique_ptr<Compressor> wrapped =
-                                   GetDefaultBuiltinCompressionManager()
-                                       ->GetCompressor({}, kLZ4Compression))
+  explicit CompressorCustomAlg(
+      std::unique_ptr<Compressor> wrapped =
+          GetBuiltinV2CompressionManager()->GetCompressor({}, kLZ4Compression))
       : CompressorWrapper(std::move(wrapped)),
         dictionary_hash_(GetSliceHash(wrapped_->GetSerializedDict())) {
     static_assert(kCompression > kLastBuiltinCompression);
@@ -750,12 +750,16 @@ struct CompressorCustomAlg : public CompressorWrapper {
 
   const char* Name() const override { return "CompressorCustomAlg"; }
 
+  CompressionType GetPreferredCompressionType() const override {
+    return kCompression;
+  }
+
   Status CompressBlock(Slice uncompressed_data, std::string* compressed_output,
                        CompressionType* out_compression_type,
                        ManagedWorkingArea* working_area) override {
     Status s = wrapped_->CompressBlock(uncompressed_data, compressed_output,
                                        out_compression_type, working_area);
-    if (*out_compression_type != kNoCompression) {
+    if (s.ok() && *out_compression_type != kNoCompression) {
       assert(*out_compression_type == kLZ4Compression);
       std::string header(/*size=*/5, 0);
       header[0] = lossless_cast<char>(kCompression);
@@ -783,9 +787,8 @@ struct CompressorCustomAlg : public CompressorWrapper {
 struct DecompressorCustomAlg : public DecompressorWrapper {
   using TypeSet = SmallEnumSet<CompressionType, kDisableCompressionOption>;
 
-  DecompressorCustomAlg(
-      std::shared_ptr<Decompressor> wrapped =
-          GetDefaultBuiltinCompressionManager()->GetDecompressor())
+  DecompressorCustomAlg(std::shared_ptr<Decompressor> wrapped =
+                            GetBuiltinV2CompressionManager()->GetDecompressor())
       : DecompressorWrapper(std::move(wrapped)),
         dictionary_hash_(GetSliceHash(wrapped_->GetSerializedDict())),
         allowed_types_(TypeSet::All()) {}
@@ -808,7 +811,8 @@ struct DecompressorCustomAlg : public DecompressorWrapper {
   }
 
   Status ExtractUncompressedSize(Args& args) override {
-    if (args.compression_type > kLastBuiltinCompression) {
+    if (args.compression_type >= kFirstCustomCompression &&
+        args.compression_type <= kLastCustomCompression) {
       assert(args.compressed_data.size() > 0);
       assert(args.compressed_data[0] ==
              lossless_cast<char>(args.compression_type));
@@ -828,7 +832,8 @@ struct DecompressorCustomAlg : public DecompressorWrapper {
   }
 
   Status DecompressBlock(const Args& args, char* uncompressed_output) override {
-    if (args.compression_type > kLastBuiltinCompression) {
+    if (args.compression_type >= kFirstCustomCompression &&
+        args.compression_type <= kLastCustomCompression) {
       // Also allowed to copy args and modify
       Args modified_args = args;
       modified_args.compression_type = kLZ4Compression;

tools/check_format_compatible.sh

@@ -137,7 +137,7 @@ EOF
 
 # To check for DB forward compatibility with loading options (old version
 # reading data from new), as well as backward compatibility
-declare -a db_forward_with_options_refs=("8.6.fb" "8.7.fb" "8.8.fb" "8.9.fb" "8.10.fb" "8.11.fb" "9.0.fb" "9.1.fb" "9.2.fb" "9.3.fb" "9.4.fb" "9.5.fb" "9.6.fb" "9.7.fb" "9.8.fb" "9.9.fb" "9.10.fb" "9.11.fb" "10.0.fb" "10.1.fb" "10.2.fb" "10.3.fb" "10.4.fb")
+declare -a db_forward_with_options_refs=("8.6.fb" "8.7.fb" "8.8.fb" "8.9.fb" "8.10.fb" "8.11.fb" "9.0.fb" "9.1.fb" "9.2.fb" "9.3.fb" "9.4.fb" "9.5.fb" "9.6.fb" "9.7.fb" "9.8.fb" "9.9.fb" "9.10.fb" "9.11.fb" "10.0.fb" "10.1.fb" "10.2.fb" "10.3.fb")
 # To check for DB forward compatibility without loading options (in addition
 # to the "with loading options" set), as well as backward compatibility
 declare -a db_forward_no_options_refs=() # N/A at the moment

tools/db_bench_tool.cc

@@ -4653,7 +4653,7 @@ class Benchmark {
         options.bottommost_compression = kZSTD;
 
         mgr = std::make_shared<RoundRobinManager>(
-            GetDefaultBuiltinCompressionManager());
+            GetBuiltinV2CompressionManager());
       } else if (!strcasecmp(FLAGS_compression_manager.c_str(), "autoskip")) {
         options.compression = FLAGS_compression_type_e;
         if (FLAGS_compression_type_e == kNoCompression) {
@@ -4662,8 +4662,8 @@ class Benchmark {
                   "autoskip");
           ErrorExit();
         }
-        mgr = CreateAutoSkipCompressionManager(
-            GetDefaultBuiltinCompressionManager());
+        mgr =
+            CreateAutoSkipCompressionManager(GetBuiltinV2CompressionManager());
       } else {
         // not defined -> exit with error
         fprintf(stderr, "Requested compression manager not supported");

tools/ldb_cmd.cc

@@ -870,8 +870,8 @@ bool LDBCommand::ParseCompressionTypeOption(
       }
       options_.compression = kZSTD;
       options_.bottommost_compression = kZSTD;
-      auto mgr = std::make_shared<RoundRobinManager>(
-          GetDefaultBuiltinCompressionManager());
+      auto mgr =
+          std::make_shared<RoundRobinManager>(GetBuiltinV2CompressionManager());
       options_.compression_manager = mgr;
 
       // Need to list zstd in the compression_name table property if it's

util/auto_skip_compressor.cc

@@ -126,6 +126,6 @@ std::unique_ptr<Compressor> AutoSkipCompressorManager::GetCompressorForSST(
 std::shared_ptr<CompressionManagerWrapper> CreateAutoSkipCompressionManager(
     std::shared_ptr<CompressionManager> wrapped) {
   return std::make_shared<AutoSkipCompressorManager>(
-      wrapped == nullptr ? GetDefaultBuiltinCompressionManager() : wrapped);
+      wrapped == nullptr ? GetBuiltinV2CompressionManager() : wrapped);
 }
 }  // namespace ROCKSDB_NAMESPACE

util/compression.cc

@@ -1043,8 +1043,7 @@ const std::shared_ptr<CompressionManager>& GetBuiltinCompressionManager(
   }
 }
 
-const std::shared_ptr<CompressionManager>&
-GetDefaultBuiltinCompressionManager() {
+const std::shared_ptr<CompressionManager>& GetBuiltinV2CompressionManager() {
   return GetBuiltinCompressionManager(2);
 }
 

util/compression.h

@@ -754,8 +754,11 @@ inline std::string CompressionTypeToString(CompressionType compression_type) {
     case kDisableCompressionOption:
       return "DisableOption";
     default: {
-      char c = lossless_cast<char>(compression_type);
-      return "Custom" + Slice(&c, 1).ToString(/*hex=*/true);
+      bool is_custom = compression_type >= kFirstCustomCompression &&
+                       compression_type <= kLastCustomCompression;
+      unsigned char c = lossless_cast<unsigned char>(compression_type);
+      return (is_custom ? "Custom" : "Reserved") +
+             ToBaseCharsString<16>(2, c, /*uppercase=*/true);
     }
   }
 }

util/compression_test.cc

@@ -158,7 +158,7 @@ class DBAutoSkip : public DBTestBase {
         rnd_(231),
         key_index_(0) {
     options.compression_manager =
-        CreateAutoSkipCompressionManager(GetDefaultBuiltinCompressionManager());
+        CreateAutoSkipCompressionManager(GetBuiltinV2CompressionManager());
     auto statistics = ROCKSDB_NAMESPACE::CreateDBStatistics();
     options.statistics = statistics;
     options.statistics->set_stats_level(StatsLevel::kExceptTimeForMutex);

util/simple_mixed_compressor.cc

@@ -17,7 +17,8 @@ namespace ROCKSDB_NAMESPACE {
 // MultiCompressorWrapper implementation
 MultiCompressorWrapper::MultiCompressorWrapper(const CompressionOptions& opts,
                                                CompressionDict&& dict) {
-  auto builtInManager = GetDefaultBuiltinCompressionManager();
+  // TODO: make the compression manager a field
+  auto builtInManager = GetBuiltinV2CompressionManager();
   const auto& compressions = GetSupportedCompressions();
   for (auto type : compressions) {
     if (type == kNoCompression) {

util/slice_test.cc

@@ -14,6 +14,7 @@
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
 #include "util/cast_util.h"
+#include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
 
@@ -410,6 +411,17 @@ TEST(UnownedPtrTest, Tests) {
   }
 }
 
+TEST(ToBaseCharsStringTest, Tests) {
+  using ROCKSDB_NAMESPACE::ToBaseCharsString;
+  // Base 16
+  ASSERT_EQ(ToBaseCharsString<16>(5, 0, true), "00000");
+  ASSERT_EQ(ToBaseCharsString<16>(5, 42, true), "0002A");
+  ASSERT_EQ(ToBaseCharsString<16>(5, 42, false), "0002a");
+  ASSERT_EQ(ToBaseCharsString<16>(2, 255, false), "ff");
+  // Base 32
+  ASSERT_EQ(ToBaseCharsString<32>(2, 255, false), "7v");
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {

util/string_util.h

@@ -40,6 +40,16 @@ inline void PutBaseChars(char** buf, size_t n, uint64_t v, bool uppercase) {
   *buf += n;
 }
 
+// Construct a string of n digits from v in base kBase
+template <size_t kBase>
+inline std::string ToBaseCharsString(size_t n, uint64_t v, bool uppercase) {
+  std::string result;
+  result.resize(n);
+  char* buf = &result[0];
+  PutBaseChars<kBase>(&buf, n, v, uppercase);
+  return result;
+}
+
 // Parse n digits from *buf in base kBase to *v and advance *buf to the
 // position after what was read. On success, true is returned. On failure,
 // false is returned, *buf is placed at the first bad character, and *v

utilities/backup/backup_engine.cc

@@ -2372,7 +2372,11 @@ IOStatus BackupEngineImpl::CopyOrCreateFile(
 
   io_s = dst_env->GetFileSystem()->NewWritableFile(dst, dst_file_options,
                                                    &dst_file, nullptr);
-  if (io_s.ok() && !src.empty()) {
+  if (!io_s.ok()) {
+    return io_s;
+  }
+
+  if (!src.empty()) {
     auto src_file_options = FileOptions(src_env_options);
     src_file_options.temperature = *src_temperature;
     io_s = src_env->GetFileSystem()->NewSequentialFile(src, src_file_options,