Add streaming precompute engine: sketch-based windowed aggregation over Prometheus/VictoriaMetrics remote write

[zzylol]

Summary

Design document: precompute_engine_design_doc.md

Introduces the Precompute Engine — a real-time streaming aggregation system that ingests raw time-series samples via Prometheus and VictoriaMetrics remote write, computes windowed sketch aggregations, and writes results directly to the ASAP store for fast query-time retrieval.

Architecture

Prometheus Remote Write       VictoriaMetrics Remote Write
 (Snappy + Protobuf)             (Zstd + Protobuf)
         |                              |
         v                              v
POST /api/v1/write           POST /api/v1/import
          \                           /
           Axum HTTP Server (:9090)
                    |
          route_decoded_samples()
           (group by series key)
                    |
             SeriesRouter (xxhash64)
           /        |        \
      Worker 0   Worker 1   Worker N-1
      (shard 0)  (shard 1)  (shard N-1)
                    |
             OutputSink.emit_batch()
                    |
             SimpleMapStore (PerKey)
                    |
            Query Engine (PromQL / SQL)

Single-machine, multi-threaded: all workers are Tokio async tasks in one process. Series are hash-partitioned across workers — no cross-shard coordination, no locks within a worker.

Key Design Choices

Pane-based incremental sliding windows
The timeline is divided into non-overlapping panes of size slide_interval. Each sliding window is composed of W = window_size / slide_interval consecutive panes. Consecutive windows share W-1 panes.

This avoids re-processing samples N×W times (naïve approach) — each sample updates exactly 1 pane. Window close merges W panes via AggregateCore::merge_with(). Non-invertible sketches (KLL, CMS, HydraKLL, MinMax, Increase) work correctly because merge is universal; subtraction-based approaches would only work for Sum.

For tumbling windows (W=1), panes degenerate to 1 pane = 1 window — no extra merges.

Supported accumulator types

Aggregation	Sub-type	Notes
SingleSubpopulation	Sum, Min/Max, Increase, KLL	per-metric, no grouping
MultipleSubpopulation	Sum, Min/Max, Increase, CMS, HydraKLL	grouped by label values

Late data handling (LateDataPolicy)

• Drop: samples behind the watermark by more than allowed_lateness_ms are discarded silently.
• ForwardToStore: late samples are emitted as mini-accumulators alongside the original window; query-time merge (SummaryMergeMultipleExec) combines them automatically.

Zero-serialization write path
Closed window accumulators flow from worker to store as in-memory Box<dyn AggregateCore> trait objects via direct function calls — no serialization, no IPC, no network hops between aggregation and storage. Contrast with the Kafka/Arroyo path where precomputes arrive hex-encoded + gzip-compressed + MessagePack-serialized.

Raw passthrough mode
pass_raw_samples = true bypasses all aggregation. Each sample is emitted as a SumAccumulator::with_sum(value) point-window, enabling store-and-forward of raw data.

Cross-Series (GROUP BY) Aggregation

The AggregationConfig has three label dimensions that control the spatial aggregation shape:

Field	Role
grouping_labels	Store GROUP BY key — labels preserved in PrecomputedOutput.key
aggregated_labels	Internal sketch sub-keys (MultipleSubpopulation only)
rollup_labels	Dropped entirely at ingest; not recoverable at query time

A config with grouping_labels: [job] and rollup_labels: [instance] means: "sum across all instances, keep one output series per job value." Multiple input series (metric{job=j1,instance=h1}, metric{job=j1,instance=h2}, ...) all contribute to the same logical output key (job=j1).

How the engine handles this across workers:

Because routing is by full series key (xxhash64(series_key) % N), two series that share a grouping_labels key but differ in rolled-up labels land on different workers:

metric{job=j1, instance=h1} -> Worker 0 -> pane accumulator keyed (job=j1)
metric{job=j1, instance=h2} -> Worker 3 -> pane accumulator keyed (job=j1)

Each worker independently closes its window and emits a separate PrecomputedOutput with key (job=j1) for the same window [0, 60s). The store appends rather than overwrites on the same (aggregation_id, key, window) tuple:

