feat(sql): support date types and calendar interval expressions

crates/frontend-sql/Cargo.toml

@@ -28,6 +28,10 @@ serde_yaml = "0.9"
 name = "synthetic_packet_trace"
 path = "tests/data_quality_check/synthetic_packet_trace.rs"
 
+[[test]]
+name = "tpch_deequ"
+path = "tests/data_quality_check/tpch_deequ.rs"
+
 [[test]]
 name = "netflow"
 path = "tests/netflow/netflow.rs"

crates/frontend-sql/src/lib.rs

@@ -51,6 +51,11 @@ pub async fn lower_sql_dialect(
         .lower(query, &accuracy)
         .await?;
     let resolved = resolve_root(&unresolved)?;
+    // Binding resolves names; schema inference also checks result types such
+    // as temporal subtraction, whose duration unit the IR cannot represent.
+    resolved
+        .output_schema()
+        .map_err(|error| SqlError::InvalidExpression(error.to_string()))?;
     Ok(resolved)
 }
 

crates/frontend-sql/src/sql/expr.rs

@@ -38,6 +38,27 @@ pub(super) fn df_expr_to_unresolved(expr: &Expr) -> Result<Unresolved, LoweringE
             None => ColumnRef::Named(col.name.clone()),
         })),
 
+        // Keep Arrow date literals equivalent to SQL CAST('YYYY-MM-DD' AS DATE),
+        // including typed nulls, without adding another canonical scalar variant.
+        Expr::Literal(
+            sv @ (datafusion::common::ScalarValue::Date32(_)
+            | datafusion::common::ScalarValue::Date64(_)),
+        ) => {
+            let text = sv.cast_to(&datafusion::arrow::datatypes::DataType::Utf8)?;
+            // Arrow formats Date64 with a time suffix; the canonical Date has
+            // no time-of-day, just like Date64 catalog registration as Date32.
+            let text = match text {
+                datafusion::common::ScalarValue::Utf8(Some(value)) => {
+                    ScalarValue::Utf8(value.split('T').next().unwrap().to_owned())
+                }
+                other => scalar_value_to_asap(&other)?,
+            };
+            Ok(Unresolved::Cast {
+                expr: Rc::new(Unresolved::Literal(text)),
+                to: asap_types::pre_asap::schema::DataType::Date,
+                try_cast: false,
+            })
+        }
         Expr::Literal(sv) => scalar_value_to_asap(sv).map(Unresolved::Literal),
 
         Expr::Alias(a) => df_expr_to_unresolved(&a.expr),
@@ -268,3 +289,37 @@ pub(super) fn split_disjuncts(expr: &Expr) -> Vec<&Expr> {
         _ => vec![expr],
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use asap_types::pre_asap::schema::DataType;
+    use datafusion::common::ScalarValue as DfScalarValue;
+
+    // Typed Arrow dates normalize to the same typed form as SQL date casts.
+    #[test]
+    fn arrow_date_literals_preserve_value_and_type() {
+        for (value, expected) in [
+            (
+                DfScalarValue::Date32(Some(0)),
+                ScalarValue::Utf8("1970-01-01".into()),
+            ),
+            (
+                DfScalarValue::Date64(Some(-86_400_000)),
+                ScalarValue::Utf8("1969-12-31".into()),
+            ),
+            (DfScalarValue::Date32(None), ScalarValue::Null),
+            (DfScalarValue::Date64(None), ScalarValue::Null),
+        ] {
+            let actual = df_expr_to_unresolved(&Expr::Literal(value)).unwrap();
+            assert_eq!(
+                actual,
+                Unresolved::Cast {
+                    expr: Rc::new(Unresolved::Literal(expected)),
+                    to: DataType::Date,
+                    try_cast: false,
+                }
+            );
+        }
+    }
+}

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

@@ -184,6 +184,29 @@ impl<'a> SqlLowerer<'a> {
         };
         let rewriter = ApplyFunctionRewrites::new(vec![Arc::new(ClickHouseBuiltinRewrite)]);
         let plan = rewriter.analyze(plan, ctx.state().options())?;
+        // Output schemas omit predicate and nested-expression types. Check the
+        // typed SQL plan before lowering erases fixed-duration units.
+        plan.apply_with_subqueries(|node| {
+            let mut schema = DFSchema::empty();
+            for input in node.inputs() {
+                schema.merge(input.schema());
+            }
+            schema.merge(node.schema());
+            node.apply_expressions(|expr| {
+                expr.apply(|nested| {
+                    if let Expr::BinaryExpr(binary) = nested {
+                        if binary.op == logical_expr::Operator::Minus
+                            && matches!(nested.get_type(&schema)?, ArrowDataType::Duration(_))
+                        {
+                            return Err(datafusion::common::DataFusionError::Plan(
+                                "temporal subtraction produces an unsupported duration type".into(),
+                            ));
+                        }
+                    }
+                    Ok(TreeNodeRecursion::Continue)
+                })
+            })
+        })?;
         let _guard = AccuracyGuard::install(accuracy.clone());
         self.lower_plan(&plan)
     }
