feat(SQL): L1 to L3 SQL lowering via DataFusion with TDD

[milindsrivastava1997]

No description provided.

[milindsrivastava1997]

Code Review — 15 items

Bugs

#1 — lower_limit maps missing fetch → n: 0 (semantic bug)
crates/lower/src/sql.rs, lower_limit:

n: eval_fetch(&limit.fetch).unwrap_or(0) as u64,

None fetch means "no LIMIT" (unlimited), but n: 0 reads as "return zero rows." QueryExpr::Limit.n should be Option<u64>, or use u64::MAX as the sentinel. A LIMIT-less Limit plan node (emitted by DataFusion when offset > 0 with no cap) becomes a zero-row filter.

---

#2 — WindowFuncKind::NthValue(0) is always wrong
crates/lower/src/sql.rs, lower_window_func_kind:

"nth_value" => Ok(WindowFuncKind::NthValue(0)),

The N argument lives in wf.args[1] but is never extracted. Every NTH_VALUE(col, N) silently becomes NthValue(0). No test catches this.

---

#3 — extract_time_range only fires when Filter directly wraps TableScan
crates/lower/src/sql.rs, lower_filter:

let inner = strip_aliases(&filter.input);
if let LogicalPlan::TableScan(scan) = inner { ...

Stacked Filter → Filter → TableScan (two separate WHERE conditions) causes the outer filter to skip time-extraction entirely. Need a test for WHERE ts > 1000 AND ts < 2000 AND value > 0 AND host = 'x' to verify.

---

#4 — push_columns_into_scan doesn't recurse through Filter
crates/lower/src/sql.rs, push_columns_into_scan:
The function only rewrites a Scan that is the direct child of Project. When the plan is Project → Filter → Scan (any WHERE clause), Scan.columns stays empty. The "empty = all columns" fallback masks this but violates the stated semantic contract.

---

#5 — lower_window silently drops all but the first window expression
crates/lower/src/sql.rs, lower_window:

let first = window.window_expr.first().ok_or_else(|| ...)?;

DataFusion emits one Window plan node for all functions sharing the same OVER clause. The second (and beyond) window function is dropped with no error. Should either error on window_expr.len() > 1 or loop and stack WindowFunc nodes.

---

#6 — UInt64 → i64 cast silently wraps on overflow
crates/lower/src/sql.rs, scalar_value_to_l3:

ScalarValue::UInt64(Some(v)) => Ok(L3Scalar::Int64(*v as i64)),

Values > i64::MAX silently become negative. Use i64::try_from(*v).map_err(|_| LoweringError::InvalidExpression(...)) instead.

---

Design Issues

#7 — AggIntent doesn't carry the aggregated column
crates/core/src/intent_algebra/expr.rs, AggIntent:
Sum, Min, Max, Avg, Stddev have no col: ColumnRef field. Consequences: output_type for Min/Max receives a dummy Float64, so MIN(ts) wrongly reports Float64. Aggregate schema columns are named agg_{i} and carry the wrong type. Adding this field is a breaking API change — better done now before L4 exists.

---

#8 — expr_to_group_key / expr_to_col_ref fall back to format!("{expr}")
crates/lower/src/sql.rs:

_ => Ok(GroupKey(format!("{expr}"))),

The DataFusion Expr Display format is not stable across versions. Should return Err(LoweringError::UnsupportedFeature(...)) for anything that isn't a plain column reference.

---

#9 — SchemaCatalog has no validation that time_column names an actual column
crates/core/src/intent_algebra/schema.rs:
TableSchema::time_column is Option<String> with no check that the named column exists in columns. A misconfigured catalog silently produces a table where time-range extraction never fires. A Result-returning constructor or a debug-mode assert would catch this early.

---

Test Gaps

#10 — No test for NthValue N extraction
Directly related to bug #2. Add a test: SELECT NTH_VALUE(value, 2) OVER (ORDER BY ts) FROM metrics and assert the resulting NthValue variant holds 2, not 0.

---

#11 — test_lower_batch_error_is_per_query only tests parse failure
crates/lower/tests/sql_lowering.rs:
The bad query "NOT VALID SQL !!!" fails at the DataFusion SQL parser, not at our lowering layer. There should also be a test where SQL parses successfully but lowering fails (e.g., SELECT array_agg(value) FROM metrics — unsupported aggregate).

---

#12 — TopK test doesn't verify by.len()
crates/lower/tests/sql_lowering.rs, test_order_by_desc_limit_becomes_topk:

assert!(by.iter().any(|c| c.0 == "host"));

Should also assert by.len() == 1 to catch spurious extra columns in the TopK by list.

---

#13 — No test for stacked Filters against a time-column table
Directly related to bug #3. Add a test with two non-time predicates plus a time predicate in a single WHERE clause that DataFusion splits into nested Filter nodes, and verify the time range is still extracted.

---

#14 — No test for multi-window-expr Window plan
Directly related to bug #5. Add a test: SELECT SUM(value) OVER (PARTITION BY region), AVG(value) OVER (PARTITION BY region) FROM metrics and assert both window functions produce output (currently only the first is kept).

---

#15 — test_unknown_table_returns_error accepts either of two error variants
crates/lower/tests/sql_lowering.rs:

assert!(matches!(err, LoweringError::DataFusion(_) | LoweringError::TableNotFound(_)));

This passes even if the error variant changes unexpectedly. Pin to DataFusion(_) (what actually fires when DataFusion rejects the unknown table at plan time) and add a comment explaining why.

[milindsrivastava1997]

Thorough review of the SQL L1→L3 lowering PR. The architecture is clean and the TDD approach is well-executed — the test suite covers the happy paths comprehensively. Comments below are ordered by severity.

Bugs / Correctness (items 1–4)
API / Design (items 5–9)
Code Quality (items 10–13)

Test-coverage gaps not tied to a specific line are listed as general comments:

• No test for LIMIT n OFFSET m (offset could silently be 0)
• No test for NOT BETWEEN low AND high normalizing to BoolOr
• No test for NOT IN (...) with negated: true
• No test for time predicate with column on the right side (1000 < ts)
• No test for DISTINCT ON (col) (see inline comment — currently silent data loss)
• No test for TryCast vs Cast producing the same IR
• No test for populate_schemas on SetOp / Merge

[milindsrivastava1997]

Bug: panic on negative LIMIT/OFFSET in debug mode.

LIMIT -1 is invalid SQL but DataFusion may still parse it. Casting a negative i64 to usize panics in debug and gives a huge value in release.

Fix:

Expr::Literal(ScalarValue::Int64(Some(v))) if *v >= 0 => Some(*v as usize),

Same guard needed for the Int32 arm on line 496.

[milindsrivastava1997]

Bug: DISTINCT ON (...) silently drops its ON-column list.

on.on_expr and on.select_expr are discarded; the emitted QueryExpr::Distinct { cols: vec![] } has no ON semantics. SELECT DISTINCT ON (region) ts, value FROM metrics would be treated as a plain DISTINCT, changing the result set.

Either:

Distinct::On(_) => return Err(LoweringError::UnsupportedFeature("DISTINCT ON".into())),

or populate cols from on.on_expr if the semantics are intentionally preserved.

[milindsrivastava1997]

Bug (silent): float/decimal time literals fall through as non-time conjuncts.

scalar_to_ms returns None for Float64, Decimal128, etc. So WHERE ts > 1000.0 is left as a Filter predicate rather than folded into time_range, with no error or warning. Add a comment here noting the limitation, or convert common float types:

ScalarValue::Float64(Some(v)) => Some(*v as i64), // ms-since-epoch stored as float

or return an InvalidExpression so callers know time extraction failed.

[milindsrivastava1997]

Bug (latent): coupling to DataFusion's Display format for aggregate names.

let output_names: Vec<String> = agg.aggr_expr.iter().map(|e| format!("{e}")).collect();

The Project schema arm uses these strings to match aggregate output columns by name. If DataFusion changes its Display impl for Expr (e.g., starts qualifying as metrics.MIN(ts) instead of MIN(metrics.ts)), name matching will silently fail and all aggregate fields will fall through to Float64.

At minimum: add a comment pinning the DataFusion version this was verified against. Better: assert in a test that format!("{e}") for a known expr matches the expected string, so a DataFusion upgrade breaks loudly rather than silently.

[milindsrivastava1997]

Design: panic! in a public library method is hostile to callers.

A downstream user calling output_type on a TopK they received from the lowerer will get an unexplained panic. Consider returning Option<L3DataType> (returning None for multi-column intents) so callers can handle it:

pub fn output_type(&self, input: &L3Field) -> Option<L3DataType> {
    match self {
        Self::TopK { .. } => None,
        // ... other arms return Some(...)
    }
}

The existing call site in output_schema already guards against TopK before calling this; returning Option makes that contract explicit in the type.

[milindsrivastava1997]

Design: two entry points map to the same DataFusion lowerer.

QueryLanguage::DataFusion (line 31) and QueryLanguage::SQL(SqlDialect::DataFusionSQL) (line 43) both pass through to SqlLowerer via separate if branches. Having two spellings of the same thing will confuse callers. A doc comment explaining why both exist (or consolidating to one canonical variant) would help.

[milindsrivastava1997]

Code quality: push_columns_into_scan only recurses through Filter.

The comment says it handles Project → Filter → Scan, which it does. But Project → Aggregate → Filter → Scan (e.g., SELECT region, COUNT(*) FROM metrics WHERE value > 0 GROUP BY region) does not get the Scan columns populated from the projection's column refs. This is currently benign because lower_aggregate doesn't call push_columns_into_scan, but the asymmetry is surprising and will bite anyone who extends the lowerer. A note in the doc comment about what topologies are and aren't handled would help.

[milindsrivastava1997]

Code quality: debug_assert! is skipped in release builds.

If a caller constructs a SchemaCatalog with an invalid time_column name and runs in release mode, the invalid catalog silently produces wrong plans (no time-range extraction). TableSchema::validate() exists and catches this — consider calling it in SchemaCatalog's construction path (or in lower_batch before processing) to enforce the invariant in all build profiles.

[milindsrivastava1997]

Code quality: extract_time_range is pub(crate) but has no direct unit tests.

All coverage goes through the integration tests, which don't exercise every branch (e.g., column on the right side of the comparison: 1000 < ts, overlapping ranges from repeated bounds, NOT BETWEEN). Given how central this function is to correctness, a dedicated unit-test module in sql.rs (or a separate tests/time_extraction.rs) covering those edge cases would add confidence.

[milindsrivastava1997]

Nit: parts.remove(0) is O(n) for the single-element fast path.

if parts.len() == 1 {
    Ok(parts.remove(0))  // shifts remaining elements (none here, but still)
}

Prefer:

if parts.len() == 1 {
    Ok(parts.pop().unwrap())
}

pop() is O(1) and more idiomatic for "take the only element".

[milindsrivastava1997]

Bug: OFFSET is silently discarded when folding Limit+Sort+Aggregate into TopK.

LIMIT k OFFSET n ORDER BY ... DESC on an aggregate will emit AggIntent::TopK and drop n. A caller expecting to page through top-k results gets the wrong slice with no error.

Fix: guard on a non-zero offset and fall through to the regular Sort + Limit path when one is present:

if let Some(k) = eval_fetch(&limit.fetch) {
    if eval_fetch(&limit.skip).unwrap_or(0) == 0 { // <-- add this guard
        let inner = strip_aliases(&limit.input);
        ...
    }
}

[milindsrivastava1997]

Correctness gap: time extraction only fires when the direct child is a TableScan.

DataFusion can produce a Filter(Filter(TableScan)) topology when it stacks multiple filter conditions. In that case the outer Filter's child is another Filter, not a TableScan, so this branch is skipped. The time bounds remain as opaque predicates and Source::Table.time_range stays None even though the SQL clearly expressed a time bound.

Consider recursing through Filter children here, or add a test that explicitly asserts the current (limited) behaviour so regressions are caught if DataFusion's planning changes.

[milindsrivastava1997]

Design: NthValue(0) as a sentinel is fragile.

This emits a WindowFuncKind::NthValue(0) that must be resolved by lower_window before it enters the tree. A debug_assert! catches the bypass in debug builds, but in release the sentinel leaks silently and schema derivation will treat it as NthValue(0) (meaning the 0th row, which is not a valid SQL index).

If lower_window_func_kind is ever called from a future code path that isn't lower_window (e.g. a PromQL path that reuses this helper), the sentinel will escape undetected in production.

Consider NthValue(Option<u64>) instead: None = unresolved (panics loudly in schema derivation if accidentally used), Some(n) = valid N.

[milindsrivastava1997]

Known gap needs a test to document current behaviour.

The comment correctly describes the limitation (Project → Aggregate → * → Scan leaves Scan.columns empty), but there is no test asserting this. Without a test, a future fix or DataFusion upgrade could change the behaviour silently in either direction.

Add a test like:

// SELECT region, COUNT(*) FROM metrics GROUP BY region
// → Scan.columns must be empty ("all columns" semantics) because push_columns_into_scan
// does not propagate through Aggregate.
let source = find_source(&result).unwrap();
let Source::Table { columns, .. } = source else { panic!() };
assert!(columns.is_empty(), "aggregate topology: columns should be empty (all-columns scan)");

This makes the gap visible in CI and signals intent for whoever implements column-pruning later.

[milindsrivastava1997]

Design: invalid catalog fails all queries, even those that don't touch the bad table.

If catalog.tables contains table A with a bad time_column and table B that is valid, a batch of queries against B only will still return Err(InvalidExpression) for every entry. This is surprising and could obscure the real problem in logs.

Two options:

1. Fail all (current): keep it, but document the behaviour explicitly and add a test that asserts it.
2. Fail selectively: validate lazily per query inside SqlLowerer::lower, so only queries that reference the invalid table fail.

At minimum, the error message should name the offending table (it does — good) and the call site should make the batch-wide behaviour visible in tests.

[milindsrivastava1997]

Design debt: output_names couples core IR to DataFusion's internal naming convention — this is a pre-L4 blocker.

The output_names field is populated with DataFusion's own schema field names (e.g. "MIN(metrics.ts)"). An enclosing Project node resolves column references against these names, so any non-DataFusion lowerer (PromQL, future dialects) that emits an Aggregate must produce DataFusion-style names or column lookup silently falls back to Float64.

The TODO comment's proposed fix (emit a Project on top of every Aggregate to rename DataFusion's names to user-visible aliases) is the right direction. Recommend creating a tracking issue before L4 is built, since the schema contract between Aggregate and its parent Project will be much harder to change once L4 rule matching depends on field names.

[milindsrivastava1997]

Design debt: string sentinel "*" for wildcard is fragile — recommend fixing before L4 consumes AggIntent::col().

Any column whose name happens to be "*" (unusual but not rejected by the type system) will be silently misclassified as the wildcard sentinel, causing col() to return None and schema derivation to fall back to Float64 instead of the real type.

The TODO's recommendation (Option<ColumnRef> where None = wildcard / no column) is the right fix and is a small change:

// In AggIntent:
Sum { col: Option<ColumnRef> },
// In agg_col():
fn agg_col(args: &[Expr]) -> Option<ColumnRef> { ... }

This propagates naturally through col() and removes the is_wildcard() method entirely. Since L4 rules will inspect AggIntent::col() to determine sketch parameters, fixing this now avoids a breaking change later.