You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "goldmedal (via GitHub)" <gi...@apache.org> on 2025/03/25 17:47:21 UTC
[PR] Introduce selection vector repartitioning [datafusion]
goldmedal opened a new pull request, #15423:
URL: https://github.com/apache/datafusion/pull/15423
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Closes #15420
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Rich-T-kid (via GitHub)" <gi...@apache.org>.
Rich-T-kid commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-5707880564
run benchmarks
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2764511786
FYI @ctsk this seems potentially related to some of the work you are doing as well
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2848428399
Thank you. This is a foundation for many late materialization optimizations 👏🏼 I have some high-level suggestions, looking forward to hearing your thoughts.
1. Use the term `(selection) bitmap` instead of `selection vector` to avoid confusion. I believe `selection vector` commonly refers (in several recent papers) to vectors of valid indices like `[1, 3, 9, ...]`. See https://db.cs.cmu.edu/papers/2021/ngom-damon2021.pdf
2. Perhaps after this PR—and before implementing any optimization using the selection bitmap—we should first extend `ExecutionPlan` to include related properties like `handles_filtered_input()` and `output_filtered_batches()`, to indicate whether an operator can process batches with metadata filter columns (bitmap/selection vector). We should also add an optimizer pass to validate that if an operator outputs filtered batches, its downstream operator must be able to handle them.
I believe similar optimizations apply to most cardinality-reducing operators, so it’s better to be cautious from the start.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2013706099
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +326,71 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| Arc::clone(f))
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ "selection",
+ DataType::Boolean,
+ false,
+ )));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).into_iter().map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let select_vector = hash_vector
+ .iter()
+ .map(|&hash| hash == partition as u64)
+ .collect::<Vec<_>>();
+ let new_col: ArrayRef =
+ Arc::new(BooleanArray::from(select_vector)) as ArrayRef;
+ let mut columns = batch
+ .columns()
+ .iter()
+ .map(|c| c.clone())
+ .collect::<Vec<_>>();
+ columns.push(new_col);
+
+ let mut options = RecordBatchOptions::new();
+ options = options.with_row_count(Some(batch.num_rows()));
+ let batch = RecordBatch::try_new_with_options(
+ batch.schema(),
Review Comment:
You should probably add the column to the schema (e.g. a `__selection_vector` of boolean type).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "zebsme (via GitHub)" <gi...@apache.org>.
zebsme commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2019843158
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +334,70 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectionVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ hash_buffer.clear();
+ hash_buffer.resize(batch.num_rows(), 0);
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(Arc::clone)
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ SELECTION_FILED_NAME,
+ DataType::Boolean,
+ false,
+ )));
+ let schema = Arc::new(arrow::datatypes::Schema::new(fields));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let selection_vector = Arc::new(
+ hash_vector
+ .iter()
+ .map(|&hash| Some(hash == partition as u64))
+ .collect::<BooleanArray>(),
+ );
+ let mut columns =
+ batch.columns().iter().map(Arc::clone).collect::<Vec<_>>();
Review Comment:
Maybe you can write code as following shows:
```
let mut columns = batch.columns().to_vec();
columns.push(selection_vector);
```
already tested on `many_to_many_hash_select_vector`
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -193,8 +194,16 @@ enum BatchPartitionerState {
num_partitions: usize,
next_idx: usize,
},
+ HashSelectionVector {
+ random_state: ahash::RandomState,
+ exprs: Vec<Arc<dyn PhysicalExpr>>,
+ num_partitions: usize,
+ hash_buffer: Vec<u64>,
+ },
}
+pub static SELECTION_FILED_NAME: &str = "__selection";
Review Comment:
should be SELECTION_FIELD_NAME
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2051721176
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -320,6 +336,68 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectionVector {
+ random_state,
+ exprs,
+ num_partitions,
+ } => {
+ let mut hash_buffer = vec![];
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ hash_buffer.resize(batch.num_rows(), 0);
+ create_hashes(&arrays, random_state, &mut hash_buffer)?;
+
+ let hash_vector = UInt64Array::from(hash_buffer)
+ .unary_mut(|a| a % *num_partitions as u64)
+ .unwrap();
+
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(Arc::clone)
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ SELECTION_FIELD_NAME,
+ DataType::Boolean,
+ false,
+ )));
+ let schema = Arc::new(arrow::datatypes::Schema::new(fields));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let selection_array = arrow_ord::cmp::eq(
+ &hash_vector,
+ &UInt64Array::from(vec![partition as u64; batch.num_rows()]),
+ )
+ .unwrap();
+ let selection_vector = Arc::new(selection_array);
Review Comment:
Is there a more efficient way to do it? 🤔 In the profile result, it spends many CPU time here.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2052394398
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -320,6 +336,68 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectionVector {
+ random_state,
+ exprs,
+ num_partitions,
+ } => {
+ let mut hash_buffer = vec![];
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ hash_buffer.resize(batch.num_rows(), 0);
+ create_hashes(&arrays, random_state, &mut hash_buffer)?;
+
+ let hash_vector = UInt64Array::from(hash_buffer)
+ .unary_mut(|a| a % *num_partitions as u64)
+ .unwrap();
+
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(Arc::clone)
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ SELECTION_FIELD_NAME,
+ DataType::Boolean,
+ false,
+ )));
+ let schema = Arc::new(arrow::datatypes::Schema::new(fields));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let selection_array = arrow_ord::cmp::eq(
+ &hash_vector,
+ &UInt64Array::from(vec![partition as u64; batch.num_rows()]),
+ )
+ .unwrap();
+ let selection_vector = Arc::new(selection_array);
Review Comment:
Thanks @Dandandan
After using `Scalar`, I think the CPU time has decreased.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2764263842
Isn't it always better partitioning on this selection vectors in case of hash-rep 🤔 What is the reason of keeping the old strategy ?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2013703175
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +326,71 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| Arc::clone(f))
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ "selection",
+ DataType::Boolean,
+ false,
+ )));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).into_iter().map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let select_vector = hash_vector
+ .iter()
+ .map(|&hash| hash == partition as u64)
Review Comment:
You can `collect` this directly to a `BooleanArray`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2758775798
https://github.com/apache/datafusion/pull/15339 It looks like the join plan is being changed.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2019935247
##########
datafusion/sqllogictest/test_files/join.slt.part:
##########
@@ -1389,6 +1389,112 @@ physical_plan
14)------------------FilterExec: y@1 = x@0
15)--------------------DataSourceExec: partitions=1, partition_sizes=[1]
+# always use hash repartition
+statement ok
+set datafusion.optimizer.hash_join_single_partition_threshold = 0;
+
+query TT
+explain
+SELECT * FROM
+(SELECT x+1 AS col0, y+1 AS col1 FROM PAIRS WHERE x == y)
+JOIN f
+ON col0 = f.a
+JOIN s
+ON col1 = s.b
+----
+logical_plan
+01)Inner Join: col1 = CAST(s.b AS Int64)
+02)--Inner Join: col0 = CAST(f.a AS Int64)
+03)----Projection: CAST(pairs.x AS Int64) + Int64(1) AS col0, CAST(pairs.y AS Int64) + Int64(1) AS col1
+04)------Filter: pairs.y = pairs.x
+05)--------TableScan: pairs projection=[x, y]
+06)----TableScan: f projection=[a]
+07)--TableScan: s projection=[b]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=8192
+02)--HashJoinExec: mode=Partitioned, join_type=Inner, on=[(col1@1, CAST(s.b AS Int64)@1)], projection=[col0@0, col1@1, a@2, b@3]
+03)----ProjectionExec: expr=[col0@1 as col0, col1@2 as col1, a@0 as a]
+04)------CoalesceBatchesExec: target_batch_size=8192
+05)--------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(CAST(f.a AS Int64)@1, col0@0)], projection=[a@0, col0@2, col1@3]
+06)----------CoalesceBatchesExec: target_batch_size=8192
+07)------------RepartitionExec: partitioning=Hash([CAST(f.a AS Int64)@1], 16), input_partitions=1
+08)--------------ProjectionExec: expr=[a@0 as a, CAST(a@0 AS Int64) as CAST(f.a AS Int64)]
+09)----------------DataSourceExec: partitions=1, partition_sizes=[1]
+10)----------CoalesceBatchesExec: target_batch_size=8192
+11)------------RepartitionExec: partitioning=Hash([col0@0], 16), input_partitions=16
+12)--------------ProjectionExec: expr=[CAST(x@0 AS Int64) + 1 as col0, CAST(y@1 AS Int64) + 1 as col1]
+13)----------------RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1
+14)------------------CoalesceBatchesExec: target_batch_size=8192
+15)--------------------FilterExec: y@1 = x@0
+16)----------------------DataSourceExec: partitions=1, partition_sizes=[1]
+17)----CoalesceBatchesExec: target_batch_size=8192
+18)------RepartitionExec: partitioning=Hash([CAST(s.b AS Int64)@1], 16), input_partitions=1
+19)--------ProjectionExec: expr=[b@0 as b, CAST(b@0 AS Int64) as CAST(s.b AS Int64)]
+20)----------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+set datafusion.optimizer.prefer_hash_selection_vector_partitioning_agg = true;
+
+# TODO: The selection vector partitioning should be used for the hash join.
+# After fix https://github.com/apache/datafusion/issues/15382
Review Comment:
I didn't implement the planner for the hash join to avoid making this PR huge and complex. I think #15382 will implement the required parts.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2760774554
> #15339 It looks like the join plan is being changed.
You should be able to get the test back by also setting `hash_join_single_partition_threshold` to `0` / a low value.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2013701354
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +326,71 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
Review Comment:
You can `collect` this directly to a `BooleanArray`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-3055027587
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.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2852012982
BTW I have not had a chance to look at this PR yet, but the high level idea of pushdown / late materialization is also being explored under the name "dynamic filtering" -- it isn't quite the same mechanism as selection vectors but there are parallels
- #15512
- And parquet filter pushdown is similar: https://github.com/apache/arrow-rs/issues/7456
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2763356410
I'm working on HashAggregate https://github.com/goldmedal/datafusion/pull/3 based on this PR.
I found we shouldn't use only one config, `prefer_hash_selection_vector_partitioning`, to control whether to use the selection vector or not. For testing and benchmark convenience, it's better to split it into two configs, one for `hash join` and another for HashAggregate.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #15423: Introduce selection vector repartitioning
URL: https://github.com/apache/datafusion/pull/15423
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-5707896564
Benchmark for [this request](https://github.com/apache/datafusion/pull/15423#issuecomment-5707880564) failed before finishing (Kubernetes reason: `BackoffLimitExceeded`).
Benchmarks requested: `clickbench_partitioned`
<details><summary>Runner log (last 40 lines)</summary>
```
2026-09-17T03:12:09.523486Z INFO runner starting benchmark runner bench_type=Datafusion, pr_url=https://github.com/apache/datafusion/pull/15423, benchmarks=clickbench_partitioned
2026-09-17T03:12:09.588055Z INFO benchmark_controller::runner::bench_datafusion === Cloning PR branch ===
2026-09-17T03:12:09.588176Z INFO benchmark_controller::runner::shell running command cmd=git, args=["clone", "--depth=200", "https://github.com/apache/datafusion.git", "/workspace/datafusion-branch"], cwd="/"
2026-09-17T03:12:14.597471Z INFO benchmark_controller::runner::shell running command cmd=git, args=["fetch", "origin", "refs/pull/15423/head:feat/selection-vector-partition", "main"], cwd="/workspace/datafusion-branch"
2026-09-17T03:12:24.600269Z INFO benchmark_controller::runner::shell running command cmd=git, args=["checkout", "feat/selection-vector-partition"], cwd="/workspace/datafusion-branch"
2026-09-17T03:12:29.601910Z INFO benchmark_controller::runner::shell running command cmd=git, args=["merge-base", "HEAD", "origin/main"], cwd="/workspace/datafusion-branch"
2026-09-17T03:12:34.604244Z ERROR runner benchmark failed error=git merge-base: git merge-base HEAD origin/main exited with code 1
stdout:
stderr:
```
</details>
<details><summary>Kubernetes message</summary>
```
Job has reached the specified backoff limit
```
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-5707904291
Benchmark for [this request](https://github.com/apache/datafusion/pull/15423#issuecomment-5707880564) failed before finishing (Kubernetes reason: `BackoffLimitExceeded`).
Benchmarks requested: `tpcds`
<details><summary>Runner log (last 40 lines)</summary>
```
2026-09-17T03:13:03.335733Z INFO runner starting benchmark runner bench_type=Datafusion, pr_url=https://github.com/apache/datafusion/pull/15423, benchmarks=tpcds
2026-09-17T03:13:03.415173Z INFO benchmark_controller::runner::bench_datafusion === Cloning PR branch ===
2026-09-17T03:13:03.415310Z INFO benchmark_controller::runner::shell running command cmd=git, args=["clone", "--depth=200", "https://github.com/apache/datafusion.git", "/workspace/datafusion-branch"], cwd="/"
2026-09-17T03:13:08.423940Z INFO benchmark_controller::runner::shell running command cmd=git, args=["fetch", "origin", "refs/pull/15423/head:feat/selection-vector-partition", "main"], cwd="/workspace/datafusion-branch"
2026-09-17T03:13:23.431204Z INFO benchmark_controller::runner::shell running command cmd=git, args=["checkout", "feat/selection-vector-partition"], cwd="/workspace/datafusion-branch"
2026-09-17T03:13:28.433219Z INFO benchmark_controller::runner::shell running command cmd=git, args=["merge-base", "HEAD", "origin/main"], cwd="/workspace/datafusion-branch"
2026-09-17T03:13:33.435599Z ERROR runner benchmark failed error=git merge-base: git merge-base HEAD origin/main exited with code 1
stdout:
stderr:
```
</details>
<details><summary>Kubernetes message</summary>
```
Job has reached the specified backoff limit
```
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2753725939
This starts to look nice @goldmedal
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2763850029
> I'm working on HashAggregate [goldmedal#3](https://github.com/goldmedal/datafusion/pull/3) based on this PR. I found we shouldn't use only one config, `prefer_hash_selection_vector_partitioning`, to control whether to use the selection vector or not. For testing and benchmark convenience, it's better to split it into two configs, one for `hash join` and another for HashAggregate.
Makes sense to me đź‘Ť
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2758703787
After rebasing to the main, the hash join no longer uses `RepartitionExec`. Remove the test by [3a151dd](https://github.com/apache/datafusion/pull/15423/commits/3a151dd042da1c114f0e832123c6fee087eb73d2)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2758966240
also c.c. @zebsme
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2761416899
> You should be able to get the test back by also setting `datafusion.optimizer.hash_join_single_partition_threshold` to `0` / a low value.
Thanks. It works. I also added the test for the normal hash partition for RepartitionExec.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2051728746
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -320,6 +336,68 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectionVector {
+ random_state,
+ exprs,
+ num_partitions,
+ } => {
+ let mut hash_buffer = vec![];
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ hash_buffer.resize(batch.num_rows(), 0);
+ create_hashes(&arrays, random_state, &mut hash_buffer)?;
+
+ let hash_vector = UInt64Array::from(hash_buffer)
+ .unary_mut(|a| a % *num_partitions as u64)
+ .unwrap();
+
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(Arc::clone)
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ SELECTION_FIELD_NAME,
+ DataType::Boolean,
+ false,
+ )));
+ let schema = Arc::new(arrow::datatypes::Schema::new(fields));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let selection_array = arrow_ord::cmp::eq(
+ &hash_vector,
+ &UInt64Array::from(vec![partition as u64; batch.num_rows()]),
+ )
+ .unwrap();
+ let selection_vector = Arc::new(selection_array);
Review Comment:
One way to speed up the comparison here is to use a `Scalar` for the rhs instead of comparing two arrays.
https://docs.rs/arrow/latest/arrow/array/struct.Scalar.html
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2848568844
Thanks @2010YOUY01 for the suggestions.
> 1. Use the term `(selection) bitmap` instead of `selection vector` to avoid confusion. I believe `selection vector` commonly refers (in several recent papers) to vectors of valid indices like `[1, 3, 9, ...]`. See https://db.cs.cmu.edu/papers/2021/ngom-damon2021.pdf
Agreed. The `selection bitmap` is a better name for the field. We can use `array.as_boolean().values().set_indices()` to get the true selection vector from a column. I did it in my implementation for hash-aggregation https://github.com/goldmedal/datafusion/pull/4#discussion_r2051711975
> 2. Perhaps after this PR—and before implementing any optimization using the selection bitmap—we should first extend `ExecutionPlan` to include related properties like `handles_filtered_input()` and `output_filtered_batches()`, to indicate whether an operator can process batches with metadata filter columns (bitmap/selection vector). We should also add an optimizer pass to validate that if an operator outputs filtered batches, its downstream operator must be able to handle them.
>
Providing the API to check the selection bitmap makes sense to me đź‘Ť. I'll consider how to do it.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2764422078
> I think to support this selection vector, the executors need to be updated to interpret an additional metadata column. However, since executors are part of the public interface that some downstream projects might depend on directly, it's better to ensure backward compatibility.
I agree with @2010YOUY01. It would be a breaking change for `RepartitionExec`.
After this PR, we still have the following-up works for #15382 and #15383. Maybe we can use the selection vector by default, but I also prefer to keep the original strategy for backward compatibility for a while.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2766646972
I did the benchmarks for HashAggregate https://github.com/apache/datafusion/issues/15383#issuecomment-2766551662
It seems that HashAggregate is slower in the selection vector mode 🤔
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "zebsme (via GitHub)" <gi...@apache.org>.
zebsme commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2019843158
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +334,70 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectionVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ hash_buffer.clear();
+ hash_buffer.resize(batch.num_rows(), 0);
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(Arc::clone)
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ SELECTION_FILED_NAME,
+ DataType::Boolean,
+ false,
+ )));
+ let schema = Arc::new(arrow::datatypes::Schema::new(fields));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let selection_vector = Arc::new(
+ hash_vector
+ .iter()
+ .map(|&hash| Some(hash == partition as u64))
+ .collect::<BooleanArray>(),
+ );
+ let mut columns =
+ batch.columns().iter().map(Arc::clone).collect::<Vec<_>>();
Review Comment:
Maybe this could be written as follows:
```
let mut columns = batch.columns().to_vec();
columns.push(selection_vector);
```
already tested in `many_to_many_hash_select_vector`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2013701354
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +326,71 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
Review Comment:
You can `collect` this directly to a `BooleanArray`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2764361563
> Isn't it always better partitioning on this selection vectors in case of hash-rep 🤔 What is the reason of keeping the old strategy ?
I think to support this selection vector, the executors need to be updated to interpret an additional metadata column. However, since executors are part of the public interface that some downstream projects might depend on directly, it's better to ensure backward compatibility.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2014723481
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +326,71 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| Arc::clone(f))
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ "selection",
+ DataType::Boolean,
+ false,
+ )));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).into_iter().map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let select_vector = hash_vector
+ .iter()
+ .map(|&hash| hash == partition as u64)
Review Comment:
Nice suggestion! Thanks.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2864278552
I think this is a draft so marking it as such to try and make it clearer what PRs are waiting on review
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "zebsme (via GitHub)" <gi...@apache.org>.
zebsme commented on code in PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#discussion_r2019843158
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
@@ -316,6 +334,70 @@ impl BatchPartitioner {
Ok((partition, batch))
});
+ Box::new(it)
+ }
+ BatchPartitionerState::HashSelectionVector {
+ random_state,
+ exprs,
+ num_partitions,
+ hash_buffer,
+ } => {
+ let timer = self.timer.timer();
+ let arrays = exprs
+ .iter()
+ .map(|expr| expr.evaluate(&batch)?.into_array(batch.num_rows()))
+ .collect::<Result<Vec<_>>>()?;
+
+ hash_buffer.clear();
+ hash_buffer.resize(batch.num_rows(), 0);
+ create_hashes(&arrays, random_state, hash_buffer)?;
+
+ let hash_vector = hash_buffer
+ .iter()
+ .map(|hash| *hash % *num_partitions as u64)
+ .collect::<Vec<_>>();
+ let mut fields = batch
+ .schema()
+ .fields()
+ .iter()
+ .map(Arc::clone)
+ .collect::<Vec<_>>();
+ fields.push(Arc::new(Field::new(
+ SELECTION_FILED_NAME,
+ DataType::Boolean,
+ false,
+ )));
+ let schema = Arc::new(arrow::datatypes::Schema::new(fields));
+ // Finished building index-arrays for output partitions
+ timer.done();
+
+ // Borrowing partitioner timer to prevent moving `self` to closure
+ let partitioner_timer = &self.timer;
+
+ let it = (0..*num_partitions).map(move |partition| {
+ // Tracking time required for repartitioned batches construction
+ let _timer = partitioner_timer.timer();
+ let selection_vector = Arc::new(
+ hash_vector
+ .iter()
+ .map(|&hash| Some(hash == partition as u64))
+ .collect::<BooleanArray>(),
+ );
+ let mut columns =
+ batch.columns().iter().map(Arc::clone).collect::<Vec<_>>();
Review Comment:
Maybe you could write code as following shows:
```
let mut columns = batch.columns().to_vec();
columns.push(selection_vector);
```
already tested on `many_to_many_hash_select_vector`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "goldmedal (via GitHub)" <gi...@apache.org>.
goldmedal commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-2764434126
The CI failure isn't related to this PR. It could be fixed by https://github.com/apache/datafusion/pull/15494
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Introduce selection vector repartitioning [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #15423:
URL: https://github.com/apache/datafusion/pull/15423#issuecomment-5707905929
Benchmark for [this request](https://github.com/apache/datafusion/pull/15423#issuecomment-5707880564) failed before finishing (Kubernetes reason: `BackoffLimitExceeded`).
Benchmarks requested: `tpch`
<details><summary>Runner log (last 40 lines)</summary>
```
2026-09-17T03:13:20.575817Z INFO runner starting benchmark runner bench_type=Datafusion, pr_url=https://github.com/apache/datafusion/pull/15423, benchmarks=tpch
2026-09-17T03:13:20.635718Z INFO benchmark_controller::runner::bench_datafusion === Cloning PR branch ===
2026-09-17T03:13:20.635865Z INFO benchmark_controller::runner::shell running command cmd=git, args=["clone", "--depth=200", "https://github.com/apache/datafusion.git", "/workspace/datafusion-branch"], cwd="/"
2026-09-17T03:13:25.645713Z INFO benchmark_controller::runner::shell running command cmd=git, args=["fetch", "origin", "refs/pull/15423/head:feat/selection-vector-partition", "main"], cwd="/workspace/datafusion-branch"
2026-09-17T03:13:35.648746Z INFO benchmark_controller::runner::shell running command cmd=git, args=["checkout", "feat/selection-vector-partition"], cwd="/workspace/datafusion-branch"
2026-09-17T03:13:40.650540Z INFO benchmark_controller::runner::shell running command cmd=git, args=["merge-base", "HEAD", "origin/main"], cwd="/workspace/datafusion-branch"
2026-09-17T03:13:45.652499Z ERROR runner benchmark failed error=git merge-base: git merge-base HEAD origin/main exited with code 1
stdout:
stderr:
```
</details>
<details><summary>Kubernetes message</summary>
```
Job has reached the specified backoff limit
```
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org