@@ -1613,6 +1636,13 @@ fn lower_agg_intent(expr: &Expr) -> Result<AggIntent<ColumnRef>, LoweringError>
                     "DISTINCT {name}"
                 )));
             }
+            // Cardinality carries one column; dropping extra DISTINCT arguments
+            // would silently change tuple cardinality into single-column cardinality.
+            if matches!(semantic, AggSemantic::Count) && agg_fn.distinct && agg_fn.args.len() != 1 {
+                return Err(LoweringError::UnsupportedAggregate(
+                    "multi-column COUNT(DISTINCT)".into(),
+                ));
+            }
             // Value reducers (`reducer_col`) require a real column — `SUM(a*b)`
             // is rejected, not silently reduced over a probe column. Quantile
             // and CountDistinct reduce a column too, so they take the same path:

crates/frontend-sql/src/sql/types.rs

@@ -53,6 +53,24 @@ pub(super) fn scalar_value_to_asap(sv: &DfScalarValue) -> Result<ScalarValue, Lo
             Ok(ScalarValue::Utf8(s.clone()))
         }
         DfScalarValue::Boolean(Some(b)) => Ok(ScalarValue::Boolean(*b)),
+        // All three of DataFusion's interval scalars land on one canonical
+        // shape; the narrower two simply leave the fields they do not carry
+        // at zero.
+        DfScalarValue::IntervalYearMonth(Some(months)) => Ok(ScalarValue::Interval {
+            months: *months,
+            days: 0,
+            nanos: 0,
+        }),
+        DfScalarValue::IntervalDayTime(Some(v)) => Ok(ScalarValue::Interval {
+            months: 0,
+            days: v.days,
+            nanos: i64::from(v.milliseconds) * 1_000_000,
+        }),
+        DfScalarValue::IntervalMonthDayNano(Some(v)) => Ok(ScalarValue::Interval {
+            months: v.months,
+            days: v.days,
+            nanos: v.nanoseconds,
+        }),
         _ if sv.is_null() => Ok(ScalarValue::Null),
         _ => Err(LoweringError::InvalidExpression(format!(
             "unsupported scalar: {sv:?}"
@@ -72,6 +90,8 @@ pub(super) fn arrow_to_dtype(dt: &ArrowDataType) -> Result<DataType, LoweringErr
         ArrowDataType::Utf8 | ArrowDataType::LargeUtf8 => Ok(DataType::Utf8),
         ArrowDataType::Boolean => Ok(DataType::Bool),
         ArrowDataType::Timestamp(_, _) => Ok(DataType::Timestamp),
+        ArrowDataType::Date32 | ArrowDataType::Date64 => Ok(DataType::Date),
+        ArrowDataType::Interval(_) => Ok(DataType::Interval),
         ArrowDataType::List(element) => Ok(DataType::List {
             element: Box::new(Column::new(
                 element.name(),
@@ -155,6 +175,18 @@ pub(super) fn dtype_to_arrow(dt: &DataType) -> ArrowDataType {
         DataType::Timestamp => {
             ArrowDataType::Timestamp(datafusion::arrow::datatypes::TimeUnit::Millisecond, None)
         }
+        // Deliberately narrowing: `Date64` lowers to `DataType::Date` and comes
+        // back as `Date32`. Both spell the same calendar date and nothing in
+        // the planner reads the width; a catalog that wants `Date64` back would
+        // need a second variant carrying no planning information.
+        DataType::Date => ArrowDataType::Date32,
+        // Only reachable through a hand-built schema: `Interval` types a
+        // literal, and no catalog declares a column with it. Mapped to the
+        // same three-field shape `ScalarValue::Interval` carries rather than
+        // left to panic.
+        DataType::Interval => {
+            ArrowDataType::Interval(datafusion::arrow::datatypes::IntervalUnit::MonthDayNano)
+        }
     }
 }
 
@@ -172,6 +204,79 @@ pub(super) fn schema_to_arrow(schema: &Schema) -> ArrowSchema {
 mod tests {
     use super::*;
 
+    /// Both Arrow date widths bridge to the one canonical `Date`, and it
+    /// registers back as `Date32` — the documented narrowing.
+    #[test]
+    fn both_arrow_date_widths_bridge_to_date() {
+        assert_eq!(
+            arrow_to_dtype(&ArrowDataType::Date32).unwrap(),
+            DataType::Date
+        );
+        assert_eq!(
+            arrow_to_dtype(&ArrowDataType::Date64).unwrap(),
+            DataType::Date
+        );
+        assert_eq!(dtype_to_arrow(&DataType::Date), ArrowDataType::Date32);
+    }
+
+    /// Every Arrow interval width shares the canonical calendar interval type.
+    #[test]
+    fn interval_types_round_trip_through_the_catalog_bridge() {
+        use datafusion::arrow::datatypes::IntervalUnit;
+        for unit in [
+            IntervalUnit::YearMonth,
+            IntervalUnit::DayTime,
+            IntervalUnit::MonthDayNano,
+        ] {
+            assert_eq!(
+                arrow_to_dtype(&ArrowDataType::Interval(unit)).unwrap(),
+                DataType::Interval
+            );
+        }
+        assert_eq!(
+            arrow_to_dtype(&dtype_to_arrow(&DataType::Interval)).unwrap(),
+            DataType::Interval
+        );
+    }
+
+    /// All three of DataFusion's interval scalars carry into the one canonical
+    /// three-field shape, with the fields they do not spell left at zero.
+    #[test]
+    fn every_datafusion_interval_scalar_carries_across() {
+        use datafusion::arrow::datatypes::{IntervalDayTime, IntervalMonthDayNano};
+
+        assert_eq!(
+            scalar_value_to_asap(&DfScalarValue::IntervalYearMonth(Some(14))).unwrap(),
+            ScalarValue::Interval {
+                months: 14,
+                days: 0,
+                nanos: 0
+            }
+        );
+        assert_eq!(
+            scalar_value_to_asap(&DfScalarValue::IntervalDayTime(Some(IntervalDayTime::new(
+                30, 500
+            ))))
+            .unwrap(),
+            ScalarValue::Interval {
+                months: 0,
+                days: 30,
+                nanos: 500_000_000
+            }
+        );
+        assert_eq!(
+            scalar_value_to_asap(&DfScalarValue::IntervalMonthDayNano(Some(
+                IntervalMonthDayNano::new(1, 2, 3)
+            )))
+            .unwrap(),
+            ScalarValue::Interval {
+                months: 1,
+                days: 2,
+                nanos: 3
+            }
+        );
+    }
+
     /// Nested map values and value nullability survive catalog registration.
     #[test]
     fn nested_map_schema_round_trip() {

crates/frontend-sql/tests/bgp_jan2024_workload/bgp_jan2024_workload.rs

@@ -173,15 +173,8 @@ async fn corpus_lowering_matches_the_pinned_aggregate_tally() {
     // not all -- as the issue itself flags, `splitByChar(...)[-1]`-style
     // calls (and a couple of other array/map-index uses) now plan far enough
     // to hit the same pre-existing map/array-index `NotImplemented` gap, and
-    // two `toStartOfInterval(...)` queries plan far enough to hit a
-    // different pre-existing gap: `types::scalar_value_to_asap` doesn't yet
-    // convert an `INTERVAL x unit` literal (`DfScalarValue::
-    // IntervalMonthDayNano`), so those two land in `Other` via
-    // `LoweringError::InvalidExpression` instead. Both are companion gaps
-    // this issue's scope explicitly doesn't chase down (see its "known
-    // caveat" section) -- getting these functions' *names* to lower to a
-    // structurally correct `FunctionCall` node is what's in scope here, not
-    // array/map indexing or interval-literal conversion.
+    // the two `toStartOfInterval(...)` queries now lower end to end because
+    // interval literal conversion is supported.
     // `argMax` support (issue #232 -- `AggIntent::Extension`, catalog-driven
     // `RewriteKind::PassThrough`) clears the "unknown function: argmax" `Plan`
     // failure for all 3 corpus occurrences: every one has both arguments as
@@ -198,7 +191,10 @@ async fn corpus_lowering_matches_the_pinned_aggregate_tally() {
     // entry, so the query still fails at the first unknown-function name it
     // hits, just no longer `laginframe`. Out of scope for #267, same as
     // `splitByChar`'s array-indexing companion gap above.
-    expect(Category::Lowered, 152);
+    // 152 -> 154: `ScalarValue::Interval` (this branch) converts the
+    // `INTERVAL x unit` literal the two `toStartOfInterval(...)` queries
+    // carry.
+    expect(Category::Lowered, 154);
     expect(Category::Plan, 40);
     expect(Category::Schema, 0);
     expect(Category::Parse, 0);
@@ -212,8 +208,8 @@ async fn corpus_lowering_matches_the_pinned_aggregate_tally() {
     // during typed planning because the Map adapter rejects array inputs.
     expect(Category::NotImplemented, 0);
     expect(Category::UnsupportedFeature, 6);
-    // Two `toStartOfInterval(...)` queries -- see the `toStartOfInterval`
-    // note above; a pre-existing `INTERVAL`-literal conversion gap, not a
-    // ClickHouse scalar-builtin catalog gap.
-    expect(Category::Other, 2);
+    // Was 2: the two `toStartOfInterval(...)` queries whose `INTERVAL`-literal
+    // conversion gap the `toStartOfInterval` note above describes. Both now
+    // lower end to end and are counted in `Lowered`.
+    expect(Category::Other, 0);
 }