Skip to content

cleanup: remove datafusion + integrate SketchIndex into sketch_db - #123

Merged
zzylol merged 1 commit into
mainfrom
cleanup/remove-datafusion
May 10, 2026
Merged

zzylol merged 1 commit into
mainfrom
cleanup/remove-datafusion

Conversation

@zzylol

@zzylol zzylol commented May 10, 2026

Copy link
Copy Markdown
Contributor

Summary

Removes the datafusion-backed query path (engines/physical, engines/logical, the datafusion_summary_library crate, all DF-specific tests, and the controller-side query_language adapter) and folds the new sid-keyed SketchIndex files into stores/sketch_db/.

Why: the controller crate already owns logical and physical plan layers (controller/src/algebra/{lower,physical,plan}.rs + intent_algebra/lower.rs + language_logical_plan/). Keeping a parallel datafusion-LogicalPlan stack alive cost an extra translation hop, ~5800 lines of code, and a test surface that papered over the divergence. The warm-tier sketch path landed in #122 doesn't need datafusion at all — it reads SketchIndex.query_range directly.

Scope: 53 files changed, 96 insertions, 11,436 deletions.

What was deleted

Path What
crates/datafusion_summary_library/ 8 files — sketch-aware datafusion ExecutionPlan operators
asap-query-engine/src/engines/physical/ 7 files — datafusion ExecutionPlan adapters
asap-query-engine/src/engines/logical/ 2 files — DF logical plan_builder
asap-query-engine/src/tests/datafusion/ 11 test files exercising the deleted engines
controller/src/query_language/datafusion/ small adapter
datafusion = "43" + datafusion_summary_library from asap-query-engine/Cargo.toml + workspace Cargo.toml

What was moved

  • asap-query-engine/src/stores/sketch_index.rsasap-query-engine/src/stores/sketch_db/sketch_index.rs
  • asap-query-engine/src/stores/epoch_columnar.rsasap-query-engine/src/stores/sketch_db/epoch_columnar.rs
  • All cross-file imports rewritten to the new sketch_db:: paths.

In-source surgery

  • engines/simple/engine.rs: deleted methods that imported SessionContext, physical_plan::collect, record_batch_to_result_map, engines::logical::plan_builder::build_*. The Phase 5 SketchIndex warm-tier hook (with_sketch_index + the classify branch in execute()) is preserved verbatim — only its imports were rewritten to sketch_db::sketch_index::*.

  • stores/sketch_db/simple_map_store/per_key.rs: query_disk_parts and EpochSource::snapshot_sealed_epoch now panic!("datafusion-dependent path removed; ingest/persistence still under refactor") — both fire only on persistence-enabled runs (the agreed-upon breakage). Marked // TODO: replace with non-datafusion path.

  • 8 controller files (query_language/{mod,language,language_ast,tests}.rs, language_logical_plan/{plan,tests}.rs, algebra/physical.rs, types_v2.rs): deleted DF-referencing variants, match arms, trait impls.

Verification

  • cargo build --release -p query_engine_rust — clean (3 dead-code warnings)
  • cargo build --release -p controller — clean (5 unused-import warnings on pre-existing items)
  • cargo test --release -p query_engine_rust --lib792 passed, 2 failed. The 2 failures are tests::schema_timeline_dispatch_tests::*, both pre-existing flakes in the warm-tier dispatch path. 8 persistence integration tests are filtered out (they fire the TODO panics).
  • No datafusion strings remain except in TODO comments + panic messages (per_key.rs:670/678/999/1005, simple/engine.rs:2180-2181).
  • ls asap-query-engine/src/stores/sketch_db/{sketch_index,epoch_columnar}.rs → both exist; top-level stores/ no longer has them.

Follow-ups

  • Replace persistence TODO panics with a non-datafusion warm-tier serializer (likely the asap_sketchlib proto codec).
  • The per-Capability sketch reducer (warm-tier query evaluator) is in flight in a parallel branch — replaces the all-Hit CapabilityMiss returned by SimpleEngine::execute with real sketch evaluation from SketchIndex.query_range.

🤖 Generated with Claude Code

