fix(query-engine): give range-query merges the CMS/KLL batch fast path (#614)

* fix(query-engine): give range-query merges the CMS/KLL batch fast path (#596)

NaiveMerger::merge_all (range path) only did a sequential merge_with
fold, never the merge_multiple batch fast path that
SimpleEngine::merge_accumulators (instant path) already had for
CountMinSketch/DatasketchesKLL. The two folds also disagreed on error
handling: SimpleEngine's warned and silently kept the partial result,
NaiveMerger's aborted via `?`.

Extract the shared "try batch merge, else fold" logic into
engines::merge_utils::merge_accumulators_batch and have both callers
delegate to it, so both now take the fast path and both abort on the
first merge_with failure instead of one of them dropping data
silently. Adds AccumulatorError::MergeFailed for the new abort case.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WDS34V2EexpbihCfH9SDWh

* fix(query-engine): restore move-based single-accumulator path, dedupe test fixtures

Two follow-ups from review of the #596 fix:

- merge_accumulators_batch always cloned the first accumulator, even
  for the common single-bucket case, regressing the move-based
  shortcut SimpleEngine::merge_accumulators had before this refactor.
  Restore it at the call site (NaiveMerger can't do the same since it
  merges from a borrowed Vec it keeps for the next slide()).
- The new regression tests duplicated PoisonableAccumulator,
  cms_from_matrix, and oracle_sequential_fold verbatim between
  window_merger.rs and simple_engine/mod.rs; moved them into
  tests::test_utilities::merge_fixtures, shared by both.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WDS34V2EexpbihCfH9SDWh

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

Author: milindsrivastava1997

asap-query-engine/src/engines/merge_utils.rs

@@ -0,0 +1,56 @@
+//! Shared accumulator-batch merge fold, used by both the instant-query path
+//! (`SimpleEngine::merge_accumulators`) and the range-query path
+//! (`NaiveMerger::merge_all`), so the two stay behaviorally identical.
+//!
+//! Tries a batch merge for accumulator types that support one (currently
+//! `DatasketchesKLL` and `CountMinSketch`), falling back to a sequential
+//! pairwise fold otherwise or if the batch merge itself fails. The fold
+//! aborts on the first `merge_with` error instead of skipping it, so a
+//! caller can't get a silently-partial merge back as `Ok`.
+
+use crate::data_model::{AggregateCore, AggregationType};
+use crate::precompute_operators::count_min_sketch_accumulator::CountMinSketchAccumulator;
+use crate::precompute_operators::datasketches_kll_accumulator::DatasketchesKLLAccumulator;
+use tracing::warn;
+
+/// Precondition: `accumulators` is non-empty. Callers already special-case
+/// the empty slice to report their own error (`AccumulatorError::EmptySlice`,
+/// `"No buckets to merge"`), so this doesn't repeat that check.
+pub(crate) fn merge_accumulators_batch(
+    accumulators: &[Box<dyn AggregateCore>],
+) -> Result<Box<dyn AggregateCore>, Box<dyn std::error::Error + Send + Sync>> {
+    debug_assert!(
+        !accumulators.is_empty(),
+        "merge_accumulators_batch requires at least one accumulator"
+    );
+
+    if accumulators.len() == 1 {
+        return Ok(accumulators[0].clone_boxed_core());
+    }
+
+    let accumulator_type = accumulators[0].get_accumulator_type();
+
+    if accumulator_type == AggregationType::DatasketchesKLL {
+        match DatasketchesKLLAccumulator::merge_multiple(accumulators) {
+            Ok(merged) => return Ok(Box::new(merged)),
+            Err(e) => warn!(
+                "Batch merge failed: {}. Falling back to sequential merge.",
+                e
+            ),
+        }
+    } else if accumulator_type == AggregationType::CountMinSketch {
+        match CountMinSketchAccumulator::merge_multiple(accumulators) {
+            Ok(merged) => return Ok(Box::new(merged)),
+            Err(e) => warn!(
+                "Batch merge failed: {}. Falling back to sequential merge.",
+                e
+            ),
+        }
+    }
+
+    let mut result = accumulators[0].clone_boxed_core();
+    for accumulator in &accumulators[1..] {
+        result = result.merge_with(accumulator.as_ref())?;
+    }
+    Ok(result)
+}

asap-query-engine/src/engines/mod.rs

@@ -1,4 +1,5 @@
 pub mod logical;
+pub(crate) mod merge_utils;
 pub mod physical;
 pub mod query_result;
 pub mod simple_engine;