fix: improve inner join cardinality estimation without distinct stats

When distinct count statistics are absent (the common case), estimate
inner join cardinality as max(left_rows, right_rows) instead of using
the formula (L * R) / max(L, R) = min(L, R).

The old estimate severely underestimates FK joins: warehouse(5) ⋈
catalog_sales(1.4M) was estimated as 5 rows, causing the optimizer
to put the 1.4M-row table on the hash join build side.

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,17 +635,27 @@ 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 selectivity
+    // denominator falls back to max(num_rows_left, num_rows_right), which
+    // gives cardinality = min(L, R). This severely underestimates FK joins
+    // (e.g. warehouse(5) ⋈ catalog_sales(1.4M) → 5 instead of 1.4M).
+    // In this case, use max(L, R) directly as a better heuristic.
+    if !has_any_distinct {
+        return Some(Precision::Inexact(*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))
         }
         Precision::Inexact(value) if value > 0 => {
             Some(Precision::Inexact((left_num_rows * right_num_rows) / value))
         }
-        // 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
-        // overestimation using just the cartesian product).
+        // Selectivity is zero (one side has no non-null values), so the join
+        // produces no rows.
+        Precision::Exact(0) => Some(Precision::Exact(0)),
+        Precision::Inexact(0) => Some(Precision::Inexact(0)),
         _ => None,
     }
 }
@@ -2159,22 +2173,24 @@ mod tests {
                 Some(Inexact(10)),
             ),
             // range(left) > range(right)
+            // Without distinct stats, use max(L, R) = max(10, 10) = 10
             (
                 (10, Inexact(6), Inexact(10), Absent, Absent),
                 (10, Inexact(8), Inexact(10), Absent, Absent),
-                Some(Inexact(20)),
+                Some(Inexact(10)),
             ),
             // range(right) > range(left)
             (
                 (10, Inexact(8), Inexact(10), Absent, Absent),
                 (10, Inexact(6), Inexact(10), Absent, Absent),
-                Some(Inexact(20)),
+                Some(Inexact(10)),
             ),
             // range(left) > len(left), range(right) > len(right)
+            // Without distinct stats, min(10, 20) = 10, so (10*20)/10 = 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 +2217,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,10 +2239,11 @@ mod tests {
                 Some(Inexact(10)),
             ),
             // range(left) = 1, range(right) = 1
+            // Without distinct stats, use max(L, R) = 10
             (
                 (10, Inexact(1), Inexact(1), Absent, Absent),
                 (10, Inexact(1), Inexact(1), Absent, Absent),
-                Some(Inexact(100)),
+                Some(Inexact(10)),
             ),
             //
             // Edge cases
@@ -2275,17 +2293,18 @@ mod tests {
                 (10, Inexact(0), Inexact(10), Absent, Absent),
                 Some(Inexact(0)),
             ),
-            // distinct(left) = 0, distinct(right) = 0
+            // distinct(left) = 0, distinct(right) = 0: no matching keys possible
             (
                 (10, Inexact(1), Inexact(10), Inexact(0), Absent),
                 (10, Inexact(1), Inexact(10), Inexact(0), Absent),
-                None,
+                Some(Inexact(0)),
             ),
             // Inexact row count < exact null count with absent distinct count
+            // Without distinct stats, use max(L, R) = 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)),
             ),
         ];