Skip to content

Commit 738aabd

Browse files
feat(frontend-sql): model SQL window frame clauses (ROWS/RANGE BETWEEN ...)
SQLWindowFunc previously dropped the frame clause entirely — a default frame, ROWS BETWEEN 2 PRECEDING AND CURRENT ROW, and ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING all lowered to identical IR, so every frame-sensitive window function (SUM, AVG, COUNT, MIN, MAX, FIRST_VALUE, LAST_VALUE, NTH_VALUE) with a non-default frame was silently wrong. Adds WindowFrame/WindowFrameUnits/WindowFrameBound to the canonical IR and threads it through lowering, CSE, resolve, and DAG export. GROUPS frames are rejected explicitly (unused in this repo's SQL corpora, unsupported downstream). Fixes #268 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1 parent ec38b28 commit 738aabd

8 files changed

Lines changed: 178 additions & 11 deletions

File tree

‎crates/frontend-sql/src/sql/mod.rs‎

Lines changed: 46 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,9 @@ use datafusion::logical_expr::expr::AggregateFunction;
3838
use datafusion::logical_expr::expr_rewriter::FunctionRewrite;
3939
use datafusion::logical_expr::{
4040
self, lit, AggregateUDF, Case, Distinct, Expr, JoinType, LogicalPlan, ScalarUDF, ScalarUDFImpl,
41-
Signature, SimpleAggregateUDF, TypeSignature, Volatility, WindowFunctionDefinition,
41+
Signature, SimpleAggregateUDF, TypeSignature, Volatility,
42+
WindowFrameBound as DfWindowFrameBound, WindowFrameUnits as DfWindowFrameUnits,
43+
WindowFunctionDefinition,
4244
};
4345
use datafusion::optimizer::analyzer::function_rewrite::ApplyFunctionRewrites;
4446
use datafusion::optimizer::{AnalyzerRule, OptimizerConfig};
@@ -48,7 +50,7 @@ use asap_sql_function_catalog::{AggSemantic, Arity, RewriteKind};
4850
use asap_types::pre_asap::agg_intent::AggIntent;
4951
use asap_types::pre_asap::query_expr::{
5052
GroupKeys, Predicate, ProjectItem, Reduction, SortKey, Source,
51-
UnresolvedQueryExpr as Unresolved,
53+
UnresolvedQueryExpr as Unresolved, WindowFrame, WindowFrameBound, WindowFrameUnits,
5254
};
5355
use asap_types::pre_asap::schema::{DataType, Schema};
5456
use asap_types::pre_asap::{
@@ -65,7 +67,7 @@ mod types;
6567
pub use types::SqlCatalog;
6668

6769
use self::expr::df_expr_to_unresolved;
68-
use self::types::{arrow_to_dtype, schema_to_arrow};
70+
use self::types::{arrow_to_dtype, scalar_value_to_asap, schema_to_arrow};
6971

7072
std::thread_local! {
7173
static ACCURACY: std::cell::RefCell<AccuracyTarget> =
@@ -545,8 +547,8 @@ impl<'a> SqlLowerer<'a> {
545547
})
546548
}
547549

548-
/// `func(args) OVER (PARTITION BY … ORDER BY …)`. One window function per
549-
/// plan node; window frames are not modelled yet (default frame assumed).
550+
/// `func(args) OVER (PARTITION BY … ORDER BY … ROWS/RANGE BETWEEN …)`. One
551+
/// window function per plan node.
550552
fn lower_window(&self, window: &logical_expr::Window) -> Result<Unresolved, LoweringError> {
551553
if window.window_expr.len() > 1 {
552554
return Err(LoweringError::UnsupportedFeature(format!(
@@ -601,6 +603,7 @@ impl<'a> SqlLowerer<'a> {
601603
})
602604
})
603605
.collect::<Result<Vec<_>, _>>()?;
606+
let frame = lower_window_frame(&wf.window_frame)?;
604607
// The window plan's schema is `[input fields …, window output]`; the last
605608
// field is the window column's name (what an enclosing Project references).
606609
let output_name = window
@@ -614,6 +617,7 @@ impl<'a> SqlLowerer<'a> {
614617
args,
615618
partition_by: partition_by.into(),
616619
order_by,
620+
frame,
617621
output_name,
618622
child,
619623
})
@@ -1665,6 +1669,43 @@ fn lower_window_func_kind(fun: &WindowFunctionDefinition) -> Result<WindowFuncKi
16651669
}
16661670
}
16671671

