feat(ingest): consume asap-precompute-rs for envelope/delta/merge logic

Cargo.toml

@@ -17,6 +17,22 @@ members = [
 edition = "2021"
 version = "0.1.0"
 
+# Phase 3 step 3: backend ingest path consumes `asap-precompute-rs`,
+# which (today) path-deps `asap_sketchlib` from `../../asap_sketchlib`
+# (see `ASAPCollector/asap-precompute-rs/Cargo.toml`). The backend
+# crates depend on `asap_sketchlib` via a git URL. Without this
+# `[patch]`, cargo treats the two sources as distinct crates and the
+# `asap_sketchlib::sketches::*` types from the two sides won't unify
+# at the FFI boundary (the wrapper-`inner()` → backend-accumulator
+# bridge in `precompute_operators::edge_runtime_adapter`).
+#
+# Pointing both at the local checkout is the same pattern downstream
+# Strategy-B adapter consumers will use; flip this to a `git = "..."`
+# entry once `asap-precompute-rs` itself moves to a git-sourced
+# `asap_sketchlib`.
+[patch."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/ProjectASAP/asap_sketchlib"]
+asap_sketchlib = { path = "../asap_sketchlib" }
+
 [workspace.dependencies]
 # Shared external deps (used by 2+ crates)
 serde = { version = "1.0", features = ["derive"] }

README.md

@@ -66,6 +66,58 @@ ASAPQuery has four main components: the **asap-planner-rs** generates sketch con
 # Note: Arroyo fork lives at https://github.com/ProjectASAP/arroyo
 ```
 
+### Ingest path consumes asap-precompute-rs
+
+Phase 3 step 3 of the ASAP edge-framework migration (see
+`docs/design-asap-edge-framework.md` and ADR-0002). The backend's
+ingest path now delegates the **shared** envelope-parsing,
+sketch-reconstruction, and sketch-merge logic to
+[`asap-precompute-rs`](https://github.com/ProjectASAP/ASAPCollector/tree/main/asap-precompute-rs)
+— the host-neutral Rust edge runtime — so the same code runs in
+agents (Rust shims) and the backend.
+
+**What moved out of this repo (to asap-precompute-rs):**
+
+- Envelope wire-format parsing (`SketchEnvelope` runtime view; was
+  inlined in every backend `*Accumulator::from_sketchlib_proto_bytes`).
+- Per-sketch state extraction from the
+  `asap_sketchlib::proto::sketchlib::SketchEnvelope.sketch_state`
+  oneof.
+- Sketch reconstruction (DDSketch + KLL today;
+  HLL / CountSketch / CountMinSketch are tracked under
+  [ProjectASAP/ASAPCollector#243](https://github.com/ProjectASAP/ASAPCollector/issues/243)).
+- Cross-runtime sketch merge (asap-precompute-rs's `Sketch::merge`).
+
+**What stays in this repo:**
+
+- Query-side engine — PromQL aggregation, storage, query planning.
+- Backend's per-accumulator query-side surface (`AggregateCore`,
+  `query_statistic`, `MergeableAccumulator`, ...).
+- Sparse-delta application
+  (`apply_modified_otlp_delta_bytes` /
+  `*Accumulator::apply_proto_delta_bytes`) — `asap_sketchlib`
+  doesn't yet expose the `compute_delta` family upstream
+  (Go's `sketchlib-go` has it; tracked upstream), so
+  asap-precompute-rs's wrappers fall back to "always full" delta
+  encoding. Backend's typed-delta apply is independent and stays.
+
+**Bridge layer:**
+[`asap-query-engine/src/precompute_operators/edge_runtime_adapter.rs`](asap-query-engine/src/precompute_operators/edge_runtime_adapter.rs)
+re-exports the asap-precompute-rs runtime view types
+(`SketchEnvelope`, `Encoding`, `SketchType`, the `Sketch` trait
+family) and provides the
+`reconstruct_via_runtime` / `unwrap_envelope_state` /
+`encode_ddsketch_envelope` / `merge_ddsketches_via_runtime`
+helpers used by the backend's `decode_modified_otlp_sketch_bytes`
+hot path.
+
+**Acceptance tests:**
+[`asap-query-engine/tests/edge_runtime_consumes_precompute_rs.rs`](asap-query-engine/tests/edge_runtime_consumes_precompute_rs.rs)
+contains round-trip + structural tests proving asap-precompute-rs
+sits in the backend's ingest path. HLL / CountSketch / CountMinSketch
+tests are present and `#[ignore]`'d with a comment pointing at issue
+#243.
+
 ## Coming soon
 
 1. Drop-in ASAPQuery artifact that works with your existing pre-configured Prometheus-Grafana stack

asap-query-engine/Cargo.toml

@@ -62,6 +62,26 @@ reqwest = { version = "0.11", default-features = false, features = ["json", "rus
 tracing-appender = "0.2"
 elastic_dsl_utilities.workspace = true
 asap_sketchlib = { git = "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/ProjectASAP/asap_sketchlib", branch = "main" }
+# Phase 3 step 3: backend ingest path consumes asap-precompute-rs for the
+# SHARED envelope-parsing / delta-apply / sketch-reconstruction / merge
+# logic. Query-side engine (PromQL aggregation, storage, query planner)
+# is unaffected. See README §"Ingest path consumes asap-precompute-rs"
+# and `docs/adr/ADR-phase3-asap-precompute-rs.md`.
+# `asap-precompute-rs` lives in the `ASAPCollector` repo as a sibling
+# crate alongside non-Rust submodules (telegraf / opentelemetry-go /
+# opentelemetry-collector / etc.) that cargo can't recursively fetch
+# during a `git` source resolution because the parent repo's submodule
+# pins reference commits not reachable via `git fetch <commit>`. The
+# `asap-precompute-rs` Cargo.toml itself path-deps `asap_sketchlib`
+# (`../../asap_sketchlib`) — that path is meaningful only inside the
+# checkout, not from cargo's git cache.
+#
+# We therefore consume `asap-precompute-rs` via a sibling path dep,
+# matching how `asap-precompute-rs` itself path-deps `asap_sketchlib`.
+# Downstream CI / reproducible builds: clone `ASAPCollector` next to
+# `ASAPQuery-backend` (sibling directories under `~/repos/`) before
+# `cargo build`. The path is relative so any matching layout works.
+asap-precompute-rs = { path = "../../ASAPCollector/asap-precompute-rs" }
 # Persistence layer (SimpleMapStore parts / manifest / Tier-2 cache)
 moka = { version = "0.12", features = ["sync"] }
 memmap2 = "0.9"

asap-query-engine/src/drivers/ingest/otel.rs

@@ -846,18 +846,75 @@ fn decode_modified_otlp_sketch_bytes(
 
     match encoding {
         ENCODING_PROTO => match kind {
+            // Phase 3 step 3: DDSketch and KLL envelope-parsing /
+            // sketch reconstruction route through the shared
+            // `edge_runtime_adapter`, which delegates to
+            // `asap-precompute-rs`'s `Sketch` trait. Backend's
+            // accumulator wraps the result. Byte parity with Go is
+            // covered by `asap_sketchlib` PRs #40 (DDSketch) and #41
+            // (KLL).
+            //
+            // HLL / CountSketch / CountMinSketch byte parity is
+            // tracked under ProjectASAP/ASAPCollector#243 — until it
+            // lands those three sketches keep using the backend's
+            // existing per-accumulator decoder.
+            SketchKind::DdSketch => {
+                use crate::precompute_operators::edge_runtime_adapter::{
+                    reconstruct_via_runtime, ReconstructedSketch, SketchType as RtSketchType,
+                };
+                // Prefer the asap-precompute-rs runtime path (envelope-
+                // wrapped bytes, the canonical edge-framework wire format).
+                // If the input is a bare `DdSketchState` (as some unit-test
+                // / pre-envelope agent payloads still emit, mirrored by the
+                // PR #14 contract on `from_sketchlib_proto_bytes`), the
+                // adapter returns an error decoding the envelope — fall
+                // back to the backend's native decoder which already
+                // accepts both shapes.
+                match reconstruct_via_runtime(RtSketchType::DDSketch, bytes) {
+                    Ok(ReconstructedSketch::DdSketch(inner)) => {
+                        Ok(Box::new(DDSketchAccumulator { inner }))
+                    }
+                    Ok(_) => Err(
+                        "edge_runtime_adapter returned non-DDSketch reconstruction".into(),
+                    ),
+                    Err(_) => Ok(Box::new(DDSketchAccumulator::from_sketchlib_proto_bytes(
+                        bytes,
+                    )?)),
+                }
+            }
+            SketchKind::Kll => {
+                use crate::precompute_operators::edge_runtime_adapter::{
+                    reconstruct_via_runtime, ReconstructedSketch, SketchType as RtSketchType,
+                };
+                // Same envelope-vs-bare-state handling as DDSketch above.
+                // Backend's KLL accumulator owns the wire-format-aligned
+                // `KllSketch` rather than the high-throughput `KLL<f64>`
+                // that asap-precompute-rs's `KLLWrapper` wraps internally
+                // — when the adapter succeeds, bridge by re-feeding the
+                // wrapper's snapshot bytes through backend's existing
+                // decoder. The envelope work (decode + state extraction
+                // + reconstruction) has already happened in the runtime
+                // adapter; this final step just reshapes into backend's
+                // accumulator type. On envelope-decode failure (bare
+                // state bytes) fall through to the native decoder.
+                match reconstruct_via_runtime(RtSketchType::KLLSketch, bytes) {
+                    Ok(ReconstructedSketch::Kll { snapshot_bytes }) => Ok(Box::new(
+                        DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(&snapshot_bytes)?,
+                    )),
+                    Ok(_) => Err(
+                        "edge_runtime_adapter returned non-KLL reconstruction".into(),
+                    ),
+                    Err(_) => Ok(Box::new(
+                        DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(bytes)?,
+                    )),
+                }
+            }
             SketchKind::CountMin => Ok(Box::new(
                 CountMinSketchAccumulator::from_sketchlib_proto_bytes(bytes)?,
             )),
             SketchKind::CountSketch => Ok(Box::new(
                 CountSketchAccumulator::from_sketchlib_proto_bytes(bytes)?,
             )),
-            SketchKind::Kll => Ok(Box::new(
-                DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(bytes)?,
-            )),
-            SketchKind::DdSketch => Ok(Box::new(DDSketchAccumulator::from_sketchlib_proto_bytes(
-                bytes,
-            )?)),
             SketchKind::Hll => Ok(Box::new(HllSketchAccumulator::from_sketchlib_proto_bytes(
                 bytes,
             )?)),