You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/22 05:03:42 UTC
(datafusion) branch main updated: fix: account for row duplication and padding in join null counts (#25861)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new e12901dadf fix: account for row duplication and padding in join null counts (#25861)
e12901dadf is described below
commit e12901dadfd462231a093627449d3a7b51d45962
Author: Kumar Ujjawal <uj...@gmail.com>
AuthorDate: Thu Oct 22 03:09:21 2026 +0000
fix: account for row duplication and padding in join null counts (#25861)
## Which issue does this PR close?
Related to #25620. Split from #25648 following [the review
request](https://github.com/apache/datafusion/pull/25648#discussion_r4122769097).
This PR can be reviewed and merged independently.
## Rationale for this change
Joins repeat or drop input rows, and outer joins add NULL padding.
Copying each input column's null count unchanged can therefore produce
inconsistent output statistics. For example, a left join can repeat a
NULL in a non-key left column many times, while an unmatched left row
introduces NULLs in every right column.
## What changes are included in this PR?
- Estimate inner and outer join null counts from the rows carrying each
side's values and the rows added as NULL padding.
- Estimate unmatched input rows using the existing semi-join estimator.
Inner-join row counts count matching pairs, so they cannot alone
determine how many input rows are unmatched.
- Account for NULL keys under ordinary and null-safe equality, while
retaining unknown contributions and conservative precision.
- Rescale null counts when a statistics provider replaces the row
estimate. Cover projected joins and swapped inputs.
The semi/anti-join estimator itself is unchanged. The rules follow join
semantics and do not contain benchmark-specific cases.
## What is the testing strategy for this PR?
Added unit tests cover NULL padding, repeated matches, unknown counts,
null-safe keys, projection, input swapping, and provider rescaling.
Existing cardinality assertions remain in place.
Passed on this independent branch:
- `cargo test --profile ci -p datafusion-physical-plan --lib` (2,369
tests)
- `cargo clippy --profile ci -p datafusion-physical-plan --all-targets
--all-features -- -D warnings`
- `cargo fmt --all` and `git diff --check`
Benchmarks were not rerun for the split. This extracts the join
implementation already present in #25648; its earlier combined
measurements are not isolated measurements of this PR.
## Are there any user-facing changes?
Join row and null-count estimates can change, which can affect EXPLAIN
output and plan choices. No public API changes.
---
datafusion/physical-plan/src/joins/utils.rs | 674 +++++++++++++++++++--
.../physical-plan/src/operator_statistics/mod.rs | 159 ++++-
2 files changed, 766 insertions(+), 67 deletions(-)
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index ef62dcf2f7..5136fa9a81 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -444,8 +444,11 @@ struct PartialJoinStatistics {
/// column-level statistics (distinct counts, min/max values) of the join keys.
/// - **Column statistics**: Combines column statistics from both inputs. For join types
/// that preserve all columns (Inner, Left, Right, Full), statistics from both sides
-/// are concatenated. For semi/anti joins, the preserved side's statistics are
-/// normalized as subset estimates.
+/// are concatenated, except null counts: a join repeats and drops rows, so null
+/// counts are scaled to the output rows, NULL-padded rows of an outer join add
+/// NULLs, and join keys under `NullEqualsNothing` keep only the NULLs of unmatched
+/// rows. For semi/anti joins, the preserved side's statistics are normalized as
+/// subset estimates.
/// - **Byte size**: For semi/anti joins, sums normalized column byte-size estimates
/// when every output column has one. Other join types return `Precision::Absent`
/// because join output size is difficult to estimate without knowing the actual data.
@@ -462,9 +465,9 @@ struct PartialJoinStatistics {
/// - Does not account for selectivity of arbitrary join filter expressions
/// (e.g., `(t1.v1 + t2.v1) % 2 = 0`). Such filters, common in NestedLoopJoinExec,
/// are not factored into the cardinality estimation.
-/// - Column statistics for inner/outer joins are simply combined from inputs
-/// without adjusting for join selectivity (acknowledged in the code as
-/// needing "filter selectivity analysis").
+/// - Apart from null counts, column statistics for inner/outer joins are simply
+/// combined from inputs without adjusting for join selectivity (acknowledged in
+/// the code as needing "filter selectivity analysis").
pub(crate) fn estimate_join_statistics(
left_stats: Statistics,
right_stats: Statistics,
@@ -523,45 +526,107 @@ fn estimate_join_cardinality(
match join_type {
JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full => {
- let ij_cardinality = estimate_inner_join_cardinality(
- Statistics {
- num_rows: left_stats.num_rows,
- total_byte_size: Precision::Absent,
- column_statistics: left_key_stats,
- },
- Statistics {
- num_rows: right_stats.num_rows,
- total_byte_size: Precision::Absent,
- column_statistics: right_key_stats,
- },
- )?;
-
- // The cardinality for inner join can also be used to estimate
- // the cardinality of left/right/full outer joins as long as it
- // it is greater than the minimum cardinality constraints of these
- // joins (so that we don't underestimate the cardinality).
- let cardinality = match join_type {
- JoinType::Inner => ij_cardinality,
- JoinType::Left => ij_cardinality.max(&left_stats.num_rows),
- JoinType::Right => ij_cardinality.max(&right_stats.num_rows),
- JoinType::Full => ij_cardinality
- .max(&left_stats.num_rows)
- .add(&ij_cardinality.max(&right_stats.num_rows))
- .sub(&ij_cardinality),
- _ => unreachable!(),
+ let left_keys_stats = Statistics {
+ num_rows: left_stats.num_rows,
+ total_byte_size: Precision::Absent,
+ column_statistics: left_key_stats,
};
+ let right_keys_stats = Statistics {
+ num_rows: right_stats.num_rows,
+ total_byte_size: Precision::Absent,
+ column_statistics: right_key_stats,
+ };
+ // For a single null-safe key, every left NULL matches every right
+ // NULL. With more keys their correlation is unknown.
+ let null_matches =
+ if null_equality == NullEquality::NullEqualsNull && on.len() == 1 {
+ left_keys_stats.column_statistics[0]
+ .null_count
+ .multiply(&right_keys_stats.column_statistics[0].null_count)
+ .to_inexact()
+ } else {
+ Precision::Absent
+ };
+ let inner_rows =
+ *estimate_inner_join_cardinality(&left_keys_stats, &right_keys_stats)?
+ .get_value()?;
+ let inner_rows =
+ inner_rows.max(null_matches.get_value().copied().unwrap_or(0));
+ let left_rows = left_stats.num_rows.get_value().copied();
+ let right_rows = right_stats.num_rows.get_value().copied();
+ let left_preserved = matches!(join_type, JoinType::Left | JoinType::Full);
+ let right_preserved = matches!(join_type, JoinType::Right | JoinType::Full);
+ let unmatched_left = if left_preserved {
+ estimate_join_unmatched_rows(
+ &left_keys_stats,
+ &right_keys_stats,
+ inner_rows,
+ null_equality,
+ )?
+ } else {
+ 0
+ };
+ let unmatched_right = if right_preserved {
+ estimate_join_unmatched_rows(
+ &right_keys_stats,
+ &left_keys_stats,
+ inner_rows,
+ null_equality,
+ )?
+ } else {
+ 0
+ };
+ let left_side = JoinSideRows {
+ input: left_rows,
+ own: inner_rows.saturating_add(unmatched_left),
+ padded: right_preserved.then_some(unmatched_right),
+ preserved: left_preserved,
+ };
+ let right_side = JoinSideRows {
+ input: right_rows,
+ own: inner_rows.saturating_add(unmatched_right),
+ padded: left_preserved.then_some(unmatched_left),
+ preserved: right_preserved,
+ };
+ let mut left_column_statistics = left_stats.column_statistics;
+ let mut right_column_statistics = right_stats.column_statistics;
+ let (left_keys, right_keys): (Vec<_>, Vec<_>) = on_column_indices
+ .iter()
+ .flatten()
+ .map(|&(left, right)| {
+ (
+ (left, right_column_statistics[right].null_count),
+ (right, left_column_statistics[left].null_count),
+ )
+ })
+ .unzip();
+ estimate_join_null_counts(
+ &mut left_column_statistics,
+ &left_side,
+ &left_keys,
+ null_equality,
+ null_matches,
+ );
+ estimate_join_null_counts(
+ &mut right_column_statistics,
+ &right_side,
+ &right_keys,
+ null_equality,
+ null_matches,
+ );
Some(PartialJoinStatistics {
- num_rows: *cardinality.get_value()?,
+ num_rows: inner_rows
+ .saturating_add(unmatched_left)
+ .saturating_add(unmatched_right),
total_byte_size: Precision::Absent,
- // We don't do anything specific here, just combine the existing
- // statistics which might yield subpar results (although it is
- // true, esp regarding min/max). For a better estimation, we need
- // filter selectivity analysis first.
- column_statistics: left_stats
- .column_statistics
+ // Apart from the null counts estimated above, we just combine
+ // the existing statistics, which might yield subpar results
+ // (although it is true, esp regarding min/max). For a better
+ // estimation, we need filter selectivity analysis first.
+ column_statistics: left_column_statistics
.into_iter()
- .chain(right_stats.column_statistics)
+ .chain(right_column_statistics)
.collect(),
})
}
@@ -758,6 +823,122 @@ fn normalize_semi_anti_join_key_null_count(
}
}
+/// Estimates input rows with no match. Inner-join rows count pairs, so they
+/// only bound the number of matching input rows when the keys repeat.
+fn estimate_join_unmatched_rows(
+ input: &Statistics,
+ other: &Statistics,
+ inner_rows: usize,
+ null_equality: NullEquality,
+) -> Option<usize> {
+ let rows = *input.num_rows.get_value()?;
+ // The semi estimate uses non-NULL NDV overlap, so it can miss NULL
+ // matches. With one null-safe key, every input NULL has a match if the
+ // other side has any NULLs. This bounds matched input rows, not pairs.
+ let matched_nulls = match (
+ null_equality,
+ input.column_statistics.as_slice(),
+ other.column_statistics.as_slice(),
+ ) {
+ (NullEquality::NullEqualsNull, [input], [other])
+ if other.null_count.get_value().is_some_and(|&count| count > 0) =>
+ {
+ input.null_count.get_value().copied().unwrap_or(0)
+ }
+ _ => 0,
+ };
+ let matched = estimate_semi_join_cardinality(
+ &input.num_rows,
+ &other.num_rows,
+ &input.column_statistics,
+ &other.column_statistics,
+ null_equality,
+ )
+ .unwrap_or(inner_rows)
+ .max(matched_nulls)
+ .min(inner_rows);
+ // A NULL in any key prevents a match under ordinary equality. The same
+ // holds for null-safe equality when the other key has no NULLs.
+ let unmatched_nulls = input
+ .column_statistics
+ .iter()
+ .zip(&other.column_statistics)
+ .filter(|(_, other)| {
+ null_equality == NullEquality::NullEqualsNothing
+ || other.null_count == Precision::Exact(0)
+ })
+ .filter_map(|(input, _)| input.null_count.get_value().copied())
+ .max()
+ .unwrap_or(0)
+ .min(rows);
+ Some(rows.saturating_sub(matched).max(unmatched_nulls))
+}
+
+/// Row counts for one side of a join that keeps both sides' columns.
+struct JoinSideRows {
+ /// Rows of this side's input, when known.
+ input: Option<usize>,
+ /// Output rows that carry this side's values.
+ own: usize,
+ /// Estimated output rows where this side's columns are NULL padding, or
+ /// `None` when the join never pads this side.
+ padded: Option<usize>,
+ /// Whether the join keeps this side's unmatched rows.
+ preserved: bool,
+}
+
+/// Estimates the null counts of one side's columns in the output of a join
+/// that keeps both sides' columns. A join repeats and drops rows, so a null
+/// count scales with the rows that carry the column, and padding adds NULLs.
+/// NULL keys never match under `NullEqualsNothing`, so only the unmatched rows
+/// of a preserved side keep them, each once. The same holds for null-safe keys
+/// whose opposite key has exactly zero NULLs. Each `keys` entry contains the
+/// column index and the opposite key's input null count.
+fn estimate_join_null_counts(
+ column_statistics: &mut [ColumnStatistics],
+ side: &JoinSideRows,
+ keys: &[(usize, Precision<usize>)],
+ null_equality: NullEquality,
+ null_matches: Precision<usize>,
+) {
+ for (idx, column_stats) in column_statistics.iter_mut().enumerate() {
+ let null_count = column_stats.null_count;
+ let nulls_cannot_match = keys.iter().any(|&(key, opposite_nulls)| {
+ key == idx
+ && (null_equality == NullEquality::NullEqualsNothing
+ || opposite_nulls == Precision::Exact(0))
+ });
+ let own_nulls = if nulls_cannot_match {
+ if side.preserved && null_count != Precision::Exact(0) {
+ // A single output partition can emit all unmatched rows of
+ // a broadcast side, so the count is exact only overall.
+ null_count.to_inexact()
+ } else {
+ Precision::Exact(0)
+ }
+ } else if null_count == Precision::Exact(0) {
+ // A join cannot add values, so a column without nulls keeps none.
+ null_count
+ } else if keys.iter().any(|&(key, _)| key == idx) {
+ if side.preserved {
+ null_matches.max(&null_count).to_inexact()
+ } else {
+ null_matches
+ }
+ } else {
+ side.input.map_or(Precision::Absent, |rows| {
+ scale_subset_count(null_count, rows, side.own)
+ })
+ };
+ // Padding is an estimate. An unknown contribution from matched rows
+ // stays unknown rather than being treated as zero.
+ column_stats.null_count = match side.padded {
+ Some(padded) => own_nulls.add(&Precision::Inexact(padded)),
+ None => own_nulls,
+ };
+ }
+}
+
// Scale a column-level count to an estimated row subset. Rounding up keeps a
// small non-zero count from disappearing solely because the subset is small.
fn scale_subset_count(
@@ -798,11 +979,11 @@ fn total_byte_size_from_column_statistics(
/// a very conservative implementation that can quickly give up if there is not
/// enough input statistics.
fn estimate_inner_join_cardinality(
- left_stats: Statistics,
- right_stats: Statistics,
+ left_stats: &Statistics,
+ right_stats: &Statistics,
) -> Option<Precision<usize>> {
// Immediately return if inputs considered as non-overlapping
- if let Some(estimation) = estimate_disjoint_inputs(&left_stats, &right_stats) {
+ if let Some(estimation) = estimate_disjoint_inputs(left_stats, right_stats) {
return Some(estimation);
}
@@ -817,10 +998,11 @@ fn estimate_inner_join_cardinality(
..
} = right_stats;
- if left_num_rows == Precision::Exact(0) || right_num_rows == Precision::Exact(0) {
+ if *left_num_rows == Precision::Exact(0) || *right_num_rows == Precision::Exact(0) {
return Some(Precision::Exact(0));
}
- if left_num_rows == Precision::Inexact(0) || right_num_rows == Precision::Inexact(0) {
+ if *left_num_rows == Precision::Inexact(0) || *right_num_rows == Precision::Inexact(0)
+ {
return Some(Precision::Inexact(0));
}
@@ -831,8 +1013,8 @@ fn estimate_inner_join_cardinality(
.iter()
.zip(right_column_statistics.iter())
{
- 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 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);
if max_distinct.get_value().is_some() {
// Seems like there are a few implementations of this algorithm that implement
@@ -3433,12 +3615,12 @@ mod tests {
assert_eq!(
estimate_inner_join_cardinality(
- Statistics {
+ &Statistics {
num_rows: Inexact(left_num_rows),
total_byte_size: Absent,
column_statistics: left_col_stats.clone(),
},
- Statistics {
+ &Statistics {
num_rows: Inexact(right_num_rows),
total_byte_size: Absent,
column_statistics: right_col_stats.clone(),
@@ -3465,9 +3647,18 @@ mod tests {
partial_join_stats.clone().map(|s| Inexact(s.num_rows)),
expected_cardinality.clone()
);
+ // NULL keys never match, so the key columns hold no NULLs.
+ let expected_column_statistics: Vec<_> = [left_col_stats, right_col_stats]
+ .concat()
+ .into_iter()
+ .map(|mut column_stats| {
+ column_stats.null_count = Exact(0);
+ column_stats
+ })
+ .collect();
assert_eq!(
partial_join_stats.map(|s| s.column_statistics),
- expected_cardinality.map(|_| [left_col_stats, right_col_stats].concat())
+ expected_cardinality.map(|_| expected_column_statistics)
);
}
Ok(())
@@ -3489,15 +3680,15 @@ mod tests {
// produces a representable cardinality.
assert_eq!(
estimate_inner_join_cardinality(
- statistics(Inexact(large_row_count), Inexact(1)),
- statistics(Inexact(3), Inexact(3)),
+ &statistics(Inexact(large_row_count), Inexact(1)),
+ &statistics(Inexact(3), Inexact(3)),
),
Some(Inexact(large_row_count))
);
assert_eq!(
estimate_inner_join_cardinality(
- statistics(Exact(large_row_count), Exact(1)),
- statistics(Exact(3), Exact(3)),
+ &statistics(Exact(large_row_count), Exact(1)),
+ &statistics(Exact(3), Exact(3)),
),
Some(Exact(large_row_count))
);
@@ -3506,8 +3697,8 @@ mod tests {
// estimate and mark it as inexact.
assert_eq!(
estimate_inner_join_cardinality(
- statistics(Exact(usize::MAX), Exact(1)),
- statistics(Exact(2), Exact(1)),
+ &statistics(Exact(usize::MAX), Exact(1)),
+ &statistics(Exact(2), Exact(1)),
),
Some(Inexact(usize::MAX))
);
@@ -3529,12 +3720,12 @@ mod tests {
// count is 200, so we are going to pick it.
assert_eq!(
estimate_inner_join_cardinality(
- Statistics {
+ &Statistics {
num_rows: Inexact(400),
total_byte_size: Absent,
column_statistics: left_col_stats,
},
- Statistics {
+ &Statistics {
num_rows: Inexact(400),
total_byte_size: Absent,
column_statistics: right_col_stats,
@@ -3563,12 +3754,12 @@ mod tests {
assert_eq!(
estimate_inner_join_cardinality(
- Statistics {
+ &Statistics {
num_rows: Inexact(100),
total_byte_size: Absent,
column_statistics: left_col_stats,
},
- Statistics {
+ &Statistics {
num_rows: Inexact(100),
total_byte_size: Absent,
column_statistics: right_col_stats,
@@ -3594,12 +3785,14 @@ mod tests {
// Join on a=c, b=d (ignore x/y)
// Right column d has NDV=2500 but only 2000 rows, so NDV is capped
// to 2000. join_selectivity = max(500, 2000) = 2000.
- // Inner cardinality = (1000 * 2000) / 2000 = 1000
+ // Inner cardinality = (1000 * 2000) / 2000 = 1000. The semi-join
+ // estimates match 500 rows from each input, leaving 500 left rows
+ // and 1500 right rows unmatched.
let cases = vec![
(JoinType::Inner, 1000),
- (JoinType::Left, 1000),
- (JoinType::Right, 2000),
- (JoinType::Full, 2000),
+ (JoinType::Left, 1500),
+ (JoinType::Right, 2500),
+ (JoinType::Full, 3000),
];
let left_col_stats = vec![
@@ -3635,8 +3828,9 @@ mod tests {
)
.unwrap();
assert_eq!(partial_join_stats.num_rows, expected_num_rows);
+ // `test_join_null_counts` covers the null counts.
assert_eq!(
- partial_join_stats.column_statistics,
+ without_null_counts(partial_join_stats.column_statistics),
[left_col_stats.clone(), right_col_stats.clone()].concat()
);
}
@@ -3644,6 +3838,354 @@ mod tests {
Ok(())
}
+ fn without_null_counts(
+ mut column_statistics: Vec<ColumnStatistics>,
+ ) -> Vec<ColumnStatistics> {
+ for column_stats in &mut column_statistics {
+ column_stats.null_count = Absent;
+ }
+ column_statistics
+ }
+
+ #[test]
+ fn test_join_null_counts_duplicate_matches() {
+ // Left: fifty 1s and fifty NULLs. Right: ten 1s. The fifty
+ // NULL-key rows stay unmatched regardless of the number of pairs.
+ let column = |nulls| ColumnStatistics {
+ null_count: Exact(nulls),
+ distinct_count: Exact(1),
+ ..ColumnStatistics::new_unknown()
+ };
+ let on = vec![(
+ Arc::new(Column::new("k", 0)) as _,
+ Arc::new(Column::new("k", 0)) as _,
+ )];
+ for join_type in [JoinType::Left, JoinType::Right, JoinType::Full] {
+ let (left, right) = if join_type == JoinType::Right {
+ (
+ create_stats(Some(10), vec![column(0)], false),
+ create_stats(Some(100), vec![column(50)], false),
+ )
+ } else {
+ (
+ create_stats(Some(100), vec![column(50)], false),
+ create_stats(Some(10), vec![column(0)], false),
+ )
+ };
+ let stats = estimate_join_cardinality(
+ &join_type,
+ left,
+ right,
+ &on,
+ NullEquality::NullEqualsNothing,
+ )
+ .unwrap();
+ assert_eq!(
+ stats.column_statistics[0].null_count,
+ Inexact(50),
+ "{join_type}"
+ );
+ assert_eq!(
+ stats.column_statistics[1].null_count,
+ Inexact(50),
+ "{join_type}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_join_null_counts_null_safe_keys() {
+ // Every non-null key is 1. Twenty left NULLs match the one right
+ // NULL, giving twenty NULLs in each output key for every join type.
+ let column = |nulls| ColumnStatistics {
+ null_count: Exact(nulls),
+ distinct_count: Exact(1),
+ ..ColumnStatistics::new_unknown()
+ };
+ let on = vec![(
+ Arc::new(Column::new("k", 0)) as _,
+ Arc::new(Column::new("k", 0)) as _,
+ )];
+ for join_type in [
+ JoinType::Inner,
+ JoinType::Left,
+ JoinType::Right,
+ JoinType::Full,
+ ] {
+ let stats = estimate_join_cardinality(
+ &join_type,
+ create_stats(Some(100), vec![column(20)], false),
+ create_stats(Some(10), vec![column(1)], false),
+ &on,
+ NullEquality::NullEqualsNull,
+ )
+ .unwrap();
+ assert_eq!(
+ stats.column_statistics[0].null_count,
+ Inexact(20),
+ "{join_type}"
+ );
+ assert_eq!(
+ stats.column_statistics[1].null_count,
+ Inexact(20),
+ "{join_type}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_join_null_counts_null_safe_multikey() {
+ // The key order differs from the column order on both sides. A NULL
+ // can only match if the corresponding key on the other side has NULLs.
+ let column = |null_count| ColumnStatistics {
+ null_count,
+ distinct_count: Exact(10),
+ ..ColumnStatistics::new_unknown()
+ };
+ let on = vec![
+ (
+ Arc::new(Column::new("a", 1)) as _,
+ Arc::new(Column::new("a", 1)) as _,
+ ),
+ (
+ Arc::new(Column::new("b", 2)) as _,
+ Arc::new(Column::new("b", 0)) as _,
+ ),
+ ];
+ for (join_type, opposite_nulls, expected) in [
+ (
+ JoinType::Inner,
+ Exact(0),
+ [Exact(0), Exact(0), Exact(0), Exact(0)],
+ ),
+ (
+ JoinType::Inner,
+ Inexact(0),
+ [Absent, Exact(0), Exact(0), Absent],
+ ),
+ (
+ JoinType::Inner,
+ Absent,
+ [Absent, Exact(0), Exact(0), Absent],
+ ),
+ (
+ JoinType::Left,
+ Exact(0),
+ [Inexact(20), Exact(0), Inexact(20), Inexact(20)],
+ ),
+ (
+ JoinType::Right,
+ Exact(0),
+ [Inexact(20), Inexact(20), Inexact(20), Exact(0)],
+ ),
+ (
+ JoinType::Full,
+ Exact(0),
+ [Inexact(40), Inexact(20), Inexact(40), Inexact(20)],
+ ),
+ ] {
+ let stats = estimate_join_cardinality(
+ &join_type,
+ create_stats(
+ Some(100),
+ vec![column(Exact(5)), column(Exact(20)), column(Exact(0))],
+ true,
+ ),
+ create_stats(
+ Some(100),
+ vec![column(Exact(20)), column(opposite_nulls)],
+ true,
+ ),
+ &on,
+ NullEquality::NullEqualsNull,
+ )
+ .unwrap();
+ let null_counts: Vec<_> = stats.column_statistics[1..]
+ .iter()
+ .map(|column| column.null_count)
+ .collect();
+ assert_eq!(null_counts, expected, "{join_type} {opposite_nulls:?}");
+ if join_type == JoinType::Inner {
+ // A non-key column is still scaled, not made null-free.
+ assert_eq!(stats.column_statistics[0].null_count, Inexact(50));
+ }
+ }
+ }
+
+ #[test]
+ fn test_join_null_counts_null_safe_unmatched_rows() {
+ let on = vec![(
+ Arc::new(Column::new("k", 0)) as _,
+ Arc::new(Column::new("k", 0)) as _,
+ )];
+ // Ninety NULL keys match every row on the other side. Only the ten
+ // distinct non-NULL keys are unmatched, even when matches duplicate.
+ for other_rows in [1, 3] {
+ for join_type in [JoinType::Left, JoinType::Right, JoinType::Full] {
+ let input = create_stats(
+ Some(100),
+ vec![create_column_stats(
+ Exact(1),
+ Exact(10),
+ Exact(10),
+ Exact(90),
+ )],
+ true,
+ );
+ let other = create_stats(
+ Some(other_rows),
+ vec![create_column_stats(
+ Absent,
+ Absent,
+ Exact(0),
+ Exact(other_rows),
+ )],
+ true,
+ );
+ let (left, right) = if join_type == JoinType::Right {
+ (other, input)
+ } else {
+ (input, other)
+ };
+ let stats = estimate_join_cardinality(
+ &join_type,
+ left,
+ right,
+ &on,
+ NullEquality::NullEqualsNull,
+ )
+ .unwrap();
+ let null_pairs = 90 * other_rows;
+ assert_eq!(stats.num_rows, null_pairs + 10, "{join_type}");
+ let expected = if join_type == JoinType::Right {
+ [Inexact(null_pairs + 10), Inexact(null_pairs)]
+ } else {
+ [Inexact(null_pairs), Inexact(null_pairs + 10)]
+ };
+ let null_counts: Vec<_> = stats
+ .column_statistics
+ .iter()
+ .map(|column| column.null_count)
+ .collect();
+ assert_eq!(null_counts, expected, "{join_type}");
+ }
+ }
+ }
+
+ #[test]
+ fn test_join_null_counts_unknown_padded_column() {
+ // All rows match. No padding does not prove that the right-hand
+ // non-key column has no NULLs: its input null count is unknown.
+ let key = ColumnStatistics {
+ null_count: Exact(0),
+ distinct_count: Exact(1),
+ ..ColumnStatistics::new_unknown()
+ };
+ let on = vec![(
+ Arc::new(Column::new("k", 0)) as _,
+ Arc::new(Column::new("k", 0)) as _,
+ )];
+ let stats = estimate_join_cardinality(
+ &JoinType::Left,
+ create_stats(Some(100), vec![key.clone()], false),
+ create_stats(Some(1), vec![key, ColumnStatistics::new_unknown()], false),
+ &on,
+ NullEquality::NullEqualsNothing,
+ )
+ .unwrap();
+ assert_eq!(stats.column_statistics[2].null_count, Absent);
+ }
+
+ #[test]
+ fn test_join_null_counts() {
+ // Each side has a key column `k` and another column `x`, both with
+ // nulls. The inner join estimates 100 pairs, while the semi joins
+ // estimate 98 matched left rows and 9 matched right rows. A
+ // padded side's counts are estimates even when no padding is expected.
+ let left_col_stats = vec![
+ create_column_stats(Inexact(0), Inexact(100), Inexact(100), Exact(20)),
+ create_column_stats(Inexact(0), Inexact(100), Inexact(100), Exact(50)),
+ ];
+ let right_col_stats = vec![
+ create_column_stats(Inexact(0), Inexact(100), Inexact(10), Exact(1)),
+ create_column_stats(Inexact(0), Inexact(100), Inexact(10), Exact(2)),
+ ];
+ // The null counts of left `k` and `x`, then of right `k` and `x`. A
+ // preserved key keeps its NULLs, but only as an estimate, because
+ // per-partition statistics cannot place the unmatched rows.
+ use NullEquality::{NullEqualsNothing, NullEqualsNull};
+ let cases = [
+ (
+ JoinType::Inner,
+ NullEqualsNothing,
+ [Exact(0), Inexact(5), Exact(0), Inexact(20)],
+ ),
+ (
+ JoinType::Inner,
+ NullEqualsNull,
+ [Inexact(20), Inexact(5), Inexact(20), Inexact(20)],
+ ),
+ (
+ JoinType::Left,
+ NullEqualsNothing,
+ [Inexact(20), Inexact(51), Inexact(902), Inexact(922)],
+ ),
+ (
+ JoinType::Left,
+ NullEqualsNull,
+ [Inexact(20), Inexact(50), Inexact(920), Inexact(920)],
+ ),
+ (
+ JoinType::Right,
+ NullEqualsNothing,
+ [Inexact(1), Inexact(6), Inexact(1), Inexact(21)],
+ ),
+ (
+ JoinType::Full,
+ NullEqualsNothing,
+ [Inexact(21), Inexact(52), Inexact(903), Inexact(923)],
+ ),
+ (
+ JoinType::Full,
+ NullEqualsNull,
+ [Inexact(20), Inexact(50), Inexact(920), Inexact(920)],
+ ),
+ ];
+ let join_on = vec![(
+ Arc::new(Column::new("k", 0)) as _,
+ Arc::new(Column::new("k", 0)) as _,
+ )];
+ for (join_type, null_equality, expected) in cases {
+ let partial_join_stats = estimate_join_cardinality(
+ &join_type,
+ create_stats(Some(1000), left_col_stats.clone(), false),
+ create_stats(Some(10), right_col_stats.clone(), false),
+ &join_on,
+ null_equality,
+ )
+ .unwrap();
+ let null_counts: Vec<_> = partial_join_stats
+ .column_statistics
+ .iter()
+ .map(|column_stats| column_stats.null_count)
+ .collect();
+ assert_eq!(null_counts, expected, "{join_type} {null_equality:?}");
+ }
+
+ // Padding does not determine how many matched values are NULL.
+ let mut unknown_right_col_stats = right_col_stats;
+ unknown_right_col_stats[1].null_count = Absent;
+ let partial_join_stats = estimate_join_cardinality(
+ &JoinType::Left,
+ create_stats(Some(1000), left_col_stats, false),
+ create_stats(Some(10), unknown_right_col_stats, false),
+ &join_on,
+ NullEqualsNothing,
+ )
+ .unwrap();
+ assert_eq!(partial_join_stats.column_statistics[3].null_count, Absent);
+ }
+
#[test]
fn test_join_cardinality_key_order() -> Result<()> {
// Reversing join key order should not change estimated cardinality
@@ -3701,7 +4243,7 @@ mod tests {
assert_eq!(stats_ba.num_rows, stats_ab.num_rows);
assert_eq!(stats_ba.column_statistics, stats_ab.column_statistics);
assert_eq!(
- stats_ab.column_statistics,
+ without_null_counts(stats_ab.column_statistics),
[left_col_stats, right_col_stats].concat()
);
@@ -3772,7 +4314,7 @@ mod tests {
.unwrap();
assert_eq!(partial_join_stats.num_rows, expected_num_rows);
assert_eq!(
- partial_join_stats.column_statistics,
+ without_null_counts(partial_join_stats.column_statistics),
[left_col_stats.clone(), right_col_stats.clone()].concat()
);
}
diff --git a/datafusion/physical-plan/src/operator_statistics/mod.rs b/datafusion/physical-plan/src/operator_statistics/mod.rs
index 9696bdf721..d63442b3c3 100644
--- a/datafusion/physical-plan/src/operator_statistics/mod.rs
+++ b/datafusion/physical-plan/src/operator_statistics/mod.rs
@@ -604,9 +604,34 @@ fn rescale_byte_size(stats: &mut Statistics, new_num_rows: Precision<usize>) {
};
}
+/// Rescale column null counts to `new_num_rows`, keeping each null fraction
+/// from the original estimate. A count stays exact only when the row count
+/// does not change, or when it is zero, and never exceeds the new row count.
+fn rescale_null_counts(stats: &mut Statistics, new_num_rows: Precision<usize>) {
+ let old_rows = stats.num_rows;
+ if old_rows == new_num_rows {
+ return;
+ }
+ for column_stats in &mut stats.column_statistics {
+ column_stats.null_count = match (
+ column_stats.null_count,
+ old_rows.get_value(),
+ new_num_rows.get_value(),
+ ) {
+ (Precision::Exact(0), _, _) => Precision::Exact(0),
+ (null_count, Some(&old), Some(&new)) if old > 0 => null_count
+ .map(|nulls| (nulls as f64 * new as f64 / old as f64).round() as usize)
+ .to_inexact()
+ .min(&new_num_rows),
+ (null_count, _, _) => null_count.to_inexact(),
+ };
+ }
+}
+
/// Overrides the operator's built-in output statistics (computed from the
/// pre-resolved `child_stats`) with `num_rows`, rescaling `total_byte_size`
-/// proportionally. Used by providers that refine only the row count.
+/// and null counts proportionally. Used by providers that refine only the row
+/// count.
fn computed_with_row_count(
plan: &dyn ExecutionPlan,
num_rows: Precision<usize>,
@@ -619,6 +644,7 @@ fn computed_with_row_count(
let mut base = Arc::unwrap_or_clone(
plan.statistics_from_inputs(&child_base, &StatisticsArgs::new())?,
);
+ rescale_null_counts(&mut base, num_rows);
rescale_byte_size(&mut base, num_rows);
Ok(StatisticsResult::Computed(ExtendedStatistics::new(base)))
}
@@ -2416,6 +2442,137 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_join_null_counts_follow_projection_and_swap() -> Result<()> {
+ // Left: 1000 rows, NDV 100, NULLs: 20 in `a`, 50 in `b`.
+ // Right: 10 rows, NDV 10, NULLs: 1 in `a`, 2 in `b`.
+ // The inner join estimates 100 pairs. The semi join estimates 98
+ // matching left rows, so the left join pads 902 right-side rows.
+ let schema = make_schema();
+ let source = |num_rows: usize, ndv: usize, null_counts: [usize; 2]| {
+ let col_stats = null_counts
+ .iter()
+ .map(|&null_count| ColumnStatistics {
+ null_count: Precision::Exact(null_count),
+ distinct_count: Precision::Exact(ndv),
+ ..ColumnStatistics::new_unknown()
+ })
+ .collect();
+ Arc::new(MockSourceExec::with_column_stats(
+ Arc::clone(&schema),
+ Precision::Exact(num_rows),
+ col_stats,
+ )) as Arc<dyn ExecutionPlan>
+ };
+ let on: JoinOn = vec![(
+ Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>,
+ Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>,
+ )];
+ // The projection outputs right `b`, left `a`, then right `a`.
+ let join = HashJoinExec::try_new(
+ source(1000, 100, [20, 50]),
+ source(10, 10, [1, 2]),
+ on,
+ None,
+ &JoinType::Left,
+ Some(vec![3, 0, 2]),
+ PartitionMode::CollectLeft,
+ NullEquality::NullEqualsNothing,
+ false,
+ )?;
+ // Swapping the inputs turns the join into a right join with the same
+ // output columns.
+ let swapped = join.swap_inputs(PartitionMode::CollectLeft)?;
+ let registry =
+ StatisticsRegistry::with_providers(vec![Arc::new(JoinStatisticsProvider)]);
+ // The preserved key keeps its 20 NULLs only as an estimate, since
+ // per-partition statistics cannot place the unmatched rows.
+ let expected = vec![
+ Precision::Inexact(922),
+ Precision::Inexact(20),
+ Precision::Inexact(902),
+ ];
+ for plan in [Arc::new(join) as Arc<dyn ExecutionPlan>, swapped] {
+ let built_in = StatisticsContext::new()
+ .compute(plan.as_ref(), &StatisticsArgs::new())?;
+ let per_partition = StatisticsContext::new().compute(
+ plan.as_ref(),
+ &StatisticsArgs::new().with_partition(Some(0)),
+ )?;
+ let provided = compute(®istry, plan.as_ref())?;
+ for stats in [built_in.as_ref(), per_partition.as_ref()] {
+ let null_counts: Vec<_> = stats
+ .column_statistics
+ .iter()
+ .map(|column_stats| column_stats.null_count)
+ .collect();
+ assert_eq!(null_counts, expected, "{}", plan.name());
+ }
+ // The provider uses 1000 output rows instead of 1002 and rescales
+ // the counts to that estimate after applying the same projection.
+ let null_counts: Vec<_> = provided
+ .base
+ .column_statistics
+ .iter()
+ .map(|column| column.null_count)
+ .collect();
+ assert_eq!(
+ null_counts,
+ vec![
+ Precision::Inexact(920),
+ Precision::Inexact(20),
+ Precision::Inexact(900)
+ ]
+ );
+ }
+ Ok(())
+ }
+
+ #[test]
+ fn test_rescale_null_counts() {
+ let null_counts = |stats: &Statistics| -> Vec<_> {
+ stats
+ .column_statistics
+ .iter()
+ .map(|column_stats| column_stats.null_count)
+ .collect()
+ };
+ let mut stats = Statistics {
+ num_rows: Precision::Inexact(5000),
+ total_byte_size: Precision::Absent,
+ column_statistics: [
+ Precision::Exact(0),
+ Precision::Exact(100),
+ Precision::Inexact(4000),
+ Precision::Inexact(6000),
+ Precision::Absent,
+ ]
+ .into_iter()
+ .map(|null_count| ColumnStatistics {
+ null_count,
+ ..ColumnStatistics::new_unknown()
+ })
+ .collect(),
+ };
+ let unchanged = null_counts(&stats);
+ rescale_null_counts(&mut stats, Precision::Inexact(5000));
+ assert_eq!(null_counts(&stats), unchanged);
+
+ // A provider that lowers the row count keeps each null fraction, and
+ // an inconsistent count cannot exceed the new row count.
+ rescale_null_counts(&mut stats, Precision::Inexact(1000));
+ assert_eq!(
+ null_counts(&stats),
+ vec![
+ Precision::Exact(0),
+ Precision::Inexact(20),
+ Precision::Inexact(800),
+ Precision::Inexact(1000),
+ Precision::Absent,
+ ]
+ );
+ }
+
#[test]
fn test_join_provider_multi_key_ndv() -> Result<()> {
// Multi-key join: ON a.a = b.a AND a.b = b.b
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org