Size-tolerant deserialization for compaction service response's counts array (#14807)

Summary:

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<T, kSize> 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.

Differential Revision: D106321320

Author: Hui Xiao

db/compaction/compaction_service_job.cc

@@ -777,6 +777,40 @@ static std::unordered_map<std::string, OptionTypeInfo>
           OptionTypeFlags::kNone}},
 };
 
+namespace {
+
+// Like OptionTypeInfo::Array<T, kSize> but deserialization tolerates
+// size mismatches: extra elements beyond kSize are silently ignored,
+// missing elements are zero-filled. Serialization and comparison are
+// inherited from Array<T, kSize> unchanged; only the parse
+// (deserialize) callback is overridden.
+template <typename T, size_t kSize>
+OptionTypeInfo SizeTolerantDeserializeArray(int offset,
+                                            OptionVerificationType verification,
+                                            OptionTypeFlags flags,
+                                            const OptionTypeInfo& elem_info,
+                                            char separator = ':') {
+  return OptionTypeInfo::Array<T, kSize>(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<T> parsed;
+        Status s =
+            ParseVector<T>(opts, elem_info, separator, name, value, &parsed);
+        if (!s.ok()) {
+          return Status(std::move(s));
+        }
+        auto* arr = static_cast<std::array<T, kSize>*>(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<std::string, OptionTypeInfo>
     compaction_stats_type_info = {
         {"micros",
@@ -868,12 +902,20 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {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<int>(CompactionReason::kNumOfReasons) - 1>(
              offsetof(struct InternalStats::CompactionStats, counts),
              OptionVerificationType::kNormal, OptionTypeFlags::kNone,

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 <memory>
+
 #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<size_t>(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;