fix: propagate column statistics through CAST in join key expressions

When join keys contain CAST expressions (e.g. CAST(id AS Float64)),
the cardinality estimator could not extract column statistics because
it only handled plain Column references. This caused unknown stats,
leading to poor join ordering (e.g. putting a 1.4M-row fact table
on the hash join build side instead of a 5-row dimension table).

Extract the underlying column index through numeric CAST expressions,
since casting can only reduce (never increase) distinct count, making
the source column's stats a valid upper bound.

TPC-DS Q99: 10.4s → ~60ms.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Dandandan

datafusion/physical-plan/src/joins/utils.rs

@@ -460,6 +460,28 @@ pub(crate) fn estimate_join_statistics(
     })
 }
 
+/// Extract the column index from a join key expression for statistics lookup.
+/// Handles plain `Column` references and `CAST(column AS numeric_type)`
+/// expressions. Casting can only merge values (many-to-one), never split
+/// them, so the source column's distinct count is always a valid upper
+/// bound for the cast result's distinct count.
+fn column_index_for_stats(expr: &Arc<dyn PhysicalExpr>) -> Option<usize> {
+    if let Some(col) = expr.as_any().downcast_ref::<Column>() {
+        return Some(col.index());
+    }
+    if let Some(cast) = expr
+        .as_any()
+        .downcast_ref::<datafusion_physical_expr::expressions::CastExpr>()
+        && let Some(col) = cast.expr.as_any().downcast_ref::<Column>()
+    {
+        let target = cast.cast_type();
+        if target.is_numeric() {
+            return Some(col.index());
+        }
+    }
+    None
+}
+
 // Estimate the cardinality for the given join with input statistics.
 fn estimate_join_cardinality(
     join_type: &JoinType,
@@ -470,13 +492,10 @@ fn estimate_join_cardinality(
     let (left_col_stats, right_col_stats) = on
         .iter()
         .map(|(left, right)| {
-            match (
-                left.as_any().downcast_ref::<Column>(),
-                right.as_any().downcast_ref::<Column>(),
-            ) {
-                (Some(left), Some(right)) => (
-                    left_stats.column_statistics[left.index()].clone(),
-                    right_stats.column_statistics[right.index()].clone(),
+            match (column_index_for_stats(left), column_index_for_stats(right)) {
+                (Some(left_idx), Some(right_idx)) => (
+                    left_stats.column_statistics[left_idx].clone(),
+                    right_stats.column_statistics[right_idx].clone(),
                 ),
                 _ => (
                     ColumnStatistics::new_unknown(),