Skip to content

Commit 3d3d67b

Browse files
zzylolclaude
andcommitted
feat(pass2): share year-aligned window segments across queries (Q60)
Pass 2's window-composition rule gains shared segments: approximate single-estimate targets over windows of one scan (same estimate up to accuracy, same grouping, no filters) share one summary per segment of a grid whose width is the gcd of every window's lookback and offset. Each query merges the segments its range covers; composition builds identical segments, which the identical-expression rule merges. Merging disjoint KLLs is exact, so each segment is sized for the strictest consumer and no accuracy is split. Only the all-shared form is offered (Q62), as its own Stage 1 variant (Sharing::WindowSegments, "· shared segments") in which each grouped target has the segment form as its only alternative. A group is skipped when its windows are identical, the union has more than MAX_PANES segments, or its queries recur at different cadences. Example 3a gets one candidate with five 1-year KLL segments; 4a, repeated monthly, can also maintain them at ingestion time (Q61). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent cfafcf1 commit 3d3d67b

12 files changed

Lines changed: 431 additions & 81 deletions

File tree

‎crates/devtools/src/bin/stage_pipeline.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,9 @@
2020
// the queries as written and, when Pass 2's identical-expression rule
2121
// merges something, again with identical sub-DAGs shared ("· shared
2222
// input"), and, when the summary-capability rule applies, again with
23-
// one summary sized for its strictest consumer ("· shared summary"); in
23+
// one summary sized for its strictest consumer ("· shared summary"),
24+
// and, when queries read windows of one scan on a common grid, again
25+
// with one summary per shared segment ("· shared segments"); in
2426
// enumeration order, only those with a written physical candidate. A
2527
// repeating query's mergeable alternatives also come in tumbling panes
2628
// (Pass 2's window-composition rule), e.g. "Q1 Kll · tumbling 1m panes";
@@ -267,6 +269,7 @@ fn stage_pipeline(
267269
Sharing::Independent => "",
268270
Sharing::IdenticalExpressions => " · shared input",
269271
Sharing::SummaryCapability => " · shared summary",
272+
Sharing::WindowSegments => " · shared segments",
270273
};
271274
let written = candidate
272275
.physical

‎crates/devtools/tests/stage_pipeline.rs‎

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -183,13 +183,15 @@ fn example3b_lists_tumbling_candidates() {
183183
}
184184
}
185185

186-
/// Example 4, Pattern A repeated monthly: the same 486 candidates as the ad
187-
/// hoc batch (3a), and none maintained at ingestion time. The windows (1–5 y)
188-
/// are longer than the month between runs and no pane width fits, so
189-
/// nothing is maintainable. The selected plan is 3a's, now amortized over
190-
/// monthly runs instead of one run per hour of horizon.
186+
/// Example 4, Pattern A repeated monthly: the same 487 logical candidates
187+
/// as the ad hoc batch (3a), 486 independent or sharing input plus the one
188+
/// sharing 1-year segments (Q60). The windows (1–5 y) are longer than the
189+
/// month between runs and no pane width fits, so only the shared segments
190+
/// can be maintained at ingestion time (Q61): 488 plans. The selected plan
191+
/// is 3a's, now amortized over monthly runs instead of one run per hour of
192+
/// horizon.
191193
#[test]
192-
fn example4a_repeats_monthly_with_nothing_maintainable() {
194+
fn example4a_repeats_monthly_with_only_segments_maintainable() {
193195
let once = generate(&[
194196
"--example",
195197
"planner-layering-3a",
@@ -205,10 +207,14 @@ fn example4a_repeats_monthly_with_nothing_maintainable() {
205207
let physical = monthly["stage2_physical_asap"]["candidates"]
206208
.as_array()
207209
.unwrap();
208-
assert_eq!(physical.len(), 486);
209-
assert!(physical
210+
assert_eq!(physical.len(), 488);
211+
let maintained: Vec<_> = physical
210212
.iter()
211-
.all(|p| !p["label"].as_str().unwrap().contains("ingestion time")));
213+
.map(|p| p["label"].as_str().unwrap())
214+
.filter(|label| label.contains("ingestion time"))
215+
.collect();
216+
assert_eq!(maintained.len(), 1, "{maintained:?}");
217+
assert!(maintained[0].ends_with("shared segments · ingestion time: Kll ×5 panes"));
212218
let selected = |d: &Value| {
213219
d["stage3_selection"]["selected"]
214220
.as_str()

‎crates/integration-tests/tests/planner_layering_common/mod.rs‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -575,6 +575,17 @@ pub fn query_option(
575575
}
576576
}
577577

578+
/// Whether `dag` shares a summary build across queries: Pass 2's
579+
/// shared-segment rule (Q60), whose segments several queries merge.
580+
pub fn shares_segments(dag: &impl ExportedDag, query_roots: &[LogicalASAPNodeId]) -> bool {
581+
cross_query_nodes(dag, query_roots).into_iter().any(|id| {
582+
matches!(
583+
dag.payload(id),
584+
LogicalASAPOperatorPayload::SummaryAgg { .. }
585+
)
586+
})
587+
}
588+
578589
/// Nodes reachable from more than one query root.
579590
pub fn cross_query_nodes(
580591
dag: &impl ExportedDag,

‎crates/integration-tests/tests/planner_layering_example3.rs‎

Lines changed: 44 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -158,13 +158,17 @@ fn stage1_a_keeps_five_independent_klls() {
158158
assert_eq!(sizes.len(), 1, "one ε, one KLL size: {sizes:?}");
159159
}
160160

161-
/// The identical-expression rule shares only raw input (scan, range, shift), never a quantile or summary, and keeps the unshared variant.
161+
/// The identical-expression rule shares only raw input (scan, range, shift), never a quantile or summary, and keeps the unshared variant. (The shared-segment candidate shares its segments; see below.)
162162
#[test]
163163
fn stage1_a_identical_expression_rule_shares_only_raw_input() {
164164
let run = run_a();
165165
assert!(run.logical.iter().any(|c| c.shared_input));
166166
assert!(run.logical.iter().any(|c| !c.shared_input));
167-
for c in &run.logical {
167+
for c in run
168+
.logical
169+
.iter()
170+
.filter(|c| !shares_segments(&c.dag, &c.query_roots))
171+
{
168172
for id in cross_query_nodes(&c.dag, &c.query_roots) {
169173
let kind = relational(c.dag.payload(id));
170174
assert!(
@@ -176,24 +180,35 @@ fn stage1_a_identical_expression_rule_shares_only_raw_input() {
176180
}
177181
}
178182

179-
/// One Exponential Histogram of KLLs over [T − 5y, T] serves all five queries, each through its own merge and p99 estimate.
183+
/// The windows' boundaries lie on a 1-year grid, so one set of five 1-year
184+
/// KLL segments over [T − 5y, T] serves all five queries (Q60, in place of
185+
/// the spec's Exponential Histogram): each query merges the segments its
186+
/// range covers and estimates p99 from its merge.
180187
#[test]
181-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram"]
182-
fn stage1_a_window_composition_adds_one_eh_for_all_five() {
188+
fn stage1_a_shared_segments_serve_all_five() {
183189
let run = run_a();
184-
let found = run.logical.iter().find(|c| {
185-
windowed_builds(c).iter().any(|(_, a, form, readers)| {
186-
*a == SketchAlgorithm::Kll
187-
&& matches!(form, WindowForm::ExponentialHistogram { horizon_ms } if *horizon_ms >= 5 * YEAR_MS)
188-
&& readers.len() == 5
189-
})
190-
});
191-
let c = found.expect("a candidate with one EH of KLLs read by all five queries");
192-
let (eh, ..) = windowed_builds(c)
193-
.into_iter()
194-
.find(|(_, _, form, _)| matches!(form, WindowForm::ExponentialHistogram { .. }))
195-
.unwrap();
196-
let estimates = estimates_of(&c.dag, eh);
190+
let c = run
191+
.logical
192+
.iter()
193+
.find(|c| shares_segments(&c.dag, &c.query_roots))
194+
.expect("a candidate sharing window segments");
195+
let segments = windowed_builds(c);
196+
assert_eq!(segments.len(), 5, "{}", c.id);
197+
for (_, algorithm, form, _) in &segments {
198+
assert_eq!(*algorithm, SketchAlgorithm::Kll);
199+
assert_eq!(*form, WindowForm::Tumbling { length_ms: YEAR_MS });
200+
}
201+
let read: BTreeSet<usize> = segments.iter().flat_map(|(.., r)| r.clone()).collect();
202+
assert_eq!(read, (0..5).collect());
203+
// Each query merges as many segments as its range has years.
204+
for (q, (_, lookback, _)) in PATTERN_A.iter().enumerate() {
205+
let covering = segments.iter().filter(|(.., r)| r.contains(&q)).count() as u64;
206+
assert_eq!(covering, lookback / YEAR_MS, "q{}", q + 1);
207+
}
208+
let estimates: std::collections::BTreeMap<_, _> = segments
209+
.iter()
210+
.flat_map(|(build, ..)| estimates_of(&c.dag, *build))
211+
.collect();
197212
assert_eq!(estimates.len(), 5, "one estimate per query");
198213
for (estimate, statistic) in estimates {
199214
assert_eq!(statistic, SketchStatistic::Quantile { q: 0.99 });
@@ -211,16 +226,14 @@ fn stage1_a_window_composition_adds_one_eh_for_all_five() {
211226
}
212227
}
213228

214-
/// The shared EH candidate is added next to the independent KLL candidates, not instead of them.
229+
/// The shared-segment candidate is added next to the independent KLL candidates, not instead of them.
215230
#[test]
216-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram"]
217231
fn stage1_a_keeps_independent_and_shared_window_summaries() {
218232
let run = run_a();
219-
let shared = run.logical.iter().any(|c| {
220-
windowed_builds(c).iter().any(|(_, _, f, r)| {
221-
matches!(f, WindowForm::ExponentialHistogram { .. }) && r.len() == 5
222-
})
223-
});
233+
let shared = run
234+
.logical
235+
.iter()
236+
.any(|c| shares_segments(&c.dag, &c.query_roots));
224237
let independent = run.logical.iter().any(|c| {
225238
let builds = windowed_builds(c);
226239
builds.len() == 5
@@ -233,7 +246,7 @@ fn stage1_a_keeps_independent_and_shared_window_summaries() {
233246

234247
/// Pass 2 adds a candidate for each way of grouping two queries onto one shared window summary.
235248
#[test]
236-
#[ignore = "needs Pass 2 window composition (#580): partial groupings"]
249+
#[ignore = "partial and pairwise groupings are not generated, only all-shared segments (Q62)"]
237250
fn stage1_a_window_composition_groups_every_pair() {
238251
let run = run_a();
239252
let groups: BTreeSet<_> = run
@@ -282,7 +295,11 @@ fn stage3_a_shared_scan_is_not_costlier() {
282295
};
283296
let cost = |c: &Logical| run.cost(&run.physical_of(c).next().unwrap().id);
284297
let mut compared = 0;
285-
for shared in run.logical.iter().filter(|c| c.shared_input) {
298+
for shared in run
299+
.logical
300+
.iter()
301+
.filter(|c| c.shared_input && !shares_segments(&c.dag, &c.query_roots))
302+
{
286303
let separate = run
287304
.logical
288305
.iter()

‎crates/integration-tests/tests/planner_layering_example4.rs‎

Lines changed: 54 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,22 @@ fn eh_options(run: &Run) -> BTreeMap<Materialization, (String, PhysicalASAPNodeI
9898
options_of(run, None, 5)
9999
}
100100

101+
/// The physical candidates of the shared-segment candidate (Q60), keyed by
102+
/// the materialization of its segments, with the newest segment's id.
103+
fn segment_options(run: &Run) -> BTreeMap<Materialization, (String, LogicalASAPNodeId)> {
104+
let logical = run
105+
.logical
106+
.iter()
107+
.find(|c| shares_segments(&c.dag, &c.query_roots))
108+
.expect("a candidate sharing window segments");
109+
run.physical_of(logical)
110+
.map(|p| {
111+
let (segment, ..) = sketch_builds(&p.dag)[0];
112+
(materialization(p, segment), (p.id.clone(), segment))
113+
})
114+
.collect()
115+
}
116+
101117
fn tumbling() -> Option<WindowForm> {
102118
Some(WindowForm::Tumbling {
103119
length_ms: PATTERN_B_INTERVAL_MS,
@@ -165,7 +181,7 @@ fn stage2_materialized_output_covers_its_consumers() {
165181

166182
/// The shared EH candidate yields A1 (query time, kept), A2 (ingestion time) and A3 (not materialized).
167183
#[test]
168-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram; and A3, not materialized for several consumers (Q44)"]
184+
#[ignore = "shared segments (Q60) are not kept for one batch, and A3 is not generated (Q44); a one-off batch has no ingestion-time option (Q61)"]
169185
fn stage2_a_shared_eh_has_three_materialization_options() {
170186
let found: BTreeSet<_> = eh_options(&run_promql(&once())).into_keys().collect();
171187
assert_eq!(
@@ -176,43 +192,53 @@ fn stage2_a_shared_eh_has_three_materialization_options() {
176192

177193
/// With data at rest, A2 is not generated; A1 and A3 remain.
178194
#[test]
179-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram"]
195+
#[ignore = "shared segments (Q60) are not kept for one batch, and A3 is not generated (Q44)"]
180196
fn stage2_a_at_rest_drops_the_ingestion_time_option() {
181197
let found: BTreeSet<_> = eh_options(&run_promql(&at_rest())).into_keys().collect();
182198
assert_eq!(found, BTreeSet::from([QueryTimeKept, NotMaterialized]));
183199
}
184200

185-
/// A materialized shared EH is one build node read by all five queries, charged once.
201+
/// The shared segments (Q60, in place of the spec's EH) are five builds,
202+
/// each built once, read by all five queries through their merges and
203+
/// charged once, in every materialization: at query time for the one-off
204+
/// batch, and also at ingestion time when it repeats monthly (Q61).
186205
#[test]
187-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram"]
188-
fn stage2_a_materialized_eh_is_built_once_for_all_consumers() {
189-
let run = run_promql(&once());
190-
for (m, (id, build)) in eh_options(&run) {
191-
if m == NotMaterialized {
192-
continue;
206+
fn stage2_a_segments_are_built_once_for_all_consumers() {
207+
for (workload, expected) in [
208+
(once(), BTreeSet::from([NotMaterialized])),
209+
(monthly(), BTreeSet::from([IngestionTime, NotMaterialized])),
210+
] {
211+
let run = run_promql(&workload);
212+
let options = segment_options(&run);
213+
assert_eq!(options.keys().copied().collect::<BTreeSet<_>>(), expected);
214+
for (id, _) in options.values() {
215+
let p = run.physical(id);
216+
let segments = sketch_builds(&p.dag);
217+
assert_eq!(segments.len(), 5, "{id}");
218+
let read: BTreeSet<usize> = segments
219+
.iter()
220+
.flat_map(|(b, ..)| readers(&p.dag, &p.query_roots, *b))
221+
.collect();
222+
assert_eq!(read.len(), 5, "{id}");
223+
let merges = p
224+
.dag
225+
.nodes
226+
.iter()
227+
.filter(|n| matches!(n.payload, LogicalASAPOperatorPayload::SummaryMerge))
228+
.count();
229+
assert_eq!(merges, 5, "{id}");
230+
if let Some(cost) = run.selection.costs.get(id) {
231+
for (b, ..) in &segments {
232+
assert!(cost.per_node.contains_key(b), "{id}");
233+
}
234+
}
193235
}
194-
let p = run.physical(&id);
195-
let ehs = sketch_builds(&p.dag)
196-
.into_iter()
197-
.filter(|(b, ..)| {
198-
matches!(
199-
window_form(&p.dag, *b),
200-
WindowForm::ExponentialHistogram { .. }
201-
)
202-
})
203-
.count();
204-
assert_eq!(ehs, 1, "{id}");
205-
assert_eq!(readers(&p.dag, &p.query_roots, build).len(), 5, "{id}");
206-
assert!(
207-
run.selection.costs[&id].per_node.contains_key(&build),
208-
"{id}"
209-
);
210236
}
211237
}
212238

213239
/// A3 rebuilds the EH for each of the five queries, so it costs more than A1, which builds it once.
214240
#[test]
215-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram; and A3 (Q44)"]
241+
#[ignore = "A3, not materialized for several consumers, is not generated (Q44)"]
216242
fn stage3_a_rebuilding_per_query_costs_more_than_building_once() {
217243
let run = run_promql(&once());
218244
let options = eh_options(&run);
@@ -227,7 +253,7 @@ fn stage3_a_rebuilding_per_query_costs_more_than_building_once() {
227253

228254
/// As given (run once, ad hoc), A1 is the cheapest of the three: A2 maintains years of history for one batch.
229255
#[test]
230-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram"]
256+
#[ignore = "a one-off batch has no ingestion-time option (Q61), and A3 is not generated (Q44)"]
231257
fn stage3_a_once_adhoc_prefers_the_query_time_eh() {
232258
let run = run_promql(&once());
233259
let options = eh_options(&run);
@@ -239,7 +265,7 @@ fn stage3_a_once_adhoc_prefers_the_query_time_eh() {
239265

240266
/// Repeated monthly and predictable, A2's maintenance is shared by many batches, so A2 gains on A1.
241267
#[test]
242-
#[ignore = "needs Pass 2 window composition (#580): Exponential Histogram, maintainable over a monthly window"]
268+
#[ignore = "a one-off batch has no ingestion-time option to compare with (Q61)"]
243269
fn stage3_a_monthly_amortizes_ingestion_time_maintenance() {
244270
let ratio = |workload: PlanningWorkload| {
245271
let run = run_promql(&workload);

‎crates/logical-optimizer/src/pass1/logical_candidates.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -697,7 +697,10 @@ fn realize(
697697
};
698698
let state = match window {
699699
WindowForm::Whole => build(child)?,
700-
WindowForm::Tumbling { pane_ms } => tumbling_state(&child, pane_ms, build)?,
700+
WindowForm::Tumbling { pane_ms }
701+
| WindowForm::Segments {
702+
segment_ms: pane_ms,
703+
} => tumbling_state(&child, pane_ms, build)?,
701704
};
702705
let evaluation = match query {
703706
Some(query) => ASAPOp::SummaryEstimate {

‎crates/logical-optimizer/src/pass2/identical_expressions.rs‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ use asap_types::ir::{OperatorNode, QueryRoot};
1616
use asap_types::workload::{MetricType, RootDemand};
1717

1818
use super::summary_capability::share_summary_capability;
19-
use super::window_composition::add_window_forms;
19+
use super::window_composition::{add_window_forms, share_window_segments};
2020
use crate::pass1::logical_candidates::{
2121
enumerate_local_logical_candidates, LocalLogicalCandidates, LogicalCandidateError,
2222
};
@@ -34,6 +34,11 @@ pub enum Sharing {
3434
/// are sized for their strictest consumer, so composition builds
3535
/// identical producers, which are merged.
3636
SummaryCapability,
37+
/// The shared-segment rule on top of the identical-expression rule
38+
/// ([`share_window_segments`]): queries over windows of one scan merge
39+
/// one shared summary per segment, and the identical segments are merged
40+
/// after composition.
41+
WindowSegments,
3742
}
3843

3944
impl Sharing {
@@ -56,7 +61,9 @@ pub struct SharingVariant<Id> {
5661
/// last is skipped when it would repeat the identical-expression variant.
5762
/// In every variant, the window-composition rule adds tumbling forms of
5863
/// mergeable alternatives for repeating queries (`demand[i]` is the demand
59-
/// of `roots[i]`; a root without one gets none).
64+
/// of `roots[i]`; a root without one gets none). Last, the shared-segment
65+
/// variant when windows of one scan can share segments: only that form, not
66+
/// the tumbling forms, for the targets it groups.
6067
pub fn stage1_logical_candidates<Id: Clone>(
6168
roots: Vec<(Id, QueryRoot)>,
6269
metric_types: &BTreeMap<String, MetricType>,
@@ -74,6 +81,7 @@ pub fn stage1_logical_candidates<Id: Clone>(
7481
});
7582
}
7683
let base = &variants.last().expect("the independent variant").inventory;
84+
let segments = share_window_segments(base, demand)?;
7785
if let Some(capability) = share_summary_capability(base)? {
7886
if capability.resized || variants.len() == 1 {
7987
variants.push(SharingVariant {
@@ -85,6 +93,12 @@ pub fn stage1_logical_candidates<Id: Clone>(
8593
for variant in &mut variants {
8694
add_window_forms(&mut variant.inventory, demand);
8795
}
96+
if let Some(inventory) = segments {
97+
variants.push(SharingVariant {
98+
sharing: Sharing::WindowSegments,
99+
inventory,
100+
});
101+
}
88102
Ok(variants)
89103
}
90104

‎crates/logical-optimizer/src/pass2/mod.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,8 @@
22
//! strictest accuracy requirement of its readers. The stage pipeline applies
33
//! the identical-expression rule ([`identical_expressions`]), the
44
//! summary-capability rule ([`summary_capability`]) and the
5-
//! window-composition rule's tumbling windows ([`window_composition`]).
5+
//! window-composition rule's tumbling windows and shared segments
6+
//! ([`window_composition`]).
67
78
pub mod identical_expressions;
89
pub mod reconciliation;

0 commit comments

Comments
 (0)