comphead commented on code in PR #25491:
URL: https://github.com/apache/datafusion/pull/25491#discussion_r4052388937
##########
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:
A positional `bool` on a function that already carries
`#[expect(clippy::too_many_arguments)]`, now at 13 parameters, with both
ordinary call sites passing a bare `false` (`exec.rs:1750`, `exec.rs:1773`). It
gates seven separate branches in the body (`2945`, `2947`, `2954`, `2989`,
`3038`, `3055`, `3253`).
Once the row-index accounting moves into `new_join_hashmap` and the scans
collapse, what is left is the empty-batch placeholder, the copy admission and
the scratch admission. Passing `Option<&PreparedAdmission>` or a small
`BuildMode` enum instead names the intent, keeps the ordinary call sites from
reading as a mystery `false`, and matches C-BOOL-ARG.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -2866,49 +2933,75 @@ async fn collect_left_input(
let is_phj_candidate = is_perfect_hash_join_candidate(&on_left, &schema)?;
- let initial = BuildSideState::try_new(
+ let mut state = BuildSideState::try_new(
metrics,
reservation,
on_left.clone(),
&schema,
should_compute_dynamic_filters || is_phj_candidate,
)?;
- let state = left_stream
- .try_fold(initial, |mut state, batch| async move {
- // Update accumulators if computing bounds
- if let Some(ref mut accumulators) = state.bounds_accumulators {
- for accumulator in accumulators {
- accumulator.update_batch(&batch)?;
- }
+ let mut concat_values = if prepared {
Review Comment:
The prepared path walks the batches three times and throws away the result
of the first walk.
1. `check_byte_concat_sizes` fills `concat_values` per batch during ingest
(`prepared.rs:226-240`), and the totals are never read. The vector exists only
for the i32 overflow side effect.
2. `exec.rs:2989` re-walks every batch and recomputes the identical
`utf8_value_span` inside `prepared_copy_bytes`. By construction
`sum_over_batches(utf8_value_span(col)) == concat_values[col]`, so the byte
spans are computed twice.
3. `exec.rs:3056` walks the batches a third time for `max(num_rows)`.
All three fold into the ingest loop as a running max and a running
per-column byte total, which yields both the overflow check and the copy bound
from one traversal.
Also worth renaming: `concat_values` holds byte counts, not values.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -2942,9 +3035,61 @@ async fn collect_left_input(
// Use `u32` indices for the JoinHashMap when num_rows ≤ u32::MAX,
otherwise use the
// `u64` indice variant
// Arc is used instead of Box to allow sharing with
SharedBuildAccumulator for hash map pushdown
+ if prepared {
+ // new_join_hashmap accounts for buckets but not its row-index
chain.
Review Comment:
This is fixing a real accounting bug, but only for the prepared path. It
should go into `new_join_hashmap` so every hash join gets it.
`JoinHashMapU32::with_capacity(cap)` allocates `next: vec![0; cap]`
(`joins/join_hash_map.rs:157-162`, and `JoinHashMapU64` at `:237-242`).
`new_join_hashmap` (`exec.rs:2856`) only reserves `estimate_memory_size::<(u32,
u64)>(num_rows, size_of::<JoinHashMapU32>())`, and that helper covers hashbrown
buckets plus the struct itself and nothing else
(`datafusion/common/src/utils/memory.rs:99-120`). There is no later `try_grow`
for the chain: the only other map accounting in this file is
`metrics.build_mem_used.add(array_map.size())` on the `ArrayMap` branch.
So every ordinary `CollectLeft` and `Partitioned` build under-reserves
`num_rows * 4` bytes today, `num_rows * 8` above `u32::MAX`. A 100M-row build
side is 400 MB the pool never sees.
Moving the `try_grow` next to the existing bucket `try_grow` inside
`new_join_hashmap` fixes it for all joins, deletes this `if prepared` block,
and turns the hand-computed `row_indices` term in the tests into an assertion
about shared code rather than about a prepared-only special case.
##########
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> {
Review Comment:
This reimplements `ArrayData::get_slice_memory_size()`, which is already
wrapped for `RecordBatch` in this crate as `pub(crate) trait GetSlicedSize`
(`physical-plan/src/spill/spill_manager.rs:223-230`), documented as exactly
"bytes needed if we materialized exactly this slice into fresh buffers". It is
already used by `sorts/sort.rs:941`, `sorts/multi_level_merge.rs:700`,
`spill/in_progress_spill_file.rs:111` and
`aggregates/grouped_hash_stream.rs:1271`.
Arrow computes the same thing arm for arm (arrow-data 60,
`data.rs:587-656`): `Utf8` gives `(len+1)*4` offsets plus
`offsets[len]-offsets[0]` values plus `ceil(len/8)` nulls, `Boolean` gives
`ceil(len/8)`, fixed-width gives `len*width`, and `NullArray` gives 0 via
`BufferSpec::AlwaysNull`. That is this match, line for line.
Reusing it collapses the body to `batch.get_sliced_size()? + 64 *
buffer_count` for the alignment padding, and drops `utf8_value_span`. It also
decouples the size function from the type allowlist in `prepared_key_indices`:
when that later admits `Binary` or `LargeUtf8`, nothing here needs to change.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -2075,6 +2138,9 @@ impl ExecutionPlan for HashJoinExec {
cache: _,
} = self;
+ if prepared_build.is_some() {
+ return plan_err!("HashJoinExec with a prepared build cannot be
serialized");
Review Comment:
`not_impl_err!` fits better than `plan_err!` here. The plan is valid, it
just cannot be encoded, which is the "missing feature" case the repo reserves
`not_impl_err!` for. `plan_err!` reads as though the user built an invalid plan.
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs:
##########
@@ -0,0 +1,787 @@
+// 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.
+
+//! Regressions for immutable preparation, task isolation and buffer ownership.
+
+use super::*;
+use crate::{
+ common,
+ empty::EmptyExec,
+ filter::FilterExec,
+ memory::MemoryStream,
+ sorts::sort::SortExec,
+ statistics::{ChildStats, StatisticsArgs, StatisticsContext},
+ stream::RecordBatchStreamAdapter,
+ test::TestMemoryExec,
+};
+use arrow::array::{Array, Int64Array, LargeStringArray, StringViewArray};
+use arrow::compute::kernels::sort::SortOptions;
+use arrow_schema::Field;
+use datafusion_common::assert_batches_eq;
+use datafusion_common::utils::memory::get_record_batch_memory_size;
+use datafusion_execution::memory_pool::GreedyMemoryPool;
+use datafusion_expr::Operator;
+use datafusion_physical_expr::expressions::BinaryExpr;
+use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr};
+use futures::{StreamExt, stream};
+
+fn batch(keys: Vec<Option<i64>>) -> RecordBatch {
+ let rows = keys.len();
+ RecordBatch::try_new(
+ Arc::new(Schema::new(vec![
+ Field::new("key", DataType::Int64, true),
+ Field::new("payload", DataType::Int64, false),
+ ])),
+ vec![
+ Arc::new(Int64Array::from(keys)),
+ Arc::new(Int64Array::from_iter_values(
+ (0..rows).map(|i| i as i64 + 10),
+ )),
+ ],
+ )
+ .unwrap()
+}
+
+fn join(build_schema: SchemaRef, probe: RecordBatch) -> Result<HashJoinExec> {
+ let right =
+ TestMemoryExec::try_new_exec(&[vec![probe.clone()]], probe.schema(),
None)?;
+ HashJoinExecBuilder::new(
+ Arc::new(EmptyExec::new(build_schema)),
+ right,
+ vec![(
+ Arc::new(Column::new("key", 0)),
+ Arc::new(Column::new("key", 0)),
+ )],
+ JoinType::Inner,
+ )
+ .with_partition_mode(PartitionMode::CollectLeft)
+ .build()
+}
+
+fn input(batches: Vec<RecordBatch>, schema: SchemaRef) ->
SendableRecordBatchStream {
+ Box::pin(MemoryStream::try_new(batches, schema, None).unwrap())
+}
+
+async fn prepare(
+ join: &HashJoinExec,
+ batches: Vec<RecordBatch>,
+ pool: Arc<dyn MemoryPool>,
+) -> Result<Arc<PreparedHashJoinBuild>> {
+ join.prepare_build(
+ input(batches, join.left().schema()),
+ pool,
+ Arc::new(ConfigOptions::default()),
+ )
+ .await
+}
+
+fn buffer_addresses(arrays: &[ArrayRef]) -> Vec<usize> {
+ arrays
+ .iter()
+ .flat_map(|array| {
+ let data = array.to_data();
+ data.buffers()
+ .iter()
+ .chain(data.nulls().map(|nulls| nulls.inner().inner()))
+ .filter(|buffer| buffer.capacity() != 0)
+ .map(|buffer| buffer.data_ptr().as_ptr() as usize)
+ .collect::<Vec<_>>()
+ })
+ .collect()
+}
+
+async fn run(join: &HashJoinExec) -> Result<Vec<RecordBatch>> {
+ let partitions =
(0..join.properties().output_partitioning().partition_count()).map(
+ |partition| async move {
+ common::collect(join.execute(partition,
Arc::new(TaskContext::default()))?)
+ .await
+ },
+ );
+ Ok(futures::future::try_join_all(partitions)
+ .await?
+ .into_iter()
+ .flatten()
+ .collect())
+}
+
+fn with_probe_filter(join: HashJoinExec) -> Result<HashJoinExec> {
+ let filter = HashJoinExec::create_dynamic_filter(&join.on);
+ let probe = Arc::new(FilterExec::try_new(
+ Arc::clone(&filter) as _,
+ Arc::clone(join.right()),
+ )?);
+ join.builder()
+ .with_new_children(vec![Arc::clone(join.left()), probe])?
+ .with_dynamic_filter(Some(HashJoinExecDynamicFilter {
+ filter,
+ build_accumulator: OnceLock::new(),
+ }))
+ .build()
+}
+
+fn payload_filter(op: Operator) -> JoinFilter {
+ JoinFilter::new(
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("build_payload", 0)),
+ op,
+ Arc::new(Column::new("probe_payload", 1)),
+ )),
+ vec![
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Right,
+ },
+ ],
+ Arc::new(Schema::new(vec![
+ Field::new("build_payload", DataType::Int64, true),
+ Field::new("probe_payload", DataType::Int64, true),
+ ])),
+ )
+}
+
+#[tokio::test]
+async fn prepared_build_placeholder_properties_and_reset() -> Result<()> {
+ let ascending = |name, index| PhysicalSortExpr {
+ expr: Arc::new(Column::new(name, index)),
+ options: SortOptions::default(),
+ };
+ let source_schema =
+ Arc::new(Schema::new(vec![Field::new("key", DataType::Int64, false)]));
+ let placeholder = ProjectionExec::try_new(
+ vec![
+ (
+ Arc::new(Column::new("key", 0)) as PhysicalExprRef,
+ "key".into(),
+ ),
+ (
+ Arc::new(Column::new("key", 0)) as PhysicalExprRef,
+ "payload".into(),
+ ),
+ ],
+ Arc::new(EmptyExec::new(source_schema)),
+ )?;
+ let left: Arc<dyn ExecutionPlan> = Arc::new(SortExec::new(
+ LexOrdering::new(vec![ascending("payload", 1)]).unwrap(),
+ Arc::new(placeholder),
+ ));
+ let build = RecordBatch::try_new(
+ left.schema(),
+ vec![
+ Arc::new(Int64Array::from(vec![1, 1])),
+ Arc::new(Int64Array::from(vec![2, 1])),
+ ],
+ )?;
+ let probe_schema =
+ Arc::new(Schema::new(vec![Field::new("key", DataType::Int64, false)]));
+ let probe = RecordBatch::try_new(
+ Arc::clone(&probe_schema),
+ vec![Arc::new(Int64Array::from(vec![1]))],
+ )?;
+ let right: Arc<dyn ExecutionPlan> = Arc::new(SortExec::new(
+ LexOrdering::new(vec![ascending("key", 0)]).unwrap(),
+ TestMemoryExec::try_new_exec(&[vec![probe]], probe_schema, None)?,
+ ));
+ let base = HashJoinExecBuilder::new(
+ left,
+ right,
+ vec![(
+ Arc::new(Column::new("key", 0)),
+ Arc::new(Column::new("key", 0)),
+ )],
+ JoinType::Inner,
+ )
+ .with_partition_mode(PartitionMode::CollectLeft)
+ .build()?;
+ let build_key: PhysicalExprRef = Arc::new(Column::new("key", 0));
+ let build_payload: PhysicalExprRef = Arc::new(Column::new("payload", 1));
+ let asserted_order = [ascending("key", 2), ascending("payload", 1)];
+ let base_properties = base.properties().equivalence_properties();
+ assert!(
+ base_properties
+ .eq_group()
+ .exprs_equal(&build_key, &build_payload)
+ );
+ assert!(base_properties.ordering_satisfy(asserted_order.clone())?);
+
+ let prepared =
+ prepare(&base, vec![build], Arc::new(GreedyMemoryPool::new(1 <<
20))).await?;
+ let attached = base.builder().with_prepared_build(prepared).build()?;
+ let properties = attached.properties().equivalence_properties();
+ assert!(
+ !properties
+ .eq_group()
+ .exprs_equal(&build_key, &build_payload)
+ );
+ assert!(properties.ordering_satisfy([ascending("key", 2)])?);
+ assert!(!properties.ordering_satisfy(asserted_order)?);
+ assert_eq!(base.child_stats_requests(None)[0], ChildStats::At(None));
+ assert_eq!(attached.child_stats_requests(None)[0], ChildStats::Skip);
+ assert_eq!(attached.child_stats_requests(Some(0))[0], ChildStats::Skip);
+ let stats = StatisticsContext::new().compute(&attached,
&StatisticsArgs::new())?;
+ assert_ne!(stats.num_rows.get_value(), Some(&0));
+ let output = run(&attached).await?;
+ assert_batches_eq!(
+ [
+ "+-----+---------+-----+",
+ "| key | payload | key |",
+ "+-----+---------+-----+",
+ "| 1 | 2 | 1 |",
+ "| 1 | 1 | 1 |",
+ "+-----+---------+-----+",
+ ],
+ &output
+ );
+ // Resetting the unused projection/sort must retain the prepared rows.
+ let reset = crate::execution_plan::reset_plan_states(Arc::new(attached))?;
+ assert_eq!(
+ common::collect(reset.execute(0,
Arc::new(TaskContext::default()))?).await?,
+ output
+ );
+ Ok(())
+}
+
+#[tokio::test]
+async fn prepared_build_projection_pushdown_preserves_rows() -> Result<()> {
+ let build = batch(vec![Some(1)]);
+ let base = join(build.schema(), build.clone())?;
+ let prepared =
+ prepare(&base, vec![build], Arc::new(GreedyMemoryPool::new(1 <<
20))).await?;
+ let task = Arc::new(base.builder().with_prepared_build(prepared).build()?);
+ let projection = ProjectionExec::try_new(
+ vec![
+ (
+ Arc::new(Column::new("key", 0)) as PhysicalExprRef,
+ "build_key".to_owned(),
+ ),
+ (
+ Arc::new(Column::new("key", 2)) as PhysicalExprRef,
+ "probe_key".to_owned(),
+ ),
+ ],
+ Arc::clone(&task) as _,
+ )?;
+ let transformed = task
+ .try_swapping_with_projection(&projection)?
+ .unwrap_or_else(|| Arc::new(projection));
+ let output =
+ common::collect(transformed.execute(0,
Arc::new(TaskContext::default()))?)
+ .await?;
+ assert_batches_eq!(
+ [
+ "+-----------+-----------+",
+ "| build_key | probe_key |",
+ "+-----------+-----------+",
+ "| 1 | 1 |",
+ "+-----------+-----------+",
+ ],
+ &output
+ );
+ Ok(())
+}
+
+#[tokio::test]
+async fn prepared_build_reuses_data_with_independent_dynamic_filters() ->
Result<()> {
+ let build = batch(vec![Some(1), Some(1), None, Some(-3)]);
+ let base = join(build.schema(), batch(vec![Some(1)]))?;
+ let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 20));
+ let prepared = prepare(&base, vec![build.clone()],
Arc::clone(&pool)).await?;
+ let bytes = pool.reserved();
+ assert_eq!(prepared.num_rows(), 4);
+ assert_eq!(bytes, prepared.reserved_bytes());
+ assert!(bytes > 0);
+
+ let mut plans = Vec::new();
+ for (op, probe) in [
+ (Operator::Gt, batch(vec![Some(1), Some(2), None])),
+ (Operator::Lt, batch(vec![Some(1), Some(1)])),
+ ] {
+ let right = TestMemoryExec::try_new_exec(
+ &[
+ vec![probe.slice(0, 1)],
+ vec![probe.slice(1, probe.num_rows() - 1)],
+ ],
+ probe.schema(),
+ None,
+ )?;
+ let task = base
+ .builder()
+ .with_new_children(vec![Arc::clone(base.left()), right])?
+ .with_filter(Some(payload_filter(op)))
+ .build()?;
+ let task = with_probe_filter(task)?
+ .builder()
+ .with_prepared_build(Arc::clone(&prepared))
+ .build()?;
+ plans.push(task);
+ }
+ let cancelled = base
+ .builder()
+ .with_prepared_build(Arc::clone(&prepared))
+ .build()?;
+ drop(cancelled.execute(0, Arc::new(TaskContext::default()))?);
+ drop(cancelled);
+ assert_eq!(pool.reserved(), bytes);
+ let (first, second) = tokio::join!(run(&plans[0]), run(&plans[1]));
+ let output = [first?, second?];
+ for (index, plan) in plans.iter().enumerate() {
+ assert_batches_eq!(
+ [
+ "+-----+---------+-----+---------+",
+ "| key | payload | key | payload |",
+ "+-----+---------+-----+---------+",
+ [
+ "| 1 | 11 | 1 | 10 |",
+ "| 1 | 10 | 1 | 11 |"
+ ][index],
+ "+-----+---------+-----+---------+",
+ ],
+ &output[index]
+ );
+ let metrics = plan.metrics().unwrap();
+ assert_eq!(
+ metrics.sum_by_name("build_input_rows").unwrap().as_usize(),
+ 0
+ );
+ assert_eq!(metrics.output_rows(), Some(1));
+ let dynamic = plan.dynamic_filter.as_ref().unwrap();
+
assert!(futures::poll!(Box::pin(dynamic.filter.wait_complete())).is_ready());
+ }
+ assert_eq!(pool.reserved(), bytes);
+ drop(plans);
+ drop(prepared);
+ assert_eq!(pool.reserved(), 0);
+ Ok(())
+}
+
+#[tokio::test]
+async fn prepared_build_errors_and_cancellation_release_reservations() ->
Result<()> {
+ let build = batch(vec![Some(1), Some(2)]);
+ let base = join(build.schema(), build.clone())?;
+ let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 20));
+ let failing = stream::iter(vec![
+ Ok(build.clone()),
+ datafusion_common::exec_err!("producer failed"),
+ ]);
+ let result = base
+ .prepare_build(
+ Box::pin(RecordBatchStreamAdapter::new(build.schema(), failing)),
+ Arc::clone(&pool),
+ Arc::new(ConfigOptions::default()),
+ )
+ .await;
+ assert!(result.is_err());
+ assert_eq!(pool.reserved(), 0);
+ let pending =
stream::iter(vec![Ok(build.clone())]).chain(stream::pending());
+ let mut preparation = Box::pin(base.prepare_build(
+ Box::pin(RecordBatchStreamAdapter::new(build.schema(), pending)),
+ Arc::clone(&pool),
+ Arc::new(ConfigOptions::default()),
+ ));
+ assert!(futures::poll!(&mut preparation).is_pending());
+ assert!(pool.reserved() > 0);
+ drop(preparation);
+ assert_eq!(pool.reserved(), 0);
+ Ok(())
+}
+
+#[tokio::test]
+async fn prepared_build_accounts_for_hash_map_row_indices() -> Result<()> {
Review Comment:
This test, `prepared_composite_keys_admit_hash_and_null_mask_scratch`
(`:689`), `prepared_null_keys_admit_materialized_validity` (`:752`) and the
admission half of `plain_bytes.rs` are one test with four inputs: compute
`peak`, assert `peak - 1` yields `ResourcesExhausted` with `reserved() == 0`,
assert `peak` succeeds with `reserved() == peak`.
Collapsing them into one table-driven case list, the way
`prepared_build_empty_and_all_null_inputs` (`:652`) already loops, keeps the
coverage and drops roughly 100 lines.
Separately, each of them recomputes the production formula (`retained +
buckets + row_indices + scratch`) in the test body. A test that re-derives the
implementation's arithmetic can only detect that the formula was applied, not
that it is correct. Moving the row-index term into `new_join_hashmap` removes
it from all four.
##########
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(
Review Comment:
`prepare_build` takes `&self` but the prepared data depends only on the left
schema, the left key columns, `null_equality`, `random_state` and `config`.
`self.left` is validated and never executed.
The result is that callers construct a throwaway join just to reach this
method, then `with_prepared_build` replaces `left` with a second `EmptyExec`
(`prepared.rs:263`). Both test helpers do exactly that (`tests.rs:91-105`, and
`proto/tests/cases/plans/joins.rs`).
Not blocking, since the `&self` form gets `validate` compatibility checking
for free. But the required call sequence (build a placeholder join, prepare,
then attach to the real one) is not obvious from the signature and is not in
the rustdoc. Either document it, or expose
`PreparedHashJoinBuild::try_new(schema, on, null_equality, config, pool,
stream)` and leave `validate` as the compatibility gate.
##########
datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs:
##########
@@ -0,0 +1,787 @@
+// 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.
+
+//! Regressions for immutable preparation, task isolation and buffer ownership.
+
+use super::*;
+use crate::{
+ common,
+ empty::EmptyExec,
+ filter::FilterExec,
+ memory::MemoryStream,
+ sorts::sort::SortExec,
+ statistics::{ChildStats, StatisticsArgs, StatisticsContext},
+ stream::RecordBatchStreamAdapter,
+ test::TestMemoryExec,
+};
+use arrow::array::{Array, Int64Array, LargeStringArray, StringViewArray};
+use arrow::compute::kernels::sort::SortOptions;
+use arrow_schema::Field;
+use datafusion_common::assert_batches_eq;
+use datafusion_common::utils::memory::get_record_batch_memory_size;
+use datafusion_execution::memory_pool::GreedyMemoryPool;
+use datafusion_expr::Operator;
+use datafusion_physical_expr::expressions::BinaryExpr;
+use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr};
+use futures::{StreamExt, stream};
+
+fn batch(keys: Vec<Option<i64>>) -> RecordBatch {
+ let rows = keys.len();
+ RecordBatch::try_new(
+ Arc::new(Schema::new(vec![
+ Field::new("key", DataType::Int64, true),
+ Field::new("payload", DataType::Int64, false),
+ ])),
+ vec![
+ Arc::new(Int64Array::from(keys)),
+ Arc::new(Int64Array::from_iter_values(
+ (0..rows).map(|i| i as i64 + 10),
+ )),
+ ],
+ )
+ .unwrap()
+}
+
+fn join(build_schema: SchemaRef, probe: RecordBatch) -> Result<HashJoinExec> {
+ let right =
+ TestMemoryExec::try_new_exec(&[vec![probe.clone()]], probe.schema(),
None)?;
+ HashJoinExecBuilder::new(
+ Arc::new(EmptyExec::new(build_schema)),
+ right,
+ vec![(
+ Arc::new(Column::new("key", 0)),
+ Arc::new(Column::new("key", 0)),
+ )],
+ JoinType::Inner,
+ )
+ .with_partition_mode(PartitionMode::CollectLeft)
+ .build()
+}
+
+fn input(batches: Vec<RecordBatch>, schema: SchemaRef) ->
SendableRecordBatchStream {
+ Box::pin(MemoryStream::try_new(batches, schema, None).unwrap())
+}
+
+async fn prepare(
+ join: &HashJoinExec,
+ batches: Vec<RecordBatch>,
+ pool: Arc<dyn MemoryPool>,
+) -> Result<Arc<PreparedHashJoinBuild>> {
+ join.prepare_build(
+ input(batches, join.left().schema()),
+ pool,
+ Arc::new(ConfigOptions::default()),
+ )
+ .await
+}
+
+fn buffer_addresses(arrays: &[ArrayRef]) -> Vec<usize> {
+ arrays
+ .iter()
+ .flat_map(|array| {
+ let data = array.to_data();
+ data.buffers()
+ .iter()
+ .chain(data.nulls().map(|nulls| nulls.inner().inner()))
+ .filter(|buffer| buffer.capacity() != 0)
+ .map(|buffer| buffer.data_ptr().as_ptr() as usize)
+ .collect::<Vec<_>>()
+ })
+ .collect()
+}
+
+async fn run(join: &HashJoinExec) -> Result<Vec<RecordBatch>> {
+ let partitions =
(0..join.properties().output_partitioning().partition_count()).map(
+ |partition| async move {
+ common::collect(join.execute(partition,
Arc::new(TaskContext::default()))?)
+ .await
+ },
+ );
+ Ok(futures::future::try_join_all(partitions)
+ .await?
+ .into_iter()
+ .flatten()
+ .collect())
+}
+
+fn with_probe_filter(join: HashJoinExec) -> Result<HashJoinExec> {
+ let filter = HashJoinExec::create_dynamic_filter(&join.on);
+ let probe = Arc::new(FilterExec::try_new(
+ Arc::clone(&filter) as _,
+ Arc::clone(join.right()),
+ )?);
+ join.builder()
+ .with_new_children(vec![Arc::clone(join.left()), probe])?
+ .with_dynamic_filter(Some(HashJoinExecDynamicFilter {
+ filter,
+ build_accumulator: OnceLock::new(),
+ }))
+ .build()
+}
+
+fn payload_filter(op: Operator) -> JoinFilter {
+ JoinFilter::new(
+ Arc::new(BinaryExpr::new(
+ Arc::new(Column::new("build_payload", 0)),
+ op,
+ Arc::new(Column::new("probe_payload", 1)),
+ )),
+ vec![
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Left,
+ },
+ ColumnIndex {
+ index: 1,
+ side: JoinSide::Right,
+ },
+ ],
+ Arc::new(Schema::new(vec![
+ Field::new("build_payload", DataType::Int64, true),
+ Field::new("probe_payload", DataType::Int64, true),
+ ])),
+ )
+}
+
+#[tokio::test]
+async fn prepared_build_placeholder_properties_and_reset() -> Result<()> {
+ let ascending = |name, index| PhysicalSortExpr {
+ expr: Arc::new(Column::new(name, index)),
+ options: SortOptions::default(),
+ };
+ let source_schema =
+ Arc::new(Schema::new(vec![Field::new("key", DataType::Int64, false)]));
+ let placeholder = ProjectionExec::try_new(
+ vec![
+ (
+ Arc::new(Column::new("key", 0)) as PhysicalExprRef,
+ "key".into(),
+ ),
+ (
+ Arc::new(Column::new("key", 0)) as PhysicalExprRef,
+ "payload".into(),
+ ),
+ ],
+ Arc::new(EmptyExec::new(source_schema)),
+ )?;
+ let left: Arc<dyn ExecutionPlan> = Arc::new(SortExec::new(
+ LexOrdering::new(vec![ascending("payload", 1)]).unwrap(),
+ Arc::new(placeholder),
+ ));
+ let build = RecordBatch::try_new(
+ left.schema(),
+ vec![
+ Arc::new(Int64Array::from(vec![1, 1])),
+ Arc::new(Int64Array::from(vec![2, 1])),
+ ],
+ )?;
+ let probe_schema =
+ Arc::new(Schema::new(vec![Field::new("key", DataType::Int64, false)]));
+ let probe = RecordBatch::try_new(
+ Arc::clone(&probe_schema),
+ vec![Arc::new(Int64Array::from(vec![1]))],
+ )?;
+ let right: Arc<dyn ExecutionPlan> = Arc::new(SortExec::new(
+ LexOrdering::new(vec![ascending("key", 0)]).unwrap(),
+ TestMemoryExec::try_new_exec(&[vec![probe]], probe_schema, None)?,
+ ));
+ let base = HashJoinExecBuilder::new(
+ left,
+ right,
+ vec![(
+ Arc::new(Column::new("key", 0)),
+ Arc::new(Column::new("key", 0)),
+ )],
+ JoinType::Inner,
+ )
+ .with_partition_mode(PartitionMode::CollectLeft)
+ .build()?;
+ let build_key: PhysicalExprRef = Arc::new(Column::new("key", 0));
+ let build_payload: PhysicalExprRef = Arc::new(Column::new("payload", 1));
+ let asserted_order = [ascending("key", 2), ascending("payload", 1)];
+ let base_properties = base.properties().equivalence_properties();
+ assert!(
+ base_properties
+ .eq_group()
+ .exprs_equal(&build_key, &build_payload)
+ );
+ assert!(base_properties.ordering_satisfy(asserted_order.clone())?);
+
+ let prepared =
+ prepare(&base, vec![build], Arc::new(GreedyMemoryPool::new(1 <<
20))).await?;
+ let attached = base.builder().with_prepared_build(prepared).build()?;
+ let properties = attached.properties().equivalence_properties();
+ assert!(
+ !properties
+ .eq_group()
+ .exprs_equal(&build_key, &build_payload)
+ );
+ assert!(properties.ordering_satisfy([ascending("key", 2)])?);
+ assert!(!properties.ordering_satisfy(asserted_order)?);
+ assert_eq!(base.child_stats_requests(None)[0], ChildStats::At(None));
+ assert_eq!(attached.child_stats_requests(None)[0], ChildStats::Skip);
+ assert_eq!(attached.child_stats_requests(Some(0))[0], ChildStats::Skip);
+ let stats = StatisticsContext::new().compute(&attached,
&StatisticsArgs::new())?;
+ assert_ne!(stats.num_rows.get_value(), Some(&0));
+ let output = run(&attached).await?;
+ assert_batches_eq!(
+ [
+ "+-----+---------+-----+",
+ "| key | payload | key |",
+ "+-----+---------+-----+",
+ "| 1 | 2 | 1 |",
+ "| 1 | 1 | 1 |",
+ "+-----+---------+-----+",
+ ],
+ &output
+ );
+ // Resetting the unused projection/sort must retain the prepared rows.
+ let reset = crate::execution_plan::reset_plan_states(Arc::new(attached))?;
+ assert_eq!(
+ common::collect(reset.execute(0,
Arc::new(TaskContext::default()))?).await?,
+ output
+ );
+ Ok(())
+}
+
+#[tokio::test]
+async fn prepared_build_projection_pushdown_preserves_rows() -> Result<()> {
+ let build = batch(vec![Some(1)]);
+ let base = join(build.schema(), build.clone())?;
+ let prepared =
+ prepare(&base, vec![build], Arc::new(GreedyMemoryPool::new(1 <<
20))).await?;
+ let task = Arc::new(base.builder().with_prepared_build(prepared).build()?);
+ let projection = ProjectionExec::try_new(
+ vec![
+ (
+ Arc::new(Column::new("key", 0)) as PhysicalExprRef,
+ "build_key".to_owned(),
+ ),
+ (
+ Arc::new(Column::new("key", 2)) as PhysicalExprRef,
+ "probe_key".to_owned(),
+ ),
+ ],
+ Arc::clone(&task) as _,
+ )?;
+ let transformed = task
+ .try_swapping_with_projection(&projection)?
+ .unwrap_or_else(|| Arc::new(projection));
+ let output =
+ common::collect(transformed.execute(0,
Arc::new(TaskContext::default()))?)
+ .await?;
+ assert_batches_eq!(
+ [
+ "+-----------+-----------+",
+ "| build_key | probe_key |",
+ "+-----------+-----------+",
+ "| 1 | 1 |",
+ "+-----------+-----------+",
+ ],
+ &output
+ );
+ Ok(())
+}
+
+#[tokio::test]
+async fn prepared_build_reuses_data_with_independent_dynamic_filters() ->
Result<()> {
Review Comment:
Three coverage gaps around the central claim of the PR, that a shared
immutable build yields the same answers as a private one.
**Equivalence.** No test asserts that a prepared build produces the same
output as an ordinary `CollectLeft` build over the same data. Every test here
checks a row count or a hand-written expected batch. Running the same `on`,
filter and probe through both paths and comparing sorted output is the one test
that would catch a mis-shared `JoinLeftData`, and it is cheap.
**Concurrency.** This test uses `tokio::join!` under the default
current-thread runtime, so the two plans never actually run in parallel. The
production scenario is N executor tasks probing one build simultaneously.
`#[tokio::test(flavor = "multi_thread", worker_threads = 4)]` with several
plans would exercise it.
**Fuzz.** `datafusion/core/tests/fuzz_cases/join_fuzz.rs` already exists. A
prepared-versus-normal property over random build and probe data (nulls,
duplicates, skewed keys, empty build, empty probe, varying batch sizes) is
exactly what that infrastructure is for, and joins are on the repo's list of
operators that warrant it.
##########
datafusion/physical-plan/src/joins/hash_join/exec.rs:
##########
@@ -2942,9 +3035,61 @@ async fn collect_left_input(
// Use `u32` indices for the JoinHashMap when num_rows ≤ u32::MAX,
otherwise use the
// `u64` indice variant
// Arc is used instead of Box to allow sharing with
SharedBuildAccumulator for hash map pushdown
+ if prepared {
+ // new_join_hashmap accounts for buckets but not its row-index
chain.
+ let index_width = if num_rows > u32::MAX as usize {
+ size_of::<u64>()
+ } else {
+ size_of::<u32>()
+ };
+ let bytes = num_rows.checked_mul(index_width).ok_or_else(|| {
+ datafusion_common::exec_datafusion_err!(
+ "Prepared hash-join row-index size overflow"
+ )
+ })?;
+ reservation.try_grow(bytes)?;
+ }
let mut hashmap = new_join_hashmap(num_rows, &mut reservation,
&metrics)?;
- let mut hashes_buffer = Vec::new();
+ let scratch_reservation = reservation.new_empty();
+ let mut hashes_buffer = if prepared {
+ let rows =
batches.iter().map(RecordBatch::num_rows).max().unwrap_or(0);
+ // Combining nullable keys can hold an old and a new validity
+ // bitmap at once. NullArray also materializes logical
validity.
+ let mask_count = if null_equality ==
NullEquality::NullEqualsNothing {
Review Comment:
This hardcodes `update_hash`'s internal temporaries from the outside. It
models `matchable_join_keys` (`joins/utils.rs:2272-2290`): among the allowed
types only `NullArray` materializes in `logical_nulls()`, and
`NullBuffer::union_many` holds at most two intermediates while folding. That is
correct today, but nothing connects the two, and `update_hash` can grow a
temporary without anything here failing.
The tests restate the same arithmetic (`tests.rs:764`: `2 *
(1001usize.div_ceil(8) + 64)`, and `tests.rs:796`), so they confirm the formula
was applied rather than that it is right.
Either drop a comment in `matchable_join_keys` pointing back here, or give
up the precision and admit a flat `on_left.len() + 1` masks. The saved bytes do
not look worth the coupling to another function's body.
--
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]
