viirya commented on code in PR #25491:
URL: https://github.com/apache/datafusion/pull/25491#discussion_r4052500116
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -1625,6 +1660,9 @@ impl ExecutionPlan for HashJoinExec {
partition: usize,
context: Arc<TaskContext>,
) -> Result<SendableRecordBatchStream> {
+ if let Some(prepared) = &self.prepared_build {
Review Comment:
This looks redundant next to the `validate` in `build()`, and I expect
someone will eventually delete it as dead weight. It is not redundant — it
catches direct field mutation that bypasses the builder, which your own test
exercises:
```rust
let mut mutated = attached.builder().build()?;
mutated.on = keys;
```
Worth a comment saying so, e.g. "fields such as `on` are writable within the
crate, so re-check here: `build()` cannot be the only gate".
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs:
##########
@@ -0,0 +1,326 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Explicit immutable build reuse for embedding executors.
+
+use super::*;
+use arrow::array::{Array, AsArray};
+use datafusion_common::exec_datafusion_err;
+use datafusion_execution::memory_pool::MemoryPool;
+
+/// An immutable, fully prepared broadcast build, independent of any probe
task.
+///
+/// Created by [`HashJoinExec::prepare_build`]. The embedding executor owns
cache
+/// identity, admission, single-flight coordination, cancellation and eviction.
+/// This object retains its input buffers and memory reservation until its last
+/// lease is dropped; it never retains an input stream or task context.
Prepared
+/// builds support fixed-width and UTF-8 build columns, with direct-column keys
+/// and non-spilling INNER joins. Residual conditions belong to each consuming
+/// join; null-aware joins remain unsupported.
+///
+/// Hash-join gathers copy supported build columns into output buffers,
+/// including contiguous selections. Output batches can therefore outlive this
+/// object without retaining unaccounted cached payload. View, dictionary and
+/// nested build columns remain unsupported. UTF-8 and fixed-size binary keys
+/// use hash-table membership filters instead of copying range or IN-list
values.
+pub struct PreparedHashJoinBuild {
Review Comment:
This is where an identity token would live if you take up point 1 in the
main review — something like `snapshot: Option<Arc<str>>`, compared in
`validate` and never interpreted. Flagging here so the thread hangs off the
struct rather than only in the summary.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -3104,18 +3250,37 @@ async fn collect_left_input(
&& !left_values.is_empty()
&& left_values[0].logical_null_count() > 0;
+ if prepared {
+ drop(batches);
+ let retained = RecordBatchMemoryCounter::new().count_batch(&batch);
+ let allowance = input_bytes.checked_add(copy_bytes).ok_or_else(|| {
+ datafusion_common::exec_datafusion_err!(
+ "Prepared hash-join payload size overflow"
+ )
+ })?;
+ if retained > allowance {
Review Comment:
`allowance` is hand-computed from Arrow's current buffer layout and 64-byte
alignment rounding. I worked through the multi-batch case and the margin looks
sufficient (roughly 192 bytes of padding budget per batch, and `4*(rows_i+1)`
summed over two or more batches dominates `4*(total+1)`), so I do not think
this fires today.
But it binds us to Arrow's allocation behaviour. If a future arrow-rs
changes capacity rounding — say to power-of-two growth — users hit a hard
`Internal error` on a perfectly valid input, with a message asking them to
report a bug.
Would you consider degrading instead of failing?
```rust
if retained > allowance {
debug_assert!(false, "prepared concat exceeded admitted bound");
reservation.try_grow(retained - allowance)?; // let the pool decide
} else {
reservation.shrink(allowance - retained);
}
```
That keeps the invariant loud in debug builds and in CI, while in release an
estimate that drifts becomes "we reserved a bit more memory" rather than a
failed query.
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs:
##########
@@ -0,0 +1,326 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Explicit immutable build reuse for embedding executors.
+
+use super::*;
+use arrow::array::{Array, AsArray};
+use datafusion_common::exec_datafusion_err;
+use datafusion_execution::memory_pool::MemoryPool;
+
+/// An immutable, fully prepared broadcast build, independent of any probe
task.
+///
+/// Created by [`HashJoinExec::prepare_build`]. The embedding executor owns
cache
+/// identity, admission, single-flight coordination, cancellation and eviction.
+/// This object retains its input buffers and memory reservation until its last
+/// lease is dropped; it never retains an input stream or task context.
Prepared
+/// builds support fixed-width and UTF-8 build columns, with direct-column keys
+/// and non-spilling INNER joins. Residual conditions belong to each consuming
+/// join; null-aware joins remain unsupported.
+///
+/// Hash-join gathers copy supported build columns into output buffers,
+/// including contiguous selections. Output batches can therefore outlive this
+/// object without retaining unaccounted cached payload. View, dictionary and
+/// nested build columns remain unsupported. UTF-8 and fixed-size binary keys
+/// use hash-table membership filters instead of copying range or IN-list
values.
+pub struct PreparedHashJoinBuild {
+ build: Arc<JoinBuildData>,
+ keys: Vec<usize>,
+ null_equality: NullEquality,
+}
+
+impl fmt::Debug for PreparedHashJoinBuild {
+ /// Describe immutable metadata without dumping table contents.
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PreparedHashJoinBuild")
+ .field("schema", &self.build.batch.schema())
+ .field("keys", &self.keys)
+ .field("rows", &self.num_rows())
+ .field("reserved_bytes", &self.reserved_bytes())
+ .finish()
+ }
+}
+
+impl PreparedHashJoinBuild {
+ /// Return the retained build reservation, excluding all per-probe state.
+ pub fn reserved_bytes(&self) -> usize {
+ self.build.reservation.size()
+ }
+
+ /// Return the complete build row count, including duplicate and null keys.
+ pub fn num_rows(&self) -> usize {
+ self.build.batch.num_rows()
+ }
+
+ /// Create independent mutable state for one consuming join.
+ pub(super) fn probe_data(&self, probe_threads: usize) -> JoinLeftData {
+ JoinLeftData {
+ build: Arc::clone(&self.build),
+ null_aware_mark_scope_map: None,
+ null_value_scope_map: None,
+ visited_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ null_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ probe_completion: ProbeCompletion::new(probe_threads),
+ build_side_has_null: false,
+ _probe_reservation: self.build.reservation.new_empty(),
+ }
+ }
+
+ /// Validate the build descriptor and current execution restrictions
without
+ /// consuming input or modifying either plan. Cache identity is
caller-owned.
+ pub(super) fn validate(&self, join: &HashJoinExec) -> Result<()> {
+ let keys = prepared_key_indices(join)?;
+ if join.left.schema() != self.build.batch.schema()
+ || keys != self.keys
+ || join.null_equality != self.null_equality
+ {
+ return plan_err!(
+ "Prepared hash-join build does not match schema, keys or null
equality"
+ );
+ }
+ if let Some(filter) = &join.dynamic_filter {
+ let filter_keys = filter.filter.children();
+ if filter_keys.len() != join.on.len()
+ || filter_keys
+ .iter()
+ .zip(&join.on)
+ .any(|(filter_key, (_, probe_key))| {
+ filter_key.as_ref() != probe_key.as_ref()
+ })
+ {
+ return plan_err!(
+ "Prepared hash-join dynamic filter keys do not match probe
keys"
+ );
+ }
+ }
+ Ok(())
+ }
+}
+
+impl HashJoinExec {
+ /// Prepare one immutable build using an embedding executor's durable pool.
+ ///
+ /// The supplied stream must own its native buffers independently of
producer
+ /// task cleanup. This method consumes only that stream, never `self.left`,
+ /// and reserves retained data, hash buckets and row-index chains against
+ /// `pool`. The caller must keep original producer allocations charged
until
+ /// its stream releases them. `config` controls ordinary perfect-map and
+ /// dynamic-filter choices. UTF-8 and fixed-size binary keys
+ /// retain hash membership only:
+ /// range bounds and IN-list literals would allocate unaccounted key
copies.
+ ///
+ /// Validates eligibility and the stream schema before polling. On error or
+ /// future cancellation, all work and reservations are dropped; no
partially
+ /// prepared object is returned. Concurrent preparation/cache publication
is
+ /// the caller's responsibility. Bounds and membership are prepared once,
but
+ /// each consuming join publishes them into its own dynamic filter.
Review Comment:
Two things about this doc block:
1. The most important caller obligation — that snapshot identity is theirs
to get right, and that getting it wrong yields silently wrong results — is not
stated here at all. It only appears in the PR description. Could it lead the
doc block?
2. "range bounds and IN-list literals would allocate unaccounted key copies"
reads as though only byte keys copy. `InListExpr::try_new_from_array`
materialises one `ScalarValue` + `lit()` per **row** (not per distinct value)
for every type, including `Int64` — it is just bounded there by
`hash_join_inlist_pushdown_max_size` (128 KB). The real reason to exclude byte
keys is that their copy volume is unbounded, not that they are the only types
that copy. Worth stating precisely, or a reader will assume the numeric path
has no unaccounted allocation.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -2854,6 +2920,7 @@ async fn collect_left_input(
null_equality: NullEquality,
null_aware: Option<NullAwareMode>,
array_map_created_count: Count,
+ prepared: bool,
Review Comment:
`collect_left_input` already carries
`#[expect(clippy::too_many_arguments)]`; this makes it a 13th parameter and
adds six `if prepared` sites scattered through the hottest function in the hash
join. Some of them — the validity-mask scratch sizing, the row-index chain
top-up — are subtle enough that I had to read them twice to confirm they do not
affect the normal path.
Would you consider pulling the admission arithmetic into its own type, e.g.
a `PreparedAdmission` holding `concat_values` / `input_bytes` / `copy_bytes`
with `observe_batch` / `admit_copies` / `settle`, and passing
`Option<PreparedAdmission>`? Six `if prepared` collapse to three `if let
Some(..)`, and the accounting math ends up somewhere it can be unit-tested on
its own — which, given how carefully tuned it is, seems worth having.
Not a blocker, but this function is on every hash join's path and the review
cost of the current shape is real.
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs:
##########
@@ -0,0 +1,326 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Explicit immutable build reuse for embedding executors.
+
+use super::*;
+use arrow::array::{Array, AsArray};
+use datafusion_common::exec_datafusion_err;
+use datafusion_execution::memory_pool::MemoryPool;
+
+/// An immutable, fully prepared broadcast build, independent of any probe
task.
+///
+/// Created by [`HashJoinExec::prepare_build`]. The embedding executor owns
cache
+/// identity, admission, single-flight coordination, cancellation and eviction.
+/// This object retains its input buffers and memory reservation until its last
+/// lease is dropped; it never retains an input stream or task context.
Prepared
+/// builds support fixed-width and UTF-8 build columns, with direct-column keys
+/// and non-spilling INNER joins. Residual conditions belong to each consuming
+/// join; null-aware joins remain unsupported.
+///
+/// Hash-join gathers copy supported build columns into output buffers,
+/// including contiguous selections. Output batches can therefore outlive this
+/// object without retaining unaccounted cached payload. View, dictionary and
+/// nested build columns remain unsupported. UTF-8 and fixed-size binary keys
+/// use hash-table membership filters instead of copying range or IN-list
values.
+pub struct PreparedHashJoinBuild {
+ build: Arc<JoinBuildData>,
+ keys: Vec<usize>,
+ null_equality: NullEquality,
+}
+
+impl fmt::Debug for PreparedHashJoinBuild {
+ /// Describe immutable metadata without dumping table contents.
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PreparedHashJoinBuild")
+ .field("schema", &self.build.batch.schema())
+ .field("keys", &self.keys)
+ .field("rows", &self.num_rows())
+ .field("reserved_bytes", &self.reserved_bytes())
+ .finish()
+ }
+}
+
+impl PreparedHashJoinBuild {
+ /// Return the retained build reservation, excluding all per-probe state.
+ pub fn reserved_bytes(&self) -> usize {
+ self.build.reservation.size()
+ }
+
+ /// Return the complete build row count, including duplicate and null keys.
+ pub fn num_rows(&self) -> usize {
+ self.build.batch.num_rows()
+ }
+
+ /// Create independent mutable state for one consuming join.
+ pub(super) fn probe_data(&self, probe_threads: usize) -> JoinLeftData {
+ JoinLeftData {
+ build: Arc::clone(&self.build),
+ null_aware_mark_scope_map: None,
+ null_value_scope_map: None,
+ visited_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ null_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ probe_completion: ProbeCompletion::new(probe_threads),
+ build_side_has_null: false,
+ _probe_reservation: self.build.reservation.new_empty(),
Review Comment:
`self.build.reservation.new_empty()` inherits the pool passed to
`prepare_build` — the caller's long-lived cache pool — not the consuming task's
pool.
Harmless today: INNER joins allocate no bitmap, so this reservation stays at
zero. But `probe_data` is exactly where outer-join support would be added, and
at that point the visited bitmap would be charged to the cache pool instead of
the task pool, which is the wrong lifetime. Either a comment marking the
hazard, or having `probe_data` take the consumer's pool now (ignored while it
is always zero), would keep the interface honest.
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs:
##########
@@ -0,0 +1,326 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Explicit immutable build reuse for embedding executors.
+
+use super::*;
+use arrow::array::{Array, AsArray};
+use datafusion_common::exec_datafusion_err;
+use datafusion_execution::memory_pool::MemoryPool;
+
+/// An immutable, fully prepared broadcast build, independent of any probe
task.
+///
+/// Created by [`HashJoinExec::prepare_build`]. The embedding executor owns
cache
+/// identity, admission, single-flight coordination, cancellation and eviction.
+/// This object retains its input buffers and memory reservation until its last
+/// lease is dropped; it never retains an input stream or task context.
Prepared
+/// builds support fixed-width and UTF-8 build columns, with direct-column keys
+/// and non-spilling INNER joins. Residual conditions belong to each consuming
+/// join; null-aware joins remain unsupported.
+///
+/// Hash-join gathers copy supported build columns into output buffers,
+/// including contiguous selections. Output batches can therefore outlive this
+/// object without retaining unaccounted cached payload. View, dictionary and
+/// nested build columns remain unsupported. UTF-8 and fixed-size binary keys
+/// use hash-table membership filters instead of copying range or IN-list
values.
+pub struct PreparedHashJoinBuild {
+ build: Arc<JoinBuildData>,
+ keys: Vec<usize>,
+ null_equality: NullEquality,
+}
+
+impl fmt::Debug for PreparedHashJoinBuild {
+ /// Describe immutable metadata without dumping table contents.
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PreparedHashJoinBuild")
+ .field("schema", &self.build.batch.schema())
+ .field("keys", &self.keys)
+ .field("rows", &self.num_rows())
+ .field("reserved_bytes", &self.reserved_bytes())
+ .finish()
+ }
+}
+
+impl PreparedHashJoinBuild {
+ /// Return the retained build reservation, excluding all per-probe state.
+ pub fn reserved_bytes(&self) -> usize {
+ self.build.reservation.size()
+ }
+
+ /// Return the complete build row count, including duplicate and null keys.
+ pub fn num_rows(&self) -> usize {
+ self.build.batch.num_rows()
+ }
+
+ /// Create independent mutable state for one consuming join.
+ pub(super) fn probe_data(&self, probe_threads: usize) -> JoinLeftData {
+ JoinLeftData {
+ build: Arc::clone(&self.build),
+ null_aware_mark_scope_map: None,
+ null_value_scope_map: None,
+ visited_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ null_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ probe_completion: ProbeCompletion::new(probe_threads),
+ build_side_has_null: false,
+ _probe_reservation: self.build.reservation.new_empty(),
+ }
+ }
+
+ /// Validate the build descriptor and current execution restrictions
without
+ /// consuming input or modifying either plan. Cache identity is
caller-owned.
+ pub(super) fn validate(&self, join: &HashJoinExec) -> Result<()> {
+ let keys = prepared_key_indices(join)?;
+ if join.left.schema() != self.build.batch.schema()
+ || keys != self.keys
+ || join.null_equality != self.null_equality
+ {
+ return plan_err!(
+ "Prepared hash-join build does not match schema, keys or null
equality"
+ );
+ }
+ if let Some(filter) = &join.dynamic_filter {
+ let filter_keys = filter.filter.children();
+ if filter_keys.len() != join.on.len()
+ || filter_keys
+ .iter()
+ .zip(&join.on)
+ .any(|(filter_key, (_, probe_key))| {
+ filter_key.as_ref() != probe_key.as_ref()
+ })
+ {
+ return plan_err!(
+ "Prepared hash-join dynamic filter keys do not match probe
keys"
+ );
+ }
+ }
+ Ok(())
+ }
+}
+
+impl HashJoinExec {
+ /// Prepare one immutable build using an embedding executor's durable pool.
+ ///
+ /// The supplied stream must own its native buffers independently of
producer
+ /// task cleanup. This method consumes only that stream, never `self.left`,
+ /// and reserves retained data, hash buckets and row-index chains against
+ /// `pool`. The caller must keep original producer allocations charged
until
+ /// its stream releases them. `config` controls ordinary perfect-map and
+ /// dynamic-filter choices. UTF-8 and fixed-size binary keys
+ /// retain hash membership only:
+ /// range bounds and IN-list literals would allocate unaccounted key
copies.
+ ///
+ /// Validates eligibility and the stream schema before polling. On error or
+ /// future cancellation, all work and reservations are dropped; no
partially
+ /// prepared object is returned. Concurrent preparation/cache publication
is
+ /// the caller's responsibility. Bounds and membership are prepared once,
but
+ /// each consuming join publishes them into its own dynamic filter.
+ pub async fn prepare_build(
+ &self,
+ input: SendableRecordBatchStream,
+ pool: Arc<dyn MemoryPool>,
+ config: Arc<ConfigOptions>,
+ ) -> Result<Arc<PreparedHashJoinBuild>> {
+ let keys = prepared_key_indices(self)?;
+ let schema = self.left.schema();
+ if input.schema() != schema {
+ return plan_err!(
+ "Prepared hash-join input schema does not match build schema"
+ );
+ }
+ let byte_keys = keys.iter().any(|&key| {
+ matches!(
+ schema.field(key).data_type(),
+ DataType::Utf8 | DataType::FixedSizeBinary(_)
+ )
+ });
+ // Range accumulation and IN-list publication materialize ScalarValue
+ // copies of byte keys. Hash membership borrows the admitted table
instead.
+ let config = if byte_keys {
+ let mut config = config.as_ref().clone();
+ config.optimizer.hash_join_inlist_pushdown_max_size = 0;
+ Arc::new(config)
+ } else {
+ config
+ };
+ let metrics_set = ExecutionPlanMetricsSet::new();
+ let metrics = BuildProbeJoinMetrics::new(0, &metrics_set);
+ let count = MetricBuilder::new(&metrics_set)
+ .counter(ARRAY_MAP_CREATED_COUNT_METRIC_NAME, 0);
+ let reservation =
MemoryConsumer::new("PreparedHashJoinBuild").register(&pool);
+ let data = collect_left_input(
+ self.random_state.random_state().clone(),
+ input,
+ self.on.iter().map(|(left, _)| Arc::clone(left)).collect(),
+ metrics,
+ reservation,
+ false,
+ 0,
+ !byte_keys,
+ config,
+ self.null_equality,
+ None,
+ count,
+ true,
+ )
+ .await?;
+ Ok(Arc::new(PreparedHashJoinBuild {
+ build: data.build,
+ keys,
+ null_equality: self.null_equality,
+ }))
+ }
+}
+
+/// Bound copy allocations, including validity, offsets and alignment. Aliased
+/// columns count separately because concatenation materializes each column.
+pub(super) fn prepared_copy_bytes(batch: &RecordBatch) -> Result<usize> {
+ let rows = batch.num_rows();
+ batch.columns().iter().try_fold(0usize, |total, array| {
+ let values = match array.data_type() {
+ DataType::Utf8 => rows
+ .checked_add(1)
+ .and_then(|len| len.checked_mul(4))
+ .and_then(|offsets|
utf8_value_span(array.as_ref()).checked_add(offsets))
+ // UTF-8 has a third allocation for offsets.
+ .and_then(|bytes| bytes.checked_add(64)),
+ DataType::Null => Some(0),
+ DataType::Boolean => Some(rows.div_ceil(8)),
+ DataType::FixedSizeBinary(width) => usize::try_from(*width)
+ .ok()
+ .and_then(|width| width.checked_mul(rows)),
+ ty => ty
+ .primitive_width()
+ .and_then(|width| width.checked_mul(rows)),
+ };
+ values
+ .and_then(|bytes| bytes.checked_add(rows.div_ceil(8)))
+ .and_then(|bytes| bytes.checked_add(2 * 64))
+ .and_then(|bytes| total.checked_add(bytes))
+ .ok_or_else(|| exec_datafusion_err!("Prepared hash-join copy size
overflow"))
+ })
+}
+
+/// Include values hidden by nulls but exclude bytes outside a sliced array.
+fn utf8_value_span(array: &dyn Array) -> usize {
+ let offsets = array.as_string::<i32>().value_offsets();
+ (offsets[offsets.len() - 1] - offsets[0]) as usize
+}
+
+/// Add `batch`'s byte-column spans to `totals`, one entry per schema column,
before
+/// an eventual single-batch concat. Reject offset overflow before allocating
+/// the value buffer; fixed-width columns leave their totals unchanged.
+pub(super) fn check_byte_concat_sizes(
+ batch: &RecordBatch,
+ totals: &mut [usize],
+) -> Result<()> {
+ for (array, total) in batch.columns().iter().zip(totals) {
+ if matches!(array.data_type(), DataType::Utf8) {
+ *total = total.checked_add(utf8_value_span(array.as_ref()))
+ .filter(|&sum| i32::try_from(sum).is_ok())
+ .ok_or_else(|| exec_datafusion_err!(
+ "Prepared hash-join UTF-8 column exceeds its offset limit;
a compact build is required"
+ ))?;
+ }
+ }
+ Ok(())
+}
+
+impl HashJoinExecBuilder {
+ /// Attach a fully prepared build to a fresh, compatible join execution.
+ ///
+ /// [`Self::build`] validates compatibility. Attaching resets the build
future
+ /// and execution metrics. A previously attached task-local dynamic filter
+ /// keeps its expression handle (also referenced by the probe plan), while
+ /// its build-report accumulator is reset. The caller must provide a fresh
+ /// filter expression/probe plan for each independent task.
+ /// Residual filters also remain consumer-local, so compatible INNER joins
+ /// may use different predicates with the same prepared data.
+ /// The resulting join plan retains the build lease. A caller retaining a
+ /// dynamic-filter expression beyond that plan must retain a prepared-build
+ /// lease alongside it, because membership filters can reference build
data.
+ /// Attach after child-rewriting physical optimizations. The unused left
+ /// subtree is replaced with an empty schema placeholder so plan resets
+ /// preserve it. Probe-only rewrites must retain the attached join's
`left()`.
+ /// Replacing that child or changing to incompatible join keys fails; other
+ /// incompatible join-mode/type changes fail validation in `build`.
Review Comment:
Fourteen lines of prose carrying six distinct constraints. One of them is a
correctness obligation:
> A caller retaining a dynamic-filter expression beyond that plan must
retain a prepared-build lease alongside it, because membership filters can
reference build data.
That is real — `PushdownStrategy::InList` holds an `Arc` to the build key
array and `PushdownStrategy::Map` holds the whole hash map — and it is
currently line 10 of a paragraph. Could this be restructured, e.g. a `# Caller
contract` section with the obligations as bullets, separated from the
descriptive text? As written it is hard for a reader to confirm they have not
missed one.
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared.rs:
##########
@@ -0,0 +1,326 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Explicit immutable build reuse for embedding executors.
+
+use super::*;
+use arrow::array::{Array, AsArray};
+use datafusion_common::exec_datafusion_err;
+use datafusion_execution::memory_pool::MemoryPool;
+
+/// An immutable, fully prepared broadcast build, independent of any probe
task.
+///
+/// Created by [`HashJoinExec::prepare_build`]. The embedding executor owns
cache
+/// identity, admission, single-flight coordination, cancellation and eviction.
+/// This object retains its input buffers and memory reservation until its last
+/// lease is dropped; it never retains an input stream or task context.
Prepared
+/// builds support fixed-width and UTF-8 build columns, with direct-column keys
+/// and non-spilling INNER joins. Residual conditions belong to each consuming
+/// join; null-aware joins remain unsupported.
+///
+/// Hash-join gathers copy supported build columns into output buffers,
+/// including contiguous selections. Output batches can therefore outlive this
+/// object without retaining unaccounted cached payload. View, dictionary and
+/// nested build columns remain unsupported. UTF-8 and fixed-size binary keys
+/// use hash-table membership filters instead of copying range or IN-list
values.
+pub struct PreparedHashJoinBuild {
+ build: Arc<JoinBuildData>,
+ keys: Vec<usize>,
+ null_equality: NullEquality,
+}
+
+impl fmt::Debug for PreparedHashJoinBuild {
+ /// Describe immutable metadata without dumping table contents.
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PreparedHashJoinBuild")
+ .field("schema", &self.build.batch.schema())
+ .field("keys", &self.keys)
+ .field("rows", &self.num_rows())
+ .field("reserved_bytes", &self.reserved_bytes())
+ .finish()
+ }
+}
+
+impl PreparedHashJoinBuild {
+ /// Return the retained build reservation, excluding all per-probe state.
+ pub fn reserved_bytes(&self) -> usize {
+ self.build.reservation.size()
+ }
+
+ /// Return the complete build row count, including duplicate and null keys.
+ pub fn num_rows(&self) -> usize {
+ self.build.batch.num_rows()
+ }
+
+ /// Create independent mutable state for one consuming join.
+ pub(super) fn probe_data(&self, probe_threads: usize) -> JoinLeftData {
+ JoinLeftData {
+ build: Arc::clone(&self.build),
+ null_aware_mark_scope_map: None,
+ null_value_scope_map: None,
+ visited_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ null_indices_bitmap: Mutex::new(BooleanBufferBuilder::new(0)),
+ probe_completion: ProbeCompletion::new(probe_threads),
+ build_side_has_null: false,
+ _probe_reservation: self.build.reservation.new_empty(),
+ }
+ }
+
+ /// Validate the build descriptor and current execution restrictions
without
+ /// consuming input or modifying either plan. Cache identity is
caller-owned.
+ pub(super) fn validate(&self, join: &HashJoinExec) -> Result<()> {
+ let keys = prepared_key_indices(join)?;
+ if join.left.schema() != self.build.batch.schema()
+ || keys != self.keys
+ || join.null_equality != self.null_equality
+ {
+ return plan_err!(
+ "Prepared hash-join build does not match schema, keys or null
equality"
+ );
+ }
+ if let Some(filter) = &join.dynamic_filter {
+ let filter_keys = filter.filter.children();
+ if filter_keys.len() != join.on.len()
+ || filter_keys
+ .iter()
+ .zip(&join.on)
+ .any(|(filter_key, (_, probe_key))| {
+ filter_key.as_ref() != probe_key.as_ref()
+ })
+ {
+ return plan_err!(
+ "Prepared hash-join dynamic filter keys do not match probe
keys"
+ );
+ }
+ }
+ Ok(())
+ }
+}
+
+impl HashJoinExec {
+ /// Prepare one immutable build using an embedding executor's durable pool.
+ ///
+ /// The supplied stream must own its native buffers independently of
producer
+ /// task cleanup. This method consumes only that stream, never `self.left`,
+ /// and reserves retained data, hash buckets and row-index chains against
+ /// `pool`. The caller must keep original producer allocations charged
until
+ /// its stream releases them. `config` controls ordinary perfect-map and
+ /// dynamic-filter choices. UTF-8 and fixed-size binary keys
+ /// retain hash membership only:
+ /// range bounds and IN-list literals would allocate unaccounted key
copies.
+ ///
+ /// Validates eligibility and the stream schema before polling. On error or
+ /// future cancellation, all work and reservations are dropped; no
partially
+ /// prepared object is returned. Concurrent preparation/cache publication
is
+ /// the caller's responsibility. Bounds and membership are prepared once,
but
+ /// each consuming join publishes them into its own dynamic filter.
+ pub async fn prepare_build(
+ &self,
+ input: SendableRecordBatchStream,
+ pool: Arc<dyn MemoryPool>,
+ config: Arc<ConfigOptions>,
+ ) -> Result<Arc<PreparedHashJoinBuild>> {
+ let keys = prepared_key_indices(self)?;
+ let schema = self.left.schema();
+ if input.schema() != schema {
+ return plan_err!(
+ "Prepared hash-join input schema does not match build schema"
+ );
+ }
+ let byte_keys = keys.iter().any(|&key| {
+ matches!(
+ schema.field(key).data_type(),
+ DataType::Utf8 | DataType::FixedSizeBinary(_)
+ )
+ });
+ // Range accumulation and IN-list publication materialize ScalarValue
+ // copies of byte keys. Hash membership borrows the admitted table
instead.
+ let config = if byte_keys {
+ let mut config = config.as_ref().clone();
+ config.optimizer.hash_join_inlist_pushdown_max_size = 0;
+ Arc::new(config)
+ } else {
+ config
+ };
+ let metrics_set = ExecutionPlanMetricsSet::new();
+ let metrics = BuildProbeJoinMetrics::new(0, &metrics_set);
+ let count = MetricBuilder::new(&metrics_set)
+ .counter(ARRAY_MAP_CREATED_COUNT_METRIC_NAME, 0);
+ let reservation =
MemoryConsumer::new("PreparedHashJoinBuild").register(&pool);
+ let data = collect_left_input(
+ self.random_state.random_state().clone(),
+ input,
+ self.on.iter().map(|(left, _)| Arc::clone(left)).collect(),
+ metrics,
+ reservation,
+ false,
+ 0,
+ !byte_keys,
+ config,
+ self.null_equality,
+ None,
+ count,
+ true,
+ )
+ .await?;
+ Ok(Arc::new(PreparedHashJoinBuild {
+ build: data.build,
+ keys,
+ null_equality: self.null_equality,
+ }))
+ }
+}
+
+/// Bound copy allocations, including validity, offsets and alignment. Aliased
+/// columns count separately because concatenation materializes each column.
+pub(super) fn prepared_copy_bytes(batch: &RecordBatch) -> Result<usize> {
+ let rows = batch.num_rows();
+ batch.columns().iter().try_fold(0usize, |total, array| {
+ let values = match array.data_type() {
+ DataType::Utf8 => rows
+ .checked_add(1)
+ .and_then(|len| len.checked_mul(4))
+ .and_then(|offsets|
utf8_value_span(array.as_ref()).checked_add(offsets))
+ // UTF-8 has a third allocation for offsets.
+ .and_then(|bytes| bytes.checked_add(64)),
+ DataType::Null => Some(0),
+ DataType::Boolean => Some(rows.div_ceil(8)),
+ DataType::FixedSizeBinary(width) => usize::try_from(*width)
+ .ok()
+ .and_then(|width| width.checked_mul(rows)),
+ ty => ty
+ .primitive_width()
+ .and_then(|width| width.checked_mul(rows)),
+ };
+ values
+ .and_then(|bytes| bytes.checked_add(rows.div_ceil(8)))
+ .and_then(|bytes| bytes.checked_add(2 * 64))
+ .and_then(|bytes| total.checked_add(bytes))
+ .ok_or_else(|| exec_datafusion_err!("Prepared hash-join copy size
overflow"))
+ })
+}
+
+/// Include values hidden by nulls but exclude bytes outside a sliced array.
+fn utf8_value_span(array: &dyn Array) -> usize {
+ let offsets = array.as_string::<i32>().value_offsets();
+ (offsets[offsets.len() - 1] - offsets[0]) as usize
+}
+
+/// Add `batch`'s byte-column spans to `totals`, one entry per schema column,
before
+/// an eventual single-batch concat. Reject offset overflow before allocating
+/// the value buffer; fixed-width columns leave their totals unchanged.
+pub(super) fn check_byte_concat_sizes(
+ batch: &RecordBatch,
+ totals: &mut [usize],
+) -> Result<()> {
+ for (array, total) in batch.columns().iter().zip(totals) {
+ if matches!(array.data_type(), DataType::Utf8) {
+ *total = total.checked_add(utf8_value_span(array.as_ref()))
+ .filter(|&sum| i32::try_from(sum).is_ok())
+ .ok_or_else(|| exec_datafusion_err!(
+ "Prepared hash-join UTF-8 column exceeds its offset limit;
a compact build is required"
+ ))?;
+ }
+ }
+ Ok(())
+}
+
+impl HashJoinExecBuilder {
+ /// Attach a fully prepared build to a fresh, compatible join execution.
+ ///
+ /// [`Self::build`] validates compatibility. Attaching resets the build
future
+ /// and execution metrics. A previously attached task-local dynamic filter
+ /// keeps its expression handle (also referenced by the probe plan), while
+ /// its build-report accumulator is reset. The caller must provide a fresh
+ /// filter expression/probe plan for each independent task.
+ /// Residual filters also remain consumer-local, so compatible INNER joins
+ /// may use different predicates with the same prepared data.
+ /// The resulting join plan retains the build lease. A caller retaining a
+ /// dynamic-filter expression beyond that plan must retain a prepared-build
+ /// lease alongside it, because membership filters can reference build
data.
+ /// Attach after child-rewriting physical optimizations. The unused left
+ /// subtree is replaced with an empty schema placeholder so plan resets
+ /// preserve it. Probe-only rewrites must retain the attached join's
`left()`.
+ /// Replacing that child or changing to incompatible join keys fails; other
+ /// incompatible join-mode/type changes fail validation in `build`.
+ pub fn with_prepared_build(mut self, prepared: Arc<PreparedHashJoinBuild>)
-> Self {
+ // This removes the ignored child's ordering/equivalences and preserves
+ // its identity through plan resets.
+ self.exec.left =
Arc::new(crate::empty::EmptyExec::new(self.exec.left.schema()));
+ self.reset_prepared_runtime_state();
+ self.exec.prepared_build = Some(prepared);
+ self.preserve_properties = false;
+ self
+ }
+}
+
+/// Validate the narrow prepared-build contract and return ordered build key
+/// indices. Reads schemas/expressions only; invalid columns are rejected
before
+/// any array access. Matching probe types preserve hashing and equality
semantics.
+/// INNER joins need no shared build-match bitmap. Residual predicates are
+/// evaluated by each consumer after hash lookup.
+fn prepared_key_indices(join: &HashJoinExec) -> Result<Vec<usize>> {
+ if join.join_type != JoinType::Inner
+ || join.mode != PartitionMode::CollectLeft
+ || join.null_aware
+ {
+ return plan_err!("Prepared hash-join builds require a CollectLeft
INNER join");
+ }
+ let left = join.left.schema();
+ let right = join.right.schema();
+ if left.fields().iter().any(|field| {
+ let ty = field.data_type();
+ ty.primitive_width().is_none()
+ && !matches!(
+ ty,
+ DataType::Null
+ | DataType::Boolean
+ | DataType::FixedSizeBinary(_)
+ | DataType::Utf8
+ )
+ }) {
+ return plan_err!(
+ "Prepared hash-join builds require fixed-width or UTF-8 build
columns"
+ );
+ }
Review Comment:
This allowlist and the match arms in `prepared_copy_bytes` cover exactly the
same set — I checked each against `DataType::primitive_width`. Nothing enforces
that, though. Adding `LargeUtf8` here without touching `prepared_copy_bytes`
would silently undercount memory rather than fail to compile. A cross-reference
comment in both directions would make the coupling visible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
