diff --git a/crates/frontend-sql/Cargo.toml b/crates/frontend-sql/Cargo.toml
index 57062f5d..60bec136 100644
--- a/crates/frontend-sql/Cargo.toml
+++ b/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"
diff --git a/crates/frontend-sql/src/lib.rs b/crates/frontend-sql/src/lib.rs
index a2296e70..5e2d5f05 100644
--- a/crates/frontend-sql/src/lib.rs
+++ b/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)
 }
 
diff --git a/crates/frontend-sql/src/sql/expr.rs b/crates/frontend-sql/src/sql/expr.rs
index 6f3a2614..27ca85be 100644
--- a/crates/frontend-sql/src/sql/expr.rs
+++ b/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,
+                }
+            );
+        }
+    }
+}
diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs
index dc9116eb..5e085a06 100644
--- a/crates/frontend-sql/src/sql/mod.rs
+++ b/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:
diff --git a/crates/frontend-sql/src/sql/types.rs b/crates/frontend-sql/src/sql/types.rs
index 020399f3..3cc2b924 100644
--- a/crates/frontend-sql/src/sql/types.rs
+++ b/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() {
diff --git a/crates/frontend-sql/tests/bgp_jan2024_workload/bgp_jan2024_workload.rs b/crates/frontend-sql/tests/bgp_jan2024_workload/bgp_jan2024_workload.rs
index 09134a26..302a349a 100644
--- a/crates/frontend-sql/tests/bgp_jan2024_workload/bgp_jan2024_workload.rs
+++ b/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);
 }
