Skip to content

feat(engine): retire legacy SketchStore read path; SketchIndex only #166

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zzylol merged 1 commit into from
May 13, 2026
Merged

feat(engine): retire legacy SketchStore read path; SketchIndex only #166

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 42 additions & 92 deletions data_plane/src/query_engines/asap_query_engine/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1044,101 +1044,51 @@ impl ASAPQueryEngine {
params.is_exact_query
);

// M2.3.5b — when a `SketchIndex` is attached AND the agg
// config is resolvable, read precomputes from SketchIndex
// (sid-keyed) instead of the legacy SketchStore. DualWriteSink
// mirrors writes to both, so the data is identical; this just
// moves the read off the agg_id-keyed store. When sketch_index
// is `None` (tests without one) or the agg_cfg is missing
// (eviction race), fall back to the legacy store.
if let Some(idx) = self.sketch_index.as_ref() {
let cfg = self.streaming_config_snapshot();
if let Some(agg_cfg) = cfg.get_aggregation_config(params.aggregation_id) {
let raw = idx.query_precomputes_by_agg(
&params.metric,
agg_cfg.aggregation_type,
params.start_timestamp,
params.end_timestamp,
);
let result: TimestampedBucketsMap = if params.is_exact_query {
// Sliding-window mode requires bit-exact (start,
// end) match. SketchIndex's range query returns
// any windows fully within [start, end] — filter
// post-hoc to recover the exact semantics the
// legacy `query_precomputed_output_exact` had.
raw.into_iter()
.map(|(k, v)| {
let filtered: Vec<_> = v
.into_iter()
.filter(|((s, e), _)| {
*s == params.start_timestamp
&& *e == params.end_timestamp
})
.collect();
(k, filtered)
// M2.3.6f — engine reads precomputes from SketchIndex only.
// The legacy `Store::query_precomputed_output*` fallback has
// been retired now that DualWriteSink (M2.3.4b) → SketchIndexSink
// (M2.3.6a) writes exclusively to SketchIndex and
// BackfillService (M2.3.6e) mirrors replays there too.
//
// Tests that don't attach a SketchIndex now get `Ok(empty)`
// here. Anything deeper than smoke-test coverage was already
// setting one (M2.3.5b made it mandatory in production).
let Some(idx) = self.sketch_index.as_ref() else {
return Ok(TimestampedBucketsMap::new());
};
let cfg = self.streaming_config_snapshot();
let Some(agg_cfg) = cfg.get_aggregation_config(params.aggregation_id) else {
return Ok(TimestampedBucketsMap::new());
};
let raw = idx.query_precomputes_by_agg(
&params.metric,
agg_cfg.aggregation_type,
params.start_timestamp,
params.end_timestamp,
);
let result: TimestampedBucketsMap = if params.is_exact_query {
// Sliding-window mode requires bit-exact (start, end)
// match. SketchIndex's range query returns any windows
// fully within [start, end] — filter post-hoc to recover
// the exact semantics the retired
// `query_precomputed_output_exact` had.
raw.into_iter()
.map(|(k, v)| {
let filtered: Vec<_> = v
.into_iter()
.filter(|((s, e), _)| {
*s == params.start_timestamp
&& *e == params.end_timestamp
})
.filter(|(_, v)| !v.is_empty())
.collect()
} else {
raw
};
return Ok(result);
}
}

let store_query_start_time = Instant::now();

let result = if params.is_exact_query {
debug!(
"Sliding window query: Looking for exact window [{}, {}]",
params.start_timestamp, params.end_timestamp
);
let res = self.store.query_precomputed_output_exact(
&params.metric,
params.aggregation_id,
params.start_timestamp,
params.end_timestamp,
);
if let Ok(ref outputs) = res {
let store_query_duration = store_query_start_time.elapsed();
debug!(
"Sliding window exact query took: {:.2}ms, found {} unique keys",
store_query_duration.as_secs_f64() * 1000.0,
outputs.len()
);
}
res
.collect();
(k, filtered)
})
.filter(|(_, v)| !v.is_empty())
.collect()
} else {
debug!(
"Tumbling window query: range [{}, {}]",
params.start_timestamp, params.end_timestamp
);
let res = self.store.query_precomputed_output(
&params.metric,
params.aggregation_id,
params.start_timestamp,
params.end_timestamp,
);
if res.is_ok() {
let store_query_duration = store_query_start_time.elapsed();
debug!(
"Tumbling window range query took: {:.2}ms",
store_query_duration.as_secs_f64() * 1000.0
);
}
res
raw
};

result.map_err(|e| {
format!(
"Error querying store for metric {}, agg {}, range [{}, {}]: {}",
params.metric,
params.aggregation_id,
params.start_timestamp,
params.end_timestamp,
e
)
})
Ok(result)
}

/// Executes the full store query plan and returns merged results
Expand Down
39 changes: 27 additions & 12 deletions data_plane/src/tests/schema_timeline_dispatch_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,23 +98,35 @@ fn build_engine(
streaming_config: Arc<StreamingConfig>,
schemas: Arc<SchemaRegistry>,
store: Arc<dyn Store>,
sketch_index: Arc<crate::stores::sketch_db::index::SketchIndex>,
_query_for_agg_id: u64,
) -> ASAPQueryEngine {
let hot_reload = HotReloadStreamingConfig::from_arc(streaming_config);
ASAPQueryEngine::new_with_hot_reload(store, hot_reload, 1)
.with_schema_registry(schemas)
.with_sketch_index(sketch_index)
}

/// Insert a single `SumAccumulator` window at `ts` into `agg_id`.
/// Uses `(ts, ts)` for the window start/end pair to match the
/// existing test-utility pattern in `engine_factories` — the engine
/// treats those as single-point buckets aligned to the tumbling
/// window, so a query whose range contains `ts` picks up the data.
fn seed_sum_at(store: &SketchStore, agg_id: u64, ts: u64, host: &str, sum: f64) {
/// Mirrors the live ingest path's M2.3.6e write-path: data lands in
/// BOTH the legacy SketchStore AND the new SketchIndex so the
/// engine's M2.3.6f read path (SketchIndex-only) sees it.
fn seed_sum_at(
store: &SketchStore,
sketch_index: &crate::stores::sketch_db::index::SketchIndex,
streaming_config: &StreamingConfig,
agg_id: u64,
ts: u64,
host: &str,
sum: f64,
) {
let key = Some(KeyByLabelValues {
labels: vec![host.to_string()]});
let output = PrecomputedOutput::new(ts, ts, key, agg_id);
let acc = SumAccumulator::with_sum(sum);
if let Some(agg_cfg) = streaming_config.get_aggregation_config(agg_id) {
sketch_index.ingest_precompute_for_agg_config(agg_cfg, &output, &acc);
}
store
.insert_precomputed_output(output, Box::new(acc))
.expect("seed insert must succeed");
Expand Down Expand Up @@ -148,10 +160,11 @@ fn sum_query_across_reconfigure_boundary_returns_combined_full_result() {
// per segment:
// agg_1's sub-range is `[QUERY_START_MS, BOUNDARY_MS]`
// agg_2's sub-range is `[BOUNDARY_MS, QUERY_TIME_MS]`
seed_sum_at(&store, 1, AGG1_SAMPLE_MS, "A", 10.0);
seed_sum_at(&store, 2, AGG2_SAMPLE_MS, "A", 20.0);
let sketch_index = Arc::new(crate::stores::sketch_db::index::SketchIndex::new());
seed_sum_at(&store, &sketch_index, &streaming_config, 1, AGG1_SAMPLE_MS, "A", 10.0);
seed_sum_at(&store, &sketch_index, &streaming_config, 2, AGG2_SAMPLE_MS, "A", 20.0);

let engine = build_engine(streaming_config, schemas, store, 2);
let engine = build_engine(streaming_config, schemas, store, sketch_index, 2);

let (_labels, qr) = engine
.handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC)
Expand Down Expand Up @@ -201,9 +214,10 @@ fn sum_query_with_purged_segment_returns_partial_with_warnings() {
));
// Only agg_2 has data; agg_1's data is assumed gone with the
// Purged classification.
seed_sum_at(&store, 2, AGG2_SAMPLE_MS, "A", 20.0);
let sketch_index = Arc::new(crate::stores::sketch_db::index::SketchIndex::new());
seed_sum_at(&store, &sketch_index, &streaming_config, 2, AGG2_SAMPLE_MS, "A", 20.0);

let engine = build_engine(streaming_config, schemas, store, 2);
let engine = build_engine(streaming_config, schemas, store, sketch_index, 2);

let (_labels, qr) = engine
.handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC)
Expand Down Expand Up @@ -251,9 +265,10 @@ fn single_schema_query_falls_through_to_default_path() {
streaming_config.clone(),
CleanupPolicy::NoCleanup,
));
seed_sum_at(&store, 7, QUERY_TIME_MS, "A", 42.0);
let sketch_index = Arc::new(crate::stores::sketch_db::index::SketchIndex::new());
seed_sum_at(&store, &sketch_index, &streaming_config, 7, QUERY_TIME_MS, "A", 42.0);

let engine = build_engine(streaming_config, schemas, store, 7);
let engine = build_engine(streaming_config, schemas, store, sketch_index, 7);

let (_labels, qr) = engine
.handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC)
Expand Down