store[(agg_id, key=(j1), [0,60s))] -> [acc_worker0, acc_worker3]

Query-time SummaryMergeMultipleExec merges all entries for the same key and window via AggregateCore::merge_with(), producing the correct combined result. No ingest-time cross-worker coordination is needed.

Eventual consistency property:

Workers have independent watermarks. For a standard Prometheus scrape (all instances delivered in one HTTP batch to route_batch()), all workers receive their samples in the same round-trip and close the window on the same flush cycle. The incompleteness window — the time between the first and last worker emitting for the same cross-series window — is typically in the millisecond range (bounded by worker task scheduling jitter on the shared Tokio thread pool).

For staggered multi-source producers (different series arriving from different remote-write clients at different times), the incompleteness window can be longer, bounded by the spread of producer arrival times. In both cases the result is eventually consistent: once all contributing workers have emitted, the store holds a complete set of accumulators and queries return the correct merged value.

This deferred-merge design is intentional: it avoids any cross-worker coordination or global watermark synchronization at ingest time, keeping the shared-nothing worker architecture intact. The store's append-multiple-per-window design and the query-time merge already handle the fan-in correctly for both the cross-series aggregation case and the ForwardToStore late-data case.

Performance (measured on this machine)

Hardware

• CPU: 2x Intel Xeon E5-2630 v3 @ 2.40 GHz (8 cores / 16 threads per socket, 32 logical CPUs total, max turbo 3.2 GHz)
• L3 cache: 20 MiB per socket (40 MiB total)
• RAM: 125 GiB

All benchmarks run with cargo run --release. Engine and client run in the same process.

Batch latency — 1,000 samples in one HTTP request (raw mode, 4 workers)

Payload (snappy)	HTTP round-trip
12,253 bytes	6.1 ms

Raw-mode flush throughput — 1,000 requests × 10,000 samples (workers = senders, NoopOutputSink)

Measures time from first request sent until all samples are flushed (emitted) from the engine, isolating engine processing throughput from store I/O. Flush lag behind send completion was ≤ 12 ms at all worker counts — the engine flushes as fast as it receives.

Workers	Send (samples/sec)	Flush E2E (samples/sec)	Speedup
1	465,039	465,039	1.00x
2	918,189	917,248	1.97x
4	1,810,115	1,806,470	3.88x
8	4,377,675	4,356,324	9.37x
16	8,990,338	8,900,489	19.13x

Near-linear scaling to 16 workers. At 16 workers the engine sustains ~8.9M samples/sec flush throughput on this hardware.

Note on super-linear speedup at 8× and 16× workers: The benchmark uses workers = senders — sender concurrency scales with worker count. The 1-worker baseline is bottlenecked by a single sender: one CPU core for Snappy compression, one in-flight HTTP connection, one channel to fill. At 16 workers, 16 concurrent senders parallelize compression across 16 cores and pipeline 16 HTTP connections simultaneously, pushing aggregate send throughput well past what a single sender can achieve. The apparent 9.37×/19.13× speedup therefore reflects sender-side parallelism as much as engine-side scaling. A clean engine-scaling measurement would fix sender count and vary only worker count; the 1-worker baseline is an artificially low reference point because the sender saturates before the engine does.

For comparison, the same 4-worker raw-mode test with StoreOutputSink (writing 10M samples to SimpleMapStore) achieves 105,293 samples/sec E2E — ~17x lower, confirming the store's RwLock is the bottleneck in the stored path, not the engine itself.

Windowed aggregation throughput — 200 requests × 5,000 samples, 50 series, 4 workers, 4 concurrent senders (NoopOutputSink)

Config	Send (samples/sec)	E2E (samples/sec)	Batch latency
Tumbling 10s Sum (W=1)	987,143	660,814	24.7 ms
Sliding 30s/10s Sum (W=3)	937,540	637,977	17.4 ms
Sliding 60s/10s Sum (W=6)	1,005,785	668,677	22.1 ms

E2E throughput is nearly identical across W=1 and W=6, confirming the pane-based optimization: each sample touches exactly 1 pane regardless of window overlap.

Worker scalability — Sliding 30s/10s Sum (W=3), 200 requests × 5,000 samples, 100 series (NoopOutputSink)

Workers	Send (samples/sec)	E2E (samples/sec)	Speedup vs 1 worker
1	247,805	220,451	1.00x
2	476,275	384,591	1.74x
4	902,488	621,468	2.82x
8	1,873,707	966,405	4.38x
16	4,160,976	1,347,133	6.11x

Windowed aggregation scales less than raw mode because window close merges (W-1 per window) and pane BTreeMap operations add per-worker CPU work that the raw path skips. E2E gains also flatten at high worker counts due to Tokio task scheduling overhead on this 32-logical-CPU machine.

New Files

File	Purpose
precompute_engine/mod.rs	Orchestrator, Axum HTTP server, flush timer
precompute_engine/config.rs	PrecomputeEngineConfig, LateDataPolicy
precompute_engine/worker.rs	Per-shard processing: watermark, pane buffering, window close, emit; unit tests for all correctness properties
precompute_engine/series_router.rs	xxhash64 routing: series key -> worker index
precompute_engine/series_buffer.rs	Per-series BTreeMap sample buffer with watermark
precompute_engine/window_manager.rs	Tumbling/sliding window alignment, closure detection, pane enumeration
precompute_engine/accumulator_factory.rs	AccumulatorUpdater trait + factory for all sketch types
precompute_engine/output_sink.rs	OutputSink trait + StoreOutputSink, NoopOutputSink, CapturingOutputSink (for testing)
precompute_engine/precompute_engine_design_doc.md	Full design document
bin/precompute_engine.rs	Standalone CLI binary
bin/test_e2e_precompute.rs	In-process E2E test: ingest -> aggregate -> PromQL query, batch latency and throughput benchmarks, worker scalability benchmark
drivers/ingest/prometheus_remote_write.rs	Prometheus remote write decoder (Snappy + protobuf)
drivers/ingest/victoriametrics_remote_write.rs	VictoriaMetrics remote write decoder (Zstd + protobuf)

Configuration

# Standalone binary
cargo run --bin precompute_engine -- \
  --streaming-config streaming_config.yaml \
  --ingest-port 9090 \
  --num-workers 4 \
  --allowed-lateness-ms 5000 \
  --flush-interval-ms 1000 \
  --late-data-policy drop

# Embedded in main binary (shares store with Kafka consumer path)
query_engine_rust --enable-prometheus-remote-write ...

Validation

Unit tests — worker.rs (6 tests via CapturingOutputSink)

CapturingOutputSink stores all (PrecomputedOutput, Box<dyn AggregateCore>) emits in a Mutex<Vec<...>> and exposes drain() / len(). Tests call process_samples and flush_all directly on the Worker struct without spinning up any Tokio tasks or HTTP servers.

Test	What it verifies
test_raw_mode_forwarding	3 samples → 3 emits; each has start == end == ts and SumAccumulator.sum == value
test_tumbling_window_correctness	Samples at t=1s/5s/9s sum to 6; window [0,10s) closes on t=10s sample; start=0, end=10000, sum=6
test_sliding_window_pane_sharing	Single sample at t=15s in 30s/10s window → 2 emits for [0,30s) and [10s,40s), both sum=42 via shared pane snapshot/take
test_groupby_separate_emits_per_series	Two series (host=A, host=B) on the same worker produce 2 independent MultipleSumAccumulator emits — no ingest-time cross-series merge
test_late_data_drop	Sample behind watermark - allowed_lateness_ms with Drop policy → 0 emits
test_late_data_forward_to_store	Late sample for an evicted pane with ForwardToStore policy → 1 emit as mini-accumulator with correct window bounds and sum

E2E tests — test_e2e_precompute.rs

• Engine starts in-process, sends Prometheus remote-write batches over HTTP, queries results via PromQL HTTP endpoint, validates aggregated values match expected.
• Raw-mode verification: 3 samples sent, looked up directly in store.

Static checks

• cargo check -p query_engine_rust passes
• Clippy clean (cargo clippy -p query_engine_rust)