From 7c80a5ac85f69822948c413cc435515889b60205 Mon Sep 17 00:00:00 2001 From: Hui Xiao Date: Thu, 9 Jul 2026 10:37:35 -0700 Subject: [PATCH] Size-tolerant deserialization for compaction service response's counts array (#14807) Summary: Pull Request resolved: https://github.com/facebook/rocksdb/pull/14807 The "counts" array in CompactionServiceResult (one compaction count per CompactionReason) is non-critical information. During cross-process remote compaction the primary and the remote worker can run different RocksDB versions and therefore disagree on kNumOfReasons, so the serialized counts array can have a different length than the reader expects. The plain OptionTypeInfo::Array parser rejects any element-count mismatch, which fails deserialization of the entire result over this non-critical field. This makes deserialization of "counts" size-tolerant via SizeTolerantDeserializeArray, which overrides only the parse callback: extra trailing elements are dropped and missing trailing elements are zero-filled, so a length mismatch no longer fails deserialization. Serialization and comparison are inherited from Array unchanged. The serialized width is intentionally kept at kNumOfReasons - 1 here, so this change does not alter what is written -- only the read path becomes tolerant. Widening the serialized width to the full kNumOfReasons is done as a separate follow-up change, which must not land until this tolerant read has been deployed to every primary (including rollback targets), so that no reader still rejects a wider result. == Test plan == **Legend** CompactionReason is an enum whose last entry is kNumOfReasons (currently 21). CompactionStats::counts is an int[kNumOfReasons] holding one count per reason; it is serialized into the remote compaction result and read back by the primary. - "narrow" = kNumOfReasons - 1 counts (omits the newest reason); what current code writes. - "wide" = kNumOfReasons counts (includes the newest reason). - strict reader: accepts only its exact expected element count; any other count fails deserialization (the plain OptionTypeInfo::Array behavior). - tolerant reader: accepts any count -- more elements than expected are dropped, fewer are zero-filled; never fails on a count mismatch (this change, SizeTolerantDeserializeArray). **Unit test** CompactionServiceTest.CompatCheckCountsWidthTolerance -- verifies that deserializing "counts" tolerates a serialized element count equal to, one greater than, or one less than the reader's expected width (extras dropped, missing zero-filled), and that the binary serializes exactly the expected number of elements. **Manual cross-version verification (ldb remote_compaction)** Test 1: baseline (unchanged behavior) -- expect-narrow strict-read primary reads narrow counts written by a worker ``` $ strict_narrow remote_compaction_worker --db=$DB --job_dir=$JOB & $ strict_narrow remote_compaction_primary --db=$DB --job_dir=$JOB remote_compaction_worker: OK (4913 bytes result) remote_compaction_primary: OK $ grep 'remote compaction result' $DB/LOG [default] [JOB 3] Received remote compaction result, output path: /tmp/cs_run/output, files: 000022.sst ``` => PASS: the primary parsed and installed the remote result. Test 2: expect-narrow tolerant-read primary reads the narrow counts a worker writes ``` $ strict_narrow remote_compaction_worker --db=$DB --job_dir=$JOB & $ tolerant_narrow remote_compaction_primary --db=$DB --job_dir=$JOB remote_compaction_worker: OK (4912 bytes result) remote_compaction_primary: OK $ grep 'remote compaction result' $DB/LOG [default] [JOB 3] Received remote compaction result, output path: /tmp/cs_run/output, files: 000022.sst ``` => PASS: the primary parsed and installed the remote result. Test 3: expect-narrow tolerant-read primary reads the wide counts a worker writes (extra reason dropped) ``` $ tolerant_wide remote_compaction_worker --db=$DB --job_dir=$JOB & $ tolerant_narrow remote_compaction_primary --db=$DB --job_dir=$JOB remote_compaction_worker: OK (4915 bytes result) remote_compaction_primary: OK $ grep 'remote compaction result' $DB/LOG [default] [JOB 3] Received remote compaction result, output path: /tmp/cs_run/output, files: 000022.sst ``` => PASS: the primary parsed and installed the remote result. Test 4: expect-wide, tolerant primary reads the narrow counts a worker writes (missing reason zero-filled) ``` $ strict_narrow remote_compaction_worker --db=$DB --job_dir=$JOB & $ tolerant_wide remote_compaction_primary --db=$DB --job_dir=$JOB remote_compaction_worker: OK (4911 bytes result) remote_compaction_primary: OK $ grep 'remote compaction result' $DB/LOG [default] [JOB 3] Received remote compaction result, output path: /tmp/cs_run/output, files: 000022.sst ``` => PASS: the primary parsed and installed the remote result. Test 5: expect-wide, tolerant primary reads a wide counts a worker write ``` $ tolerant_wide remote_compaction_worker --db=$DB --job_dir=$JOB & $ tolerant_wide remote_compaction_primary --db=$DB --job_dir=$JOB remote_compaction_worker: OK (4917 bytes result) remote_compaction_primary: OK $ grep 'remote compaction result' $DB/LOG [default] [JOB 3] Received remote compaction result, output path: /tmp/cs_run/output, files: 000022.sst ``` => PASS: the primary parsed and installed the remote result. Reviewed By: jaykorean Differential Revision: D106321320 fbshipit-source-id: fc1bd66a4d11ac6bf38dff6aa214ebafab75f3d0 --- db/compaction/compaction_service_job.cc | 52 ++++++++- db/compaction/compaction_service_test.cc | 129 +++++++++++++++++++++++ 2 files changed, 176 insertions(+), 5 deletions(-) diff --git a/db/compaction/compaction_service_job.cc b/db/compaction/compaction_service_job.cc index 9ff2bd8d16..40709c917d 100644 --- a/db/compaction/compaction_service_job.cc +++ b/db/compaction/compaction_service_job.cc @@ -777,6 +777,40 @@ static std::unordered_map OptionTypeFlags::kNone}}, }; +namespace { + +// Like OptionTypeInfo::Array but deserialization tolerates +// size mismatches: extra elements beyond kSize are silently ignored, +// missing elements are zero-filled. Serialization and comparison are +// inherited from Array unchanged; only the parse +// (deserialize) callback is overridden. +template +OptionTypeInfo SizeTolerantDeserializeArray(int offset, + OptionVerificationType verification, + OptionTypeFlags flags, + const OptionTypeInfo& elem_info, + char separator = ':') { + return OptionTypeInfo::Array(offset, verification, flags, elem_info, + separator) + .SetParseFunc([elem_info, separator]( + const ConfigOptions& opts, const std::string& name, + const std::string& value, void* addr) { + std::vector parsed; + Status s = + ParseVector(opts, elem_info, separator, name, value, &parsed); + if (!s.ok()) { + return Status(std::move(s)); + } + auto* arr = static_cast*>(addr); + for (size_t i = 0; i < kSize; ++i) { + (*arr)[i] = (i < parsed.size()) ? parsed[i] : T{}; + } + return Status::OK(); + }); +} + +} // namespace + static std::unordered_map compaction_stats_type_info = { {"micros", @@ -868,12 +902,20 @@ static std::unordered_map {offsetof(struct InternalStats::CompactionStats, count), OptionType::kUInt64T, OptionVerificationType::kNormal, OptionTypeFlags::kNone}}, + // The remote worker may serialize a different number of + // CompactionReason counts than this primary expects (RocksDB versions + // can disagree on kNumOfReasons). Deserialization tolerates the + // mismatch: extra trailing elements are dropped, missing ones are + // zero-filled. + // + // The serialized width is intentionally kept at kNumOfReasons - 1 (the + // reduced width) so this change does NOT alter what is written -- only + // the read path becomes tolerant. Widening the serialized width to the + // full kNumOfReasons is a separate, later change that must not land + // until this tolerant read has been deployed to every primary, + // including rollback targets. {"counts", - OptionTypeInfo::Array< - /* In release 11.2, a new compaction reason was added. This broken - * the reader and writer. To unblock release 11.1, we temporarily - * reduce the count array size to the old one. TODO add a proper - * serialization and deserialization method. */ + SizeTolerantDeserializeArray< int, static_cast(CompactionReason::kNumOfReasons) - 1>( offsetof(struct InternalStats::CompactionStats, counts), OptionVerificationType::kNormal, OptionTypeFlags::kNone, diff --git a/db/compaction/compaction_service_test.cc b/db/compaction/compaction_service_test.cc index 094ff01352..139bbf1301 100644 --- a/db/compaction/compaction_service_test.cc +++ b/db/compaction/compaction_service_test.cc @@ -3,6 +3,8 @@ // COPYING file in the root directory) and Apache 2.0 License // (found in the LICENSE.Apache file in the root directory). +#include + #include "db/db_test_util.h" #include "file/file_util.h" #include "port/stack_trace.h" @@ -12,6 +14,71 @@ namespace ROCKSDB_NAMESPACE { +namespace { + +// Append ":extra_value" to every "counts=...;" field in a serialized +// CompactionServiceResult string to simulate a worker with more +// CompactionReason enums than the reader. Loops because the serialized +// result contains multiple CompactionStats structs (output_level_stats, +// proximal_level_stats), each with its own "counts=" field. +void AppendToCountsFields(std::string* s, const std::string& extra_value) { + const std::string prefix = "counts="; + size_t pos = s->find(prefix); + while (pos != std::string::npos) { + // Find the semicolon that terminates this field's value. + size_t semi = s->find(';', pos + prefix.size()); + assert(semi != std::string::npos); + // Insert ":extra_value" just before the semicolon. + s->insert(semi, ":" + extra_value); + // Advance past the inserted text to find the next occurrence. + pos = s->find(prefix, semi + extra_value.size() + 2); + } +} + +// Remove the last colon-separated element from every "counts=...;" field +// to simulate a worker with fewer CompactionReason enums than the reader. +// Loops for the same reason as AppendToCountsFields (multiple counts fields). +void RemoveLastFromCountsFields(std::string* s) { + const std::string prefix = "counts="; + size_t pos = s->find(prefix); + while (pos != std::string::npos) { + size_t val_begin = pos + prefix.size(); + size_t semi = s->find(';', val_begin); + assert(semi != std::string::npos); + // Find the last ':' separator within the value. + size_t last_sep = s->rfind(':', semi - 1); + assert(last_sep != std::string::npos && last_sep >= val_begin); + // Erase from the last ':' up to (not including) the ';'. + // e.g. "counts=0:3:5;" -> erase ":5" -> "counts=0:3;" + s->erase(last_sep, semi - last_sep); + pos = s->find(prefix, last_sep); + } +} + +// Count the colon-separated elements in the first "counts=...;" field of a +// serialized CompactionServiceResult. Used to assert the serialized width +// (how many CompactionReason counts this binary writes). +size_t CountFirstCountsFieldElements(const std::string& s) { + const std::string prefix = "counts="; + size_t pos = s.find(prefix); + assert(pos != std::string::npos); + size_t val_begin = pos + prefix.size(); + size_t semi = s.find(';', val_begin); + assert(semi != std::string::npos); + if (semi == val_begin) { + return 0; + } + size_t count = 1; + for (size_t i = val_begin; i < semi; ++i) { + if (s[i] == ':') { + ++count; + } + } + return count; +} + +} // namespace + class MyTestCompactionService : public CompactionService { public: MyTestCompactionService( @@ -1609,6 +1676,68 @@ TEST_F(CompactionServiceTest, InvalidResultFallsBackToLocal) { my_cs->GetFinalCompactionServiceJobStatus()); } +TEST_F(CompactionServiceTest, CompatCheckCountsWidthTolerance) { + // Number of per-CompactionReason counts this binary serializes and expects. + constexpr size_t kExpected = + static_cast(CompactionReason::kNumOfReasons) - 1; + constexpr size_t kLastIndex = kExpected - 1; + constexpr int kFirstCount = 3; + constexpr int kLastCount = 5; + + CompactionServiceResult result; + result.status = Status::OK(); + result.output_level = 2; + result.output_path = "/tmp/output"; + result.internal_stats.output_level_stats.counts[0] = kFirstCount; + result.internal_stats.output_level_stats.counts[kLastIndex] = kLastCount; + std::string serialized; + ASSERT_OK(result.Write(&serialized)); + + // Guard the produced format: this binary writes exactly kExpected counts. + // That element count is what a remote worker hands to the primary to read, + // so pinning it catches an accidental change to what is written. + ASSERT_EQ(CountFirstCountsFieldElements(serialized), kExpected); + + // Peer agrees on kNumOfReasons: every written count is read back unchanged. + { + CompactionServiceResult parsed; + ASSERT_OK(CompactionServiceResult::Read(serialized, &parsed)); + ASSERT_OK(parsed.status); + const auto& counts = parsed.internal_stats.output_level_stats.counts; + ASSERT_EQ(counts[0], kFirstCount); + ASSERT_EQ(counts[kLastIndex], kLastCount); + } + + // Peer serialized one more count than this binary knows (a worker with an + // extra CompactionReason): the surplus trailing count is ignored, and the + // counts this binary understands are read unchanged. + { + std::string from_wider_peer = serialized; + AppendToCountsFields(&from_wider_peer, "7"); + ASSERT_EQ(CountFirstCountsFieldElements(from_wider_peer), kExpected + 1); + CompactionServiceResult parsed; + ASSERT_OK(CompactionServiceResult::Read(from_wider_peer, &parsed)); + ASSERT_OK(parsed.status); + const auto& counts = parsed.internal_stats.output_level_stats.counts; + ASSERT_EQ(counts[0], kFirstCount); + ASSERT_EQ(counts[kLastIndex], kLastCount); + } + + // Peer serialized one fewer count than this binary knows (a worker missing + // the newest CompactionReason): the absent trailing count reads as zero. + { + std::string from_narrower_peer = serialized; + RemoveLastFromCountsFields(&from_narrower_peer); + ASSERT_EQ(CountFirstCountsFieldElements(from_narrower_peer), kExpected - 1); + CompactionServiceResult parsed; + ASSERT_OK(CompactionServiceResult::Read(from_narrower_peer, &parsed)); + ASSERT_OK(parsed.status); + const auto& counts = parsed.internal_stats.output_level_stats.counts; + ASSERT_EQ(counts[0], kFirstCount); + ASSERT_EQ(counts[kLastIndex], 0); + } +} + TEST_F(CompactionServiceTest, SubCompaction) { Options options = CurrentOptions(); options.max_subcompactions = 10;