fix(ddx-ad): to_sql writes a NULL whose type has no kind as a bare NULL

crates/ddx-ad/src/to_sql/reader.rs

@@ -184,10 +184,13 @@ struct Reader<'p> {
 }
 
 impl Reader<'_> {
-    /// [`ty`], or `None` for the plan's `null` type: NULLs of no type.
+    /// [`ty`], or `None` for NULLs of no type: the plan's `null` type, as
+    /// DataFusion's producer writes it, or a type with no kind, as DuckDB's
+    /// writes it for a bare `NULL`.
     fn ty_or_none(&self, t: &SubstraitType) -> Result<Option<Type>> {
         match &t.kind {
             Some(Kind::UserDefined(u)) if Some(u.type_reference) == self.null_type => Ok(None),
+            None => Ok(None),
             _ => ty(t).map(Some),
         }
     }
@@ -1702,6 +1705,32 @@ mod tests {
     use super::*;
     use crate::to_sql::{to_sql, Dialect};
 
+    #[test]
+    fn a_null_whose_type_has_no_kind_is_written_bare() {
+        // DuckDB's producer writes a bare NULL so, and folds a
+        // `CAST(NULL AS BIGINT)` into one (#235).
+        use crate::functions::Extensions;
+        let null = substrait::proto::Expression {
+            rex_type: Some(RexType::Literal(SubstraitLiteral {
+                nullable: true,
+                type_variation_reference: 0,
+                literal_type: Some(LiteralType::Null(SubstraitType { kind: None })),
+            })),
+        };
+        let rel = crate::emit::project(crate::emit::read_step("t", vec!["a".into()]), vec![null]);
+        let functions = Functions::from_plan(&Plan::default()).unwrap();
+        let plan = crate::emit::plan(
+            rel,
+            vec!["a".into(), "n".into()],
+            &Extensions::new(&functions),
+        );
+        let written = to_sql(&plan, Dialect::DuckDb).unwrap();
+        assert!(
+            written.contains("NULL AS") && !written.contains("CAST(NULL"),
+            "{written}"
+        );
+    }
+
     #[test]
     fn every_canonical_name_is_one_ddx_writes() {
         assert!(syntax("frobnicate").is_none());

python/ddxdb/tests/test_duckdb.py

@@ -504,6 +504,33 @@ def test_a_window_over_an_aggregate_is_differentiated(ad):
 # Refusals.
 
 
+@pytest.mark.parametrize(
+    "u, loss, source",
+    [
+        ("SELECT CASE WHEN i = 1 THEN NULL ELSE i END AS k, val FROM w", "SUM(val * val * k)", "u"),
+        ("SELECT i AS k, CASE WHEN i = 1 THEN NULL ELSE val END AS val FROM w", "SUM(val * val)", "u"),
+        (
+            "SELECT CASE WHEN i = 1 THEN CAST(NULL AS BIGINT) ELSE i END AS k, val FROM w",
+            "SUM(u.val * b.val)",
+            "u JOIN b ON u.k = b.i",
+        ),
+    ],
+)
+def test_a_bare_null_is_differentiated(ad, con, u, loss, source):
+    # DuckDB's producer gives a bare NULL a type of no kind, and folds a
+    # CAST(NULL AS ...) into one; the step's SQL writes it as a bare NULL.
+    # The gradient is DataFusion's for the same SQL.
+    from datafusion import SessionContext
+    from ddxdb import ad as df
+
+    sql = f"WITH u AS ({u}), loss AS (SELECT {loss} AS l FROM {source}) SELECT i, grad(loss, val) AS g FROM w"
+    ctx = SessionContext()
+    for name in ("w", "b"):
+        ctx.from_arrow(con.execute(f"SELECT * FROM {name}").arrow(), name)
+    want = {r["i"]: r["g"] for r in df.sql(ctx, sql).to_arrow_table().to_pylist()}
+    assert dict(pairs(ad.sql(con, sql))) == pytest.approx(want)
+
+
 def test_a_repeated_column_is_refused_with_a_reason(ad, con):
     # DuckDB's producer drops a column that repeats another; ddx cannot tell
     # which is which, and says so rather than guess.