1672+
/// Map DataFusion's resolved `WindowFrame` (issue #268) to the canonical
1673+
/// [`WindowFrame`]. DataFusion's planner always fills in the SQL-standard
1674+
/// default frame before the logical plan is built, so this never sees an
1675+
/// "absent" frame — only `ROWS`/`RANGE`/`GROUPS` with concrete bounds.
1676+
/// `GROUPS` is rejected: no query in this repo's SQL corpora uses it, and
1677+
/// nothing downstream interprets frame semantics yet, so it isn't worth
1678+
/// modelling untested.
1679+
fn lower_window_frame(
1680+
frame: &datafusion::logical_expr::WindowFrame,
1681+
) -> Result<WindowFrame, LoweringError> {
1682+
let units = match frame.units {
1683+
DfWindowFrameUnits::Rows => WindowFrameUnits::Rows,
1684+
DfWindowFrameUnits::Range => WindowFrameUnits::Range,
1685+
DfWindowFrameUnits::Groups => {
1686+
return Err(LoweringError::UnsupportedFeature(
1687+
"window frame unit: GROUPS".into(),
1688+
))
1689+
}
1690+
};
1691+
let bound = |b: &DfWindowFrameBound| -> Result<WindowFrameBound, LoweringError> {
1692+
Ok(match b {
1693+
DfWindowFrameBound::Preceding(v) => {
1694+
WindowFrameBound::Preceding(scalar_value_to_asap(v)?)
1695+
}
1696+
DfWindowFrameBound::CurrentRow => WindowFrameBound::CurrentRow,
1697+
DfWindowFrameBound::Following(v) => {
1698+
WindowFrameBound::Following(scalar_value_to_asap(v)?)
1699+
}
1700+
})
1701+
};
1702+
Ok(WindowFrame {
1703+
units,
1704+
start_bound: bound(&frame.start_bound)?,
1705+
end_bound: bound(&frame.end_bound)?,
1706+
})
1707+
}
1708+
16681709
// ── Issue #225, item 3: DataFusion registry drift detection ────────────────
16691710
//
16701711
// `asap_sql_function_catalog::NATIVE_FUNCTIONS` is hand-maintained data

‎crates/frontend-sql/tests/sql_lowering.rs‎

Lines changed: 67 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ use asap_frontend_sql::{lower_sql, lower_sql_dialect, SqlCatalog};
99
use asap_types::pre_asap::schema::{Column, DataType, Schema};
1010
use asap_types::pre_asap::{
1111
AggIntent, CompareOpKind, GroupKeys, JoinKind, QueryExpr, Reduction, ScalarValue, Source,
12-
WindowFuncKind,
12+
WindowFrameBound, WindowFrameUnits, WindowFuncKind,
1313
};
1414
use asap_types::types::AccuracyTarget;
1515
use asap_types::workload::SqlDialect;
@@ -722,6 +722,72 @@ async fn window_aggregate_lowers_to_windowfunc() {
722722
assert_eq!(args, &vec![QueryExpr::Column(3)], "SUM(bytes) → arg col 3");
723723
}
724724

725+
// ── Window frames (issue #268) ───────────────────────────────────────────────
726+
727+
/// The frame clause must actually reach the IR, not just the display string:
728+
/// three window frames that differ semantically must lower to different
729+
/// `SQLWindowFunc.frame` values.
730+
#[tokio::test]
731+
async fn window_frame_is_captured_not_dropped() {
732+
let default_frame =
733+
lower("SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts) FROM metrics")
734+
.await;
735+
let two_preceding = lower(
736+
"SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts \
737+
ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) FROM metrics",
738+
)
739+
.await;
740+
let unbounded_following = lower(
741+
"SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts \
742+
ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) FROM metrics",
743+
)
744+
.await;
745+
746+
let frame_of = |qe: &QueryExpr| {
747+
let QueryExpr::SQLWindowFunc { frame, .. } = find_windowfunc(qe).unwrap() else {
748+
unreachable!();
749+
};
750+
frame.clone()
751+
};
752+
let (a, b, c) = (
753+
frame_of(&default_frame),
754+
frame_of(&two_preceding),
755+
frame_of(&unbounded_following),
756+
);
757+
assert_ne!(a, b, "default frame vs ROWS 2 PRECEDING must differ");
758+
assert_ne!(
759+
a, c,
760+
"default frame vs ROWS CURRENT..UNBOUNDED FOLLOWING must differ"
761+
);
762+
assert_ne!(b, c);
763+
764+
assert_eq!(b.units, WindowFrameUnits::Rows);
765+
assert_eq!(
766+
b.start_bound,
767+
WindowFrameBound::Preceding(ScalarValue::Int64(2))
768+
);
769+
assert_eq!(b.end_bound, WindowFrameBound::CurrentRow);
770+
771+
assert_eq!(c.start_bound, WindowFrameBound::CurrentRow);
772+
assert_eq!(c.end_bound, WindowFrameBound::Following(ScalarValue::Null));
773+
}
774+
775+
/// `GROUPS` frames aren't in this repo's SQL corpora and nothing downstream
776+
/// interprets frame semantics yet — rejected explicitly rather than silently
777+
/// mis-lowered.
778+
#[tokio::test]
779+
async fn groups_frame_is_rejected() {
780+
let err = lower_sql(
781+
"SELECT service, SUM(latency) OVER (PARTITION BY service ORDER BY ts \
782+
GROUPS BETWEEN 2 PRECEDING AND CURRENT ROW) FROM metrics",
783+
&catalog(),
784+
AccuracyTarget::Exact,
785+
)
786+
.await
787+
.expect_err("GROUPS frame unit must be rejected");
788+
assert!(format!("{err}").contains("GROUPS"), "got {err}");
789+
}
790+
725791
// ── Nested query functions: derived tables / inline views (issue #27) ───────────
726792

