fix: extend column-existence validator to 20+ transforms

The validator added in 94c19f8 only covered xf.fill_forward,
xf.fill_backward, xf.fill_constant, and xf.cast. Every other
transform that names input columns by string (xf.distinct,
xf.sort, xf.project, xf.drop, xf.keep, xf.rename, xf.aggregate,
xf.pivot, xf.unpivot, xf.row_hash, xf.url.parse, xf.ip.parse,
xf.cdc.scd1, xf.cdc.scd2, xf.cdc.compare, the whole window family,
join leftKey) slipped through and only got caught by DuckDB at
runtime as "Binder Error: Referenced column not found".

User-reported: an xf.distinct configured with a column name that
didn't exist in the upstream CSV (typo / stale config) ran for
85 ms before failing inside DuckDB with the cryptic runtime error
instead of failing at planner time with a clear "column 'X' not
found in upstream" message.

Validator only fires when the upstream's column set is actually
known (declared by autodetect on the source, propagated through
the existing derive_output_columns). Joins skip right-side keys
because lookup-input columns aren't propagated through the planner
today; only main-side leftKey is checked. False negatives are
preferable to false positives here.

Author: SouravRoy-ETL

crates/duckdb-engine/src/plan.rs

@@ -1561,12 +1561,29 @@ fn validate_column_refs(
             c, case_hint
         ))
     };
-    match component_id {
-        "xf.fill_forward" | "xf.fill_backward" | "xf.fill_constant" => {
-            if let Some(c) = p.get("column").and_then(JsonValue::as_str) {
+    // Helper for components whose props expose a single "column" key.
+    let check_single_col = |p: &JsonValue| -> Result<(), String> {
+        if let Some(c) = p.get("column").and_then(JsonValue::as_str) {
+            let c = c.trim();
+            if !c.is_empty() {
                 check(c)?;
             }
         }
+        Ok(())
+    };
+    let check_list = |key: &str| -> Result<(), String> {
+        for c in columns_list(p, key) {
+            let c = c.trim();
+            if !c.is_empty() {
+                check(c)?;
+            }
+        }
+        Ok(())
+    };
+    match component_id {
+        "xf.fill_forward" | "xf.fill_backward" | "xf.fill_constant" => {
+            check_single_col(p)?;
+        }
         "xf.cast" => {
             // Multi-row form
             if let Some(arr) = p.get("casts").or_else(|| p.get("columns")).and_then(JsonValue::as_array) {
@@ -1579,11 +1596,119 @@ fn validate_column_refs(
                     }
                 }
             }
-            // Single-row form
-            if let Some(c) = p.get("column").and_then(JsonValue::as_str) {
-                let c = c.trim();
-                if !c.is_empty() {
-                    check(c)?;
+            check_single_col(p)?;
+        }
+        "xf.distinct" | "xf.drop" | "xf.keep" | "xf.unpivot" | "xf.row_hash" => {
+            check_list("columns")?;
+        }
+        "xf.project" => {
+            check_list("columns")?;
+            check_list("keep")?;
+        }
+        "xf.sort" => {
+            // orderBy is either an array of column-name strings or
+            // an array of {column, direction} objects. Validate both.
+            if let Some(arr) = p.get("orderBy").and_then(JsonValue::as_array) {
+                for entry in arr {
+                    let c = entry
+                        .as_str()
+                        .map(|s| s.to_string())
+                        .or_else(|| {
+                            entry
+                                .get("column")
+                                .and_then(JsonValue::as_str)
+                                .map(|s| s.to_string())
+                        });
+                    if let Some(c) = c {
+                        let c = c.trim();
+                        if !c.is_empty() {
+                            check(c)?;
+                        }
+                    }
+                }
+            }
+        }
+        "xf.rename" => {
+            if let Some(arr) = p.get("renames").and_then(JsonValue::as_array) {
+                for entry in arr {
+                    if let Some(c) = entry.get("from").and_then(JsonValue::as_str) {
+                        let c = c.trim();
+                        if !c.is_empty() {
+                            check(c)?;
+                        }
+                    }
+                }
+            }
+        }
+        "xf.aggregate" => {
+            check_list("groupBy")?;
+            // aggregateColumns: [{column, fn}, ...] - check the column field.
+            if let Some(arr) = p.get("aggregateColumns").and_then(JsonValue::as_array) {
+                for entry in arr {
+                    if let Some(c) = entry.get("column").and_then(JsonValue::as_str) {
+                        let c = c.trim();
+                        if !c.is_empty() {
+                            check(c)?;
+                        }
+                    }
+                }
+            }
+        }
+        "xf.pivot" => {
+            check_single_col(p)?;
+            for key in ["pivotColumn", "valueColumn", "valuesColumn"] {
+                if let Some(c) = p.get(key).and_then(JsonValue::as_str) {
+                    let c = c.trim();
+                    if !c.is_empty() {
+                        check(c)?;
+                    }
+                }
+            }
+            check_list("groupBy")?;
+        }
+        "xf.url.parse" | "xf.ip.parse" => {
+            check_single_col(p)?;
+        }
+        "xf.cdc.scd1" | "xf.cdc.scd2" | "xf.cdc.compare" => {
+            check_list("naturalKey")?;
+            check_list("compareColumns")?;
+        }
+        // Window family: partitionBy + orderBy are upstream columns.
+        // `column` is the column the function operates on (lead/lag/
+        // first/last) - present on a subset.
+        "xf.window"
+        | "xf.rownum"
+        | "xf.rank"
+        | "xf.denserank"
+        | "xf.lead"
+        | "xf.lag"
+        | "xf.first"
+        | "xf.last"
+        | "xf.ntile"
+        | "xf.rank.filter"
+        | "xf.cumulative"
+        | "xf.aggwin" => {
+            check_list("partitionBy")?;
+            check_list("orderBy")?;
+            check_single_col(p)?;
+        }
+        // Join keys on the left side. Right-side keys reference the
+        // lookup input, whose columns we don't currently propagate
+        // through the planner; skip those rather than emit a false
+        // positive.
+        "xf.join"
+        | "xf.join.left"
+        | "xf.join.right"
+        | "xf.join.full"
+        | "xf.join.cross"
+        | "xf.semi"
+        | "xf.anti" => {
+            if let Some(s) = p.get("leftKey").and_then(JsonValue::as_str) {
+                for k in s.split(',') {
+                    let k = k.trim();
+                    if !k.is_empty() {
+                        check(k)?;
+                    }
                 }
             }
         }