fix: Deduplicate collect_left_input physical expression evaluation

[jonathanc-n]

Which issue does this PR close?

• Closes #.

Rationale for this change

I tried some optimizations in #16719 but there is a large slow down with concatenating arrays twice

What changes are included in this PR?

Couple of changes:

• We return early in collect_left_input
• concatenate and pass the entire batch into update_hash to avoid duplicate left value calculation

Are these changes tested?

Are there any user-facing changes?

[jonathanc-n]

@alamb Would you be able to run some benchmarks on this please?

[jonathanc-n]

This will cause failures on SymmetricHashJoin as it will compare empty arrays. @nuno-faria If you have the time, would you like to work on this?

[nuno-faria]

Yeah I can take a look. Could you provide an example that uses a SymmetricHashJoin?

[jonathanc-n]

If you add this code:

if batches.len() == 0 {
    return Ok(JoinLeftData::new(
        Box::new(JoinHashMapU32::with_capacity(0)),
        RecordBatch::new_empty(schema),
        Vec::new(),
        Mutex::new(BooleanBufferBuilder::new(0)),
        AtomicUsize::new(probe_threads_count),
        reservation,
    ));
};

and run sqllogictests it should direct you to the queries

[nuno-faria]

I think I fixed it by updating the equal_rows_arr function. What would be the easiest way to contribute, since I don't have push permissions to jonathanc-n/datafusion?

[jonathanc-n]

I think you can just open a pul request from another branch, or you could include it in your other pull request that you have open, and I can update this branch after it merges

[alamb]

🤖 ./gh_compare_branch.sh Benchmark Script Running
Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing optimize-hj (399d8d5) to f38f52f diff using: tpch_mem clickbench_partitioned clickbench_extended
Results will be posted here when complete

[alamb]

🤖: Benchmark completed

Details