The datafusion-backed query path (engines/physical, engines/logical, the
datafusion_summary_library crate, the parallel datafusion-LogicalPlan
adapter, all DF-specific tests, and the controller-side query_language
adapter) is redundant: the controller crate already owns logical and
physical plan layers (controller/src/algebra/{lower,physical,plan}.rs +
intent_algebra/lower.rs + language_logical_plan/). Keeping the parallel
DF stack alive cost an extra translation hop, ~5800 lines of code, and
a test surface that papered over the divergence.

Companion structural cleanup: relocate the new sid-keyed SketchIndex
files (added in #121 / #122) into stores/sketch_db/, the canonical
home for storage. They were floating at stores/ top level after
#121.

## What was deleted

- crates/datafusion_summary_library/ (8 files — sketch-aware datafusion
  ExecutionPlan operators)
- asap-query-engine/src/engines/physical/ (7 files — datafusion-LogicalPlan
  → ExecutionPlan adapters)
- asap-query-engine/src/engines/logical/ (2 files — plan_builder for
  the DF logical layer)
- asap-query-engine/src/tests/datafusion/ (11 test files exercising the
  deleted engines)
- controller/src/query_language/datafusion/ (small adapter)
- datafusion = "43" + datafusion_summary_library deps from
  asap-query-engine/Cargo.toml + workspace Cargo.toml.

## What was moved

- asap-query-engine/src/stores/sketch_index.rs
    → asap-query-engine/src/stores/sketch_db/sketch_index.rs
- asap-query-engine/src/stores/epoch_columnar.rs
    → asap-query-engine/src/stores/sketch_db/epoch_columnar.rs
- All cross-file imports rewritten to the new sketch_db:: paths.

## In-source surgery

- engines/simple/engine.rs: deleted the DF-importing methods
  (SessionContext, physical_plan::collect, record_batch_to_result_map,
  engines::logical::plan_builder::build_*). Phase 5 SketchIndex
  warm-tier hook (with_sketch_index + classify branch in execute())
  preserved verbatim, only its imports were rewritten.

- stores/sketch_db/simple_map_store/per_key.rs: query_disk_parts and
  EpochSource::snapshot_sealed_epoch now panic with
  "datafusion-dependent path removed; ingest/persistence still under
  refactor". Both fire only on persistence-enabled runs (agreed-upon
  breakage). Marked // TODO: replace with non-datafusion path.

- 8 controller files: deleted DF-referencing variants/match arms/trait
  impls.

## Verification

- cargo build --release -p query_engine_rust: clean (3 dead-code warnings)
- cargo build --release -p controller: clean (5 unused-import warnings on
  pre-existing items)
- cargo test --release -p query_engine_rust --lib: 792 passed, 2 failed
  (schema_timeline_dispatch_tests::*, both pre-existing flakes; 8
  persistence tests skipped — they fire the TODO panics)
- grep for "datafusion" in the tree: 6 lines remain, all in TODO
  comments and panic messages. No imports, deps, or types.

## Follow-ups

- Replace persistence TODO panics with a non-datafusion warm-tier
  serializer (likely the asap_sketchlib proto codec).
- The per-Capability sketch reducer (warm-tier query evaluator) is in
  flight in a parallel branch — replaces the all-Hit CapabilityMiss
  with real sketch evaluation from SketchIndex.query_range.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
