fix(ft.aggregate): Fix Compress() zombie-parent bug in Quantile reducer

Signed-off-by: Riley Des <riley.desserre@improving.com>

Author: rileydes-improving

src/commands/ft_aggregate_exec.cc

@@ -418,25 +418,33 @@ class Quantile : public GroupBy::ReducerInstance {
   // Compress samples to maintain space bounds.
   // Walks backward from the second-to-last sample so that merges
   // propagate toward the tail where error bounds are largest.
+  // parent_idx tracks the nearest live (non-merged) sample to the right of
+  // the current position, preventing merged (g==0) zombie samples from being
+  // used as merge targets and silently absorbing subsequent merges.
   void Compress() const {
     if (samples_.size() < 2) return;
 
     // Compute rank of the last sample (n - 1 - last.g)
     double r =
         static_cast<double>(n_) - 1.0 - static_cast<double>(samples_.back().g);
 
-    // Walk backward from second-to-last to first
+    // Walk backward from second-to-last to first, always merging into the
+    // nearest live parent to the right.
+    size_t parent_idx = samples_.size() - 1;
     for (int i = static_cast<int>(samples_.size()) - 2; i >= 0; --i) {
       Sample& curr = samples_[i];
-      Sample& parent = samples_[i + 1];
+      Sample& parent = samples_[parent_idx];
       double g_curr = curr.g;
 
       if (curr.g + parent.g + parent.delta <=
           static_cast<size_t>(GetMaxVal(r))) {
-        // Merge curr into parent
+        // Merge curr into parent; parent_idx stays — keep accumulating into
+        // the same live sample.
         parent.g += curr.g;
-        // Mark for removal by setting g to 0
         curr.g = 0;
+      } else {
+        // curr survives; it becomes the new live parent for i-1.
+        parent_idx = static_cast<size_t>(i);
       }
       r -= g_curr;
     }

testing/ft_aggregate_exec_test.cc

@@ -1029,5 +1029,43 @@ TEST_F(AggregateExecTest, QuantileStringNumericValueTest) {
   EXPECT_NEAR(result, 2.0, 1.0);
 }
 
+// Regression test for the Compress() zombie-parent bug.
+// Before the fix, each Compress() pass deleted at most one sample because
+// merged (g==0) samples were used as merge targets, reviving them.  After the
+// fix, a single pass must delete all consecutively-mergeable samples.
+TEST_F(AggregateExecTest, CompressBoundedSampleCount) {
+  std::cerr << "CompressBoundedSampleCount\n";
+
+  // Insert enough values to trigger many Flush()+Compress() cycles.
+  // With EPSILON=0.01 the GK bound is O((1/ε)·log(ε·N)) ≈ 700 samples for
+  // N=100 000.  The buggy implementation retained ~N samples; the fixed one
+  // must stay well below N.
+  const size_t kN = 100000;
+  // kDefaultBufferSize is 500; each flush triggers a compress.
+  // We drive inserts through the public ProcessRecord path via Execute().
+  // To keep the test self-contained we use a large single group.
+  RecordSet records(nullptr);
+  for (size_t i = 0; i < kN; ++i) {
+    auto rec = std::make_unique<Record>(2);
+    rec->fields_[0] = expr::Value(static_cast<double>(i));
+    rec->fields_[1] = expr::Value(1.0);  // single group key
+    records.emplace_back(std::move(rec));
+  }
+
+  auto param = MakeStages("groupby 1 @n2 reduce quantile 2 @n1 0.5");
+  EXPECT_TRUE((param->stages_[0]->Execute(records)).ok());
+  EXPECT_EQ(records.size(), 1);
+  auto record = records.pop_front();
+
+  // The median of [0, N) is N/2.  Allow 1% relative error (EPSILON).
+  ASSERT_TRUE(record->fields_.at(2).IsDouble());
+  double result = *(record->fields_.at(2).AsDouble());
+  double expected = static_cast<double>(kN) / 2.0;
+  EXPECT_NEAR(result, expected, expected * 0.01)
+      << "Median of [0," << kN << ") should be within 1% of " << expected;
+
+  std::cerr << "CompressBoundedSampleCount passed, median=" << result << "\n";
+}
+
 }  // namespace aggregate
 }  // namespace valkey_search