Widen compaction service counts to full kNumOfReasons

Summary:
Widen the serialized "counts" array in CompactionServiceResult from kNumOfReasons - 1 back to the full kNumOfReasons, so the count for the newest CompactionReason is included in the remote compaction result.

The width was previously reduced by one to match readers that expected the narrower array; restoring it changes what is written. This is safe only because deserialization of "counts" is now size-tolerant (a reader expecting fewer elements drops the extra, one expecting more zero-fills the missing). It must not land until that size-tolerant read has been deployed to every primary, including rollback targets so should be 1-2 releases ahead; otherwise a primary can still or fallback to running the strict reader would reject the wider result and fail deserialization of this field.

== 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: expect-narrow tolerant-read primary reads wide counts a worker write (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 2: expect-wide tolerant-read primary reads 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 3: expect-wide tolerant-read primary reads wide counts a worker writes 
```
$ 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: D111068809

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,16 +902,11 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct InternalStats::CompactionStats, count),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
           OptionTypeFlags::kNone}},
-        {"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. */
-             int, static_cast<int>(CompactionReason::kNumOfReasons) - 1>(
-             offsetof(struct InternalStats::CompactionStats, counts),
-             OptionVerificationType::kNormal, OptionTypeFlags::kNone,
-             {0, OptionType::kInt})},
+        {"counts", SizeTolerantDeserializeArray<
+                       int, static_cast<int>(CompactionReason::kNumOfReasons)>(
+                       offsetof(struct InternalStats::CompactionStats, counts),
+                       OptionVerificationType::kNormal, OptionTypeFlags::kNone,
+                       {0, OptionType::kInt})},
 };
 
 static std::unordered_map<std::string, OptionTypeInfo>

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);
+  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;