diff --git a/crates/frontend-sql/tests/data_quality_check/data/tpch_deequ_queries.sql b/crates/frontend-sql/tests/data_quality_check/data/tpch_deequ_queries.sql
new file mode 100644
index 00000000..8f2932b5
--- /dev/null
+++ b/crates/frontend-sql/tests/data_quality_check/data/tpch_deequ_queries.sql
@@ -0,0 +1,165 @@
+-- The DQC warehouse-ingestion check set, as one SQL statement per check.
+--
+-- Source: sidra's `workload/tpch_deequ.yaml`, 50 checks over TPC-H `lineitem`,
+-- each fused into one query by `pyutils.workload.backend.fuse` — every metric
+-- node becomes a CTE and the assertion node becomes the outer SELECT, so the
+-- threshold a check asserts (`completeness >= 0.99`) travels with the query
+-- rather than being stripped off it.
+--
+-- Regenerate with, from the sidra repo:
+--   uv run python pyscripts/asap-planner_test/emit_sql_set.py workload/tpch_deequ.yaml
+--
+-- ONE QUERY PER LINE, and the reader splits on newlines rather than on `;`:
+-- U-P3p's regex literal is '^[a-zA-Z ,.:;!?-]+$', which contains a semicolon,
+-- so a `;` split would fragment it. `fuse` emits single-line SQL, so the line
+-- is an exact statement boundary and no escaping is needed.
+
+-- U-P1a
+WITH metric AS (SELECT COUNT(l_shipdate) * 1.0 / COUNT(*) AS completeness FROM lineitem) SELECT completeness, completeness >= 0.99 AS ok FROM metric
+
+-- U-P1b
+WITH metric AS (SELECT COUNT(l_orderkey) * 1.0 / COUNT(*) AS orderkey, COUNT(l_partkey) * 1.0 / COUNT(*) AS partkey, COUNT(l_suppkey) * 1.0 / COUNT(*) AS suppkey, COUNT(l_linenumber) * 1.0 / COUNT(*) AS linenumber FROM lineitem) SELECT orderkey, partkey, suppkey, linenumber, orderkey = 1.0 AND partkey = 1.0 AND suppkey = 1.0 AND linenumber = 1.0 AS ok FROM metric
+
+-- U-P1c
+WITH metric AS (SELECT COUNT(l_comment) * 1.0 / COUNT(*) AS comment, COUNT(l_shipinstruct) * 1.0 / COUNT(*) AS shipinstruct, COUNT(l_shipmode) * 1.0 / COUNT(*) AS shipmode FROM lineitem) SELECT comment, shipinstruct, shipmode, comment = 1.0 AND shipinstruct = 1.0 AND shipmode = 1.0 AS ok FROM metric
+
+-- U-P1d
+WITH metric AS (SELECT COUNT(l_commitdate) * 1.0 / COUNT(*) AS commitdate, COUNT(l_receiptdate) * 1.0 / COUNT(*) AS receiptdate FROM lineitem) SELECT commitdate, receiptdate, commitdate = 1.0 AND receiptdate = 1.0 AS ok FROM metric
+
+-- U-P1e
+WITH metric AS (SELECT COUNT(l_quantity) * 1.0 / COUNT(*) AS quantity, COUNT(l_extendedprice) * 1.0 / COUNT(*) AS extendedprice, COUNT(l_discount) * 1.0 / COUNT(*) AS discount, COUNT(l_tax) * 1.0 / COUNT(*) AS tax, COUNT(l_returnflag) * 1.0 / COUNT(*) AS returnflag, COUNT(l_linestatus) * 1.0 / COUNT(*) AS linestatus FROM lineitem) SELECT quantity, extendedprice, discount, tax, returnflag, linestatus, quantity = 1.0 AND extendedprice = 1.0 AND discount = 1.0 AND tax = 1.0 AND returnflag = 1.0 AND linestatus = 1.0 AS ok FROM metric
+
+-- U-P2a
+WITH metric AS (SELECT COUNT(DISTINCT l_orderkey) * 1.0 / COUNT(*) AS distinctness FROM lineitem) SELECT distinctness, distinctness >= 0.999 AS ok FROM metric
+
+-- U-P2b
+WITH metric AS (SELECT COUNT(DISTINCT l_orderkey, l_linenumber) * 1.0 / COUNT(*) AS pk_distinctness FROM lineitem) SELECT pk_distinctness, pk_distinctness = 1.0 AS ok FROM metric
+
+-- U-P2c
+WITH metric AS (SELECT APPROX_DISTINCT(l_partkey) AS parts, APPROX_DISTINCT(l_suppkey) AS suppliers FROM lineitem) SELECT parts, suppliers, parts >= 1000 AND suppliers >= 100 AS ok FROM metric
+
+-- U-P2d
+WITH metric AS (SELECT COUNT(DISTINCT l_returnflag) AS nd_returnflag, COUNT(DISTINCT l_linestatus) AS nd_linestatus, COUNT(DISTINCT l_shipmode) AS nd_shipmode, COUNT(DISTINCT l_shipinstruct) AS nd_shipinstruct, COUNT(DISTINCT l_linenumber) AS nd_linenumber, COUNT(DISTINCT l_quantity) AS nd_quantity FROM lineitem) SELECT nd_returnflag, nd_linestatus, nd_shipmode, nd_shipinstruct, nd_linenumber, nd_quantity, nd_returnflag = 3 AND nd_linestatus = 2 AND nd_shipmode = 7 AND nd_shipinstruct = 4 AND nd_linenumber = 7 AND nd_quantity = 50 AS ok FROM metric
+
+-- U-P2e
+WITH metric AS (SELECT APPROX_DISTINCT(l_shipdate) AS nd_shipdate, APPROX_DISTINCT(l_commitdate) AS nd_commitdate FROM lineitem) SELECT nd_shipdate, nd_commitdate, nd_shipdate BETWEEN 2300 AND 2800 AND nd_commitdate BETWEEN 2250 AND 2750 AS ok FROM metric
+
+-- U-P3a
+WITH metric AS (SELECT AVG(CASE WHEN l_quantity BETWEEN 1 AND 50 THEN 1.0 ELSE 0.0 END) AS in_range FROM lineitem) SELECT in_range, in_range = 1.0 AS ok FROM metric
+
+-- U-P3b
+WITH metric AS (SELECT AVG(CASE WHEN l_discount BETWEEN 0.00 AND 0.10 THEN 1.0 ELSE 0.0 END) AS discount_ok, AVG(CASE WHEN l_tax BETWEEN 0.00 AND 0.08 THEN 1.0 ELSE 0.0 END) AS tax_ok FROM lineitem) SELECT discount_ok, tax_ok, discount_ok = 1.0 AND tax_ok = 1.0 AS ok FROM metric
+
+-- U-P3c
+WITH metric AS (SELECT AVG(CASE WHEN l_returnflag IN ('A', 'N', 'R') THEN 1.0 ELSE 0.0 END) AS returnflag_ok, AVG(CASE WHEN l_linestatus IN ('O', 'F') THEN 1.0 ELSE 0.0 END) AS linestatus_ok FROM lineitem) SELECT returnflag_ok, linestatus_ok, returnflag_ok = 1.0 AND linestatus_ok = 1.0 AS ok FROM metric
+
+-- U-P3d
+WITH metric AS (SELECT AVG(CASE WHEN REGEXP_LIKE(l_shipmode, '^(AIR|FOB|MAIL|RAIL|REG AIR|SHIP|TRUCK)$') THEN 1.0 ELSE 0.0 END) AS shipmode_ok, AVG(CASE WHEN REGEXP_LIKE(l_shipinstruct, '^(COLLECT COD|DELIVER IN PERSON|NONE|TAKE BACK RETURN)$') THEN 1.0 ELSE 0.0 END) AS shipinstruct_ok FROM lineitem) SELECT shipmode_ok, shipinstruct_ok, shipmode_ok = 1.0 AND shipinstruct_ok = 1.0 AS ok FROM metric
+
+-- U-P3e
+WITH metric AS (SELECT AVG(CASE WHEN l_extendedprice > 0.0 THEN 1.0 ELSE 0.0 END) AS price_positive, AVG(CASE WHEN l_linenumber BETWEEN 1 AND 7 THEN 1.0 ELSE 0.0 END) AS linenumber_ok FROM lineitem) SELECT price_positive, linenumber_ok, price_positive = 1.0 AND linenumber_ok = 1.0 AS ok FROM metric
+
+-- U-P3f
+WITH metric AS (SELECT MIN(LENGTH(l_comment)) AS comment_min, MAX(LENGTH(l_comment)) AS comment_max, MAX(LENGTH(l_shipmode)) AS shipmode_max FROM lineitem) SELECT comment_min, comment_max, shipmode_max, comment_min >= 1 AND comment_max <= 44 AND shipmode_max <= 10 AS ok FROM metric
+
+-- U-P3g
+WITH metric AS (SELECT AVG(CASE WHEN l_shipdate BETWEEN CAST('1992-01-01' AS DATE) AND CAST('1998-12-31' AS DATE) THEN 1.0 ELSE 0.0 END) AS in_window, AVG(CASE WHEN l_shipdate <= l_receiptdate THEN 1.0 ELSE 0.0 END) AS before_receipt FROM lineitem) SELECT in_window, before_receipt, in_window = 1.0 AND before_receipt = 1.0 AS ok FROM metric
+
+-- U-P3h
+WITH metric AS (SELECT MIN(l_orderkey) AS min_orderkey, MIN(l_partkey) AS min_partkey, MIN(l_suppkey) AS min_suppkey FROM lineitem) SELECT min_orderkey, min_partkey, min_suppkey, min_orderkey >= 1 AND min_partkey >= 1 AND min_suppkey >= 1 AS ok FROM metric
+
+-- U-P3i
+WITH metric AS (SELECT AVG(CASE WHEN l_orderkey > 0 AND l_orderkey % 32 < 8 THEN 1.0 ELSE 0.0 END) AS on_grid FROM lineitem) SELECT on_grid, on_grid = 1.0 AS ok FROM metric
+
+-- U-P3j
+WITH metric AS (SELECT AVG(CASE WHEN l_discount IN (0.00, 0.01, 0.02, 0.03, 0.04, 0.05, 0.06, 0.07, 0.08, 0.09, 0.10) THEN 1.0 ELSE 0.0 END) AS discount_on_grid, AVG(CASE WHEN l_tax IN (0.00, 0.01, 0.02, 0.03, 0.04, 0.05, 0.06, 0.07, 0.08) THEN 1.0 ELSE 0.0 END) AS tax_on_grid FROM lineitem) SELECT discount_on_grid, tax_on_grid, discount_on_grid = 1.0 AND tax_on_grid = 1.0 AS ok FROM metric
+
+-- U-P3k
+WITH metric AS (SELECT AVG(CASE WHEN l_quantity = FLOOR(l_quantity) THEN 1.0 ELSE 0.0 END) AS integral FROM lineitem) SELECT integral, integral = 1.0 AS ok FROM metric
+
+-- U-P3l
+WITH metric AS (SELECT AVG(CASE WHEN l_commitdate BETWEEN CAST('1992-01-31' AS DATE) AND CAST('1998-10-31' AS DATE) THEN 1.0 ELSE 0.0 END) AS commit_window, AVG(CASE WHEN l_receiptdate BETWEEN CAST('1992-01-03' AS DATE) AND CAST('1998-12-31' AS DATE) THEN 1.0 ELSE 0.0 END) AS receipt_window FROM lineitem) SELECT commit_window, receipt_window, commit_window = 1.0 AND receipt_window = 1.0 AS ok FROM metric
+
+-- U-P3m
+WITH metric AS (SELECT AVG(CASE WHEN l_receiptdate > l_shipdate THEN 1.0 ELSE 0.0 END) AS lag_at_least_1, AVG(CASE WHEN l_receiptdate <= l_shipdate + INTERVAL '30' DAY THEN 1.0 ELSE 0.0 END) AS lag_at_most_30 FROM lineitem) SELECT lag_at_least_1, lag_at_most_30, lag_at_least_1 = 1.0 AND lag_at_most_30 = 1.0 AS ok FROM metric
+
+-- U-P3n
+WITH metric AS (SELECT AVG(CASE WHEN (l_shipdate > CAST('1995-06-17' AS DATE)) = (l_linestatus = 'O') THEN 1.0 ELSE 0.0 END) AS linestatus_agrees, AVG(CASE WHEN (l_receiptdate <= CAST('1995-06-17' AS DATE)) = (l_returnflag IN ('R', 'A')) THEN 1.0 ELSE 0.0 END) AS returnflag_agrees FROM lineitem) SELECT linestatus_agrees, returnflag_agrees, linestatus_agrees = 1.0 AND returnflag_agrees = 1.0 AS ok FROM metric
+
+-- U-P3o
+WITH metric AS (SELECT MIN(LENGTH(l_returnflag)) AS rf_min, MAX(LENGTH(l_returnflag)) AS rf_max, MIN(LENGTH(l_linestatus)) AS ls_min, MAX(LENGTH(l_linestatus)) AS ls_max, MIN(LENGTH(l_shipinstruct)) AS si_min, MAX(LENGTH(l_shipinstruct)) AS si_max, MIN(LENGTH(l_shipmode)) AS sm_min FROM lineitem) SELECT rf_min, rf_max, ls_min, ls_max, si_min, si_max, sm_min, rf_min = 1 AND rf_max = 1 AND ls_min = 1 AND ls_max = 1 AND si_min = 4 AND si_max = 17 AND sm_min = 3 AS ok FROM metric
+
+-- U-P3p
+WITH metric AS (SELECT AVG(CASE WHEN REGEXP_LIKE(l_comment, '^[a-zA-Z ,.:;!?-]+$') THEN 1.0 ELSE 0.0 END) AS comment_charset, AVG(CASE WHEN REGEXP_LIKE(l_shipinstruct, '^[A-Z ]+$') THEN 1.0 ELSE 0.0 END) AS shipinstruct_charset FROM lineitem) SELECT comment_charset, shipinstruct_charset, comment_charset = 1.0 AND shipinstruct_charset = 1.0 AS ok FROM metric
+
+-- U-P3q
+WITH metric AS (SELECT AVG(CASE WHEN (l_extendedprice * 100) % l_quantity = 0 THEN 1.0 ELSE 0.0 END) AS recomputable FROM lineitem) SELECT recomputable, recomputable = 1.0 AS ok FROM metric
+
+-- U-P4a
+WITH metric AS (SELECT AVG(l_extendedprice) AS mean_price, APPROX_PERCENTILE_CONT(l_discount, 0.99) AS p99_discount FROM lineitem) SELECT mean_price, p99_discount, mean_price BETWEEN 30000 AND 40000 AND p99_discount <= 0.10 AS ok FROM metric
+
+-- U-P4b
+WITH metric AS (SELECT APPROX_PERCENTILE_CONT(l_quantity, 0.50) AS p50, APPROX_PERCENTILE_CONT(l_quantity, 0.90) AS p90, APPROX_PERCENTILE_CONT(l_quantity, 0.99) AS p99 FROM lineitem) SELECT p50, p90, p99, p50 BETWEEN 20 AND 31 AND p90 BETWEEN 40 AND 50 AND p99 BETWEEN 45 AND 50 AS ok FROM metric
+
+-- U-P4c
+WITH metric AS (SELECT AVG(l_quantity) AS mean_qty, STDDEV_POP(l_quantity) AS sd_qty, MIN(l_quantity) AS min_qty, MAX(l_quantity) AS max_qty FROM lineitem) SELECT mean_qty, sd_qty, min_qty, max_qty, mean_qty BETWEEN 25 AND 26 AND sd_qty BETWEEN 14 AND 15 AND min_qty >= 1 AND max_qty <= 50 AS ok FROM metric
+
+-- U-P4d
+WITH metric AS (SELECT CORR(l_quantity, l_extendedprice) AS r FROM lineitem) SELECT r, r BETWEEN 0.80 AND 1.00 AS ok FROM metric
+
+-- U-P4e
+WITH metric AS (SELECT SUM(l_quantity) AS total_qty, SUM(l_extendedprice) AS total_price FROM lineitem) SELECT total_qty, total_price, total_qty > 0 AND total_price > 0 AS ok FROM metric
+
+-- U-P4f
+WITH metric AS (SELECT AVG(l_discount) AS mean_discount, AVG(l_tax) AS mean_tax, STDDEV_POP(l_discount) AS sd_discount, STDDEV_POP(l_tax) AS sd_tax FROM lineitem) SELECT mean_discount, mean_tax, sd_discount, sd_tax, mean_discount BETWEEN 0.048 AND 0.052 AND mean_tax BETWEEN 0.038 AND 0.042 AND sd_discount BETWEEN 0.031 AND 0.032 AND sd_tax BETWEEN 0.025 AND 0.026 AS ok FROM metric
+
+-- U-P4g
+WITH metric AS (SELECT AVG(l_linenumber) AS mean_lineno, MIN(l_linenumber) AS min_lineno, MAX(l_linenumber) AS max_lineno FROM lineitem) SELECT mean_lineno, min_lineno, max_lineno, mean_lineno BETWEEN 2.9 AND 3.1 AND min_lineno >= 1 AND max_lineno <= 7 AS ok FROM metric
+
+-- U-P4h
+WITH metric AS (SELECT APPROX_PERCENTILE_CONT(l_extendedprice, 0.50) AS p50_price, APPROX_PERCENTILE_CONT(l_extendedprice, 0.95) AS p95_price FROM lineitem) SELECT p50_price, p95_price, p50_price BETWEEN 30000 AND 40000 AND p95_price BETWEEN 65000 AND 85000 AS ok FROM metric
+
+-- U-P4i
+WITH metric AS (SELECT AVG(CASE WHEN l_returnflag = 'N' THEN 1.0 ELSE 0.0 END) AS frac_new, AVG(CASE WHEN l_linestatus = 'O' THEN 1.0 ELSE 0.0 END) AS frac_open, AVG(CASE WHEN l_shipmode = 'AIR' THEN 1.0 ELSE 0.0 END) AS frac_air, AVG(CASE WHEN l_shipinstruct = 'NONE' THEN 1.0 ELSE 0.0 END) AS frac_no_instruction FROM lineitem) SELECT frac_new, frac_open, frac_air, frac_no_instruction, frac_new BETWEEN 0.45 AND 0.55 AND frac_open BETWEEN 0.45 AND 0.55 AND frac_air BETWEEN 0.13 AND 0.16 AND frac_no_instruction BETWEEN 0.23 AND 0.27 AS ok FROM metric
+
+-- U-P4j
+WITH metric AS (SELECT AVG(CASE WHEN l_commitdate < l_receiptdate THEN 1.0 ELSE 0.0 END) AS late_rate FROM lineitem) SELECT late_rate, late_rate BETWEEN 0.58 AND 0.68 AS ok FROM metric
+
+-- U-P4k
+WITH metric AS (SELECT MIN(l_extendedprice) AS min_price, MAX(l_extendedprice) AS max_price, MIN(l_discount) AS min_discount, MAX(l_discount) AS max_discount, MIN(l_tax) AS min_tax, MAX(l_tax) AS max_tax FROM lineitem) SELECT min_price, max_price, min_discount, max_discount, min_tax, max_tax, min_price >= 900 AND max_price <= 104950 AND min_discount = 0.00 AND max_discount = 0.10 AND min_tax = 0.00 AND max_tax = 0.08 AS ok FROM metric
+
+-- U-P4l
+WITH metric AS (SELECT CORR(l_discount, l_tax) AS r_discount_tax, CORR(l_quantity, l_discount) AS r_quantity_discount FROM lineitem) SELECT r_discount_tax, r_quantity_discount, r_discount_tax BETWEEN -0.02 AND 0.02 AND r_quantity_discount BETWEEN -0.02 AND 0.02 AS ok FROM metric
+
+-- U-P4m
+WITH metric AS (SELECT APPROX_PERCENTILE_CONT(l_discount, 0.25) AS d25, APPROX_PERCENTILE_CONT(l_discount, 0.75) AS d75, APPROX_PERCENTILE_CONT(l_tax, 0.25) AS t25, APPROX_PERCENTILE_CONT(l_tax, 0.75) AS t75 FROM lineitem) SELECT d25, d75, t25, t75, d25 BETWEEN 0.01 AND 0.03 AND d75 BETWEEN 0.07 AND 0.09 AND t25 BETWEEN 0.01 AND 0.03 AND t75 BETWEEN 0.05 AND 0.07 AS ok FROM metric
+
+-- U-P4n
+WITH metric AS (SELECT APPROX_PERCENTILE_CONT(l_linenumber, 0.50) AS ln50, APPROX_PERCENTILE_CONT(l_linenumber, 0.90) AS ln90 FROM lineitem) SELECT ln50, ln90, ln50 BETWEEN 2 AND 4 AND ln90 BETWEEN 5 AND 7 AS ok FROM metric
+
+-- U-P4o
+WITH metric AS (SELECT AVG(CASE WHEN l_shipdate BETWEEN CAST('1993-01-01' AS DATE) AND CAST('1993-12-31' AS DATE) THEN 1.0 ELSE 0.0 END) AS y1993, AVG(CASE WHEN l_shipdate BETWEEN CAST('1995-01-01' AS DATE) AND CAST('1995-12-31' AS DATE) THEN 1.0 ELSE 0.0 END) AS y1995, AVG(CASE WHEN l_shipdate BETWEEN CAST('1997-01-01' AS DATE) AND CAST('1997-12-31' AS DATE) THEN 1.0 ELSE 0.0 END) AS y1997 FROM lineitem) SELECT y1993, y1995, y1997, y1993 BETWEEN 0.13 AND 0.17 AND y1995 BETWEEN 0.13 AND 0.17 AND y1997 BETWEEN 0.13 AND 0.17 AS ok FROM metric
+
+-- U-P7a
+WITH metric AS (SELECT COUNT(*) AS n FROM lineitem) SELECT n, n > 0 AS ok FROM metric
+
+-- U-P8a
+WITH metric AS (SELECT AVG(CASE WHEN l_shipdate >= CAST('1998-11-01' AS DATE) THEN 1.0 ELSE 0.0 END) AS fresh FROM lineitem) SELECT fresh, fresh > 0.0 AS ok FROM metric
+
+-- U-P8b
+WITH metric AS (SELECT AVG(CASE WHEN l_shipdate <= CAST('1998-12-31' AS DATE) THEN 1.0 ELSE 0.0 END) AS not_future FROM lineitem) SELECT not_future, not_future = 1.0 AS ok FROM metric
+
+-- U-P8c
+WITH metric AS (SELECT AVG(CASE WHEN l_receiptdate >= CAST('1998-12-01' AS DATE) THEN 1.0 ELSE 0.0 END) AS recent FROM lineitem) SELECT recent, recent > 0.0 AS ok FROM metric
+
+-- U-P8d
+WITH metric AS (SELECT AVG(CASE WHEN l_shipdate < CAST('1992-02-01' AS DATE) THEN 1.0 ELSE 0.0 END) AS oldest FROM lineitem) SELECT oldest, oldest > 0.0 AS ok FROM metric
+
+-- U-P9a
+WITH metric AS (SELECT AVG(CASE WHEN l_extendedprice >= l_quantity * 900.0 THEN 1.0 ELSE 0.0 END) AS holds FROM lineitem) SELECT holds, holds = 1.0 AS ok FROM metric
+
+-- U-P9b
+WITH metric AS (SELECT AVG(CASE WHEN l_extendedprice <= l_quantity * 2099.0 THEN 1.0 ELSE 0.0 END) AS price_upper, AVG(CASE WHEN l_extendedprice * (1 - l_discount) * (1 + l_tax) >= 0.0 THEN 1.0 ELSE 0.0 END) AS charge_nonneg FROM lineitem) SELECT price_upper, charge_nonneg, price_upper = 1.0 AND charge_nonneg = 1.0 AS ok FROM metric
+
+-- U-P9c
+WITH metric AS (SELECT AVG(CASE WHEN l_extendedprice * (1 - l_discount) * (1 + l_tax) BETWEEN l_extendedprice * 0.90 AND l_extendedprice * 1.08 THEN 1.0 ELSE 0.0 END) AS in_band FROM lineitem) SELECT in_band, in_band = 1.0 AS ok FROM metric
diff --git a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs
index 5831aafb..596b6564 100644
--- a/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs
+++ b/crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs
@@ -12,10 +12,9 @@
 //! query corpus that pins the front end against regressions. Two guarantees:
 //!   1. **Totality** — every query returns `Ok` or a clean `LoweringError`,
 //!      never panics.