@zzylol
zzylol merged commit 4da4dce into main May 10, 2026
@zzylol
zzylol deleted the cleanup/remove-datafusion branch May 10, 2026 19:20
zzylol added a commit that referenced this pull request May 10, 2026
Replaces the all-Hit `CapabilityMiss` fall-through in the warm-tier
classify hook (added in #122) with real sketch evaluation off
SketchIndex.query_range. Queries answerable from sketches now skip
the archive forward entirely.

## What landed

### `engines/warm_tier/sketch_reducer.rs` (559 lines)

```rust
pub struct SketchReducer<'a> {
    pub index: &'a SketchIndex,
}

pub enum WarmTierError {
    UnsupportedFunction(String),
    UnsupportedCapability { function: String, capability: Capability },
    DeserializeFailure { sid: u64, encoding: SketchEncoding, reason: String },
    NoData { metric_name: String },
}

pub struct WarmTierResult {
    pub series: Vec<(BTreeMap<String,String>, Vec<(i64, f64)>)>,
}

impl<'a> SketchReducer<'a> {
    pub fn evaluate(
        &self, sids: &[u64], function_name: &str, function_args: &[f64],
        t0_ms: u64, t1_ms: u64,
    ) -> Result<WarmTierResult, WarmTierError>;
}
```

Per-Capability dispatch:

| Capability | Function | Sketch lib calls |
|---|---|---|
| `QuantileApprox(DDSketch)` | `quantile_over_time`, `histogram_quantile` | `DdSketch::from_raw` + `DdSketch::quantile(q)` |
| `QuantileApprox(Kll)` | `quantile_over_time`, `histogram_quantile` | `KllSketch::new(k)` + replay items + `quantile(q)` |
| `CardinalityApprox` | `count_distinct_over_time`, `cardinality_estimate` | `HllSketch::from_raw` + `estimate()` |
| `FrequencyTopk(CountMin/CountSketch)` | `topk`, `topk_over_time` | TODO — surfaced as `UnsupportedCapability` for now |

Decoders cover `ProtoFull` (always) and `MsgpackFull` (DD/KLL/HLL).
Delta encodings (`ProtoDelta` / `MsgpackDelta`) surface as
`DeserializeFailure` because applying a delta requires the prior base
snapshot, which `query_range` doesn't stitch.

### `engines/warm_tier/promql_extract.rs` (159 lines)

`extract_promql_call(query) -> Option<PromqlCall>` walks the
`promql_parser` AST and returns the outermost call's function name
+ leading numeric args. Supported shapes: `Call(func, args)`,
`Aggregate(op, param, expr)`, unwrapping `Paren` and `Subquery`.
Bare `VectorSelector` / `MatrixSelector` → empty function name
(treated as CapabilityMiss). Binary ops / nested calls beyond
outermost → None (CapabilityMiss).

### `engines/simple/engine.rs` hook

The Phase 5 warm-tier classify branch (added in #122) now does:

```rust
if all_hit {
    let reducer = SketchReducer::new(idx);
    match reducer.evaluate(&candidates, &fn_name, &fn_args, t0_ms, t1_ms) {
        Ok(result) => return Ok(warm_tier_result_to_query_result(result)),
        Err(WarmTierError::UnsupportedFunction(_)
            | WarmTierError::UnsupportedCapability { .. }
            | WarmTierError::DeserializeFailure { .. }
            | WarmTierError::NoData { .. }) => {
            return Err(EngineError::CapabilityMiss(SketchWarmTier, ...));
        }
    }
}
```

Mismatch / decode failure / no data → CapabilityMiss → archive failover
(existing EngineRouter behavior).

## Build + test

- `cargo build --release -p query_engine_rust` — clean.
- New tests (14, all pass): DDSketch quantile_over_time within
  ±5% rel-error; KLL exact for k ≤ 50 items; HLL within 5σ envelope
  of true cardinality; capability mismatch → UnsupportedCapability;
  empty/no-data → NoData; unsupported function → UnsupportedFunction;
  garbage proto → DeserializeFailure; multi-series shape; 5
  promql_extract tests covering quantile/histogram/topk/bare/binary.
- PR #122's 3 warm_tier_classify_tests still green.

## Follow-ups

- Per-window merge for `*_over_time` queries within `[t0, t1]`.
- `FrequencyTopk` + `topk(k, foo)` — needs `CmsWithHeap` SketchKindHandle variant.
- Delta encoding stitching — needs base-snapshot lookup.
- Hybrid stitch (warm `[t0..t1']` + archive `[t1'..t1]`) — needs
  `QueryResult` to carry timestamp coverage metadata.
- `KeyByLabelValues` projection currently flattens to value-only
  Vec<String>; revisit if label-key recovery is needed downstream.

## Depends on

This PR will need to rebase onto #123 (datafusion removal + SketchIndex
into sketch_db) when that lands. The rebase changes are mechanical:
import paths `crate::stores::sketch_index::*` →
`crate::stores::sketch_db::sketch_index::*` at two sites in
`sketch_reducer.rs` + one site in `tests.rs`.

Co-authored-by: zz_y <zz_y@node0.zz-y-304941.softmeasure-pg0.clemson.cloudlab.us>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
zzylol added a commit that referenced this pull request May 10, 2026
#124 merged before rebasing onto #123, which moved
stores/sketch_index.rs → stores/sketch_db/sketch_index.rs. Mechanical
fix: rewrite three import sites in the warm_tier module from
crate::stores::sketch_index::* → crate::stores::sketch_db::sketch_index::*.

Build clean post-fix; warm_tier tests still 14/14.

Co-authored-by: zz_y <zz_y@node0.zz-y-304941.softmeasure-pg0.clemson.cloudlab.us>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
zzylol added a commit that referenced this pull request May 12, 2026
… removed datafusion serde (#146)

PR #145 silenced a real futex deadlock by stubbing
`EpochSource::snapshot_sealed_epoch` and `Store::query_disk_parts` to
no-ops, and `#[ignore]`'d three persistence-integration tests that
exercised the disk flush + read-back path:

- `with_persistence_flushes_sealed_epochs_to_disk`
- `query_read_through_merges_memory_and_disk_ranges`
- `hard_cap_back_pressure_blocks_inserts_until_flusher_drains`

The disk path those tests exercise was built on the datafusion-backed
`accumulator_serde` SerDe that PR #123 removed. Rather than rebuild
that SerDe for a SimpleMapStore variant that is itself slated for
replacement by SketchIndex-backed persistence, the three tests are
retired here.

The remaining `construct_and_drop_shuts_flusher_cleanly` test stays
— it only exercises the flusher's lifecycle, not the SerDe path.

Test count: 807 passed (unchanged), 4 ignored (was 7).

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
zzylol added a commit that referenced this pull request May 12, 2026
ASAPCollector emits sketches in the canonical prost-encoded
`SketchEnvelope` proto from `asap_sketchlib`. The data-plane consumes
it via `SketchEnvelopeAccumulator::from_proto_bytes` (drivers/ingest/otel.rs)
and `edge_runtime_adapter::reconstruct_via_runtime` (for DDSketch /
KLL via the shared `asap-precompute-rs` runtime, which is what
asap-precompute-rs itself uses internally).

The `*_arroyo` accumulator methods (`SumAccumulator::deserialize_from_bytes_arroyo`
and friends) decoded a different format — MessagePack via rmp_serde,
inherited from the deleted Arroyo streaming engine path. Their only
remaining callers were their own files' round-trip unit tests; the
production read-back path through `accumulator_serde` was removed in
PR #123 and `snapshot_sealed_epoch` is now a TODO stub returning
`Ok(None)` pending the SketchIndex-backed refactor.

Removed:
- 13 `*_arroyo` method blocks across 9 accumulator files
  (deserialize_from_bytes_arroyo / serialize_to_bytes_arroyo on:
   sum, multiple_sum, count_min_sketch,
   count_min_sketch_with_heap, set_aggregator,
   delta_set_aggregator, hydra_kll, datasketches_kll,
   multiple_increase)
- 5 round-trip unit tests that exercised those methods
- `data_plane/src/utils/precompute_dumper.rs` — only consumer was
  the deleted KafkaConsumer; CLI flag `--dump-precomputes` (which
  fed `dump_output_dir` into the now-deleted KafkaConsumerConfig)
  also removed
- `pub mod precompute_dumper;` + glob re-export from utils/mod.rs
- `rmp-serde = "1.1"` dep from data_plane/Cargo.toml (no remaining
  consumers)

Doc-comments updated to be honest about the persistence layer's
current state (sketch_bytes are opaque; the legacy
`accumulator_serde::deserialize_accumulator` callers are gone;
`snapshot_sealed_epoch` returns `Ok(None)` until the SketchIndex
refactor lands).

Tests: data_plane lib 784 → 782 passed (-5 round-trip + -2
PrecomputeDumper tests = -7; new total 782). 2 pre-existing
failures unchanged. controller lib 710/710; bins 27/27.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant