Skip to content

Commit 1397f2e

Browse files
zzylolclaude
andcommitted
feat(pass2): the planner chooses the segment grid (Q67)
The shared-segment rule (Q60) offered one fixed grid, the gcd of every window's lookback and offset. It now offers two segmentations, each as its own Stage 1 variant, and Stage 3's cost decides: - split only at the windows' own boundaries (fewest segments); for Example 3a four segments, the oldest two years wide; - the even gcd grid (3a: five 1-year segments), when it differs. WindowForm::Segments carries the grid unit and a bitmask of where each segment ends, so a target's segments may differ in width; tumbling_state and the coverage proof take the widths. Sharing::WindowSegments carries the number of distinct segments, labeled "· shared segments ×4" / "×5". Each segment is still sized for the group's strictest reader. Stage 2 does not maintain a merge of panes of different widths at ingestion time or keep it: it is no stream of panes. So in 4a only the even grid has an ingestion-time option. Example 3a and 4a select the four boundary segments: they build the same rows as the five 1-year ones and merge fewer states (3a: 18,104.0039 vs 18,104.0044 per second). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 9cf2956 commit 1397f2e

9 files changed

Lines changed: 409 additions & 193 deletions

File tree

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

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,9 @@
2222
// input"), and, when the summary-capability rule applies, again with
2323
// one summary sized for its strictest consumer ("· shared summary"),
2424
// and, when queries read windows of one scan on a common grid, again
25-
// with one summary per shared segment ("· shared segments"); in
26-
// enumeration order, only those with a written physical candidate. A
25+
// with one summary per shared segment, once per segmentation, by its
26+
// number of segments ("· shared segments ×4"); in enumeration order,
27+
// only those with a written physical candidate. A
2728
// repeating query's mergeable alternatives also come in tumbling panes
2829
// (Pass 2's window-composition rule), e.g. "Q1 Kll · tumbling 1m panes";
2930
// - stage2_physical_asap: per logical candidate, its physical candidates
@@ -267,11 +268,11 @@ fn stage_pipeline(
267268
let inventory = &variant.inventory;
268269
let index = offset + choice_index(inventory, &candidate.choice) + 1;
269270
let mut label = label(inventory, &target_owners(inventory), &candidate.choice);
270-
label += match candidate.sharing {
271-
Sharing::Independent => "",
272-
Sharing::IdenticalExpressions => " · shared input",
273-
Sharing::SummaryCapability => " · shared summary",
274-
Sharing::WindowSegments => " · shared segments",
271+
label += &match candidate.sharing {
272+
Sharing::Independent => String::new(),
273+
Sharing::IdenticalExpressions => " · shared input".into(),
274+
Sharing::SummaryCapability => " · shared summary".into(),
275+
Sharing::WindowSegments { segments } => format!(" · shared segments ×{segments}"),
275276
};
276277
let written = candidate
277278
.physical

‎crates/devtools/tests/stage_pipeline.rs‎

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -170,13 +170,14 @@ fn example3b_lists_tumbling_candidates() {
170170
}
171171
}
172172

173-
/// Example 4, Pattern A repeated monthly: the same 487 logical candidates
174-
/// as the ad hoc batch (3a), 486 independent or sharing input plus the one
175-
/// sharing 1-year segments (Q60). The windows (1–5 y) are longer than the
176-
/// month between runs and no pane width fits, so only the shared segments
177-
/// can be maintained at ingestion time (Q61): 488 plans. The selected plan
178-
/// is 3a's, now amortized over monthly runs instead of one run per hour of
179-
/// horizon.
173+
/// Example 4, Pattern A repeated monthly: the same 488 logical candidates
174+
/// as the ad hoc batch (3a), 486 independent or sharing input plus the two
175+
/// sharing segments (Q60, Q67): four split at the windows' boundaries and
176+
/// five 1-year ones. The windows (1–5 y) are longer than the month between
177+
/// runs and no pane width fits, so only the even 1-year segments can be
178+
/// maintained at ingestion time (Q61); the uneven ones are no stream of
179+
/// panes: 489 plans. The selected plan is 3a's, now amortized over monthly
180+
/// runs instead of one run per hour of horizon.
180181
#[test]
181182
fn example4a_repeats_monthly_with_only_segments_maintainable() {
182183
let once = generate(&[
@@ -194,14 +195,14 @@ fn example4a_repeats_monthly_with_only_segments_maintainable() {
194195
let physical = monthly["stage2_physical_asap"]["candidates"]
195196
.as_array()
196197
.unwrap();
197-
assert_eq!(physical.len(), 488);
198+
assert_eq!(physical.len(), 489);
198199
let maintained: Vec<_> = physical
199200
.iter()
200201
.map(|p| p["label"].as_str().unwrap())
201202
.filter(|label| label.contains("ingestion time"))
202203
.collect();
203204
assert_eq!(maintained.len(), 1, "{maintained:?}");
204-
assert!(maintained[0].ends_with("shared segments · ingestion time: Kll ×5 panes"));
205+
assert!(maintained[0].ends_with("shared segments ×5 · ingestion time: Kll ×5 panes"));
205206
let selected = |d: &Value| {
206207
d["stage3_selection"]["selected"]
207208
.as_str()

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

Lines changed: 87 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -180,51 +180,103 @@ fn stage1_a_identical_expression_rule_shares_only_raw_input() {
180180
}
181181
}
182182

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.
183+
/// The planner chooses the segment grid (Q67): one candidate splits the
184+
/// windows at their boundaries into four KLL segments over [T − 5y, T], the
185+
/// oldest two years wide; another uses the even grid of five 1-year
186+
/// segments. Either serves all five queries (Q60, in place of the spec's
187+
/// Exponential Histogram): each query merges the segments its range covers
188+
/// and estimates p99 from its merge.
187189
#[test]
188190
fn stage1_a_shared_segments_serve_all_five() {
189191
let run = run_a();
190-
let c = run
192+
let candidates: Vec<_> = run
191193
.logical
192194
.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
195+
.filter(|c| shares_segments(&c.dag, &c.query_roots))
196+
.collect();
197+
let mut widths: Vec<Vec<u64>> = candidates
209198
.iter()
210-
.flat_map(|(build, ..)| estimates_of(&c.dag, *build))
199+
.map(|c| {
200+
let mut widths: Vec<u64> = windowed_builds(c)
201+
.iter()
202+
.map(|(_, _, form, _)| match form {
203+
WindowForm::Tumbling { length_ms } => length_ms / YEAR_MS,
204+
other => panic!("{}: {other:?}", c.id),
205+
})
206+
.collect();
207+
widths.sort();
208+
widths
209+
})
211210
.collect();
212-
assert_eq!(estimates.len(), 5, "one estimate per query");
213-
for (estimate, statistic) in estimates {
214-
assert_eq!(statistic, SketchStatistic::Quantile { q: 0.99 });
215-
let merged = c
216-
.dag
217-
.producers(estimate)
218-
.into_iter()
219-
.any(|p| matches!(c.dag.payload(p), LogicalASAPOperatorPayload::SummaryMerge));
220-
assert!(
221-
merged,
222-
"{}: estimate {estimate:?} does not read a merge",
223-
c.id
224-
);
211+
widths.sort();
212+
assert_eq!(widths, [vec![1, 1, 1, 1, 1], vec![1, 1, 1, 2]]);
213+
for c in candidates {
214+
let segments = windowed_builds(c);
215+
for (_, algorithm, ..) in &segments {
216+
assert_eq!(*algorithm, SketchAlgorithm::Kll);
217+
}
218+
let read: BTreeSet<usize> = segments.iter().flat_map(|(.., r)| r.clone()).collect();
219+
assert_eq!(read, (0..5).collect());
220+
// The segments each query merges add up to its range.
221+
for (q, (_, lookback, _)) in PATTERN_A.iter().enumerate() {
222+
let covered: u64 = segments
223+
.iter()
224+
.filter(|(.., r)| r.contains(&q))
225+
.map(|(_, _, form, _)| match form {
226+
WindowForm::Tumbling { length_ms } => *length_ms,
227+
_ => unreachable!(),
228+
})
229+
.sum();
230+
assert_eq!(covered, *lookback, "{} q{}", c.id, q + 1);
231+
}
232+
let estimates: std::collections::BTreeMap<_, _> = segments
233+
.iter()
234+
.flat_map(|(build, ..)| estimates_of(&c.dag, *build))
235+
.collect();
236+
assert_eq!(estimates.len(), 5, "one estimate per query");
237+
for (estimate, statistic) in estimates {
238+
assert_eq!(statistic, SketchStatistic::Quantile { q: 0.99 });
239+
let merged = c
240+
.dag
241+
.producers(estimate)
242+
.into_iter()
243+
.any(|p| matches!(c.dag.payload(p), LogicalASAPOperatorPayload::SummaryMerge));
244+
assert!(
245+
merged,
246+
"{}: estimate {estimate:?} does not read a merge",
247+
c.id
248+
);
249+
}
225250
}
226251
}
227252

253+
/// Stage 3 decides the grid: the four boundary segments cost as much to
254+
/// build as the five 1-year ones (each row lands in one segment) but merge
255+
/// fewer states, so they are cheaper, and cheapest overall.
256+
#[test]
257+
fn stage3_a_selects_the_cheaper_segment_grid() {
258+
let run = run_a();
259+
let mut costs: Vec<(usize, f64, &str)> = run
260+
.logical
261+
.iter()
262+
.filter(|c| shares_segments(&c.dag, &c.query_roots))
263+
.map(|c| {
264+
let p = run.physical_of(c).next().unwrap();
265+
(
266+
windowed_builds(c).len(),
267+
run.cost(&p.id).unwrap(),
268+
p.id.as_str(),
269+
)
270+
})
271+
.collect();
272+
costs.sort_by_key(|(segments, ..)| *segments);
273+
let [(4, four, selected), (5, five, _)] = costs[..] else {
274+
panic!("{costs:?}");
275+
};
276+
assert!(four < five, "×4 {four} vs ×5 {five}");
277+
assert_eq!(run.selection.selected, selected);
278+
}
279+
228280
/// The shared-segment candidate is added next to the independent KLL candidates, not instead of them.
229281
#[test]
230282
fn stage1_a_keeps_independent_and_shared_window_summaries() {

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

Lines changed: 25 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -97,13 +97,17 @@ fn eh_options(run: &Run) -> BTreeMap<Materialization, (String, LogicalASAPNodeId
9797
options_of(run, None, 5)
9898
}
9999

100-
/// The physical candidates of the shared-segment candidate (Q60), keyed by
101-
/// the materialization of its segments, with the newest segment's id.
102-
fn segment_options(run: &Run) -> BTreeMap<Materialization, (String, LogicalASAPNodeId)> {
100+
/// The physical candidates of the shared-segment candidate (Q60) with
101+
/// `count` segments (Q67), keyed by the materialization of its segments,
102+
/// with the newest segment's id.
103+
fn segment_options(
104+
run: &Run,
105+
count: usize,
106+
) -> BTreeMap<Materialization, (String, LogicalASAPNodeId)> {
103107
let logical = run
104108
.logical
105109
.iter()
106-
.find(|c| shares_segments(&c.dag, &c.query_roots))
110+
.find(|c| shares_segments(&c.dag, &c.query_roots) && sketch_builds(&c.dag).len() == count)
107111
.expect("a candidate sharing window segments");
108112
run.physical_of(logical)
109113
.map(|p| {
@@ -197,23 +201,31 @@ fn stage2_a_at_rest_drops_the_ingestion_time_option() {
197201
assert_eq!(found, BTreeSet::from([QueryTimeKept, NotMaterialized]));
198202
}
199203

200-
/// The shared segments (Q60, in place of the spec's EH) are five builds,
201-
/// each built once, read by all five queries through their merges and
202-
/// charged once, in every materialization: at query time for the one-off
203-
/// batch, and also at ingestion time when it repeats monthly (Q61).
204+
/// The shared segments (Q60, in place of the spec's EH), four split at the
205+
/// windows' boundaries or five on the 1-year grid (Q67), are each built
206+
/// once, read by all five queries through their merges and charged once,
207+
/// in every materialization: at query time for the one-off batch, and, for
208+
/// the even grid only, also at ingestion time when it repeats monthly
209+
/// (Q61). Uneven segments are no stream of panes.
204210
#[test]
205211
fn stage2_a_segments_are_built_once_for_all_consumers() {
206-
for (workload, expected) in [
207-
(once(), BTreeSet::from([NotMaterialized])),
208-
(monthly(), BTreeSet::from([IngestionTime, NotMaterialized])),
212+
for (workload, count, expected) in [
213+
(once(), 4, BTreeSet::from([NotMaterialized])),
214+
(once(), 5, BTreeSet::from([NotMaterialized])),
215+
(monthly(), 4, BTreeSet::from([NotMaterialized])),
216+
(
217+
monthly(),
218+
5,
219+
BTreeSet::from([IngestionTime, NotMaterialized]),
220+
),
209221
] {
210222
let run = run_promql(&workload);
211-
let options = segment_options(&run);
223+
let options = segment_options(&run, count);
212224
assert_eq!(options.keys().copied().collect::<BTreeSet<_>>(), expected);
213225
for (id, _) in options.values() {
214226
let p = run.physical(id);
215227
let segments = sketch_builds(&p.dag);
216-
assert_eq!(segments.len(), 5, "{id}");
228+
assert_eq!(segments.len(), count, "{id}");
217229
let read: BTreeSet<usize> = segments
218230
.iter()
219231
.flat_map(|(b, ..)| readers(&p.dag, &p.query_roots, *b))

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

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -711,10 +711,9 @@ fn realize(
711711
};
712712
build(child, coverage)?
713713
}
714-
WindowForm::Tumbling { pane_ms }
715-
| WindowForm::Segments {
716-
segment_ms: pane_ms,
717-
} => tumbling_state(&child, pane_ms, build)?,
714+
WindowForm::Tumbling { .. } | WindowForm::Segments { .. } => {
715+
tumbling_state(&child, window, build)?
716+
}
718717
};
719718
let evaluation = match query {
720719
Some(query) => ASAPOp::SummaryEstimate {

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

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,9 @@ pub enum Sharing {
3737
/// The shared-segment rule on top of the identical-expression rule
3838
/// ([`share_window_segments`]): queries over windows of one scan merge
3939
/// one shared summary per segment, and the identical segments are merged
40-
/// after composition.
41-
WindowSegments,
40+
/// after composition. One variant per segmentation, by its number of
41+
/// distinct segments (Q67).
42+
WindowSegments { segments: usize },
4243
}
4344

4445
impl Sharing {
@@ -62,8 +63,9 @@ pub struct SharingVariant<Id> {
6263
/// In every variant, the window-composition rule adds tumbling forms of
6364
/// mergeable alternatives for repeating queries (`demand[i]` is the demand
6465
/// 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.
66+
/// variants when windows of one scan can share segments, one per
67+
/// segmentation (Q67): only that form, not the tumbling forms, for the
68+
/// targets it groups.
6769
pub fn stage1_logical_candidates<Id: Clone>(
6870
roots: Vec<(Id, QueryRoot)>,
6971
metric_types: &BTreeMap<String, MetricType>,
@@ -93,9 +95,9 @@ pub fn stage1_logical_candidates<Id: Clone>(
9395
for variant in &mut variants {
9496
add_window_forms(&mut variant.inventory, demand);
9597
}
96-
if let Some(inventory) = segments {
98+
for (segments, inventory) in segments {
9799
variants.push(SharingVariant {
98-
sharing: Sharing::WindowSegments,
100+
sharing: Sharing::WindowSegments { segments },
99101
inventory,
100102
});
101103
}

0 commit comments

Comments
 (0)