test: e2e test for modified-OTLP CountMin sketch hot path (PR D) - #7
Merged
Merged
Conversation
Adds the first real end-to-end correctness test for the modified-OTLP sketch path landed in PR A (#5, vendoring) + PR B (#6, routing + CountMin decoder). The test exercises every public surface in the Phase 1 hot path: 1. Build a `StreamingConfig` with a single `AggregationConfig` for `http_requests_total` with `aggregation_type=CountMinSketch`, `grouping_labels=[service]`, 1-second tumbling windows. 2. Spawn a real `PrecomputeEngine` (workers + router + ingest state). 3. Spawn a real `OtlpReceiver::with_ingest_state(...)` wired to the same engine — modeling the production wiring `main.rs` does. 4. POST a real protobuf-encoded `ExportMetricsServiceRequest` over OTLP HTTP at `/v1/metrics`. The request carries a single `Metric.data = CountMinSketch{ data_points: [ CountMinSketchDataPoint { sketch: <CountMinState bytes>, encoding: COUNT_MIN_SKETCH_ENCODING_PROTO, attributes: [{service: "auth"}], time_unix_nano: 100ms } ] }` payload with a known 2x4 matrix (counts: row 0 = [1,2,3,4], row 1 = [5,6,7,8]). 5. POST a second OTLP request timestamped 2 s past epoch (past the 1 s window end) so the precompute engine's watermark advances and closes window 0. 6. Wait for the periodic flush (100ms interval) to fire. 7. Drain the `CapturingOutputSink`, find the entry for window 0, downcast the `Box<dyn AggregateCore>` to `CountMinSketchAccumulator`, read `inner.sketch()`, and assert each row matches the expected `Vec<f64>`. What this validates end-to-end: - PR A: the vendored `asap_otel_proto` crate's tonic bindings expose `Metric.data::Countminsketch` and the typed `CountMinSketchDataPoint` fields, and the prost build pipeline produces decoders that the test crate can use directly to construct a request. - PR B routing: `route_modified_otlp_sketches_to_precompute` walks the `Metric.data` oneof, flattens each per-variant data point into `ModifiedOtlpSketchDp`, matches the metric against `ingest_state.agg_configs` by name, computes the group key from the `service` attribute via `IngestState::extract_group_key_for`, and emits `WorkerMessage::AccumulatorInput` correctly. - PR B decoder: `CountMinSketchAccumulator::from_sketchlib_proto_bytes` decodes the `CountMinState` proto, picks the int64 counter path, reshapes the flat counts into `Vec<Vec<f64>>`, and constructs the underlying `CountMinSketch` via `from_legacy_matrix` — round-trip matches the original matrix exactly. - Precompute engine sketch-pane merge: the `WorkerMessage::AccumulatorInput` reaches the right worker via the (agg_id, group_key) hash, lands in the `sketch_panes` of the matching `GroupState`, and survives the watermark-driven window close to be emitted via the `OutputSink`. - Window close + sink: after the watermark advances past the 1 s window end, the worker's flush emits a `(PrecomputedOutput, Box<dyn AggregateCore>)` tuple with `start_timestamp=0` and `end_timestamp=1000` to the `CapturingOutputSink`. - Round-trip correctness: the matrix that comes out of the sink is bit-identical to what went in over OTLP, proving there is no data loss in the routing/decoder/merge path for a single-input window. Test layout ----------- - New file `asap-query-engine/tests/e2e_modified_otlp_sketch_path.rs` - Helpers: `make_count_min_agg_config(...)` — builds a tumbling-window CountMinSketch `AggregationConfig` with a `service` group `engine_config(...)` — builds a `PrecomputeEngineConfig` with a 100ms flush interval so the test does not have to wait long `build_count_min_state(...)` — builds a `CountMinState` proto from a known matrix `build_export_request(...)` — wraps a `CountMinSketchDataPoint` in a fully-formed `ExportMetricsServiceRequest` `post_otlp_http(...)` — sends a protobuf body to the OTLP HTTP endpoint at `localhost:port/v1/metrics` and asserts 2xx Validation ---------- - `cargo check -p query_engine_rust --tests`: clean - `cargo clippy -p query_engine_rust --tests -- -D warnings`: clean - `cargo fmt --check`: clean - `cargo test -p query_engine_rust --test e2e_modified_otlp_sketch_path`: `test e2e_count_min_sketch_modified_otlp_path ... ok` `test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured` - `cargo test -p query_engine_rust --lib`: 439 passed, 0 failed, 5 ignored What's still PR C territory --------------------------- - KLL / DDSketch / CountSketch / HLL e2e coverage. PR C (task #8) is re-scoped as smaller per-sketch-type follow-ups (PR C-CountSketch, PR C-HLL, PR C-KLL, PR C-DDSketch); each adds a per-type decoder and an analogous test arm to this file. - Delta transmission (`*_ENCODING_PROTO_DELTA`). PR C-delta adds per-series baseline tracking and a delta-merge codepath; this test file gets a `delta` test case once that lands. - MessagePack encoding parity (`*_ENCODING_MSGPACK` variants). PR I (task #14) adds those; this test file gets a four-way `(format, mode)` correctness assertion at that point. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
zzylol
added a commit
that referenced
this pull request
Apr 14, 2026
Replaces one-file-per-sealed-epoch + dir-per-agg + JSON manifest with
the layout every mainstream TSDB converges on:
* Unit of file is a "part" — one directory per flush tick containing
meta.bin + data.bin + index.bin. All of the tick's candidate epochs
are bundled into a single data.bin regardless of which agg-id they
came from, and are locatable via a sorted, mmap-friendly index.bin
that supports O(log N) binary search on (agg_id, start_ms).
* File count scales with flush ticks, not with epochs. One-file-per-
epoch produced ~288K files/day on a 200-agg 1-minute-window setup;
the parts layout produces ~86K files (three per tick) with a
natural group-commit of fdatasync amortized over the whole tick.
* Global manifest becomes an append-only parts_manifest.log with a
periodic binary parts_manifest.snapshot, replacing the JSON file
that was rewritten in full every tick. Size is proportional to
flush ticks, not to epochs that have ever existed, and the
snapshot is mmap-cast-to-slice on startup (zero parse).
* T2 retention becomes whole-part rm -rf on tight time ranges, since
each part covers ~flush_interval_ms of data.
* Tier-2 cache is now keyed on PartId and holds mmap'd part views;
config knob renamed segment_cache_bytes -> part_cache_bytes.
Updates flusher pseudocode, flush_and_evict narrative, query path,
recovery sequence, concurrency summary, phasing, and adds entry #7 to
the Resolved decisions table. Compaction of adjacent small parts is
noted as a v2 follow-up — the layout accommodates it cleanly but v1
ships without it since T2 + a reasonable flush interval keeps part
count well within what a binary-searched Vec<PartEntry> handles.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
4 tasks
zzylol
added a commit
that referenced
this pull request
May 12, 2026
…ema helper (#140) First sub-PR of Step γ (variant-by-variant consumer migration). Adopts the (c) bridge strategy: keep `legacy_expr::QueryExpr::Aggregate` as the L2 emit shape; add a one-way canonical-builder helper consumers call on demand. No construction site rewritten; no bridge enum variant added to legacy QueryExpr. This avoids the "child must be canonical" coupling that would otherwise force γ1 to migrate every other legacy variant (SketchAgg, WindowedAgg, TopK, etc.) in the same PR. ## What landed ### New: `controller/src/intent_algebra/aggregate_bridge.rs` (328 lines, 8 tests) ```rust pub fn bridge_aggregate_to_canonical( keys: &[ColumnRef], aggs: &[AggItem], having: &Option<legacy::Predicate>, schema: &Schema, ) -> Result<BridgedAggregate, BridgeError>; pub struct BridgedAggregate { pub by: Vec<ColumnId>, pub aggs: Vec<AggIntent>, pub having: Option<HavingPredicate>, } pub enum BridgeError { UnresolvedKey(ResolveError), HavingDeferred(QueryExprError), } ``` The `having` translation routes through `from_legacy_scalar` (Batch 2). E-deferred ScalarExpr variants (`FunctionCall`, `ScalarSubquery`, `InList`, `Between`) surface as `BridgeError::HavingDeferred(...)`. Test `bridge_having_deferred_e_variant_surfaces_error` is the contract; no in-tree construction site builds a HAVING with E-variants today. ### `column_resolution.rs` extension (+211 lines, +6 tests) ```rust pub fn output_schema_for_aggregate( input: &Schema, by: &[ColumnId], aggs: &[AggIntent], ) -> Schema; pub fn resolve_named_keys(keys: &[ColumnRef], schema: &Schema) -> Result<Vec<ColumnId>, ResolveError>; ``` Mirrors canonical `query_expr::QueryExpr::output_schema_in`'s Aggregate arm: outputs `by`-columns positionally + one column per `AggIntent::output_column(probe)`, strips `time_index`, sets `unique_keys = [by]`. This was Step β TODO #1. Step γ2-γ4 consumers descending into legacy `Aggregate.input` should pass `output_schema_for_aggregate(parent_schema, &bridged.by, &bridged.aggs)` as the inner subtree's `parent_schema`. ### Demo wire in `physical/allocator.rs::alloc_node` The legacy `QueryExpr::Aggregate { keys, aggs, having, input }` arm now calls the bridge to derive canonical-shape data and enrich `NodeAnnotation.rationale` with intent kinds + group-by column count. Emit shape stays legacy; behavior unchanged (only the rationale string carries extra info). Proves the bridge is reachable. ## Construction-site migration: 0 (intentional) Per strategy (c), all ~10 construction sites in `query_parser/{promql,sql}.rs`, `legacy_lower::lower_aggregate`, and the optimizer rewrite path still emit `legacy_expr::QueryExpr::Aggregate`. They migrate in γ7 once no legacy `input` subtree remains (SketchAgg, WindowedAgg, TopK migrations land first in γ2-γ4). ## Build + test - `cargo build --release -p controller` — clean - `cargo build --release -p query_engine_rust` — clean - `cargo test -p controller --lib` — **688 passed** (was 674; +14 new bridge + column_resolution tests) - `cargo test -p controller --bin controller` — 27 passed ## Known caveats - Canonical `QueryExpr::Aggregate.having` field is `Option<HavingPredicate>` where `HavingPredicate(pub String)`, NOT typed `Option<Predicate>` as the spec text suggested. Bridge renders the converted `Predicate` via `format!("{pred:?}")`. Round-tripping from the string is a follow-up (when canonical `having` upgrades to typed `Predicate`). - `BridgeError` can't derive `PartialEq` (QueryExprError doesn't); tests use `matches!`. Non-blocking. - `optimizer::engine::HydraConversion` (Step β TODO #7) NOT migrated — out of γ1's stated scope, deferred to a later γ sub-PR. ## Diff: 5 files, +618 / -4 Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
zzylol
added a commit
that referenced
this pull request
May 13, 2026
Post-M2.3 reorg #5 of 8. Creates a new `sketch_db::lifecycle` module and moves `SchemaEvictionService` (and friends) into `lifecycle/eviction.rs`. The sid-level lifecycle FIELDS and methods on `SketchInstanceMetadata` / `SketchStore` stay in `index/` next to the data they gate — only the schedule-driven *service* moves here. `schema/mod.rs` re-exports the eviction types under their legacy path (`sketch_db::schema::SchemaEvictionService`) so existing consumers compile unchanged. Canonical home is now `sketch_db::lifecycle::*`. This is the structural skeleton for the upcoming sub-PRs: - #6 will add `lifecycle::reconcile_from_streaming_config`, reimplementing schema/'s reconcile semantics over the sid catalog. - #7 will then delete `schema/` once the only resident is the thin re-export. 783/783 lib tests pass. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
PR D — first end-to-end correctness test for the modified-OTLP sketch hot path. Targeted at the CountMin path that lands in PR B (#6); other sketch types get parallel test arms as PR C delivers their decoders.
Branched off PR #6 (PR B) so it can be merged as soon as PR B lands. Targets
feat/decode-modified-otel-sketchesso only the new test file shows up in the diff.What it validates
Real OTLP HTTP request →
route_modified_otlp_sketches_to_precompute→CountMinSketchAccumulator::from_sketchlib_proto_bytes→WorkerMessage::AccumulatorInput→ workersketch_panesmerge → window close →CapturingOutputSink→ matrix matches expected.A 2x4
CountMinStatematrix with known values (row 0 = [1,2,3,4],row 1 = [5,6,7,8]) is encoded asCountMinStateproto bytes, wrapped in aMetric.data = CountMinSketch{ data_points: [CountMinSketchDataPoint { sketch, encoding=PROTO, attributes=[{service: "auth"}], time_unix_nano=100ms }] }payload, POSTed over real OTLP HTTP, and the resulting storedCountMinSketchAccumulatormatrix is asserted bit-identical.Validates end-to-end:
asap_otel_protoexposesMetric.data::Countminsketchand the typedCountMinSketchDataPointfields with prost build pipeline working.route_modified_otlp_sketches_to_precomputewalks the oneof, flattens per-variant data points intoModifiedOtlpSketchDp, matches by metric name, computes group key from theserviceattribute viaIngestState::extract_group_key_for, emitsWorkerMessage::AccumulatorInput.CountMinSketchAccumulator::from_sketchlib_proto_bytesdecodesCountMinState, picks int64 path, reshapes toVec<Vec<f64>>, constructs viafrom_legacy_matrix— round-trip exact.Test layout
New file
asap-query-engine/tests/e2e_modified_otlp_sketch_path.rs(310 lines) with helpers for buildingAggregationConfig,PrecomputeEngineConfig,CountMinStateproto, fullExportMetricsServiceRequest, and the OTLP HTTP client.Validation
cargo check -p query_engine_rust --tests: cleancargo clippy -p query_engine_rust --tests -- -D warnings: cleancargo fmt --check: cleancargo test --test e2e_modified_otlp_sketch_path:1 passedcargo test --lib: 439 passed, 0 failed (no regressions)What's still PR C territory
*_ENCODING_PROTO_DELTA) — PR C-delta*_ENCODING_MSGPACK) — PR I🤖 Generated with Claude Code