From 8bf6b3faef95dd4497c45cbba2e00487dbf3d6d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dani=C3=ABl=20Heres?= Date: Thu, 20 Aug 2026 21:48:10 +0200 Subject: [PATCH] fix: a join reports the size of the rows it emits A join knew how many rows it would emit but not how wide they were: every join type but semi and anti reported no total byte size at all. An operator above it reading that size finds nothing, so `hash_join_single_partition_threshold`, a byte threshold, falls back to counting rows. The width of an output row follows from the sides the join emits, which the join type already says: both sides for an inner or outer join, the preserved side for a semi or anti join, and one side plus a boolean for a mark join. Multiplying it by the estimated cardinality gives the size. Semi and anti joins keep the size they derive from their column statistics, which knows which columns survive; the width fills a gap rather than replacing a better answer. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012YiAABcW4WSqij31zz2P6c --- .../partition_statistics.rs | 6 +- datafusion/physical-plan/src/joins/utils.rs | 86 ++++++++++++++++++- 2 files changed, 88 insertions(+), 4 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/partition_statistics.rs b/datafusion/core/tests/physical_optimizer/partition_statistics.rs index 6cabcdb710393..6decb7b9092f3 100644 --- a/datafusion/core/tests/physical_optimizer/partition_statistics.rs +++ b/datafusion/core/tests/physical_optimizer/partition_statistics.rs @@ -1596,7 +1596,7 @@ mod test { // For collect left mode, the min/max values are from the entire left table and the specific partition of the right table. let expected_p0_statistics = Statistics { num_rows: Precision::Inexact(2), - total_byte_size: Precision::Absent, + total_byte_size: Precision::Inexact(32), column_statistics: vec![ // Left id column: all partitions (id 1..4) ColumnStatistics { @@ -1677,7 +1677,7 @@ mod test { // For partitioned mode, the min/max values are from the specific partition for each side. let expected_p0_statistics = Statistics { num_rows: Precision::Inexact(2), - total_byte_size: Precision::Absent, + total_byte_size: Precision::Inexact(32), column_statistics: vec![ // Left id column: partition 0 only (id 3..4) ColumnStatistics { @@ -1756,7 +1756,7 @@ mod test { // For auto mode, the min/max values are from the entire left and right tables. let expected_p0_statistics = Statistics { num_rows: Precision::Inexact(4), - total_byte_size: Precision::Absent, + total_byte_size: Precision::Inexact(64), column_statistics: vec![ // Left id column: all partitions (id 1..4) ColumnStatistics { diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs index 654d873ae1b0e..3b2326ddf96d7 100644 --- a/datafusion/physical-plan/src/joins/utils.rs +++ b/datafusion/physical-plan/src/joins/utils.rs @@ -470,12 +470,42 @@ pub(crate) fn estimate_join_statistics( join_type: &JoinType, schema: &Schema, ) -> Result { + // Width of one output row, from the sides this join emits. Without it a join reports + // no size and `hash_join_single_partition_threshold` falls back to counting rows. + let width = |stats: &Statistics| match ( + stats.total_byte_size.get_value(), + stats.num_rows.get_value(), + ) { + (Some(bytes), Some(rows)) if *rows > 0 => Some(*bytes as f64 / *rows as f64), + _ => None, + }; + // The boolean a mark join appends is one bit per row. + const MARK_COLUMN_WIDTH: f64 = 1.0 / 8.0; + let output_width = match join_type { + JoinType::LeftSemi | JoinType::LeftAnti => width(&left_stats), + JoinType::RightSemi | JoinType::RightAnti => width(&right_stats), + JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full => { + width(&left_stats) + .zip(width(&right_stats)) + .map(|(left, right)| left + right) + } + JoinType::LeftMark => width(&left_stats).map(|w| w + MARK_COLUMN_WIDTH), + JoinType::RightMark => width(&right_stats).map(|w| w + MARK_COLUMN_WIDTH), + }; + let join_stats = estimate_join_cardinality(join_type, left_stats, right_stats, on, null_equality); let (num_rows, total_byte_size, column_statistics) = match join_stats { Some(stats) => ( Precision::Inexact(stats.num_rows), - stats.total_byte_size, + match (stats.total_byte_size, output_width) { + // Only fill a gap: a size the join derived from its column statistics + // knows which columns it keeps, which an average row width cannot. + (Precision::Absent, Some(width)) => { + Precision::Inexact((stats.num_rows as f64 * width) as usize) + } + (derived, _) => derived, + }, stats.column_statistics, ), None => ( @@ -2810,6 +2840,60 @@ mod tests { Ok(()) } + /// A join reports the size of the rows it emits, so an operator above it can size + /// itself without counting rows. + #[test] + fn test_join_statistics_report_an_output_byte_size() -> Result<()> { + let sized = |rows: usize, bytes: usize| Statistics { + num_rows: Inexact(rows), + total_byte_size: Inexact(bytes), + column_statistics: vec![ColumnStatistics { + byte_size: Inexact(bytes), + ..create_column_stats(Inexact(0), Inexact(100), Inexact(100), Inexact(0)) + }], + }; + let schema = Schema::new(vec![ + Field::new("a", DataType::Int64, false), + Field::new("c", DataType::Int64, false), + ]); + let on: JoinOn = vec![( + Arc::new(Column::new("a", 0)) as _, + Arc::new(Column::new("c", 0)) as _, + )]; + let estimate = |join_type: JoinType| -> Result { + estimate_join_statistics( + // Eight bytes per row on the left, sixteen on the right. + sized(1000, 8000), + sized(2000, 32000), + &on, + NullEquality::NullEqualsNothing, + &join_type, + &schema, + ) + }; + + // An inner join emits both sides: 1000 * 2000 / 100 rows, 24 bytes each. + let inner = estimate(JoinType::Inner)?; + assert_eq!(inner.num_rows, Inexact(20000)); + assert_eq!(inner.total_byte_size, Inexact(20000 * 24)); + + // A mark join emits the left side plus one bit per row. + let mark = estimate(JoinType::LeftMark)?; + assert_eq!(mark.num_rows, Inexact(1000)); + assert_eq!(mark.total_byte_size, Inexact((1000.0 * 8.125) as usize)); + + // A semi join derives its size from the columns it keeps, which is more + // precise than an average row width, so that estimate is left alone. + let semi = estimate(JoinType::LeftSemi)?; + assert_ne!(semi.total_byte_size, Absent); + assert_eq!( + semi.total_byte_size, + total_byte_size_from_column_statistics(&semi.column_statistics) + ); + + Ok(()) + } + fn create_stats( num_rows: Option, column_stats: Vec,