No type-level distinction between order-sensitive and order-insensitive accumulator merges

[milindsrivastava1997]

MergeableAccumulator<T>::merge_accumulators(accumulators: Vec<T>) (asap-query-engine/src/data_model/traits.rs) takes a plain, unordered Vec<T>. Every accumulator implemented against it today (Sum, KLL, HLL, CountMinSketch, MinMax, SetAggregator, ...) is commutative/order-insensitive, so this has never mattered.

DeltaSetAggregatorAccumulator breaks that assumption: its correct merge is a chronological fold over added/removed deltas, not a commutative union (see #586). But nothing in the trait signature — or at the call site — distinguishes "this accumulator's merge is order-sensitive" from "this one isn't":

• The call site that gathers buckets for merging, merge_precomputed_outputs (asap-query-engine/src/engines/simple_engine/mod.rs), discards each bucket's timestamp before invoking merge (timestamped_buckets.iter().map(|(_, bucket)| bucket.clone_boxed_core())), so by the time any accumulator's merge logic runs, order information is already gone regardless of whether that accumulator type needs it.
• There is no marker, trait bound, or naming convention that would make an order-sensitive accumulator's requirements visible to someone adding a new accumulator type, or to someone touching the merge call path.
• The bug in DeltaSetAggregator: merge is order-insensitive, store returns buckets out of order, and get_keys() drops all keys on any removal #586 was able to exist (order-insensitive dedup logic, unsorted bucket delivery from the store) without any compiler or interface signal that DeltaSetAggregator's contract differed from its neighbors.

This is a systemic gap, not specific to DeltaSetAggregator: any future accumulator that needs order-sensitive/stateful-over-time merging (e.g. other "delta"-style accumulators, counter-reset-aware rate tracking, etc.) will silently inherit the same class of bug, because the current interfaces give no way to express or enforce "this accumulator's merge must see buckets in chronological order with timestamps intact."

Related: #586 (the concrete DeltaSetAggregator bugs), #588 (DeltaSetAggregator's tumbling-window requirement, a related but separate contract that's also currently unenforced).

[milindsrivastava1997]

Idea worth considering later: use two separate types, one for a raw unmerged delta and one for a merged/combined delta. Right now nothing stops added/removed from overlapping except convention. If the raw type enforced "no overlap" at compile time, that class of bug couldn't happen. Not proposing this as the fix, just flagging it as an option to weigh, since it would add real cost (a second type to plumb through serialization and storage) and would need to work for future accumulators with this same problem.

[milindsrivastava1997]

Two design options for making order-sensitive vs. order-insensitive merges distinguishable. Both work within the current Box<dyn AggregateCore> architecture, but with opposite trade-offs.

Option A: capability trait

Add a second, narrower trait that only order-sensitive accumulators implement, plus one accessor on the base trait to reach it:

pub trait OrderedMerge: AggregateCore {
    fn merge_ordered(&self, ordered: &[(i64, &dyn AggregateCore)]) -> Result<Box<dyn AggregateCore>, ...>;
}

pub trait AggregateCore: SerializableToSink + Send + Sync {
    // ...existing methods...
    fn as_ordered_merge(&self) -> Option<&dyn OrderedMerge> { None } // default: not order-sensitive
}

DeltaSetAggregatorAccumulator overrides as_ordered_merge to return Some(self) and implements merge_ordered. The one merge call site (simple_engine::merge_precomputed_outputs) branches: if let Some(om) = acc.as_ordered_merge() { om.merge_ordered(&sorted_with_timestamps) } else { self.merge_accumulators(...) }.

Before/after: nothing else changes. window_merger.rs's NaiveMerger (currently Vec<Box<dyn AggregateCore>>, folds via merge_with) stays exactly as-is — it never needs to know the split exists.

Pros: small, scoped diff (one trait, one default method, one override, one call-site branch). Every other accumulator (Sum, KLL, HLL, CountMinSketch, ...) is untouched. merge_ordered's signature forces timestamps to be supplied wherever it's actually called.
Cons: still a runtime check — a future order-sensitive accumulator that forgets to override as_ordered_merge compiles fine and silently takes the unordered path. Two parallel merge code paths to maintain.

Option B: replace Box<dyn AggregateCore> with a closed enum

AggregationType already enumerates all ~13 accumulator kinds, so this isn't a new axis — just making the type system reflect it:

pub enum Accumulator {
    OrderInsensitive(OrderInsensitiveAccumulator),
    OrderSensitive(OrderSensitiveAccumulator),
}
pub enum OrderInsensitiveAccumulator { Sum(SumAccumulator), Hll(HllAccumulator), CountMinSketch(CountMinSketchAccumulator), /* ... */ }
pub enum OrderSensitiveAccumulator { DeltaSetAggregator(DeltaSetAggregatorAccumulator) /* future order-sensitive types go here */ }

Before (today, sum_accumulator.rs:92-112, repeated near-identically in all ~13 accumulator files):

fn merge_with(&self, other: &dyn AggregateCore) -> Result<Box<dyn AggregateCore>, ...> {
    if other.get_accumulator_type() != self.get_accumulator_type() {
        return Err(format!("Cannot merge SumAccumulator with {}", other.get_accumulator_type()).into());
    }
    let other_sum = other.as_any().downcast_ref::<SumAccumulator>()
        .ok_or("Failed to downcast to SumAccumulator")?;
    let merged = Self::merge_accumulators(vec![self.clone(), other_sum.clone()])?;
    ...
}

After:

fn merge_with(a: &Accumulator, b: &Accumulator) -> Result<Accumulator, MergeError> {
    match (a, b) {
        (Accumulator::Sum(x), Accumulator::Sum(y)) => Ok(Accumulator::Sum(SumAccumulator::merge_accumulators(vec![x.clone(), y.clone()])?)),
        (Accumulator::Hll(x), Accumulator::Hll(y)) => Ok(Accumulator::Hll(HllAccumulator::merge_accumulators(vec![x.clone(), y.clone()])?)),
        // ... one line per type
        (a, b) => Err(MergeError::TypeMismatch(a.type_name(), b.type_name())), // one shared mismatch arm, not 13 copies
    }
}

But the reverse also holds — window_merger.rs's NaiveMerger, which today is fully generic over accumulator type (Vec<Box<dyn AggregateCore>>, folds via merge_with, never touched when a new accumulator type is added), now has to branch on OrderInsensitive vs. OrderSensitive and gains a "mixed kinds in one window" error case that couldn't previously be expressed.

Pros: real compiler-enforced exhaustiveness — add a 14th accumulator type and every non-wildcard match on Accumulator fails to compile until you place it in OrderInsensitive or OrderSensitive. Also deletes a large amount of existing as_any()/downcast_ref() boilerplate (165 call sites in the crate today, merge_with being the dominant repeated pattern) and the runtime "Cannot merge X with Y" checks that go with it.
Cons: touches every place currently built around dyn AggregateCore/dyn SingleSubpopulationAggregate/dyn MultipleSubpopulationAggregate polymorphism (precompute operators, WindowMerger, the two *Factory traits, accumulator_serde.rs) — a crate-wide refactor, not a scoped fix. Loses open extension (adding a type today = "new struct + impl the traits"; with the enum it's "new struct + new variant + touch every exhaustive match on it").

Recommendation

Option A for a scoped fix to this specific bug class. Option B is worth it only if the codebase is already heading away from trait-object-based accumulators for other reasons — it's a genuine net win on the downcast-boilerplate side, but it's a much bigger change than #604 alone justifies.