You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "geoffreyclaude (via GitHub)" <gi...@apache.org> on 2026/06/17 08:10:29 UTC
[PR] Fix shared TopK early exit with global prefix threshold [datafusion]
geoffreyclaude opened a new pull request, #22991:
URL: https://github.com/apache/datafusion/pull/22991
## Which issue does this PR close?
Closes #22874.
Follow-up to #22852.
## Rationale for this change
PR #22852 fixes the local all-filtered-batch path by calling `attempt_early_completion` before returning. The remaining regression is in partitioned `SortExec`: every local `TopK` shares one `TopKDynamicFilters`. One partition can establish a global threshold before another partition has filled its local heap. The second partition then sees fully rejected batches, but `heap.max()` is still `None` locally, so it cannot prove completion and keeps draining sorted input.
This patch stores the common-prefix row for the shared global threshold, and each local `TopK` checks that shared prefix before falling back to its local heap prefix. It also prevents local partition `TopK`s from marking a shared dynamic filter complete while sibling partitions can still tighten it. Single-partition behavior is unchanged.
## What changes are included in this PR?
- Store a global common-prefix threshold row in `TopKDynamicFilters`.
- Check that global prefix threshold in `attempt_early_completion` before local heap fallback.
- Keep the PR #22852 all-filtered-batch completion call.
- Also check completion on batches that pass the filter but produce zero heap replacements.
- Avoid marking shared partitioned `TopK` filters complete from individual local partitions.
- Add tests for shared-filter completion before local heap fill, equal-prefix non-completion, missing-prefix non-completion, and DESC/null prefix ordering.
## Are these changes tested?
Commands run on the final rebased branch:
- `cargo fmt --all`
- `cargo test -p datafusion-physical-plan topk --lib`
- `cargo test -p datafusion-physical-plan sort --lib`
- `cargo clippy -p datafusion-physical-plan --lib -- -D warnings`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo build --release --bin dfbench`
Benchmark command:
`target/release/dfbench sort-tpch --sorted --limit 10 --iterations 5 --path /tmp/df-topk-bench-data/tpch_sf1 -o /tmp/df-patched-rerun2-top10_sorted_tpch.json`
Key comparisons from the clean reruns:
- Patched vs pre-#15770 v48: Q8 5.60 ms to 3.98 ms, Q9 7.77 ms to 5.89 ms, Q10 10.20 ms to 8.16 ms.
- Patched vs PR #22852 fix: Q8 28.48 ms to 3.98 ms, Q9 39.47 ms to 5.89 ms, Q10 57.70 ms to 8.16 ms.
Debug proof for the bounded-read shape after this patch:
- Q8: `DataSourceExec` output_rows=81.92K, output_batches=10, files_processed=0, bytes_scanned=15.79M.
- Q9: `DataSourceExec` output_rows=81.92K, output_batches=10, files_processed=0, bytes_scanned=20.89M.
- Q10: `DataSourceExec` output_rows=81.92K, output_batches=10, files_processed=0, bytes_scanned=34.69M.
Those Q8/Q9/Q10 debug runs show the scan returns to one batch per partition instead of draining millions of rows across the remaining file ranges.
## Are there any user-facing changes?
No. This is an internal physical execution optimization fix.
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on code in PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#discussion_r3437036972
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -904,19 +904,51 @@ impl SortExec {
self.preserve_partitioning = preserve_partitioning;
Arc::make_mut(&mut self.cache).partitioning =
Self::output_partitioning_helper(&self.input, self.preserve_partitioning);
+ if self.fetch.is_some() {
+ self.rebuild_filter_for_current_partitioning();
+ }
self
}
- /// Add or reset `self.filter` to a new `TopKDynamicFilters`.
+ fn topk_emitter_count(&self) -> usize {
+ self.cache.output_partitioning().partition_count()
+ }
+
+ /// Build a new shared TopK dynamic filter wrapper for this `SortExec`.
fn create_filter(&self) -> Arc<RwLock<TopKDynamicFilters>> {
let children = self
.expr
.iter()
.map(|sort_expr| Arc::clone(&sort_expr.expr))
.collect::<Vec<_>>();
- Arc::new(RwLock::new(TopKDynamicFilters::new(Arc::new(
- DynamicFilterPhysicalExpr::new(children, lit(true)),
- ))))
+ self.create_filter_with_expr(Arc::new(DynamicFilterPhysicalExpr::new(
+ children,
+ lit(true),
+ )))
+ }
+
+ fn create_filter_with_expr(
+ &self,
+ expr: Arc<DynamicFilterPhysicalExpr>,
+ ) -> Arc<RwLock<TopKDynamicFilters>> {
+ Arc::new(RwLock::new(
+ TopKDynamicFilters::new_with_topk_emitter_count(
+ expr,
+ self.topk_emitter_count(),
+ ),
+ ))
+ }
+
+ /// Rebuild the shared TopK filter wrapper after output partitioning changes.
+ ///
+ /// The dynamic filter expression is preserved, but wrapper state such as the
+ /// shared threshold and remaining emitter count is reset for the new
+ /// partitioning.
+ fn rebuild_filter_for_current_partitioning(&mut self) {
Review Comment:
Yes, exactly: this is plan-construction / plan-rewrite state, not runtime execution state. Rebuilding the wrapper is intentional when the SortExec properties change, because the wrapper owns plan-derived state such as the shared threshold and expected emitter count. We should not carry an active runtime threshold across that kind of rebuild.
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4752758447
๐ค Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22991#issuecomment-4752736859)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4752736859-596-p2hmr 6.12.68+ #1 SMP Sat May 2 07:49:07 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
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 fix/topk-global-prefix-early-exit (560cc5b73e89935173b881ac027b988d2476f01c) to 88fb94d (merge-base) [diff](https://github.com/apache/datafusion/compare/88fb94d88251a512694cc8b37ee09f53d6916160..560cc5b73e89935173b881ac027b988d2476f01c) using: topk_sorted_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: 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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on code in PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#discussion_r3437038200
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -2764,6 +2799,71 @@ mod tests {
Ok(())
}
+ async fn emit_sort_partition(
+ sort: &Arc<SortExec>,
+ partition: usize,
+ task_ctx: Arc<TaskContext>,
+ ) -> Result<()> {
+ let _batches: Vec<RecordBatch> =
+ sort.execute(partition, task_ctx)?.try_collect().await?;
+ Ok(())
+ }
+
+ fn assert_filter_still_waiting(filter: &Arc<DynamicFilterPhysicalExpr>) {
+ let dynamic_filter_expr: Arc<dyn PhysicalExpr> =
+ Arc::<DynamicFilterPhysicalExpr>::clone(filter);
+ assert!(
+ matches!(
+ DynamicFilterTracking::classify(&dynamic_filter_expr),
+ DynamicFilterTracking::Watching(_)
+ ),
+ "the shared filter should remain watchable until every partition emits"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_preserved_topk_filter_waits_for_all_sort_partitions() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+ let partitions = vec![
+ vec![RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![3, 1, 2]))],
+ )?],
+ vec![RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![6, 4, 5]))],
+ )?],
+ ];
+ let input = TestMemoryExec::try_new_exec(&partitions, Arc::clone(&schema), None)?;
+ let sort = SortExec::new(
+ [PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0)))].into(),
+ input,
+ )
+ // `with_fetch` creates the TopK filter; preserving partitioning after
+ // that must rebuild it with one emitter per output partition.
+ .with_fetch(Some(2))
Review Comment:
Good catch. I changed `with_fetch` in 4c32f8a0c so an existing dynamic filter keeps its expression, but the `TopKDynamicFilters` wrapper is rebuilt against the current output partitioning instead of being cloned with possibly stale emitter state. I also added `test_with_fetch_rebuilds_existing_topk_filter` for the ordering where the filter exists before fetch is set.
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4752736859
run benchmark topk_sorted_tpch
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4730036450
๐ค Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22991#issuecomment-4730000666)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4730000666-584-rlxkd 6.12.68+ #1 SMP Sat May 2 07:49:07 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
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 fix/topk-global-prefix-early-exit (96ba61cb10a1fcc368235b9063a3d7e88b8b293a) to 7bb6e15 (merge-base) [diff](https://github.com/apache/datafusion/compare/7bb6e152bee1df3cd8215176742bfc1cdf300e52..96ba61cb10a1fcc368235b9063a3d7e88b8b293a) using: partitioned_topk
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: 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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4732917306
@kosiew for info, since you reviewed the [previous PR](https://github.com/apache/datafusion/pull/22852) this one builds upon
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4730000666
run benchmark partitioned_topk
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4730589198
๐ค Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22991#issuecomment-4730553583)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4730553583-585-mm5xj 6.12.68+ #1 SMP Sat May 2 07:49:07 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
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 fix/topk-global-prefix-early-exit (96ba61cb10a1fcc368235b9063a3d7e88b8b293a) to 7bb6e15 (merge-base) [diff](https://github.com/apache/datafusion/compare/7bb6e152bee1df3cd8215176742bfc1cdf300e52..96ba61cb10a1fcc368235b9063a3d7e88b8b293a) using: topk_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: 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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#discussion_r3433076781
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -1101,86 +1208,111 @@ mod tests {
assert_eq!(record_batch_store.batches_size, 0);
}
- /// Builds an `(a Int32, b Float64)` schema and a `TopK` with full sort
- /// `(a ASC, b ASC)`, input prefix `[a]`, `k = 3`, `batch_size = 2`. Used by
- /// the prefix-completion tests below to keep their per-scenario logic in focus.
- fn build_ab_prefix_topk() -> Result<(Arc<Schema>, TopK)> {
- let schema = Arc::new(Schema::new(vec![
- Field::new("a", DataType::Int32, false),
+ fn make_ab_schema() -> SchemaRef {
+ make_ab_schema_with_nullable_a(false)
+ }
+
+ fn make_ab_schema_with_nullable_a(a_nullable: bool) -> SchemaRef {
+ Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, a_nullable),
Field::new("b", DataType::Float64, false),
- ]));
+ ]))
+ }
+
+ fn make_topk_filter() -> Arc<RwLock<TopKDynamicFilters>> {
Review Comment:
Small cleanup suggestion: `make_topk_filter` and `make_shared_topk_filter` currently build the same wrapper. It might be a little easier to maintain if the single-emitter helper delegates to the shared helper so any future fixture changes only need to be made in one place.
```rust
fn make_topk_filter() -> Arc<RwLock<TopKDynamicFilters>> {
make_shared_topk_filter(1)
}
```
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on code in PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#discussion_r3437038200
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -2764,6 +2799,71 @@ mod tests {
Ok(())
}
+ async fn emit_sort_partition(
+ sort: &Arc<SortExec>,
+ partition: usize,
+ task_ctx: Arc<TaskContext>,
+ ) -> Result<()> {
+ let _batches: Vec<RecordBatch> =
+ sort.execute(partition, task_ctx)?.try_collect().await?;
+ Ok(())
+ }
+
+ fn assert_filter_still_waiting(filter: &Arc<DynamicFilterPhysicalExpr>) {
+ let dynamic_filter_expr: Arc<dyn PhysicalExpr> =
+ Arc::<DynamicFilterPhysicalExpr>::clone(filter);
+ assert!(
+ matches!(
+ DynamicFilterTracking::classify(&dynamic_filter_expr),
+ DynamicFilterTracking::Watching(_)
+ ),
+ "the shared filter should remain watchable until every partition emits"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_preserved_topk_filter_waits_for_all_sort_partitions() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+ let partitions = vec![
+ vec![RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![3, 1, 2]))],
+ )?],
+ vec![RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![6, 4, 5]))],
+ )?],
+ ];
+ let input = TestMemoryExec::try_new_exec(&partitions, Arc::clone(&schema), None)?;
+ let sort = SortExec::new(
+ [PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0)))].into(),
+ input,
+ )
+ // `with_fetch` creates the TopK filter; preserving partitioning after
+ // that must rebuild it with one emitter per output partition.
+ .with_fetch(Some(2))
Review Comment:
Good catch. I changed with_fetch in 4c32f8a0c so an existing dynamic filter keeps its expression, but the TopKDynamicFilters wrapper is rebuilt against the current output partitioning instead of being cloned with possibly stale emitter state. I also added test_with_fetch_rebuilds_existing_topk_filter for the ordering where the filter exists before fetch is set.
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4730246371
Benchmark for [this request](https://github.com/apache/datafusion/pull/22991#issuecomment-4730000666) failed.
Last 20 lines of output:
<details><summary>Click to expand</summary>
```
# Micro-Benchmarks (specific operators and features)
cancellation: How long cancelling a query takes
nlj: Benchmark for simple nested loop joins, testing various join scenarios
hj: Benchmark for simple hash joins, testing various join scenarios
smj: Benchmark for simple sort merge joins, testing various join scenarios
dict: Benchmark for dictionary-encoded group-by scenarios
compile_profile: Compile and execute TPC-H across selected Cargo profiles, reporting timing and binary size
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
Supported Configuration (Environment Variables)
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
DATA_DIR directory to store datasets
CARGO_COMMAND command that runs the benchmark binary
DATAFUSION_DIR directory to use (default /workspace/datafusion-base)
RESULTS_NAME folder where the benchmark files are stored
PREFER_HASH_JOIN Prefer hash join algorithm (default true)
SIMULATE_LATENCY Simulate object store latency to mimic S3 (default false)
DATAFUSION_* Set the given datafusion configuration
```
</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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4752858691
๐ค Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22991#issuecomment-4752736859)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
<details><summary>CPU Details (lscpu)</summary>
```
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>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and fix_topk-global-prefix-early-exit
--------------------
Benchmark run_topk_sorted_tpch.json
--------------------
โโโโโโโโโณโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ fix_topk-global-prefix-early-exit โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Q1 โ 2.12 / 2.98 ยฑ0.76 / 4.21 ms โ 2.14 / 2.83 ยฑ0.83 / 4.31 ms โ +1.06x faster โ
โ Q2 โ 2.27 / 2.68 ยฑ0.55 / 3.78 ms โ 2.33 / 2.43 ยฑ0.10 / 2.60 ms โ +1.11x faster โ
โ Q3 โ 31.59 / 32.71 ยฑ1.52 / 35.61 ms โ 31.76 / 32.87 ยฑ1.86 / 36.55 ms โ no change โ
โ Q4 โ 11.43 / 11.58 ยฑ0.13 / 11.81 ms โ 2.56 / 2.67 ยฑ0.14 / 2.95 ms โ +4.33x faster โ
โ Q5 โ 9.52 / 9.88 ยฑ0.40 / 10.61 ms โ 9.99 / 10.18 ยฑ0.21 / 10.57 ms โ no change โ
โ Q6 โ 16.67 / 16.81 ยฑ0.09 / 16.93 ms โ 17.15 / 17.33 ยฑ0.23 / 17.77 ms โ no change โ
โ Q7 โ 37.14 / 37.97 ยฑ0.70 / 38.89 ms โ 37.07 / 37.65 ยฑ0.69 / 38.92 ms โ no change โ
โ Q8 โ 3.35 / 11.89 ยฑ10.47 / 27.14 ms โ 3.33 / 4.55 ยฑ2.34 / 9.24 ms โ +2.61x faster โ
โ Q9 โ 4.73 / 4.77 ยฑ0.03 / 4.82 ms โ 4.75 / 4.78 ยฑ0.02 / 4.81 ms โ no change โ
โ Q10 โ 6.85 / 32.12 ยฑ20.97 / 53.02 ms โ 6.88 / 6.90 ยฑ0.01 / 6.91 ms โ +4.66x faster โ
โ Q11 โ 3.83 / 4.01 ยฑ0.30 / 4.61 ms โ 3.80 / 4.38 ยฑ0.87 / 6.10 ms โ 1.09x slower โ
โโโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 167.41ms โ
โ Total Time (fix_topk-global-prefix-early-exit) โ 126.56ms โ
โ Average Time (HEAD) โ 15.22ms โ
โ Average Time (fix_topk-global-prefix-early-exit) โ 11.51ms โ
โ Queries Faster โ 5 โ
โ Queries Slower โ 1 โ
โ Queries with No Change โ 5 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**topk_sorted_tpch โ base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 640.4 MiB |
| Avg memory | 106.8 MiB |
| CPU user | 0.0s |
| CPU sys | 0.0s |
| Peak spill | 0 B |
**topk_sorted_tpch โ branch**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 224.0 KiB |
| Avg memory | 37.3 KiB |
| CPU user | 0.0s |
| CPU sys | 0.0s |
| Peak spill | 0 B |
</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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "LiaCastaneda (via GitHub)" <gi...@apache.org>.
LiaCastaneda commented on code in PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#discussion_r3435811168
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -904,19 +904,51 @@ impl SortExec {
self.preserve_partitioning = preserve_partitioning;
Arc::make_mut(&mut self.cache).partitioning =
Self::output_partitioning_helper(&self.input, self.preserve_partitioning);
+ if self.fetch.is_some() {
+ self.rebuild_filter_for_current_partitioning();
+ }
self
}
- /// Add or reset `self.filter` to a new `TopKDynamicFilters`.
+ fn topk_emitter_count(&self) -> usize {
+ self.cache.output_partitioning().partition_count()
+ }
+
+ /// Build a new shared TopK dynamic filter wrapper for this `SortExec`.
fn create_filter(&self) -> Arc<RwLock<TopKDynamicFilters>> {
let children = self
.expr
.iter()
.map(|sort_expr| Arc::clone(&sort_expr.expr))
.collect::<Vec<_>>();
- Arc::new(RwLock::new(TopKDynamicFilters::new(Arc::new(
- DynamicFilterPhysicalExpr::new(children, lit(true)),
- ))))
+ self.create_filter_with_expr(Arc::new(DynamicFilterPhysicalExpr::new(
+ children,
+ lit(true),
+ )))
+ }
+
+ fn create_filter_with_expr(
+ &self,
+ expr: Arc<DynamicFilterPhysicalExpr>,
+ ) -> Arc<RwLock<TopKDynamicFilters>> {
+ Arc::new(RwLock::new(
+ TopKDynamicFilters::new_with_topk_emitter_count(
+ expr,
+ self.topk_emitter_count(),
+ ),
+ ))
+ }
+
+ /// Rebuild the shared TopK filter wrapper after output partitioning changes.
+ ///
+ /// The dynamic filter expression is preserved, but wrapper state such as the
+ /// shared threshold and remaining emitter count is reset for the new
+ /// partitioning.
+ fn rebuild_filter_for_current_partitioning(&mut self) {
Review Comment:
nit: Is it harmless to reconstruct the filter from scratch? I guess this isn't an issue if it's only called at planning time, but if this function were called during execution we could end up discarding an active `shared_threshold`.
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -2764,6 +2799,71 @@ mod tests {
Ok(())
}
+ async fn emit_sort_partition(
+ sort: &Arc<SortExec>,
+ partition: usize,
+ task_ctx: Arc<TaskContext>,
+ ) -> Result<()> {
+ let _batches: Vec<RecordBatch> =
+ sort.execute(partition, task_ctx)?.try_collect().await?;
+ Ok(())
+ }
+
+ fn assert_filter_still_waiting(filter: &Arc<DynamicFilterPhysicalExpr>) {
+ let dynamic_filter_expr: Arc<dyn PhysicalExpr> =
+ Arc::<DynamicFilterPhysicalExpr>::clone(filter);
+ assert!(
+ matches!(
+ DynamicFilterTracking::classify(&dynamic_filter_expr),
+ DynamicFilterTracking::Watching(_)
+ ),
+ "the shared filter should remain watchable until every partition emits"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_preserved_topk_filter_waits_for_all_sort_partitions() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+ let partitions = vec![
+ vec![RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![3, 1, 2]))],
+ )?],
+ vec![RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from(vec![6, 4, 5]))],
+ )?],
+ ];
+ let input = TestMemoryExec::try_new_exec(&partitions, Arc::clone(&schema), None)?;
+ let sort = SortExec::new(
+ [PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0)))].into(),
+ input,
+ )
+ // `with_fetch` creates the TopK filter; preserving partitioning after
+ // that must rebuild it with one emitter per output partition.
+ .with_fetch(Some(2))
Review Comment:
something codex highlighted was that in `with_fetch`, if a dynamic filter already exists it's cloned and reused, so it keeps the emitter count it was built with, but not necessarily the current partition count. I'm not sure if it's actually possible that there the emitter count and partition count can differ, but it seems that every other place that rebuilds the sort calls `rebuild_filter_for_current_partitioning`, should we call it there 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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on code in PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#discussion_r3437036972
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -904,19 +904,51 @@ impl SortExec {
self.preserve_partitioning = preserve_partitioning;
Arc::make_mut(&mut self.cache).partitioning =
Self::output_partitioning_helper(&self.input, self.preserve_partitioning);
+ if self.fetch.is_some() {
+ self.rebuild_filter_for_current_partitioning();
+ }
self
}
- /// Add or reset `self.filter` to a new `TopKDynamicFilters`.
+ fn topk_emitter_count(&self) -> usize {
+ self.cache.output_partitioning().partition_count()
+ }
+
+ /// Build a new shared TopK dynamic filter wrapper for this `SortExec`.
fn create_filter(&self) -> Arc<RwLock<TopKDynamicFilters>> {
let children = self
.expr
.iter()
.map(|sort_expr| Arc::clone(&sort_expr.expr))
.collect::<Vec<_>>();
- Arc::new(RwLock::new(TopKDynamicFilters::new(Arc::new(
- DynamicFilterPhysicalExpr::new(children, lit(true)),
- ))))
+ self.create_filter_with_expr(Arc::new(DynamicFilterPhysicalExpr::new(
+ children,
+ lit(true),
+ )))
+ }
+
+ fn create_filter_with_expr(
+ &self,
+ expr: Arc<DynamicFilterPhysicalExpr>,
+ ) -> Arc<RwLock<TopKDynamicFilters>> {
+ Arc::new(RwLock::new(
+ TopKDynamicFilters::new_with_topk_emitter_count(
+ expr,
+ self.topk_emitter_count(),
+ ),
+ ))
+ }
+
+ /// Rebuild the shared TopK filter wrapper after output partitioning changes.
+ ///
+ /// The dynamic filter expression is preserved, but wrapper state such as the
+ /// shared threshold and remaining emitter count is reset for the new
+ /// partitioning.
+ fn rebuild_filter_for_current_partitioning(&mut self) {
Review Comment:
Yes, exactly: this is plan-construction / plan-rewrite state, not runtime execution state. Rebuilding the wrapper is intentional when the `SortExec` properties change, because the wrapper owns plan-derived state such as the shared threshold and expected emitter count. We should not carry an active runtime threshold across that kind of rebuild.
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4730704985
๐ค Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22991#issuecomment-4730553583)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
<details><summary>CPU Details (lscpu)</summary>
```
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>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and fix_topk-global-prefix-early-exit
--------------------
Benchmark run_topk_tpch.json
--------------------
โโโโโโโโโณโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโโโโโโ
โ Query โ HEAD โ fix_topk-global-prefix-early-exit โ Change โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Q1 โ 2.25 / 2.76 ยฑ0.73 / 4.19 ms โ 2.03 / 2.55 ยฑ0.85 / 4.24 ms โ +1.08x faster โ
โ Q2 โ 10.79 / 11.44 ยฑ0.70 / 12.69 ms โ 10.52 / 10.87 ยฑ0.49 / 11.83 ms โ no change โ
โ Q3 โ 31.99 / 32.86 ยฑ0.62 / 33.75 ms โ 31.91 / 32.59 ยฑ0.79 / 34.10 ms โ no change โ
โ Q4 โ 11.65 / 12.59 ยฑ0.78 / 13.54 ms โ 11.55 / 11.64 ยฑ0.09 / 11.77 ms โ +1.08x faster โ
โ Q5 โ 9.25 / 9.37 ยฑ0.08 / 9.47 ms โ 9.71 / 9.83 ยฑ0.11 / 10.01 ms โ no change โ
โ Q6 โ 16.55 / 16.61 ยฑ0.06 / 16.69 ms โ 16.89 / 17.08 ยฑ0.21 / 17.48 ms โ no change โ
โ Q7 โ 36.49 / 37.04 ยฑ0.86 / 38.74 ms โ 37.38 / 37.97 ยฑ0.50 / 38.55 ms โ no change โ
โ Q8 โ 27.68 / 28.84 ยฑ1.27 / 30.55 ms โ 28.11 / 29.79 ยฑ0.93 / 30.70 ms โ no change โ
โ Q9 โ 35.35 / 36.57 ยฑ1.61 / 39.70 ms โ 35.09 / 35.45 ยฑ0.33 / 36.07 ms โ no change โ
โ Q10 โ 53.78 / 56.24 ยฑ2.70 / 60.92 ms โ 54.25 / 55.71 ยฑ1.33 / 57.37 ms โ no change โ
โ Q11 โ 3.82 / 3.96 ยฑ0.13 / 4.21 ms โ 3.72 / 4.75 ยฑ1.12 / 6.15 ms โ 1.20x slower โ
โโโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโโ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโณโโโโโโโโโโโ
โ Benchmark Summary โ โ
โกโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโฉ
โ Total Time (HEAD) โ 248.29ms โ
โ Total Time (fix_topk-global-prefix-early-exit) โ 248.25ms โ
โ Average Time (HEAD) โ 22.57ms โ
โ Average Time (fix_topk-global-prefix-early-exit) โ 22.57ms โ
โ Queries Faster โ 2 โ
โ Queries Slower โ 1 โ
โ Queries with No Change โ 8 โ
โ Queries with Failure โ 0 โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโดโโโโโโโโโโโ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**topk_tpch โ base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 590.2 MiB |
| Avg memory | 98.4 MiB |
| CPU user | 0.0s |
| CPU sys | 0.0s |
| Peak spill | 0 B |
**topk_tpch โ branch**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 609.4 MiB |
| Avg memory | 101.6 MiB |
| CPU user | 0.0s |
| CPU sys | 0.0s |
| Peak spill | 0 B |
</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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on code in PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#discussion_r3433340901
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -1101,86 +1208,111 @@ mod tests {
assert_eq!(record_batch_store.batches_size, 0);
}
- /// Builds an `(a Int32, b Float64)` schema and a `TopK` with full sort
- /// `(a ASC, b ASC)`, input prefix `[a]`, `k = 3`, `batch_size = 2`. Used by
- /// the prefix-completion tests below to keep their per-scenario logic in focus.
- fn build_ab_prefix_topk() -> Result<(Arc<Schema>, TopK)> {
- let schema = Arc::new(Schema::new(vec![
- Field::new("a", DataType::Int32, false),
+ fn make_ab_schema() -> SchemaRef {
+ make_ab_schema_with_nullable_a(false)
+ }
+
+ fn make_ab_schema_with_nullable_a(a_nullable: bool) -> SchemaRef {
+ Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, a_nullable),
Field::new("b", DataType::Float64, false),
- ]));
+ ]))
+ }
+
+ fn make_topk_filter() -> Arc<RwLock<TopKDynamicFilters>> {
Review Comment:
Thanks, factorized in c05846fd9
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4733864573
I opened https://github.com/apache/datafusion/pull/23003 to add a dedicated TopK benchmark over sorted TPCH data
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "geoffreyclaude (via GitHub)" <gi...@apache.org>.
geoffreyclaude commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4730553583
run benchmark topk_tpch
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew merged PR #22991:
URL: https://github.com/apache/datafusion/pull/22991
--
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] Fix shared TopK early exit with shared prefix threshold [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on PR #22991:
URL: https://github.com/apache/datafusion/pull/22991#issuecomment-4757253806
Merging with `main` to get #23053
--
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