Comparing HEAD and optimize-hj
--------------------
Benchmark clickbench_extended.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Query        ┃        HEAD ┃ optimize-hj ┃    Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ QQuery 0     │  2004.10 ms │  1973.31 ms │ no change │
│ QQuery 1     │   793.97 ms │   773.20 ms │ no change │
│ QQuery 2     │  1446.68 ms │  1453.19 ms │ no change │
│ QQuery 3     │   661.12 ms │   667.42 ms │ no change │
│ QQuery 4     │  1347.83 ms │  1365.97 ms │ no change │
│ QQuery 5     │ 14533.33 ms │ 14466.28 ms │ no change │
│ QQuery 6     │  2075.60 ms │  2008.93 ms │ no change │
│ QQuery 7     │  1911.95 ms │  1886.33 ms │ no change │
└──────────────┴─────────────┴─────────────┴───────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary          ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (HEAD)          │ 24774.58ms │
│ Total Time (optimize-hj)   │ 24594.64ms │
│ Average Time (HEAD)        │  3096.82ms │
│ Average Time (optimize-hj) │  3074.33ms │
│ Queries Faster             │          0 │
│ Queries Slower             │          0 │
│ Queries with No Change     │          8 │
│ Queries with Failure       │          0 │
└────────────────────────────┴────────────┘
--------------------
Benchmark clickbench_partitioned.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃        HEAD ┃ optimize-hj ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 0     │     2.22 ms │     2.68 ms │  1.21x slower │
│ QQuery 1     │    34.47 ms │    35.92 ms │     no change │
│ QQuery 2     │    81.50 ms │    81.90 ms │     no change │
│ QQuery 3     │    98.45 ms │    99.17 ms │     no change │
│ QQuery 4     │   579.62 ms │   590.91 ms │     no change │
│ QQuery 5     │   866.54 ms │   869.49 ms │     no change │
│ QQuery 6     │     2.30 ms │     2.26 ms │     no change │
│ QQuery 7     │    38.91 ms │    40.23 ms │     no change │
│ QQuery 8     │   847.29 ms │   861.89 ms │     no change │
│ QQuery 9     │  1142.59 ms │  1167.02 ms │     no change │
│ QQuery 10    │   261.05 ms │   254.91 ms │     no change │
│ QQuery 11    │   291.12 ms │   287.88 ms │     no change │
│ QQuery 12    │   872.36 ms │   877.67 ms │     no change │
│ QQuery 13    │  1241.12 ms │  1210.39 ms │     no change │
│ QQuery 14    │   820.82 ms │   799.62 ms │     no change │
│ QQuery 15    │   784.25 ms │   776.55 ms │     no change │
│ QQuery 16    │  1599.64 ms │  1588.05 ms │     no change │
│ QQuery 17    │  1602.58 ms │  1584.75 ms │     no change │
│ QQuery 18    │  2834.95 ms │  2852.56 ms │     no change │
│ QQuery 19    │    86.24 ms │    84.64 ms │     no change │
│ QQuery 20    │  1155.75 ms │  1150.13 ms │     no change │
│ QQuery 21    │  1306.42 ms │  1297.38 ms │     no change │
│ QQuery 22    │  2130.60 ms │  2126.87 ms │     no change │
│ QQuery 23    │  7563.14 ms │  7422.36 ms │     no change │
│ QQuery 24    │   448.13 ms │   444.55 ms │     no change │
│ QQuery 25    │   302.62 ms │   297.32 ms │     no change │
│ QQuery 26    │   433.41 ms │   434.73 ms │     no change │
│ QQuery 27    │  1545.26 ms │  1546.61 ms │     no change │
│ QQuery 28    │ 11907.40 ms │ 12994.34 ms │  1.09x slower │
│ QQuery 29    │   520.91 ms │   519.70 ms │     no change │
│ QQuery 30    │   787.16 ms │   751.71 ms │     no change │
│ QQuery 31    │   794.39 ms │   808.93 ms │     no change │
│ QQuery 32    │  2379.26 ms │  2358.10 ms │     no change │
│ QQuery 33    │  3144.81 ms │  3179.19 ms │     no change │
│ QQuery 34    │  3176.98 ms │  3237.12 ms │     no change │
│ QQuery 35    │  1260.51 ms │  1262.17 ms │     no change │
│ QQuery 36    │   126.33 ms │   118.26 ms │ +1.07x faster │
│ QQuery 37    │    50.54 ms │    50.23 ms │     no change │
│ QQuery 38    │   118.74 ms │   117.88 ms │     no change │
│ QQuery 39    │   194.10 ms │   191.10 ms │     no change │
│ QQuery 40    │    42.10 ms │    41.04 ms │     no change │
│ QQuery 41    │    38.16 ms │    38.70 ms │     no change │
│ QQuery 42    │    32.06 ms │    33.30 ms │     no change │
└──────────────┴─────────────┴─────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary          ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (HEAD)          │ 53546.82ms │
│ Total Time (optimize-hj)   │ 54490.22ms │
│ Average Time (HEAD)        │  1245.27ms │
│ Average Time (optimize-hj) │  1267.21ms │
│ Queries Faster             │          1 │
│ Queries Slower             │          2 │
│ Queries with No Change     │         40 │
│ Queries with Failure       │          0 │
└────────────────────────────┴────────────┘
--------------------
Benchmark tpch_mem_sf1.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┓
┃ Query        ┃      HEAD ┃ optimize-hj ┃       Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━┩
│ QQuery 1     │  97.18 ms │    96.38 ms │    no change │
│ QQuery 2     │  20.42 ms │    20.43 ms │    no change │
│ QQuery 3     │  32.72 ms │    32.46 ms │    no change │
│ QQuery 4     │  18.34 ms │    18.30 ms │    no change │
│ QQuery 5     │  49.77 ms │    47.80 ms │    no change │
│ QQuery 6     │  11.67 ms │    11.67 ms │    no change │
│ QQuery 7     │  84.86 ms │    90.26 ms │ 1.06x slower │
│ QQuery 8     │  24.10 ms │    24.29 ms │    no change │
│ QQuery 9     │  52.98 ms │    52.15 ms │    no change │
│ QQuery 10    │  42.28 ms │    42.36 ms │    no change │
│ QQuery 11    │  11.27 ms │    11.16 ms │    no change │
│ QQuery 12    │  34.10 ms │    34.45 ms │    no change │
│ QQuery 13    │  26.25 ms │    25.63 ms │    no change │
│ QQuery 14    │   9.79 ms │     9.74 ms │    no change │
│ QQuery 15    │  18.81 ms │    18.46 ms │    no change │
│ QQuery 16    │  17.85 ms │    17.93 ms │    no change │
│ QQuery 17    │  94.98 ms │    93.54 ms │    no change │
│ QQuery 18    │ 186.18 ms │   193.66 ms │    no change │
│ QQuery 19    │  25.09 ms │    24.75 ms │    no change │
│ QQuery 20    │  30.91 ms │    30.91 ms │    no change │
│ QQuery 21    │ 144.53 ms │   146.34 ms │    no change │
│ QQuery 22    │  14.34 ms │    14.57 ms │    no change │
└──────────────┴───────────┴─────────────┴──────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ Benchmark Summary          ┃           ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ Total Time (HEAD)          │ 1048.41ms │
│ Total Time (optimize-hj)   │ 1057.25ms │
│ Average Time (HEAD)        │   47.65ms │
│ Average Time (optimize-hj) │   48.06ms │
│ Queries Faster             │         0 │
│ Queries Slower             │         1 │
│ Queries with No Change     │        21 │
│ Queries with Failure       │         0 │
└────────────────────────────┴───────────┘

