Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-22 Thread via GitHub 


adriangb merged PR #21666:
URL: https://github.com/apache/datafusion/pull/21666


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-22 Thread via GitHub 


adriangb commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4296490447

   Thanks for the review @kosiew !
   
   @gabotechs I did try to use an existing primitive, but I couldn't quite get 
anything to fit the shape we need. If you or anyone else can improve it a 
followup is welcome, but I didn't want to hold up unblocking 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: [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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-22 Thread via GitHub 


kosiew commented on code in PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#discussion_r3122136055


##
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##


Review Comment:
   This should be updated since this PR replaced the Barrier-based coordination.



-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-21 Thread via GitHub 


kosiew commented on code in PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#discussion_r3117320962


##
datafusion/physical-plan/src/joins/hash_join/stream.rs:
##
@@ -421,6 +424,49 @@ impl HashJoinStream {
 }
 }
 
+/// Transitions state after build-side data has been collected, 
automatically
+/// reporting build data to the accumulator when one is present.
+///
+/// If a `build_accumulator` is configured, this method constructs the
+/// appropriate [`PartitionBuildData`], schedules the reporting future, and
+/// returns [`HashJoinStreamState::WaitPartitionBoundsReport`]. Otherwise 
it
+/// delegates to [`Self::state_after_build_ready`].
+fn transition_after_build_collected(
+&mut self,
+left_data: &Arc,
+) -> HashJoinStreamState {
+let Some(build_accumulator) = self.build_accumulator.as_ref() else {
+return Self::state_after_build_ready(self.join_type, 
left_data.as_ref());
+};
+
+let pushdown = left_data.membership().clone();
+let bounds = left_data
+.bounds
+.clone()
+.unwrap_or_else(|| PartitionBounds::new(vec![]));
+
+let build_data = match self.mode {
+PartitionMode::Partitioned => PartitionBuildData::Partitioned {
+partition_id: self.partition,
+pushdown,
+bounds,
+},
+PartitionMode::CollectLeft => {
+PartitionBuildData::CollectLeft { pushdown, bounds }
+}
+PartitionMode::Auto => unreachable!(
+"PartitionMode::Auto should not be present at execution time. 
This is a bug in DataFusion, please report it!"
+),
+};
+
+let acc = Arc::clone(build_accumulator);
+self.build_waiter = Some(OnceFut::new(async move {
+acc.report_build_data(build_data).await
+}));
+self.build_reported = true;

Review Comment:
   I think there is still a race here. `build_reported` gets flipped as soon as 
the `OnceFut` is created, but the future is lazy and only runs once 
`wait_for_partition_bounds_report()` actually polls it.
   
   If a parent drops this stream after `transition_after_build_collected()` 
returns, but before the waiter is ever polled, `Drop` will skip 
`report_canceled_partition()` even though nothing was delivered to the 
coordinator. That feels like it recreates the original hang, just in a narrower 
timing window.
   
   Would it make sense to only mark this as reported after the waiter completes 
successfully? Alternatively, maybe replace the bool with a small lifecycle 
state so `Drop` can still cancel something that was only scheduled but never 
observed.



##
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##
@@ -358,229 +397,283 @@ impl SharedBuildAccumulator {
 /// # Returns
 /// * `Result<()>` - Ok if successful, Err if filter update failed or mode 
mismatch
 pub(crate) async fn report_build_data(&self, data: PartitionBuildData) -> 
Result<()> {
-// Store data in the accumulator
-{
+let finalize_input = {
 let mut guard = self.inner.lock();
+self.store_build_data(&mut guard, data)?;
+self.take_finalize_input_if_ready(&mut guard)
+};
 
-match (data, &mut *guard) {
-// Partitioned mode
-(
-PartitionBuildData::Partitioned {
-partition_id,
-pushdown,
-bounds,
-},
-AccumulatedBuildData::Partitioned { partitions },
-) => {
-partitions[partition_id] = Some(PartitionData { pushdown, 
bounds });
-}
-// CollectLeft mode (store once, deduplicate across partitions)
-(
-PartitionBuildData::CollectLeft { pushdown, bounds },
-AccumulatedBuildData::CollectLeft { data },
-) => {
-// Deduplicate - all partitions report the same data in 
CollectLeft
-if data.is_none() {
-*data = Some(PartitionData { pushdown, bounds });
-}
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+
+self.wait_for_completion().await
+}
+
+pub(crate) fn report_canceled_partition(&self, partition_id: usize) {
+let finalize_input = {
+let mut guard = self.inner.lock();
+self.store_canceled_partition(&mut guard, partition_id);
+self.take_finalize_input_if_ready(&mut guard)
+};
+
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+}
+
+fn store_build_data(
+&self,
+

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-19 Thread via GitHub 


adriangb commented on code in PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#discussion_r3107093513


##
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##
@@ -358,229 +391,283 @@ impl SharedBuildAccumulator {
 /// # Returns
 /// * `Result<()>` - Ok if successful, Err if filter update failed or mode 
mismatch
 pub(crate) async fn report_build_data(&self, data: PartitionBuildData) -> 
Result<()> {
-// Store data in the accumulator
-{
+let finalize_input = {
 let mut guard = self.inner.lock();
+self.store_build_data(&mut guard, data)?;
+self.take_finalize_input_if_ready(&mut guard)
+};
 
-match (data, &mut *guard) {
-// Partitioned mode
-(
-PartitionBuildData::Partitioned {
-partition_id,
-pushdown,
-bounds,
-},
-AccumulatedBuildData::Partitioned { partitions },
-) => {
-partitions[partition_id] = Some(PartitionData { pushdown, 
bounds });
-}
-// CollectLeft mode (store once, deduplicate across partitions)
-(
-PartitionBuildData::CollectLeft { pushdown, bounds },
-AccumulatedBuildData::CollectLeft { data },
-) => {
-// Deduplicate - all partitions report the same data in 
CollectLeft
-if data.is_none() {
-*data = Some(PartitionData { pushdown, bounds });
-}
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+
+self.wait_for_completion().await
+}
+
+pub(crate) fn report_canceled_partition(&self, partition_id: usize) {
+let finalize_input = {
+let mut guard = self.inner.lock();
+self.store_canceled_partition(&mut guard, partition_id);
+self.take_finalize_input_if_ready(&mut guard)
+};
+
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+}
+
+fn store_build_data(
+&self,
+guard: &mut AccumulatorState,
+data: PartitionBuildData,
+) -> Result<()> {
+match (data, &mut guard.data) {
+(
+PartitionBuildData::Partitioned {
+partition_id,
+pushdown,
+bounds,
+},
+AccumulatedBuildData::Partitioned {
+partitions,
+completed_partitions,
+},
+) => {
+if matches!(partitions[partition_id], 
PartitionStatus::Pending) {
+*completed_partitions += 1;
 }
-// Mismatched modes - should never happen
-_ => {
-return datafusion_common::internal_err!(
-"Build data mode mismatch in report_build_data"
-);
+partitions[partition_id] =
+PartitionStatus::Reported(PartitionData { pushdown, bounds 
});
+}
+(
+PartitionBuildData::CollectLeft { pushdown, bounds },
+AccumulatedBuildData::CollectLeft {
+data,
+reported_count,
+..
+},
+) => {
+if matches!(data, PartitionStatus::Pending) {
+*data = PartitionStatus::Reported(PartitionData { 
pushdown, bounds });
 }
+*reported_count += 1;
+}
+_ => {
+return datafusion_common::internal_err!(
+"Build data mode mismatch in report_build_data"
+);
 }
 }
+Ok(())
+}
 
-// Wait for all partitions to report
-if self.barrier.wait().await.is_leader() {
-// All partitions have reported, so we can create and update the 
filter
-let inner = self.inner.lock();
-
-match &*inner {
-// CollectLeft: Simple conjunction of bounds and membership 
check
-AccumulatedBuildData::CollectLeft { data } => {
-if let Some(partition_data) = data {
-// Create membership predicate (InList for small build 
sides, hash lookup otherwise)
-let membership_expr = create_membership_predicate(
-&self.on_right,
-partition_data.pushdown.clone(),
-&HASH_JOIN_SEED,
-self.probe_schema.as_ref(),
-)?;
-
-// Create bounds

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-19 Thread via GitHub 


gabotechs commented on code in PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#discussion_r3107041470


##
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##
@@ -358,229 +391,283 @@ impl SharedBuildAccumulator {
 /// # Returns
 /// * `Result<()>` - Ok if successful, Err if filter update failed or mode 
mismatch
 pub(crate) async fn report_build_data(&self, data: PartitionBuildData) -> 
Result<()> {
-// Store data in the accumulator
-{
+let finalize_input = {
 let mut guard = self.inner.lock();
+self.store_build_data(&mut guard, data)?;
+self.take_finalize_input_if_ready(&mut guard)
+};
 
-match (data, &mut *guard) {
-// Partitioned mode
-(
-PartitionBuildData::Partitioned {
-partition_id,
-pushdown,
-bounds,
-},
-AccumulatedBuildData::Partitioned { partitions },
-) => {
-partitions[partition_id] = Some(PartitionData { pushdown, 
bounds });
-}
-// CollectLeft mode (store once, deduplicate across partitions)
-(
-PartitionBuildData::CollectLeft { pushdown, bounds },
-AccumulatedBuildData::CollectLeft { data },
-) => {
-// Deduplicate - all partitions report the same data in 
CollectLeft
-if data.is_none() {
-*data = Some(PartitionData { pushdown, bounds });
-}
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+
+self.wait_for_completion().await
+}
+
+pub(crate) fn report_canceled_partition(&self, partition_id: usize) {
+let finalize_input = {
+let mut guard = self.inner.lock();
+self.store_canceled_partition(&mut guard, partition_id);
+self.take_finalize_input_if_ready(&mut guard)
+};
+
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+}
+
+fn store_build_data(
+&self,
+guard: &mut AccumulatorState,
+data: PartitionBuildData,
+) -> Result<()> {
+match (data, &mut guard.data) {
+(
+PartitionBuildData::Partitioned {
+partition_id,
+pushdown,
+bounds,
+},
+AccumulatedBuildData::Partitioned {
+partitions,
+completed_partitions,
+},
+) => {
+if matches!(partitions[partition_id], 
PartitionStatus::Pending) {
+*completed_partitions += 1;
 }
-// Mismatched modes - should never happen
-_ => {
-return datafusion_common::internal_err!(
-"Build data mode mismatch in report_build_data"
-);
+partitions[partition_id] =
+PartitionStatus::Reported(PartitionData { pushdown, bounds 
});
+}
+(
+PartitionBuildData::CollectLeft { pushdown, bounds },
+AccumulatedBuildData::CollectLeft {
+data,
+reported_count,
+..
+},
+) => {
+if matches!(data, PartitionStatus::Pending) {
+*data = PartitionStatus::Reported(PartitionData { 
pushdown, bounds });
 }
+*reported_count += 1;
+}
+_ => {
+return datafusion_common::internal_err!(
+"Build data mode mismatch in report_build_data"
+);
 }
 }
+Ok(())
+}
 
-// Wait for all partitions to report
-if self.barrier.wait().await.is_leader() {
-// All partitions have reported, so we can create and update the 
filter
-let inner = self.inner.lock();
-
-match &*inner {
-// CollectLeft: Simple conjunction of bounds and membership 
check
-AccumulatedBuildData::CollectLeft { data } => {
-if let Some(partition_data) = data {
-// Create membership predicate (InList for small build 
sides, hash lookup otherwise)
-let membership_expr = create_membership_predicate(
-&self.on_right,
-partition_data.pushdown.clone(),
-&HASH_JOIN_SEED,
-self.probe_schema.as_ref(),
-)?;
-
-// Create bound

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-19 Thread via GitHub 


adriangb commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4276179492

   Great all we need is review from a committer now 😄 


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-19 Thread via GitHub 


RatulDawar commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4275551909

   > @RatulDawar does 
[5ad96d9](https://github.com/apache/datafusion/commit/5ad96d908292d3c17e74964854f68dd0e3a3ad43)
 help?
   
   Looks great now !
   


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-18 Thread via GitHub 


Omega359 commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4274594865

   I'm not actually a committer @adriangb ... legal at work never got back to 
approving me :( In any case I'm not knowledgeable enough about this area of the 
code to be much help I'm afraid.


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-17 Thread via GitHub 


adriangb commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4269075547

   @Omega359, @gabotechs, @kosiew wondering if one of you as a committer would 
mind reviewing?


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-17 Thread via GitHub 


adriangb commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4268080577

   @RatulDawar does 5ad96d908292d3c17e74964854f68dd0e3a3ad43 help?


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


RatulDawar commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4262811577

   @adriangb  went through the solution and the PR, makes much more sense to 
remove barrier and track states instead here, this also give a much predictable 
behaviour due to the mention of explicit states. 
   
   Just one concern with code here, reaching to a correct state here depends on 
if the person calls report_build_data. Can we have a state transition method so 
that build data is automatically reported and we would just need to call the 
state changes method ike existing state_after_build_ready.
   
   


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangb commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261415170

   @RatulDawar could you let me know what you think of this solution?


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangb commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261412314

   Okay I think this addresses the root cause with no performance regression or 
behavior changes.


-- 
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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangbot commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261342595

   🤖 Benchmark completed (GKE) | 
[trigger](https://github.com/apache/datafusion/pull/21666#issuecomment-4261207193)
   
   **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
   
   CPU Details (lscpu)
   
   ```
   Architecture:aarch64
   CPU op-mode(s):  64-bit
   Byte Order:  Little Endian
   CPU(s):  16
   On-line CPU(s) list: 0-15
   Vendor ID:   ARM
   Model name:  Neoverse-V2
   Model:   1
   Thread(s) per core:  1
   Core(s) per cluster: 16
   Socket(s):   -
   Cluster(s):  1
   Stepping:r0p1
   BogoMIPS:2000.00
   Flags:   fp asimd evtstrm aes pmull sha1 
sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 
sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 
sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm 
bf16 dgh rng bti
   L1d cache:   1 MiB (16 instances)
   L1i cache:   1 MiB (16 instances)
   L2 cache:32 MiB (16 instances)
   L3 cache:80 MiB (1 instance)
   NUMA node(s):1
   NUMA node0 CPU(s):   0-15
   Vulnerability Gather data sampling:  Not affected
   Vulnerability Indirect target selection: Not affected
   Vulnerability Itlb multihit: Not affected
   Vulnerability L1tf:  Not affected
   Vulnerability Mds:   Not affected
   Vulnerability Meltdown:  Not affected
   Vulnerability Mmio stale data:   Not affected
   Vulnerability Reg file data sampling:Not affected
   Vulnerability Retbleed:  Not affected
   Vulnerability Spec rstack overflow:  Not affected
   Vulnerability Spec store bypass: Mitigation; Speculative Store 
Bypass disabled via prctl
   Vulnerability Spectre v1:Mitigation; __user pointer 
sanitization
   Vulnerability Spectre v2:Mitigation; CSV2, BHB
   Vulnerability Srbds: Not affected
   Vulnerability Tsa:   Not affected
   Vulnerability Tsx async abort:   Not affected
   Vulnerability Vmscape:   Not affected
   ```
   
   
   
   Details
   
   
   ```
   Comparing HEAD and codex_hash-join-empty-partition-reporting
   
   Benchmark tpcds_sf1.json
   
   
┏━━━┳━━┳━━━┳━━━┓
   ┃ Query ┃ HEAD ┃ 
codex_hash-join-empty-partition-reporting ┃Change ┃
   
┡━━━╇━━╇━━━╇━━━┩
   │ QQuery 1  │  7.07 / 7.57 ±0.73 / 9.00 ms │   6.97 
/ 7.43 ±0.74 / 8.91 ms │ no change │
   │ QQuery 2  │146.94 / 148.20 ±1.03 / 149.53 ms │ 145.91 / 
146.88 ±0.82 / 147.85 ms │ no change │
   │ QQuery 3  │114.73 / 115.34 ±0.69 / 116.66 ms │ 113.82 / 
114.05 ±0.17 / 114.26 ms │ no change │
   │ QQuery 4  │1446.30 / 1484.55 ±24.78 / 1512.49 ms │ 1428.52 / 
1468.10 ±25.80 / 1498.60 ms │ no change │
   │ QQuery 5  │174.45 / 176.13 ±1.14 / 177.61 ms │ 175.16 / 
177.00 ±1.82 / 179.68 ms │ no change │
   │ QQuery 6  │   892.55 / 911.70 ±20.43 / 948.24 ms │854.85 / 
896.75 ±23.73 / 920.00 ms │ no change │
   │ QQuery 7  │348.04 / 352.12 ±2.92 / 357.04 ms │ 346.46 / 
348.34 ±1.90 / 351.83 ms │ no change │
   │ QQuery 8  │118.14 / 118.71 ±0.56 / 119.74 ms │ 118.42 / 
119.54 ±0.99 / 121.06 ms │ no change │
   │ QQuery 9  │107.55 / 110.79 ±3.53 / 117.29 ms │ 103.18 / 
106.62 ±2.30 / 108.76 ms │ no change │
   │ QQuery 10 │108.45 / 109.88 ±0.84 / 110.81 ms │ 107.59 / 
109.38 ±1.47 / 111.94 ms │ no change │
   │ QQuery 11 │ 1057.30 / 1064.65 ±7.61 / 1078.23 ms │  1040.15 / 
1049.87 ±7.29 / 1061.85 ms │ no change │
   │ QQuery 12 │   47.28 / 48.55 ±0.95 / 49.41 ms │45.67 / 
47.06 ±0.83 / 47.91 ms │ no change │
   │ QQuery 13 │415.70 / 423.17 ±4.91 / 428.64 ms │ 403.64 / 
410.71 ±3.64 / 413.85 ms │ no change │
   │ QQuery 14 │ 1026.57 / 1032.91 ±3.77 / 1037.06 ms │  1007.54 / 
1017.70 ±8.62 / 1030.60 ms │ no change │

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangbot commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261338042

   🤖 Benchmark completed (GKE) | 
[trigger](https://github.com/apache/datafusion/pull/21666#issuecomment-4261207193)
   
   **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
   
   CPU Details (lscpu)
   
   ```
   Architecture:aarch64
   CPU op-mode(s):  64-bit
   Byte Order:  Little Endian
   CPU(s):  16
   On-line CPU(s) list: 0-15
   Vendor ID:   ARM
   Model name:  Neoverse-V2
   Model:   1
   Thread(s) per core:  1
   Core(s) per cluster: 16
   Socket(s):   -
   Cluster(s):  1
   Stepping:r0p1
   BogoMIPS:2000.00
   Flags:   fp asimd evtstrm aes pmull sha1 
sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 
sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 
sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm 
bf16 dgh rng bti
   L1d cache:   1 MiB (16 instances)
   L1i cache:   1 MiB (16 instances)
   L2 cache:32 MiB (16 instances)
   L3 cache:80 MiB (1 instance)
   NUMA node(s):1
   NUMA node0 CPU(s):   0-15
   Vulnerability Gather data sampling:  Not affected
   Vulnerability Indirect target selection: Not affected
   Vulnerability Itlb multihit: Not affected
   Vulnerability L1tf:  Not affected
   Vulnerability Mds:   Not affected
   Vulnerability Meltdown:  Not affected
   Vulnerability Mmio stale data:   Not affected
   Vulnerability Reg file data sampling:Not affected
   Vulnerability Retbleed:  Not affected
   Vulnerability Spec rstack overflow:  Not affected
   Vulnerability Spec store bypass: Mitigation; Speculative Store 
Bypass disabled via prctl
   Vulnerability Spectre v1:Mitigation; __user pointer 
sanitization
   Vulnerability Spectre v2:Mitigation; CSV2, BHB
   Vulnerability Srbds: Not affected
   Vulnerability Tsa:   Not affected
   Vulnerability Tsx async abort:   Not affected
   Vulnerability Vmscape:   Not affected
   ```
   
   
   
   Details
   
   
   ```
   Comparing HEAD and codex_hash-join-empty-partition-reporting
   
   Benchmark clickbench_partitioned.json
   
   
┏━━━┳━━━┳━━━┳━━━┓
   ┃ Query ┃  HEAD ┃ 
codex_hash-join-empty-partition-reporting ┃Change ┃
   
┡━━━╇━━━╇━━━╇━━━┩
   │ QQuery 0  │  1.21 / 4.56 ±6.51 / 17.58 ms │  1.21 / 
4.48 ±6.40 / 17.28 ms │ no change │
   │ QQuery 1  │15.53 / 15.68 ±0.17 / 15.97 ms │14.97 / 
15.50 ±0.28 / 15.72 ms │ no change │
   │ QQuery 2  │44.79 / 45.06 ±0.32 / 45.59 ms │43.95 / 
44.23 ±0.20 / 44.44 ms │ no change │
   │ QQuery 3  │43.28 / 45.82 ±1.56 / 47.87 ms │43.79 / 
44.60 ±1.19 / 46.95 ms │ no change │
   │ QQuery 4  │ 300.16 / 305.00 ±3.93 / 311.02 ms │ 291.01 / 
298.90 ±6.49 / 306.78 ms │ no change │
   │ QQuery 5  │ 350.72 / 358.13 ±4.67 / 365.27 ms │ 344.86 / 
347.96 ±3.76 / 355.04 ms │ no change │
   │ QQuery 6  │  6.02 / 7.78 ±2.92 / 13.60 ms │   5.30 / 
6.27 ±0.60 / 6.94 ms │ +1.24x faster │
   │ QQuery 7  │16.97 / 17.52 ±0.66 / 18.53 ms │17.14 / 
17.38 ±0.18 / 17.64 ms │ no change │
   │ QQuery 8  │ 410.17 / 420.59 ±8.39 / 431.59 ms │ 417.02 / 
428.47 ±6.85 / 435.06 ms │ no change │
   │ QQuery 9  │ 652.25 / 661.89 ±5.75 / 668.45 ms │ 644.64 / 
655.84 ±7.14 / 665.75 ms │ no change │
   │ QQuery 10 │92.52 / 94.00 ±1.40 / 96.34 ms │92.71 / 
95.10 ±2.60 / 99.97 ms │ no change │
   │ QQuery 11 │ 105.99 / 107.24 ±1.18 / 109.13 ms │ 105.92 / 
107.73 ±1.33 / 109.64 ms │ no change │
   │ QQuery 12 │ 340.77 / 348.75 ±5.66 / 356.06 ms │ 350.33 / 
353.92 ±2.95 / 358.53 ms │ no change │
   │ QQuery 13 │455.88 / 472.88 ±14.96 / 498.25 ms │476.85 / 495.23 
±14.45 / 513.87 ms │ no change │
   │ QQuery 14 │ 344.86 / 348.07 ±

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangbot commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261225754

   🤖 Benchmark running (GKE) | 
[trigger](https://github.com/apache/datafusion/pull/21666#issuecomment-4261207193)
   **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux 
bench-c4261207193-1373-pk7t6 6.12.55+ #1 SMP Sun Feb  1 08:59:41 UTC 2026 
aarch64 GNU/Linux`
   CPU Details (lscpu)
   
   ```
   Architecture:aarch64
   CPU op-mode(s):  64-bit
   Byte Order:  Little Endian
   CPU(s):  16
   On-line CPU(s) list: 0-15
   Vendor ID:   ARM
   Model name:  Neoverse-V2
   Model:   1
   Thread(s) per core:  1
   Core(s) per cluster: 16
   Socket(s):   -
   Cluster(s):  1
   Stepping:r0p1
   BogoMIPS:2000.00
   Flags:   fp asimd evtstrm aes pmull sha1 
sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 
sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 
sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm 
bf16 dgh rng bti
   L1d cache:   1 MiB (16 instances)
   L1i cache:   1 MiB (16 instances)
   L2 cache:32 MiB (16 instances)
   L3 cache:80 MiB (1 instance)
   NUMA node(s):1
   NUMA node0 CPU(s):   0-15
   Vulnerability Gather data sampling:  Not affected
   Vulnerability Indirect target selection: Not affected
   Vulnerability Itlb multihit: Not affected
   Vulnerability L1tf:  Not affected
   Vulnerability Mds:   Not affected
   Vulnerability Meltdown:  Not affected
   Vulnerability Mmio stale data:   Not affected
   Vulnerability Reg file data sampling:Not affected
   Vulnerability Retbleed:  Not affected
   Vulnerability Spec rstack overflow:  Not affected
   Vulnerability Spec store bypass: Mitigation; Speculative Store 
Bypass disabled via prctl
   Vulnerability Spectre v1:Mitigation; __user pointer 
sanitization
   Vulnerability Spectre v2:Mitigation; CSV2, BHB
   Vulnerability Srbds: Not affected
   Vulnerability Tsa:   Not affected
   Vulnerability Tsx async abort:   Not affected
   Vulnerability Vmscape:   Not affected
   ```
   
   
   
   Comparing codex/hash-join-empty-partition-reporting 
(7011c5de4dcadd85f7cb5f4b609c9f7c42d0e1df) to 5c653be (merge-base) 
[diff](https://github.com/apache/datafusion/compare/5c653bee5da64003915f6dfeb3da15759b091a8d..7011c5de4dcadd85f7cb5f4b609c9f7c42d0e1df)
 using: clickbench_partitioned
   Results will be posted here when complete
   
   ---
   [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: [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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangbot commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261222935

   🤖 Benchmark running (GKE) | 
[trigger](https://github.com/apache/datafusion/pull/21666#issuecomment-4261207193)
   **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux 
bench-c4261207193-1374-6k5jz 6.12.55+ #1 SMP Sun Feb  1 08:59:41 UTC 2026 
aarch64 GNU/Linux`
   CPU Details (lscpu)
   
   ```
   Architecture:aarch64
   CPU op-mode(s):  64-bit
   Byte Order:  Little Endian
   CPU(s):  16
   On-line CPU(s) list: 0-15
   Vendor ID:   ARM
   Model name:  Neoverse-V2
   Model:   1
   Thread(s) per core:  1
   Core(s) per cluster: 16
   Socket(s):   -
   Cluster(s):  1
   Stepping:r0p1
   BogoMIPS:2000.00
   Flags:   fp asimd evtstrm aes pmull sha1 
sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 
sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 
sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm 
bf16 dgh rng bti
   L1d cache:   1 MiB (16 instances)
   L1i cache:   1 MiB (16 instances)
   L2 cache:32 MiB (16 instances)
   L3 cache:80 MiB (1 instance)
   NUMA node(s):1
   NUMA node0 CPU(s):   0-15
   Vulnerability Gather data sampling:  Not affected
   Vulnerability Indirect target selection: Not affected
   Vulnerability Itlb multihit: Not affected
   Vulnerability L1tf:  Not affected
   Vulnerability Mds:   Not affected
   Vulnerability Meltdown:  Not affected
   Vulnerability Mmio stale data:   Not affected
   Vulnerability Reg file data sampling:Not affected
   Vulnerability Retbleed:  Not affected
   Vulnerability Spec rstack overflow:  Not affected
   Vulnerability Spec store bypass: Mitigation; Speculative Store 
Bypass disabled via prctl
   Vulnerability Spectre v1:Mitigation; __user pointer 
sanitization
   Vulnerability Spectre v2:Mitigation; CSV2, BHB
   Vulnerability Srbds: Not affected
   Vulnerability Tsa:   Not affected
   Vulnerability Tsx async abort:   Not affected
   Vulnerability Vmscape:   Not affected
   ```
   
   
   
   Comparing codex/hash-join-empty-partition-reporting 
(7011c5de4dcadd85f7cb5f4b609c9f7c42d0e1df) to 5c653be (merge-base) 
[diff](https://github.com/apache/datafusion/compare/5c653bee5da64003915f6dfeb3da15759b091a8d..7011c5de4dcadd85f7cb5f4b609c9f7c42d0e1df)
 using: tpcds
   Results will be posted here when complete
   
   ---
   [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: [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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangbot commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261222545

   🤖 Benchmark running (GKE) | 
[trigger](https://github.com/apache/datafusion/pull/21666#issuecomment-4261207193)
   **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux 
bench-c4261207193-1375-rv9zw 6.12.55+ #1 SMP Sun Feb  1 08:59:41 UTC 2026 
aarch64 GNU/Linux`
   CPU Details (lscpu)
   
   ```
   Architecture:aarch64
   CPU op-mode(s):  64-bit
   Byte Order:  Little Endian
   CPU(s):  16
   On-line CPU(s) list: 0-15
   Vendor ID:   ARM
   Model name:  Neoverse-V2
   Model:   1
   Thread(s) per core:  1
   Core(s) per cluster: 16
   Socket(s):   -
   Cluster(s):  1
   Stepping:r0p1
   BogoMIPS:2000.00
   Flags:   fp asimd evtstrm aes pmull sha1 
sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 
sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 
sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm 
bf16 dgh rng bti
   L1d cache:   1 MiB (16 instances)
   L1i cache:   1 MiB (16 instances)
   L2 cache:32 MiB (16 instances)
   L3 cache:80 MiB (1 instance)
   NUMA node(s):1
   NUMA node0 CPU(s):   0-15
   Vulnerability Gather data sampling:  Not affected
   Vulnerability Indirect target selection: Not affected
   Vulnerability Itlb multihit: Not affected
   Vulnerability L1tf:  Not affected
   Vulnerability Mds:   Not affected
   Vulnerability Meltdown:  Not affected
   Vulnerability Mmio stale data:   Not affected
   Vulnerability Reg file data sampling:Not affected
   Vulnerability Retbleed:  Not affected
   Vulnerability Spec rstack overflow:  Not affected
   Vulnerability Spec store bypass: Mitigation; Speculative Store 
Bypass disabled via prctl
   Vulnerability Spectre v1:Mitigation; __user pointer 
sanitization
   Vulnerability Spectre v2:Mitigation; CSV2, BHB
   Vulnerability Srbds: Not affected
   Vulnerability Tsa:   Not affected
   Vulnerability Tsx async abort:   Not affected
   Vulnerability Vmscape:   Not affected
   ```
   
   
   
   Comparing codex/hash-join-empty-partition-reporting 
(7011c5de4dcadd85f7cb5f4b609c9f7c42d0e1df) to 5c653be (merge-base) 
[diff](https://github.com/apache/datafusion/compare/5c653bee5da64003915f6dfeb3da15759b091a8d..7011c5de4dcadd85f7cb5f4b609c9f7c42d0e1df)
 using: tpch
   Results will be posted here when complete
   
   ---
   [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: [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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangb commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261207193

   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: [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]

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


adriangbot commented on PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#issuecomment-4261101975

   🤖 Benchmark completed (GKE) | 
[trigger](https://github.com/apache/datafusion/pull/21666#issuecomment-4260950110)
   
   **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
   
   CPU Details (lscpu)
   
   ```
   Architecture:aarch64
   CPU op-mode(s):  64-bit
   Byte Order:  Little Endian
   CPU(s):  16
   On-line CPU(s) list: 0-15
   Vendor ID:   ARM
   Model name:  Neoverse-V2
   Model:   1
   Thread(s) per core:  1
   Core(s) per cluster: 16
   Socket(s):   -
   Cluster(s):  1
   Stepping:r0p1
   BogoMIPS:2000.00
   Flags:   fp asimd evtstrm aes pmull sha1 
sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 
sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 
sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm 
bf16 dgh rng bti
   L1d cache:   1 MiB (16 instances)
   L1i cache:   1 MiB (16 instances)
   L2 cache:32 MiB (16 instances)
   L3 cache:80 MiB (1 instance)
   NUMA node(s):1
   NUMA node0 CPU(s):   0-15
   Vulnerability Gather data sampling:  Not affected
   Vulnerability Indirect target selection: Not affected
   Vulnerability Itlb multihit: Not affected
   Vulnerability L1tf:  Not affected
   Vulnerability Mds:   Not affected
   Vulnerability Meltdown:  Not affected
   Vulnerability Mmio stale data:   Not affected
   Vulnerability Reg file data sampling:Not affected
   Vulnerability Retbleed:  Not affected
   Vulnerability Spec rstack overflow:  Not affected
   Vulnerability Spec store bypass: Mitigation; Speculative Store 
Bypass disabled via prctl
   Vulnerability Spectre v1:Mitigation; __user pointer 
sanitization
   Vulnerability Spectre v2:Mitigation; CSV2, BHB
   Vulnerability Srbds: Not affected
   Vulnerability Tsa:   Not affected
   Vulnerability Tsx async abort:   Not affected
   Vulnerability Vmscape:   Not affected
   ```
   
   
   
   Details
   
   
   ```
   Comparing HEAD and codex_hash-join-empty-partition-reporting
   
   Benchmark tpcds_sf1.json
   
   
┏━━━┳━━┳━━━┳━━━┓
   ┃ Query ┃ HEAD ┃ 
codex_hash-join-empty-partition-reporting ┃Change ┃
   
┡━━━╇━━╇━━━╇━━━┩
   │ QQuery 1  │  6.92 / 7.40 ±0.82 / 9.04 ms │   7.00 
/ 7.45 ±0.78 / 9.01 ms │ no change │
   │ QQuery 2  │144.54 / 145.39 ±0.79 / 146.62 ms │ 147.27 / 
148.61 ±0.75 / 149.50 ms │ no change │
   │ QQuery 3  │115.14 / 115.85 ±0.68 / 116.87 ms │ 114.50 / 
115.93 ±1.13 / 117.57 ms │ no change │
   │ QQuery 4  │1335.48 / 1382.00 ±36.82 / 1440.86 ms │ 1402.24 / 
1427.26 ±15.58 / 1447.66 ms │ no change │
   │ QQuery 5  │174.17 / 175.13 ±0.83 / 176.53 ms │ 172.07 / 
174.75 ±2.27 / 177.94 ms │ no change │
   │ QQuery 6  │   853.20 / 880.67 ±22.38 / 913.37 ms │863.36 / 
885.51 ±22.93 / 927.68 ms │ no change │
   │ QQuery 7  │344.50 / 348.21 ±4.06 / 355.46 ms │ 346.17 / 
347.70 ±0.89 / 348.94 ms │ no change │
   │ QQuery 8  │115.84 / 116.59 ±0.68 / 117.69 ms │ 116.73 / 
118.22 ±1.22 / 119.85 ms │ no change │
   │ QQuery 9  │102.18 / 105.91 ±2.54 / 108.50 ms │ 100.90 / 
106.70 ±7.24 / 120.08 ms │ no change │
   │ QQuery 10 │107.43 / 108.45 ±0.84 / 109.44 ms │ 106.39 / 
108.37 ±1.16 / 109.59 ms │ no change │
   │ QQuery 11 │969.56 / 981.48 ±7.26 / 988.61 ms │   954.27 / 
979.77 ±14.85 / 1000.34 ms │ no change │
   │ QQuery 12 │   45.11 / 46.95 ±1.34 / 49.09 ms │44.90 / 
45.78 ±0.72 / 46.82 ms │ no change │
   │ QQuery 13 │403.49 / 408.01 ±3.30 / 412.85 ms │ 400.55 / 
404.63 ±2.29 / 407.48 ms │ no change │
   │ QQuery 14 │ 1006.48 / 1017.03 ±8.22 / 1025.61 ms │  1010.30 / 
1015.54 ±6.32 / 1026.77 ms │ no change │

Re: [PR] Handle canceled partitioned hash join dynamic filters lazily [datafusion]

2026-04-16 Thread via GitHub 


Copilot commented on code in PR #21666:
URL: https://github.com/apache/datafusion/pull/21666#discussion_r3094166706


##
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##
@@ -358,229 +391,285 @@ impl SharedBuildAccumulator {
 /// # Returns
 /// * `Result<()>` - Ok if successful, Err if filter update failed or mode 
mismatch
 pub(crate) async fn report_build_data(&self, data: PartitionBuildData) -> 
Result<()> {
-// Store data in the accumulator
-{
+let finalize_input = {

Review Comment:
   The `report_build_data` doc comment still says "have reported (barrier 
wait)", but this method no longer uses `tokio::sync::Barrier` (it uses 
`Notify`/`CompletionState`). Please update the docs to match the current 
synchronization mechanism so the comment doesn’t mislead future changes.



##
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##
@@ -262,13 +262,40 @@ struct PartitionData {
 /// Build-side data organized by partition mode
 enum AccumulatedBuildData {
 Partitioned {
-partitions: Vec>,
+partitions: Vec,
+completed_partitions: usize,
 },
 CollectLeft {
-data: Option,
+data: PartitionStatus,
+reported_count: usize,
+expected_reports: usize,
 },
 }
 
+enum CompletionState {
+Pending,
+Finalizing,
+Ready(std::result::Result<(), String>),

Review Comment:
   `CompletionState::Ready` stores errors as `Result<(), String>`, which forces 
later callers to lose the original `DataFusionError` variant/backtrace/context. 
Consider storing `Result<(), Arc>` (or 
`datafusion_common::SharedResult<()>`) in `CompletionState` instead, so you can 
propagate `DataFusionError::Shared(...)` to all waiters without stringifying.
   ```suggestion
   Ready(std::result::Result<(), Arc>),
   ```



##
datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs:
##
@@ -358,229 +391,285 @@ impl SharedBuildAccumulator {
 /// # Returns
 /// * `Result<()>` - Ok if successful, Err if filter update failed or mode 
mismatch
 pub(crate) async fn report_build_data(&self, data: PartitionBuildData) -> 
Result<()> {
-// Store data in the accumulator
-{
+let finalize_input = {
 let mut guard = self.inner.lock();
+self.store_build_data(&mut guard, data)?;
+self.take_finalize_input_if_ready(&mut guard)
+};
 
-match (data, &mut *guard) {
-// Partitioned mode
-(
-PartitionBuildData::Partitioned {
-partition_id,
-pushdown,
-bounds,
-},
-AccumulatedBuildData::Partitioned { partitions },
-) => {
-partitions[partition_id] = Some(PartitionData { pushdown, 
bounds });
-}
-// CollectLeft mode (store once, deduplicate across partitions)
-(
-PartitionBuildData::CollectLeft { pushdown, bounds },
-AccumulatedBuildData::CollectLeft { data },
-) => {
-// Deduplicate - all partitions report the same data in 
CollectLeft
-if data.is_none() {
-*data = Some(PartitionData { pushdown, bounds });
-}
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+
+self.wait_for_completion().await
+}
+
+pub(crate) fn report_canceled_partition(&self, partition_id: usize) {
+let finalize_input = {
+let mut guard = self.inner.lock();
+self.store_canceled_partition(&mut guard, partition_id);
+self.take_finalize_input_if_ready(&mut guard)
+};
+
+if let Some(finalize_input) = finalize_input {
+self.finish(finalize_input);
+}
+}
+
+fn store_build_data(
+&self,
+guard: &mut AccumulatorState,
+data: PartitionBuildData,
+) -> Result<()> {
+match (data, &mut guard.data) {
+(
+PartitionBuildData::Partitioned {
+partition_id,
+pushdown,
+bounds,
+},
+AccumulatedBuildData::Partitioned {
+partitions,
+completed_partitions,
+},
+) => {
+if matches!(partitions[partition_id], 
PartitionStatus::Pending) {
+*completed_partitions += 1;
 }
-// Mismatched modes - should never happen
-_ => {
-return datafusion_common::internal_err!(
-"Build data mode mismatch in report_build_data"
-);
+partition