feat : support for out-of-orderness tolerance

[sherali42]

Pull Request checklist

• The commit(s) message(s) follows the contribution guidelines ?
• Tests for the changes have been added (for bug fixes / features) ?
• Docs have been added / updated (for bug fixes / features) ?

Summary

Adds out-of-order (OOO) sample ingestion to FiloDB: time-bucketed,
per-partition aggregation that accepts samples whose timestamps fall behind
the latest seen timestamp by up to a configurable tolerance window. Standard
TimeSeriesPartition rejects such samples outright; this change replaces the
hard rejection with bounded in-memory aggregation, finalizing each bucket to a
single chunk row once an event-time watermark passes it.

Opt-in per schema. No on-disk format change. Cassandra, downsampling, query
planners, and persistence are unaffected. Full design in
docs/specs/ooo-sample-ingestion.md;
ADR in doc/adr-aggregating-buffers.md.

Motivation

Standard TimeSeriesPartition.ingest enforces strict in-order arrival —
anything older than the latest chunk's endTime is dropped and counted on
outOfOrderDropped. In production this loses real data:

• Producer skew. Retries, batching, and clock drift push samples behind
  the in-order watermark by seconds to minutes.
• Delta-temporality metrics. A dropped delta sample doesn't just leave a
  hole — it permanently distorts every downstream rate() / delta() that
  crosses the gap, because nothing else re-sums the missing delta.

Design (one paragraph)

AggregatingTimeSeriesPartition extends TimeSeriesPartition and is
selected at partition-creation time when the schema declares aggregators.
Incoming samples are aggregated into fixed-width event-time buckets keyed by
ceilToBucket(sampleTs, intervalMs). A watermark — the latest sample
timestamp on the partition — advances forward only; buckets whose
timestamps fall more than oooToleranceMs behind the watermark are
finalized and handed to the existing TimeSeriesPartition.ingest pipeline
as a single row. Active (not-yet-finalized) buckets are queryable via
AggregatingRangeVector, which merges in-memory bucket rows with finalized
chunk rows. Kafka commit offsets are held back to the earliest active
bucket so crash recovery never loses samples that were aggregated but not
yet flushed.

The model is purely event-time. No wall-clock dependency in tolerance
checks, finalization, or query visibility — replay is deterministic.

What's in the change

Per-schema config (additive)

Key	Meaning
aggregators	List of name(colId), e.g. ["dSum(1)", "hSum(3)"]
aggregation-interval	Bucket width (HOCON duration, e.g. 1m)
aggregation-ooo-tolerance	Acceptance window (HOCON duration, e.g. 2m)

A non-empty aggregators list is the only trigger. Schemas without
aggregators are completely unaffected. Canonical example schema
aggregating-delta-histogram-v2 is registered in filodb-defaults.conf.

Core components

• AggregatingTimeSeriesPartition — routes ingest through
  BucketAggregationState; routes query through AggregatingRangeVector
• BucketAggregationState — TreeMap of per-bucket aggregators with object
  pooling to limit allocation pressure
• Aggregator trait + Sum/Min/Max/Last/First/Count/HistogramSum/
  HistogramLast with type-specialized add paths (no boxing on Double/Long)
• BinaryHistogram.addValuesTo — in-place histogram accumulation without
  per-sample allocation (the only new public API on existing types)
• AggregatingRangeVector + MergingRangeVectorCursor — merge finalized
  chunk rows with active in-memory bucket data, with a per-cursor scratch
  pool for histogram snapshots
• Watermark-based Kafka offset hold in TimeSeriesShard.prepareFlushGroup —
  prevents commit past samples still held in active buckets

Out of scope (intentionally)

• Per-column tolerance windows (schema-wide only)
• Late-arriving samples beyond the tolerance window (still rejected and
  counted on outOfOrderDropped)
• Re-aggregating already-finalized buckets (silently dropped to avoid
  duplicates)

Tests

• 6 new specs across core/src/test/scala/filodb.core/memstore/aggregation/
  and core/src/test/scala/filodb.core/{memstore,query}/:
  BucketAggregationStateSpec, AggregatorSpec, ColumnAggregatorSpec,
  AggregatingTimeSeriesPartitionSpec, AggregatingRangeVectorSpec,
  EventTimeWatermarkSpec
• Schema-level config coverage in ColumnSpec and SchemasSpec
• PeriodicSamplesMapperSpec updated for AggregatingRangeVector query path
• OOOAggregationBenchmark (JMH) — scalar and histogram-typed scenarios for
  ingest, query, and finalization throughput / allocation
• AggregationHotPathBenchmark — micro for the hot ingest path

Rollout / risk

• Off by default — schemas opt in by declaring aggregators. Existing
  schemas behave identically.
• No on-disk format change. No protocol change. No query-engine API change.
• Crash recovery: the Kafka offset hold guarantees in-flight bucket samples
  are re-ingested on replay; aggregation is deterministic and idempotent
  under replay.

Test plan

• Smoke a dev cluster with aggregating-delta-histogram-v2 schema and
  verify OOO samples (vs. previous outOfOrderDropped behavior)
• Verify Kafka offset hold under graceful shutdown + restart
• Compare ingest CPU / allocations against a non-aggregating schema
  (JMH locally; production canary in a low-traffic shard)
• Confirm queries against a partition with active buckets return the
  in-memory data merged with finalized chunks