[alamb]

🤖 ./gh_compare_branch.sh Benchmark Script Running
Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing optimize-hj (399d8d5) to f38f52f diff using: tpch10 tpch_mem10
Results will be posted here when complete

[alamb]

🤖: Benchmark completed

Details

Comparing HEAD and optimize-hj
--------------------
Benchmark tpch_mem_sf10.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┓
┃ Query        ┃ HEAD ┃ optimize-hj ┃       Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━┩
│ QQuery 1     │ FAIL │        FAIL │ incomparable │
│ QQuery 2     │ FAIL │        FAIL │ incomparable │
│ QQuery 3     │ FAIL │        FAIL │ incomparable │
│ QQuery 4     │ FAIL │        FAIL │ incomparable │
│ QQuery 5     │ FAIL │        FAIL │ incomparable │
│ QQuery 6     │ FAIL │        FAIL │ incomparable │
│ QQuery 7     │ FAIL │        FAIL │ incomparable │
│ QQuery 8     │ FAIL │        FAIL │ incomparable │
│ QQuery 9     │ FAIL │        FAIL │ incomparable │
│ QQuery 10    │ FAIL │        FAIL │ incomparable │
│ QQuery 11    │ FAIL │        FAIL │ incomparable │
│ QQuery 12    │ FAIL │        FAIL │ incomparable │
│ QQuery 13    │ FAIL │        FAIL │ incomparable │
│ QQuery 14    │ FAIL │        FAIL │ incomparable │
│ QQuery 15    │ FAIL │        FAIL │ incomparable │
│ QQuery 16    │ FAIL │        FAIL │ incomparable │
│ QQuery 17    │ FAIL │        FAIL │ incomparable │
│ QQuery 18    │ FAIL │        FAIL │ incomparable │
│ QQuery 19    │ FAIL │        FAIL │ incomparable │
│ QQuery 20    │ FAIL │        FAIL │ incomparable │
│ QQuery 21    │ FAIL │        FAIL │ incomparable │
│ QQuery 22    │ FAIL │        FAIL │ incomparable │
└──────────────┴──────┴─────────────┴──────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━┓
┃ Benchmark Summary          ┃        ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━┩
│ Total Time (HEAD)          │ 0.00ms │
│ Total Time (optimize-hj)   │ 0.00ms │
│ Average Time (HEAD)        │ 0.00ms │
│ Average Time (optimize-hj) │ 0.00ms │
│ Queries Faster             │      0 │
│ Queries Slower             │      0 │
│ Queries with No Change     │      0 │
│ Queries with Failure       │     22 │
└────────────────────────────┴────────┘
--------------------
Benchmark tpch_sf10.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┓
┃ Query        ┃ HEAD ┃ optimize-hj ┃       Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━┩
│ QQuery 1     │ FAIL │        FAIL │ incomparable │
│ QQuery 2     │ FAIL │        FAIL │ incomparable │
│ QQuery 3     │ FAIL │        FAIL │ incomparable │
│ QQuery 4     │ FAIL │        FAIL │ incomparable │
│ QQuery 5     │ FAIL │        FAIL │ incomparable │
│ QQuery 6     │ FAIL │        FAIL │ incomparable │
│ QQuery 7     │ FAIL │        FAIL │ incomparable │
│ QQuery 8     │ FAIL │        FAIL │ incomparable │
│ QQuery 9     │ FAIL │        FAIL │ incomparable │
│ QQuery 10    │ FAIL │        FAIL │ incomparable │
│ QQuery 11    │ FAIL │        FAIL │ incomparable │
│ QQuery 12    │ FAIL │        FAIL │ incomparable │
│ QQuery 13    │ FAIL │        FAIL │ incomparable │
│ QQuery 14    │ FAIL │        FAIL │ incomparable │
│ QQuery 15    │ FAIL │        FAIL │ incomparable │
│ QQuery 16    │ FAIL │        FAIL │ incomparable │
│ QQuery 17    │ FAIL │        FAIL │ incomparable │
│ QQuery 18    │ FAIL │        FAIL │ incomparable │
│ QQuery 19    │ FAIL │        FAIL │ incomparable │
│ QQuery 20    │ FAIL │        FAIL │ incomparable │
│ QQuery 21    │ FAIL │        FAIL │ incomparable │
│ QQuery 22    │ FAIL │        FAIL │ incomparable │
└──────────────┴──────┴─────────────┴──────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━┓
┃ Benchmark Summary          ┃        ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━┩
│ Total Time (HEAD)          │ 0.00ms │
│ Total Time (optimize-hj)   │ 0.00ms │
│ Average Time (HEAD)        │ 0.00ms │
│ Average Time (optimize-hj) │ 0.00ms │
│ Queries Faster             │      0 │
│ Queries Slower             │      0 │
│ Queries with No Change     │      0 │
│ Queries with Failure       │     22 │
└────────────────────────────┴────────┘