727793
/// Collect every `AggIntent` in the tree, root-to-leaf.

‎crates/types/src/dag_export.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -440,6 +440,7 @@ fn build(expr: &QueryExpr, nodes: &mut Vec<DagNode>, cache: &mut HashCache) -> u
440440
args,
441441
partition_by,
442442
order_by,
443+
frame,
443444
output_name,
444445
child,
445446
} => {
@@ -449,6 +450,7 @@ fn build(expr: &QueryExpr, nodes: &mut Vec<DagNode>, cache: &mut HashCache) -> u
449450
"args": args,
450451
"partition_by": partition_by,
451452
"order_by": order_by,
453+
"frame": frame,
452454
"output_name": output_name,
453455
});
454456
push_node(

‎crates/types/src/pre_asap/canonicalize.rs‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -294,7 +294,9 @@ fn try_rewrite_rownumber_topk(expr: &QueryExpr) -> Option<QueryExpr> {
294294
#[cfg(test)]
295295
mod tests {
296296
use super::*;
297-
use crate::pre_asap::query_expr::{GroupKeys, ProjectItem, Source};
297+
use crate::pre_asap::query_expr::{
298+
GroupKeys, ProjectItem, Source, WindowFrame, WindowFrameBound, WindowFrameUnits,
299+
};
298300
use crate::pre_asap::schema::{Column, DataType, Schema};
299301
use crate::types::AccuracyTarget;
300302

@@ -473,6 +475,16 @@ mod tests {
473475
}
474476
}
475477

478+
/// `ROW_NUMBER` ignores its frame clause, so the top-k rewrite doesn't care
479+
/// what's in it; any concrete frame works as fixture data.
480+
fn rownumber_frame() -> WindowFrame {
481+
WindowFrame {
482+
units: WindowFrameUnits::Rows,
483+
start_bound: WindowFrameBound::Preceding(ScalarValue::Null),
484+
end_bound: WindowFrameBound::Following(ScalarValue::Null),
485+
}
486+
}
487+
476488
/// `Filter{ rn(3) <= 5 } { SQLWindowFunc{ RowNumber, PARTITION BY region(2),
477489
/// ORDER BY col(2) DESC } { agg } }`.
478490
fn rownumber_topk(agg: QueryExpr) -> QueryExpr {
@@ -485,6 +497,7 @@ mod tests {
485497
ascending: false,
486498
nulls_first: true,
487499
}],
500+
frame: rownumber_frame(),
488501
output_name: "rn".into(),
489502
child: Rc::new(agg),
490503
};
@@ -567,6 +580,7 @@ mod tests {
567580
ascending: false,
568581
nulls_first: true,
569582
}],
583+
frame: rownumber_frame(),
570584
output_name: "rn".into(),
571585
child: Rc::new(grouped(AggIntent::Count {
572586
accuracy: AccuracyTarget::Exact,

‎crates/types/src/pre_asap/cse.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -350,6 +350,7 @@ pub(crate) fn structural_hash(node: &QueryExpr, cache: &mut HashCache) -> u64 {
350350
args,
351351
partition_by,
352352
order_by,
353+
frame,
353354
output_name,
354355
child,
355356
} => {
@@ -361,6 +362,7 @@ pub(crate) fn structural_hash(node: &QueryExpr, cache: &mut HashCache) -> u64 {
361362
args,
362363
partition_by,
363364
order_by,
365+
frame,
364366
output_name,
365367
),
366368
);
@@ -666,13 +668,15 @@ fn rebuild_children(table: &mut InternTable, expr: QueryExpr) -> QueryExpr {
666668
args,
667669
partition_by,
668670
order_by,
671+
frame,
669672
output_name,
670673
child,
671674
} => SQLWindowFunc {
672675
func,
673676
args,
674677
partition_by,
675678
order_by,
679+
frame,
676680
output_name,
677681
child: intern_child(table, child),
678682
},

‎crates/types/src/pre_asap/mod.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,8 @@ pub use query_expr::{
5656
aggregate_output_schema, AtModifier, BinaryOpKind, ColState, DataModel, GroupKeys, GroupSide,
5757
InfoMatcher, JoinKind, Predicate, ProjectItem, QueryExpr, QueryExprError, Reduction,
5858
ResolvedQueryExpr, SampleKind, SetOpKind, SortKey, Source, TimeShift, UnresolvedQueryExpr,
59-
VectorGrouping, VectorMatch, VectorMatchKind, WindowFuncKind,
59+
VectorGrouping, VectorMatch, VectorMatchKind, WindowFrame, WindowFrameBound, WindowFrameUnits,
60+
WindowFuncKind,
6061
};
6162
pub use resolve::{resolve_root, ResolveTreeError};
6263
pub use schema::{Column, ColumnId, DataType, Schema};

‎crates/types/src/pre_asap/query_expr.rs‎

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -332,6 +332,42 @@ pub enum WindowFuncKind {
332332
Max,
333333
}
334334

335+
/// A window's frame-spec (`ROWS`/`RANGE BETWEEN … AND …`) — which rows around
336+
/// the current one an analytic window function reads. `GROUPS` is rejected at
337+
/// lowering time (issue #268): every SQL corpus in this repo uses only `ROWS`,
338+
/// and nothing downstream interprets frame semantics yet, so it isn't worth
339+
/// modelling untested.
340+
///
341+
/// Meaningless (but harmless) on the rank-only and navigation functions
342+
/// (`ROW_NUMBER`/`RANK`/`DENSE_RANK`/`LAG`/`LEAD`), which ignore the frame per
343+
/// SQL semantics — DataFusion still attaches one, stored here verbatim.
344+
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
345+
pub struct WindowFrame {
346+
pub units: WindowFrameUnits,
347+
pub start_bound: WindowFrameBound,
348+
pub end_bound: WindowFrameBound,
349+
}
350+
351+
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
352+
#[serde(rename_all = "snake_case")]
353+
pub enum WindowFrameUnits {
354+
/// Boundaries count physical rows: `ROWS BETWEEN 2 PRECEDING AND CURRENT ROW`.
355+
Rows,
356+
/// Boundaries count by value-distance on the (single) `ORDER BY` column:
357+
/// `RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW`.
358+
Range,
359+
}
360+
361+
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
362+
#[serde(rename_all = "snake_case")]
363+
pub enum WindowFrameBound {
364+
/// `UNBOUNDED PRECEDING` is `Preceding(ScalarValue::Null)`.
365+
Preceding(ScalarValue),
366+
CurrentRow,
367+
/// `UNBOUNDED FOLLOWING` is `Following(ScalarValue::Null)`.
368+
Following(ScalarValue),
369+
}
370+
335371
/// A symbolic label matcher on the **info metric** side of an
336372
/// [`QueryExpr::PromqlInfoEnrich`] (issue #84). Unlike a `Scan` predicate it is not
337373
/// resolved positionally — it references the info metric's labels (`__name__`
@@ -729,16 +765,17 @@ pub enum QueryExpr<C: ColState = ColumnId> {
729765
child: Rc<QueryExpr<C>>,
730766
},
731767

732-
/// SQL analytic window function: `func(args) OVER (PARTITION BY … ORDER BY …)`.
733-
/// Output schema = child schema + one column named `output_name` (the name
734-
/// the enclosing `Project` references). Window frames are not modelled yet.
768+
/// SQL analytic window function: `func(args) OVER (PARTITION BY … ORDER BY …
769+
/// ROWS/RANGE BETWEEN …)`. Output schema = child schema + one column named
770+
/// `output_name` (the name the enclosing `Project` references).
735771
SQLWindowFunc {
736772
func: WindowFuncKind,
737773
/// Operand expressions (`LAG(value)` → `[Column(value_id)]`); empty for
738774
/// the rank-only functions (`ROW_NUMBER`/`RANK`/`DENSE_RANK`).
739775
args: Vec<QueryExpr<C>>,
740776
partition_by: GroupKeys<C>,
741777
order_by: Vec<SortKey<C>>,
778+
frame: WindowFrame,
742779
/// The output column's name — DataFusion's window-expr field name, so a
743780
/// `Project` above resolves it (cf. `Aggregate.output_names`).
744781
output_name: String,

‎crates/types/src/pre_asap/resolve.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -337,6 +337,7 @@ fn resolve(
337337
args,
338338
partition_by,
339339
order_by,
340+
frame,
340341
output_name,
341342
child,
342343
} => {
@@ -362,6 +363,7 @@ fn resolve(
362363
args,
363364
partition_by,
364365
order_by,
366+
frame: frame.clone(),
365367
output_name: output_name.clone(),
366368
child: Rc::new(child),
367369
}

0 commit comments

Comments
 (0)