fix: improve inner join cardinality estimation without distinct stats

When distinct count statistics are absent, the formula (L*R)/max_distinct
gives min(L,R), which severely underestimates FK joins (e.g.
warehouse(5) ⋈ catalog_sales(1.4M) → 5 instead of ~1.4M).

Add a floor of max(L,R) when no actual distinct count stats are
available. This prevents the optimizer from putting large fact tables
on the hash join build side.

When distinct stats ARE available, the formula is unchanged.

TPC-DS Q99: 10.4s → 59ms (157x faster).

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

Author: Dandandan

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

@@ -611,10 +611,14 @@ fn estimate_inner_join_cardinality(
     // The algorithm here is partly based on the non-histogram selectivity estimation
     // from Spark's Catalyst optimizer.
     let mut join_selectivity = Precision::Absent;
+    let mut has_any_distinct = false;
     for (left_stat, right_stat) in left_column_statistics
         .iter()
         .zip(right_column_statistics.iter())
     {
+        has_any_distinct |= left_stat.distinct_count.get_value().is_some()
+            || right_stat.distinct_count.get_value().is_some();
+
         let left_max_distinct = max_distinct_count(&left_num_rows, left_stat);
         let right_max_distinct = max_distinct_count(&right_num_rows, right_stat);
         let max_distinct = left_max_distinct.max(&right_max_distinct);
@@ -631,13 +635,28 @@ fn estimate_inner_join_cardinality(
     // of the two inputs and normalizing it by the selectivity factor.
     let left_num_rows = left_stats.num_rows.get_value()?;
     let right_num_rows = right_stats.num_rows.get_value()?;
+
+    // When no actual distinct count stats are available, the formula
+    // (L*R)/max_distinct can severely underestimate FK joins (e.g.
+    // warehouse(5) ⋈ catalog_sales(1.4M) → 5 instead of 1.4M) because
+    // max_distinct falls back to max(num_rows), giving min(L, R).
+    // Use max(L, R) as a floor to prevent this.
+    let min_cardinality = if has_any_distinct {
+        0
+    } else {
+        *left_num_rows.max(right_num_rows)
+    };
+
     match join_selectivity {
         Precision::Exact(value) if value > 0 => {
-            Some(Precision::Exact((left_num_rows * right_num_rows) / value))
+            let estimate = (left_num_rows * right_num_rows) / value;
+            Some(Precision::Inexact(estimate.max(min_cardinality)))
         }
         Precision::Inexact(value) if value > 0 => {
-            Some(Precision::Inexact((left_num_rows * right_num_rows) / value))
+            let estimate = (left_num_rows * right_num_rows) / value;
+            Some(Precision::Inexact(estimate.max(min_cardinality)))
         }
+        _ if min_cardinality > 0 => Some(Precision::Inexact(min_cardinality)),
         // Since we don't have any information about the selectivity (which is derived
         // from the number of distinct rows information) we can give up here for now.
         // And let other passes handle this (otherwise we would need to produce an
@@ -2159,6 +2178,7 @@ mod tests {
                 Some(Inexact(10)),
             ),
             // range(left) > range(right)
+            // range(left)=5, range(right)=3: formula (10*10)/5 = 20, cap max(10,10) = 10 → 20
             (
                 (10, Inexact(6), Inexact(10), Absent, Absent),
                 (10, Inexact(8), Inexact(10), Absent, Absent),
@@ -2171,10 +2191,11 @@ mod tests {
                 Some(Inexact(20)),
             ),
             // range(left) > len(left), range(right) > len(right)
+            // formula: (10*20)/20 = 10, capped at max(10,20) = 20
             (
                 (10, Inexact(1), Inexact(15), Absent, Absent),
                 (20, Inexact(1), Inexact(40), Absent, Absent),
-                Some(Inexact(10)),
+                Some(Inexact(20)),
             ),
             // Distinct count matches the range
             (
@@ -2201,6 +2222,7 @@ mod tests {
                 Some(Inexact(20)),
             ),
             // min(left) < 0 (range(left) > range(right))
+            // Without distinct stats, use max(L, R) = max(10, 10) = 10
             (
                 (10, Inexact(-5), Inexact(5), Absent, Absent),
                 (10, Inexact(1), Inexact(5), Absent, Absent),
@@ -2222,6 +2244,7 @@ mod tests {
                 Some(Inexact(10)),
             ),
             // range(left) = 1, range(right) = 1
+            // range=1 on both sides: formula (10*10)/1 = 100
             (
                 (10, Inexact(1), Inexact(1), Absent, Absent),
                 (10, Inexact(1), Inexact(1), Absent, Absent),
@@ -2282,10 +2305,12 @@ mod tests {
                 None,
             ),
             // Inexact row count < exact null count with absent distinct count
+            // Inexact row count < exact null count, no distinct stats:
+            // max(0, 10) = 10
             (
                 (0, Inexact(1), Inexact(10), Absent, Exact(5)),
                 (10, Inexact(1), Inexact(10), Absent, Absent),
-                Some(Inexact(0)),
+                Some(Inexact(10)),
             ),
         ];