-//!   2. **Full coverage** — the DataFusion SQL front end lowers **all 70**
-//!      (CTEs, `LAG` window functions, multi-argument `COUNT(DISTINCT …)`,
-//!      `STDDEV_POP`, `HAVING`, `CASE`). A regression that drops any query below
-//!      full coverage trips the ratchet.
+//!   2. **Coverage ratchet** — 61 queries lower; nine composite
+//!      COUNT(DISTINCT) queries are rejected because the IR cannot represent
+//!      tuple cardinality. Successful lowering must not silently drop keys.
 //!
 //! Schema: `packets(srcip, dstip, srcport, dstport, proto, time, pkt_len)`;
 //! flow / 5-tuple = `(srcip, dstip, srcport, dstport, proto)`.
@@ -187,7 +186,7 @@ struct Tally {
 }
 
 #[tokio::test]
-async fn corpus_lowering_is_total_and_fully_supported() {
+async fn corpus_lowering_rejects_only_unsupported_tuple_counts() {
     let cat = catalog();
     let mut t = Tally::default();
     for q in queries() {
@@ -196,7 +195,12 @@ async fn corpus_lowering_is_total_and_fully_supported() {
             Ok(_) => t.lowered += 1,
             // DataFusion surfaces parse/plan failures as `DataFusion(_)`.
             Err(LoweringError::DataFusion(_)) => t.unparseable += 1,
-            Err(_) => t.rejected += 1,
+            Err(LoweringError::UnsupportedAggregate(reason))
+                if reason == "multi-column COUNT(DISTINCT)" =>
+            {
+                t.rejected += 1
+            }
+            Err(error) => panic!("unexpected lowering failure for {q}: {error}"),
         }
     }
     eprintln!("synthetic-packet-trace SQL corpus: {t:?}");
@@ -210,12 +214,8 @@ async fn corpus_lowering_is_total_and_fully_supported() {
         "some DQC queries failed to parse/plan: {t:?}"
     );
 
-    // Full-coverage ratchet: today the SQL front end lowers ALL 70. A change
-    // that can no longer lower some query trips this deliberately.
-    assert_eq!(
-        t.lowered, 70,
-        "SQL lowering coverage regressed below full DQC coverage: {t:?}"
-    );
+    assert_eq!(t.rejected, 9, "expected nine unsupported tuple counts");
+    assert_eq!(t.lowered, 61, "SQL lowering coverage changed: {t:?}");
 }
 
 impl Tally {
@@ -247,21 +247,18 @@ async fn distinct_source_ips_is_cardinality() {
 }
 
 #[tokio::test]
-async fn multi_arg_count_distinct_flow_is_cardinality() {
-    // SP-CD-FLOW — distinct 5-tuples per source port. The multi-column
-    // `COUNT(DISTINCT srcip, dstip, srcport, dstport, proto)` is still one
-    // `Cardinality` (of the tuple), grouped by srcport.
-    let qe = lower(
+async fn multi_arg_count_distinct_flow_is_rejected() {
+    // A single-column Cardinality intent cannot represent a distinct 5-tuple.
+    let error = lower_sql(
         "SELECT srcport, COUNT(DISTINCT srcip, dstip, srcport, dstport, proto) AS n \
          FROM packets GROUP BY srcport ORDER BY n DESC",
+        &catalog(),
+        AccuracyTarget::Exact,
     )
-    .await;
-    let (by, measures) = first_aggregate(&qe).expect("expected an Aggregate");
-    assert_eq!(by.len(), 1, "grouped by srcport");
-    assert!(matches!(
-        measures.as_slice(),
-        [AggIntent::Cardinality { .. }]
-    ));
+    .await
+    .unwrap_err();
+    assert!(matches!(error, LoweringError::UnsupportedAggregate(reason)
+        if reason == "multi-column COUNT(DISTINCT)"));
 }
 
 #[tokio::test]
diff --git a/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs b/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs
new file mode 100644
index 00000000..e6b0c83c
--- /dev/null
+++ b/crates/frontend-sql/tests/data_quality_check/tpch_deequ.rs
@@ -0,0 +1,103 @@
+//! Real-world **SQL** conformance over the DQC warehouse-ingestion check set.
+//!
+//! Source: sidra's `workload/tpch_deequ.yaml` — 50 data-quality checks over
+//! TPC-H `lineitem` (completeness, uniqueness, validity, distributional, volume,
+//! freshness), each fused into one query whose outer SELECT *is* the check's
+//! assertion (`tests/data/tpch_deequ_queries.sql`). Here we only lower them.
+//!
+//! This is the second SQL corpus beside `synthetic_packet_trace.rs`, and it
+//! pins a different shape: where that one is grouped counts and window
+//! functions over a packet trace, these are ungrouped `avg(CASE WHEN …)` folds
+//! with date and interval arithmetic in the predicate — the shape a check set
+//! has when every cell answers one yes/no question about one table.
+//!
+//! Two guarantees:
+//!   1. **Totality** — every query returns `Ok` or a clean `LoweringError`,
+//!      never panics.
+//!   2. **Coverage ratchet** — 47 of 50 lower today. The exact rejected set
+//!      consists of one composite COUNT(DISTINCT) and two correlation checks,
+//!      whose aggregate semantics the canonical IR cannot represent.
+//!
+//! Schema: `lineitem`, the 16 TPC-H columns. The four `DECIMAL(15,2)` columns
+//! are declared `Float64` — the canonical `DataType` has no fixed-point type,
+//! which is the same narrowing sidra's own catalog file makes.
+
+use asap_frontend_sql::{lower_sql, SqlCatalog, SqlError as LoweringError};
+use asap_types::pre_asap::schema::{Column, DataType, Schema};
+use asap_types::types::AccuracyTarget;
+
+const CORPUS: &str = include_str!("data/tpch_deequ_queries.sql");
+
+fn col(name: &str, dtype: DataType) -> Column {
+    Column::new(name, dtype, false)
+}
+
+/// No `time_index` and no `unique_keys`: the checks do not slice by time, and
+/// declaring `(l_orderkey, l_linenumber)` as a key would let a planner fold
+/// U-P2b's `pk_distinctness = 1.0` to a constant — a check that reads no rows
+/// is no longer the check the corpus is measuring.
+fn catalog() -> SqlCatalog {
+    SqlCatalog::new().with_table(
+        "lineitem",
+        Schema::new(vec![
+            col("l_orderkey", DataType::Int64),
+            col("l_partkey", DataType::Int64),
+            col("l_suppkey", DataType::Int64),
+            col("l_linenumber", DataType::Int64),
+            col("l_quantity", DataType::Float64),
+            col("l_extendedprice", DataType::Float64),
+            col("l_discount", DataType::Float64),
+            col("l_tax", DataType::Float64),
+            col("l_returnflag", DataType::Utf8),
+            col("l_linestatus", DataType::Utf8),
+            col("l_shipdate", DataType::Date),
+            col("l_commitdate", DataType::Date),
+            col("l_receiptdate", DataType::Date),
+            col("l_shipinstruct", DataType::Utf8),
+            col("l_shipmode", DataType::Utf8),
+            col("l_comment", DataType::Utf8),
+        ]),
+    )
+}
+
+/// One query per line. Not a `;` split: U-P3p's regex literal
+/// `'^[a-zA-Z ,.:;!?-]+$'` contains a semicolon, and the generator emits
+/// single-line SQL, so the line is the exact statement boundary.
+fn queries() -> Vec<(&'static str, &'static str)> {
+    let mut id = None;
+    let mut queries = Vec::new();
+    for line in CORPUS.lines().map(str::trim) {
+        if let Some(query_id) = line.strip_prefix("-- U-") {
+            id = Some(query_id);
+        } else if !line.is_empty() && !line.starts_with("--") {
+            queries.push((id.take().expect("each query has an ID"), line));
+        }
+    }
+    queries
+}
+
+// Pin rejected IDs and error reasons so coverage swaps cannot pass the ratchet.
+#[tokio::test]
+async fn lowers_the_warehouse_ingestion_check_set() {
+    let cat = catalog();
+    let queries = queries();
+    assert_eq!(queries.len(), 50);
+    let mut rejected = Vec::new();
+    let mut lowered = 0;
+    for (id, query) in queries {
+        match lower_sql(query, &cat, AccuracyTarget::Exact).await {
+            Ok(_) => lowered += 1,
+            Err(LoweringError::UnsupportedAggregate(reason)) => rejected.push((id, reason)),
+            Err(error) => panic!("unexpected failure for U-{id}: {error}"),
+        }
+    }
+    assert_eq!(
+        rejected,
+        vec![
+            ("P2b", "multi-column COUNT(DISTINCT)".into()),
+            ("P4d", "corr".into()),
+            ("P4l", "corr".into()),
+        ]
+    );
+    assert_eq!(lowered, 47);
+}
diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs
index 4ce39b55..642c9713 100644
--- a/crates/frontend-sql/tests/sql_lowering.rs
+++ b/crates/frontend-sql/tests/sql_lowering.rs
@@ -2454,3 +2454,33 @@ async fn clickhouse_tuple_element_preserves_declared_field_metadata() {
         );
     }
 }
+
+// A multi-column DISTINCT must not silently count only the first column.
+#[tokio::test]
+async fn composite_distinct_is_rejected() {
+    let cat = SqlCatalog::new().with_table(
+        "t",
+        Schema::new(vec![
+            Column::new("a", DataType::Int64, false),
+            Column::new("b", DataType::Int64, false),
+        ]),
+    );
+    let error = lower_sql(
+        "SELECT COUNT(DISTINCT a, b) FROM t",
+        &cat,
+        AccuracyTarget::Exact,
+    )
+    .await
+    .unwrap_err();
+    assert!(
+        error.to_string().contains("multi-column COUNT(DISTINCT)"),
+        "{error}"
+    );
+    lower_sql(
+        "SELECT COUNT(DISTINCT a) FROM t",
+        &cat,
+        AccuracyTarget::Exact,
+    )
+    .await
+    .unwrap();
+}
diff --git a/crates/frontend-sql/tests/temporal_types.rs b/crates/frontend-sql/tests/temporal_types.rs
new file mode 100644
index 00000000..bdb53d24
--- /dev/null
+++ b/crates/frontend-sql/tests/temporal_types.rs
@@ -0,0 +1,116 @@
+use asap_frontend_sql::{lower_sql, SqlCatalog};
+use asap_types::{
+    pre_asap::schema::{Column, DataType, Schema},
+    types::AccuracyTarget,
+};
+fn catalog() -> SqlCatalog {
+    SqlCatalog::new().with_table(
+        "t",
+        Schema::new(vec![Column::new("d", DataType::Date, false)]),
+    )
+}
+// Unsupported fixed-duration results fail lowering instead of acquiring a float schema.
+#[tokio::test]
+async fn temporal_subtraction_rejects_unrepresentable_duration() {
+    for dtype in [DataType::Date, DataType::Timestamp] {
+        let catalog =
+            SqlCatalog::new().with_table("t", Schema::new(vec![Column::new("d", dtype, false)]));
+        let error = lower_sql(
+            "SELECT d - d AS elapsed FROM t",
+            &catalog,
+            AccuracyTarget::Exact,
+        )
+        .await
+        .unwrap_err();
+        assert!(
+            error.to_string().contains("unsupported duration type"),
+            "{error}"
+        );
+    }
+}
+// Date/interval arithmetic continues to preserve dates, including explicit interval casts.
+#[tokio::test]
+async fn date_shifts_keep_their_type() {
+    for query in [
+        "SELECT d + INTERVAL '30' DAY AS shifted FROM t",
+        "SELECT d - CAST('1 day' AS INTERVAL) AS shifted FROM t",
+    ] {
+        let node = lower_sql(query, &catalog(), AccuracyTarget::Exact)
+            .await
+            .unwrap();
+        assert_eq!(
+            node.output_schema().unwrap().columns[0].dtype,
+            DataType::Date
+        );
+    }
+}
+// Interval literals and explicit interval casts must both cross the Arrow bridge.
+#[tokio::test]
+async fn interval_cast_lowers_like_interval_literal() {
+    for query in [
+        "SELECT INTERVAL '1' DAY AS duration FROM t",
+        "SELECT CAST('1 day' AS INTERVAL) AS duration FROM t",
+    ] {
+        let node = lower_sql(query, &catalog(), AccuracyTarget::Exact)
+            .await
+            .unwrap();
+        assert_eq!(
+            node.output_schema().unwrap().columns[0].dtype,
+            DataType::Interval
+        );
+    }
+}
+
+// Unsupported durations must be rejected even when hidden by another expression.
+#[tokio::test]
+async fn nested_temporal_subtraction_is_rejected() {
+    for query in [
+        "SELECT (d - d) IS NULL FROM t",
+        "SELECT d FROM t WHERE (d - d) IS NULL",
+        "SELECT CASE WHEN true THEN INTERVAL '1 day' ELSE d - d END FROM t",
+        "SELECT d FROM t ORDER BY d - d",
+    ] {
+        let result = lower_sql(query, &catalog(), AccuracyTarget::Exact).await;
+        let error = result.expect_err(&format!("accepted unsupported duration: {query}"));
+        assert!(
+            error.to_string().contains("unsupported duration type"),
+            "{query}: {error}"
+        );
+    }
+}
+
+// Negating a calendar interval must preserve its type, including nonliteral inputs.
+#[tokio::test]
+async fn negative_intervals_keep_their_type() {
+    for query in [
+        "SELECT -INTERVAL '1 day' AS duration FROM t",
+        "SELECT -CAST('1 day' AS INTERVAL) AS duration FROM t",
+        "SELECT -(INTERVAL '1 day' + INTERVAL '2 days') AS duration FROM t",
+    ] {
+        let node = lower_sql(query, &catalog(), AccuracyTarget::Exact)
+            .await
+            .unwrap();
+        assert_eq!(
+            node.output_schema().unwrap().columns[0].dtype,
+            DataType::Interval,
+            "{query}"
+        );
+    }
+}
+
+// SQL date literal syntax keeps its date type through projection and shifts.
+#[tokio::test]
+async fn sql_date_literals_keep_their_type() {
+    for query in [
+        "SELECT DATE '2024-02-29' FROM t",
+        "SELECT DATE '1969-12-31' + INTERVAL '1 day' FROM t",
+    ] {
+        let node = lower_sql(query, &catalog(), AccuracyTarget::Exact)
+            .await
+            .unwrap();
+        assert_eq!(
+            node.output_schema().unwrap().columns[0].dtype,
+            DataType::Date
+        );
+    }
+}
diff --git a/crates/types/src/pre_asap/expr_ir.rs b/crates/types/src/pre_asap/expr_ir.rs
index b78f37b7..fd52767e 100644
--- a/crates/types/src/pre_asap/expr_ir.rs
+++ b/crates/types/src/pre_asap/expr_ir.rs
@@ -47,6 +47,15 @@ pub enum ScalarValue {
     Utf8(String),
     Boolean(bool),
     Null,
+    /// A calendar duration, in Arrow's three independent fields. Not collapsed
+    /// into one nanosecond count: a month is not a fixed number of nanoseconds,
+    /// so `INTERVAL '1' MONTH` has no faithful scalar form. DataFusion's three
+    /// interval scalars (`YearMonth`, `DayTime`, `MonthDayNano`) all lower here.
+    Interval {
+        months: i32,
+        days: i32,
+        nanos: i64,
+    },
 }
 
 /// Binary comparison operators.
diff --git a/crates/types/src/pre_asap/query_expr.rs b/crates/types/src/pre_asap/query_expr.rs
index dbc14b48..3763a553 100644
--- a/crates/types/src/pre_asap/query_expr.rs
+++ b/crates/types/src/pre_asap/query_expr.rs
@@ -1663,6 +1663,7 @@ fn infer_expr_type(
     schema: &Schema,
 ) -> Result<(DataType, bool), QueryExprError> {
     Ok(match expr {
+        QueryExpr::CurrentTimestamp => (DataType::Timestamp, false),
         QueryExpr::Column(id) => schema
             .columns
             .get(*id)
@@ -1674,6 +1675,7 @@ fn infer_expr_type(
             ScalarValue::Utf8(_) => (DataType::Utf8, false),
             ScalarValue::Boolean(_) => (DataType::Bool, false),
             ScalarValue::Null => (DataType::Null, true),
+            ScalarValue::Interval { .. } => (DataType::Interval, false),
         },
         // Boolean-valued expressions (SQL three-valued logic → nullable).
         QueryExpr::Compare { .. }
@@ -1683,13 +1685,39 @@ fn infer_expr_type(
         | QueryExpr::IsNull(_)
         | QueryExpr::IsNotNull(_)
         | QueryExpr::InList { .. } => (DataType::Bool, true),
-        QueryExpr::Arithmetic { left, right, .. } => {
+        QueryExpr::Arithmetic { op, left, right } => {
             let (lt, ln) = infer_expr_type(left, schema)?;
             let (rt, rn) = infer_expr_type(right, schema)?;
-            let dtype = if matches!(lt, DataType::Int64) && matches!(rt, DataType::Int64) {
-                DataType::Int64
-            } else {
-                DataType::Float64
+            // Temporal subtraction yields a fixed duration with a unit, not a
+            // calendar interval or a floating-point number. Until the IR can
+            // preserve that unit, fail instead of publishing a numeric schema.
+            if matches!(op, ArithmeticOpKind::Sub)
+                && matches!(lt, DataType::Date | DataType::Timestamp)
+                && matches!(rt, DataType::Date | DataType::Timestamp)
+            {
+                return Err(QueryExprError::InvalidScalarSignature(
+                    "temporal subtraction produces an unsupported duration type".into(),
+                ));
+            }
+
+            // Operand order is not checked: the orders that are not valid SQL
+            // (`Interval - Timestamp`) are rejected by the planner upstream, so
+            // a pair rule stays as small as the numeric one it sits beside.
+            let dtype = match (&lt, &rt) {
+                // SQL unary minus lowers to -1 * expression, including intervals.
+                (DataType::Int64, DataType::Interval) | (DataType::Interval, DataType::Int64)
+                    if matches!(op, ArithmeticOpKind::Mul) =>
+                {
+                    DataType::Interval
+                }
+                (DataType::Timestamp, DataType::Interval)
+                | (DataType::Interval, DataType::Timestamp) => DataType::Timestamp,
+                (DataType::Date, DataType::Interval) | (DataType::Interval, DataType::Date) => {
+                    DataType::Date
+                }
+                (DataType::Interval, DataType::Interval) => DataType::Interval,
+                (DataType::Int64, DataType::Int64) => DataType::Int64,
+                _ => DataType::Float64,
             };
             (dtype, ln || rn)
         }
@@ -1760,6 +1788,55 @@ mod tests {
         Column::new(name, dtype, nullable)
     }
 
+    /// Shifting an instant by a duration stays an instant, and shifting a date
+    /// stays a date — neither falls through to the numeric default, which is
+    /// what `l_shipdate + INTERVAL '30' DAY` would otherwise be typed as.
+    #[test]
+    fn interval_arithmetic_keeps_the_temporal_type() {
+        let schema = Schema::new(vec![
+            col("ts", DataType::Timestamp, false),
+            col("d", DataType::Date, false),
+        ]);
+        let thirty_days = || {
+            Rc::new(QueryExpr::Literal(ScalarValue::Interval {
+                months: 0,
+                days: 30,
+                nanos: 0,
+            }))
+        };
+        let shift = |column, op| QueryExpr::Arithmetic {
+            op,
+            left: Rc::new(QueryExpr::Column(column)),
+            right: thirty_days(),
+        };
+
+        assert_eq!(
+            shift(0, ArithmeticOpKind::Add)
+                .scalar_type(&schema)
+                .unwrap()
+                .0,
+            DataType::Timestamp
+        );
+        assert_eq!(
+            shift(1, ArithmeticOpKind::Sub)
+                .scalar_type(&schema)
+                .unwrap()
+                .0,
+            DataType::Date
+        );
+        assert_eq!(
+            QueryExpr::Arithmetic {
+                op: ArithmeticOpKind::Add,
+                left: thirty_days(),
+                right: thirty_days(),
+            }
+            .scalar_type(&schema)
+            .unwrap()
+            .0,
+            DataType::Interval
+        );
+    }
+
     fn scan(
         columns: Vec<Column>,
         time_index: Option<ColumnId>,
diff --git a/crates/types/src/pre_asap/schema.rs b/crates/types/src/pre_asap/schema.rs
index d1718bfe..81b4ae1f 100644
--- a/crates/types/src/pre_asap/schema.rs
+++ b/crates/types/src/pre_asap/schema.rs
@@ -85,6 +85,16 @@ pub enum DataType {
     /// Wall-clock timestamp. PromQL leaves carry exactly one of these
     /// (the `time_index` column); SQL leaves may or may not.
     Timestamp,
+    /// The type of a [`ScalarValue::Interval`] — a calendar duration, not an
+    /// instant. Carried so `infer_expr_type` can give an interval literal a
+    /// type and type `Timestamp ± Interval`; no column is ever declared with it.
+    Interval,
+    /// Calendar date with no time-of-day — SQL `DATE`, Arrow `Date32`/`Date64`.
+    /// Distinct from `Timestamp` because a `CAST(… AS DATE)` is a real type
+    /// change DataFusion keeps in the plan: collapsing the two here would make
+    /// the bridge lossy in the one direction (`dtype_to_arrow`) that registers
+    /// catalog tables.
+    Date,
     /// Variable-length sequence. The existing column contract preserves the
     /// element field name, type, and nullability. Nested fields are unqualified.
     List { element: Box<Column> },