[jonathanc-n]

Maybe we should do this again when tests are passing after #16716 is merged

[ctsk]

The duplicate evaluation of the keys on the build side is no big deal, because those expressions are simple column selections. This is done when lowering a logical plan to a physical plan: [0] [1]

[0]

datafusion/datafusion/core/src/physical_planner.rs

Lines 950 to 1058 in eb25e8d

	LogicalPlan::Join(Join {
	left: original_left,
	right: original_right,
	on: keys,
	filter,
	join_type,
	null_equality,
	schema: join_schema,
	..
	}) => {
	let [physical_left, physical_right] = children.two()?;
	// If join has expression equijoin keys, add physical projection.
	let has_expr_join_key = keys.iter().any(|(l, r)| {
	!(matches!(l, Expr::Column(_)) && matches!(r, Expr::Column(_)))
	});
	let (new_logical, physical_left, physical_right) = if has_expr_join_key {
	// TODO: Can we extract this transformation to somewhere before physical plan
	// creation?
	let (left_keys, right_keys): (Vec<_>, Vec<_>) =
	keys.iter().cloned().unzip();
	let (left, left_col_keys, left_projected) =
	wrap_projection_for_join_if_necessary(
	&left_keys,
	original_left.as_ref().clone(),
	)?;
	let (right, right_col_keys, right_projected) =
	wrap_projection_for_join_if_necessary(
	&right_keys,
	original_right.as_ref().clone(),
	)?;
	let column_on = (left_col_keys, right_col_keys);
	let left = Arc::new(left);
	let right = Arc::new(right);
	let (new_join, requalified) = Join::try_new_with_project_input(
	node,
	Arc::clone(&left),
	Arc::clone(&right),
	column_on,
	)?;
	let new_join = LogicalPlan::Join(new_join);
	// If inputs were projected then create ExecutionPlan for these new
	// LogicalPlan nodes.
	let physical_left = match (left_projected, left.as_ref()) {
	// If left_projected is true we are guaranteed that left is a Projection
	(
	true,
	LogicalPlan::Projection(Projection { input, expr, .. }),
	) => self.create_project_physical_exec(
	session_state,
	physical_left,
	input,
	expr,
	)?,
	_ => physical_left,
	};
	let physical_right = match (right_projected, right.as_ref()) {
	// If right_projected is true we are guaranteed that right is a Projection
	(
	true,
	LogicalPlan::Projection(Projection { input, expr, .. }),
	) => self.create_project_physical_exec(
	session_state,
	physical_right,
	input,
	expr,
	)?,
	_ => physical_right,
	};
	// Remove temporary projected columns
	if left_projected || right_projected {
	// Re-qualify the join schema only if the inputs were previously requalified in
	// `try_new_with_project_input`. This ensures that when building the Projection
	// it can correctly resolve field nullability and data types
	// by disambiguating fields from the left and right sides of the join.
	let qualified_join_schema = if requalified {
	Arc::new(qualify_join_schema_sides(
	join_schema,
	original_left,
	original_right,
	)?)
	} else {
	Arc::clone(join_schema)
	};
	let final_join_result = qualified_join_schema
	.iter()
	.map(Expr::from)
	.collect::<Vec<_>>();
	let projection = LogicalPlan::Projection(Projection::try_new(
	final_join_result,
	Arc::new(new_join),
	)?);
	// LogicalPlan mutated
	(Cow::Owned(projection), physical_left, physical_right)
	} else {
	// LogicalPlan mutated
	(Cow::Owned(new_join), physical_left, physical_right)
	}
	} else {
	// LogicalPlan unchanged
	(Cow::Borrowed(node), physical_left, physical_right)
	};

[1]

datafusion/datafusion/expr/src/logical_plan/builder.rs

Lines 1979 to 2039 in eb25e8d

	pub fn wrap_projection_for_join_if_necessary(
	join_keys: &[Expr],
	input: LogicalPlan,
	) -> Result<(LogicalPlan, Vec<Column>, bool)> {
	let input_schema = input.schema();
	let alias_join_keys: Vec<Expr> = join_keys
	.iter()
	.map(|key| {
	// The display_name() of cast expression will ignore the cast info, and show the inner expression name.
	// If we do not add alias, it will throw same field name error in the schema when adding projection.
	// For example:
	// input scan : [a, b, c],
	// join keys: [cast(a as int)]
	//
	// then a and cast(a as int) will use the same field name - `a` in projection schema.
	// https://github.com/apache/datafusion/issues/4478
	if matches!(key, Expr::Cast(_)) || matches!(key, Expr::TryCast(_)) {
	let alias = format!("{key}");
	key.clone().alias(alias)
	} else {
	key.clone()
	}
	})
	.collect::<Vec<_>>();
	let need_project = join_keys.iter().any(|key| !matches!(key, Expr::Column(_)));
	let plan = if need_project {
	// Include all columns from the input and extend them with the join keys
	let mut projection = input_schema
	.columns()
	.into_iter()
	.map(Expr::Column)
	.collect::<Vec<_>>();
	let join_key_items = alias_join_keys
	.iter()
	.flat_map(|expr| expr.try_as_col().is_none().then_some(expr))
	.cloned()
	.collect::<HashSet<Expr>>();
	projection.extend(join_key_items);
	LogicalPlanBuilder::from(input)
	.project(projection.into_iter().map(SelectExpr::from))?
	.build()?
	} else {
	input
	};
	let join_on = alias_join_keys
	.into_iter()
	.map(|key| {
	if let Some(col) = key.try_as_col() {
	Ok(col.clone())
	} else {
	let name = key.schema_name().to_string();
	Ok(Column::from_name(name))
	}
	})
	.collect::<Result<Vec<_>>>()?;
	Ok((plan, join_on, need_project))
	}

[jonathanc-n]

@ctsk Oh i see, I guess this pull request would be to just deduplicate the code

[jonathanc-n]

@Dandandan Would you be able to take a look when you have the time?

[github-actions]

Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.