Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4481358430 > Thanks @alamb and @Dandandan for the reviews! I'm pulling this one in now, will likely invest some more time in this operator to catch some other low hanging fruits LOL I am not sure there are many **low hanging** fruits in the operator, but I think any improvements you can make will make a big different to performance. RepartitionExec is one of the most performance critical operators we have in the system I think Here is one idea for your amusement (I think it could be a pretty big win for queries that repartition a lot of rows, esp if those rows are strings) - https://github.com/apache/datafusion/issues/11680 -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs merged PR #22010: URL: https://github.com/apache/datafusion/pull/22010 -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4474642172 Thanks @alamb and @Dandandan for the reviews! I'm pulling this one in now, will likely invest some more time in this operator to catch some other low hanging fruits -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
zhuqi-lucas commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4469398587 Nice improvement! -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3252989571
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatc
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3252985988
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
Review Comment:
No longer relevant, as this part of the code has been left untouched in the
end.
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangb commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4466084628 Very cool!! -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460641826 Those are some sweet performance numbers -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460225272 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpcds_sf1.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │56.99 / 58.21 ±1.04 / 60.07 ms │ 56.77 / 57.98 ±1.08 / 59.99 ms │ no change │ │ QQuery 2 │ 292.04 / 298.30 ±5.49 / 306.15 ms │ 285.58 / 305.41 ±11.75 / 318.19 ms │ no change │ │ QQuery 3 │71.29 / 76.51 ±3.24 / 80.46 ms │ 72.56 / 77.31 ±2.94 / 81.73 ms │ no change │ │ QQuery 4 │783.80 / 804.25 ±14.52 / 826.83 ms │ 787.12 / 824.67 ±21.68 / 848.88 ms │ no change │ │ QQuery 5 │632.55 / 647.58 ±10.82 / 663.25 ms │ 598.86 / 623.84 ±20.94 / 652.08 ms │ no change │ │ QQuery 6 │ 148.31 / 154.42 ±5.73 / 164.46 ms │ 153.77 / 160.32 ±7.32 / 173.84 ms │ no change │ │ QQuery 7 │ 829.31 / 841.31 ±9.88 / 856.13 ms │ 851.73 / 869.40 ±11.99 / 886.72 ms │ no change │ │ QQuery 8 │ 147.42 / 151.08 ±2.69 / 155.26 ms │ 150.22 / 157.78 ±3.90 / 160.83 ms │ no change │ │ QQuery 9 │234.49 / 257.44 ±17.51 / 276.96 ms │ 222.10 / 232.19 ±7.40 / 241.07 ms │ +1.11x faster │ │ QQuery 10 │ 155.97 / 159.73 ±4.34 / 168.09 ms │ 158.81 / 160.57 ±1.92 / 163.81 ms │ no change │ │ QQuery 11 │518.71 / 543.21 ±14.30 / 561.60 ms │ 517.88 / 536.85 ±13.37 / 558.45 ms │ no change │ │ QQuery 12 │ 103.99 / 109.75 ±5.93 / 118.11 ms │ 104.17 / 108.96 ±2.98 / 112.45 ms │ no change │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460191800 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark clickbench_partitioned.json ┏━━━┳┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇╇━╇━━━┩ │ QQuery 0 │ 1.52 / 5.09 ±7.00 / 19.08 ms │ 1.54 / 5.05 ±6.91 / 18.87 ms │ no change │ │ QQuery 1 │ 21.89 / 24.42 ±1.42 / 25.95 ms │ 22.22 / 24.21 ±1.31 / 26.27 ms │ no change │ │ QQuery 2 │ 47.96 / 49.11 ±1.79 / 52.63 ms │ 48.64 / 50.04 ±1.00 / 51.76 ms │ no change │ │ QQuery 3 │ 38.56 / 39.16 ±0.47 / 39.64 ms │ 38.32 / 40.06 ±1.69 / 42.36 ms │ no change │ │ QQuery 4 │ 341.04 / 351.42 ±7.42 / 361.45 ms │ 235.37 / 247.97 ±11.83 / 269.43 ms │ +1.42x faster │ │ QQuery 5 │ 379.12 / 382.16 ±3.44 / 388.47 ms │ 315.52 / 317.97 ±1.78 / 320.99 ms │ +1.20x faster │ │ QQuery 6 │ 21.51 / 22.16 ±0.80 / 23.48 ms │ 20.39 / 21.91 ±1.06 / 23.49 ms │ no change │ │ QQuery 7 │ 52.46 / 53.68 ±1.03 / 55.25 ms │ 53.44 / 56.01 ±2.98 / 61.76 ms │ no change │ │ QQuery 8 │ 508.13 / 512.85 ±3.37 / 516.94 ms │ 365.75 / 368.89 ±2.31 / 372.42 ms │ +1.39x faster │ │ QQuery 9 │ 393.39 / 403.30 ±9.01 / 414.29 ms │ 402.83 / 414.62 ±10.30 / 430.26 ms │ no change │ │ QQuery 10 │ 149.17 / 152.05 ±2.66 / 156.80 ms │ 139.44 / 141.93 ±2.66 / 147.10 ms │ +1.07x faster │ │ QQuery 11 │ 159.97 / 162.77 ±2.14 / 166.03 ms │ 153.32 / 156.35 ±2.43 / 160.62 ms │ no change │ │ QQuery 12 │ 384.87 / 390.
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460173228 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf1.json ┏━━━┳┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇╇━╇━━━┩ │ QQuery 1 │ 79.56 / 81.97 ±1.85 / 85.24 ms │ 79.83 / 81.26 ±1.69 / 83.39 ms │ no change │ │ QQuery 2 │ 81.07 / 83.04 ±1.87 / 86.20 ms │ 81.42 / 84.29 ±2.72 / 88.79 ms │ no change │ │ QQuery 3 │ 382.63 / 404.83 ±25.72 / 454.94 ms │ 330.88 / 360.87 ±15.99 / 376.79 ms │ +1.12x faster │ │ QQuery 4 │ 67.38 / 69.36 ±2.50 / 74.21 ms │ 64.34 / 68.78 ±4.03 / 74.06 ms │ no change │ │ QQuery 5 │ 614.57 / 681.29 ±34.68 / 711.72 ms │ 561.06 / 587.51 ±33.33 / 650.88 ms │ +1.16x faster │ │ QQuery 6 │ 29.02 / 31.35 ±1.73 / 33.85 ms │ 27.00 / 28.33 ±1.44 / 31.13 ms │ +1.11x faster │ │ QQuery 7 │ 655.43 / 672.57 ±15.06 / 698.14 ms │ 616.90 / 664.13 ±25.65 / 687.78 ms │ no change │ │ QQuery 8 │ 634.04 / 656.89 ±16.55 / 683.19 ms │ 604.92 / 639.78 ±27.76 / 674.08 ms │ no change │ │ QQuery 9 │ 599.65 / 626.80 ±20.45 / 661.04 ms │ 556.69 / 577.91 ±13.45 / 598.83 ms │ +1.08x faster │ │ QQuery 10 │ 115.16 / 125.34 ±7.60 / 136.35 ms │ 114.29 / 121.75 ±9.27 / 139.82 ms │ no change │ │ QQuery 11 │ 85.64 / 88.49 ±2.04 / 91.23 ms │ 84.76 / 89.74 ±2.92 / 93.18 ms │ no change │ │ QQuery 12 │ 648.30 / 677.95 ±21.35 / 703.07 ms │ 660.48 / 669.96 ±7.88 / 680.50 ms │ no change │ │ QQuery 13 │ 279.84 / 302.90 ±19.09 / 333.65 ms │ 287.20 / 298.27 ±9.26 / 315.28 m
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460101524 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4386511709) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 364.78 / 366.75 ±1.67 / 369.01 ms │ 358.36 / 361.40 ±2.40 / 364.27 ms │ no change │ │ QQuery 2 │ 1256.51 / 1320.97 ±43.17 / 1364.14 ms │ 1121.59 / 1173.06 ±40.38 / 1219.22 ms │ +1.13x faster │ │ QQuery 3 │ 1757.31 / 1803.39 ±28.63 / 1841.82 ms │ 1247.53 / 1327.84 ±43.03 / 1377.25 ms │ +1.36x faster │ │ QQuery 4 │845.21 / 877.25 ±19.55 / 899.84 ms │ 587.63 / 605.98 ±12.98 / 622.54 ms │ +1.45x faster │ │ QQuery 5 │ 2089.77 / 2156.19 ±39.55 / 2197.03 ms │ 1515.59 / 1548.80 ±25.56 / 1583.01 ms │ +1.39x faster │ │ QQuery 6 │ 137.93 / 143.26 ±3.79 / 148.21 ms │ 135.14 / 137.68 ±2.62 / 142.44 ms │ no change │ │ QQuery 7 │ 2427.79 / 2479.68 ±48.56 / 2544.70 ms │ 1819.85 / 1833.80 ±11.81 / 1855.69 ms │ +1.35x faster │ │ QQuery 8 │ 2687.47 / 2723.97 ±22.38 / 2756.85 ms │ 2081.70 / 2143.69 ±51.83 / 2230.41 ms │ +1.27x faster │ │ QQuery 9 │ 2872.66 / 2948.19 ±46.84 / 3019.35 ms │ 1997.18 / 2037.50 ±40.03 / 2113.29 ms │ +1.45x faster │ │ QQuery 10 │ 1432.53 / 1473.99 ±26.80 / 1517.14 ms │ 1302.06 / 1318.92 ±15.28 / 1341.64 ms │ +1.12x faster │ │ QQuery 11 │ 985.33 / 1000.86 ±19.76 / 1037.91 ms │ 803.64 / 874.05 ±47.61 / 925.74 ms │ +1.15x faster │ │ QQuery 12 │773.43 / 792.03 ±16.55 / 818.42 ms │ 689.61 / 795.63 ±63.37 / 853.44 ms │ no change │ │ QQuery 13 │756.91 / 783.87 ±22.27 / 814.61 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460102761 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4385685216) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 360.17 / 362.05 ±1.85 / 365.43 ms │ 356.93 / 359.69 ±2.15 / 362.92 ms │ no change │ │ QQuery 2 │ 1270.51 / 1345.26 ±57.44 / 1425.80 ms │ 1122.37 / 1173.55 ±40.05 / 1226.90 ms │ +1.15x faster │ │ QQuery 3 │ 1724.88 / 1766.32 ±23.56 / 1792.36 ms │ 1317.30 / 1354.35 ±26.43 / 1384.13 ms │ +1.30x faster │ │ QQuery 4 │860.64 / 891.22 ±18.17 / 915.78 ms │ 590.59 / 612.66 ±13.87 / 631.01 ms │ +1.45x faster │ │ QQuery 5 │ 2086.63 / 2149.82 ±38.71 / 2202.07 ms │ 1449.89 / 1507.71 ±36.46 / 1559.81 ms │ +1.43x faster │ │ QQuery 6 │ 141.67 / 145.78 ±2.07 / 147.02 ms │ 136.98 / 138.92 ±1.37 / 140.55 ms │ no change │ │ QQuery 7 │ 2405.04 / 2474.71 ±49.75 / 2532.86 ms │ 1790.30 / 1824.28 ±29.23 / 1873.00 ms │ +1.36x faster │ │ QQuery 8 │ 2726.63 / 2759.38 ±34.06 / 2801.53 ms │ 2087.74 / 2135.56 ±48.96 / 2226.01 ms │ +1.29x faster │ │ QQuery 9 │ 2864.29 / 2926.43 ±44.12 / 2994.44 ms │ 2038.66 / 2053.80 ±12.20 / 2068.33 ms │ +1.42x faster │ │ QQuery 10 │ 1437.56 / 1485.28 ±25.25 / 1509.78 ms │ 1338.68 / 1355.72 ±18.81 / 1385.92 ms │ +1.10x faster │ │ QQuery 11 │ 987.14 / 1021.70 ±27.27 / 1059.03 ms │ 839.11 / 867.91 ±20.15 / 895.25 ms │ +1.18x faster │ │ QQuery 12 │786.01 / 800.12 ±11.06 / 816.65 ms │ 725.45 / 794.42 ±42.47 / 845.15 ms │ no change │ │ QQuery 13 │715.81 / 769.15 ±27.72 / 795.64 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460101478 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4372819392) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark clickbench_partitioned.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 0 │ 1.19 / 4.67 ±6.84 / 18.34 ms │ 1.19 / 4.60 ±6.75 / 18.11 ms │ no change │ │ QQuery 1 │12.45 / 12.91 ±0.23 / 13.09 ms │ 12.37 / 12.90 ±0.65 / 14.16 ms │ no change │ │ QQuery 2 │36.17 / 36.59 ±0.39 / 37.08 ms │ 35.18 / 35.33 ±0.20 / 35.73 ms │ no change │ │ QQuery 3 │30.63 / 31.52 ±0.76 / 32.52 ms │ 30.38 / 30.50 ±0.11 / 30.65 ms │ no change │ │ QQuery 4 │ 229.08 / 235.49 ±6.78 / 247.29 ms │ 221.83 / 225.31 ±3.21 / 231.25 ms │ no change │ │ QQuery 5 │ 275.62 / 277.09 ±1.30 / 279.44 ms │ 270.59 / 274.65 ±2.63 / 278.42 ms │ no change │ │ QQuery 6 │ 7.09 / 7.64 ±0.48 / 8.24 ms │ 6.34 / 6.76 ±0.29 / 7.25 ms │ +1.13x faster │ │ QQuery 7 │13.87 / 13.95 ±0.08 / 14.04 ms │ 13.75 / 13.86 ±0.08 / 13.99 ms │ no change │ │ QQuery 8 │ 311.82 / 315.00 ±2.65 / 318.26 ms │ 307.77 / 309.90 ±2.61 / 314.62 ms │ no change │ │ QQuery 9 │ 447.21 / 451.62 ±3.37 / 456.77 ms │ 451.58 / 458.13 ±6.39 / 470.26 ms │ no change │ │ QQuery 10 │68.02 / 69.70 ±1.49 / 72.13 ms │ 69.47 / 70.87 ±1.14 / 72.73 ms │ no change │ │ QQuery 11 │78.82 / 80.22 ±0.92 / 81.42 ms │ 80.62 / 83.90 ±4.21 / 92.10 ms │ no change │ │ QQuery 12 │ 272.95 / 275.82 ±3.62 / 282.2
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460099402 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4459957969) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark clickbench_partitioned.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 0 │ 1.21 / 4.81 ±7.08 / 18.96 ms │ 1.20 / 4.55 ±6.66 / 17.86 ms │ +1.06x faster │ │ QQuery 1 │12.47 / 12.86 ±0.21 / 13.03 ms │ 12.37 / 12.44 ±0.06 / 12.55 ms │ no change │ │ QQuery 2 │35.91 / 36.53 ±0.53 / 37.46 ms │ 34.94 / 35.15 ±0.21 / 35.53 ms │ no change │ │ QQuery 3 │30.45 / 31.50 ±0.75 / 32.76 ms │ 30.09 / 30.44 ±0.42 / 31.24 ms │ no change │ │ QQuery 4 │ 231.07 / 234.39 ±4.55 / 243.34 ms │ 220.10 / 224.10 ±2.23 / 226.75 ms │ no change │ │ QQuery 5 │ 275.32 / 277.33 ±1.66 / 280.09 ms │ 271.74 / 274.83 ±2.81 / 279.41 ms │ no change │ │ QQuery 6 │ 6.92 / 7.44 ±0.39 / 8.11 ms │ 6.47 / 7.18 ±0.50 / 7.77 ms │ no change │ │ QQuery 7 │13.72 / 13.78 ±0.04 / 13.84 ms │ 13.46 / 13.52 ±0.05 / 13.58 ms │ no change │ │ QQuery 8 │ 310.07 / 315.11 ±4.79 / 323.57 ms │ 303.43 / 306.90 ±2.44 / 309.99 ms │ no change │ │ QQuery 9 │ 439.88 / 447.49 ±4.67 / 452.81 ms │ 439.60 / 454.54 ±8.55 / 466.04 ms │ no change │ │ QQuery 10 │68.13 / 69.61 ±0.92 / 70.74 ms │ 69.01 / 69.67 ±0.50 / 70.47 ms │ no change │ │ QQuery 11 │79.12 / 80.58 ±1.62 / 83.66 ms │ 79.70 / 81.89 ±1.98 / 85.44 ms │ no change │ │ QQuery 12 │ 270.75 / 273.21 ±2.37 / 276.6
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460099130 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373801306) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark clickbench_partitioned.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 0 │ 1.39 / 4.99 ±7.04 / 19.06 ms │ 1.38 / 4.80 ±6.76 / 18.32 ms │ no change │ │ QQuery 1 │13.97 / 14.99 ±0.71 / 16.16 ms │ 13.61 / 14.47 ±0.73 / 15.58 ms │ no change │ │ QQuery 2 │36.88 / 37.50 ±0.37 / 37.94 ms │ 36.32 / 37.41 ±0.80 / 38.66 ms │ no change │ │ QQuery 3 │30.75 / 31.38 ±0.60 / 32.52 ms │ 30.56 / 30.98 ±0.48 / 31.88 ms │ no change │ │ QQuery 4 │ 231.34 / 234.87 ±2.63 / 239.26 ms │ 198.15 / 200.98 ±2.60 / 204.99 ms │ +1.17x faster │ │ QQuery 5 │ 284.64 / 288.45 ±3.61 / 294.82 ms │ 261.70 / 264.95 ±3.19 / 270.37 ms │ +1.09x faster │ │ QQuery 6 │ 8.98 / 9.79 ±0.72 / 11.06 ms │ 8.93 / 9.32 ±0.34 / 9.78 ms │ no change │ │ QQuery 7 │17.28 / 17.67 ±0.36 / 18.32 ms │ 17.14 / 17.39 ±0.14 / 17.50 ms │ no change │ │ QQuery 8 │ 325.86 / 330.15 ±3.56 / 335.07 ms │ 284.94 / 288.82 ±2.39 / 291.68 ms │ +1.14x faster │ │ QQuery 9 │362.82 / 377.72 ±11.10 / 396.47 ms │ 358.25 / 367.30 ±6.29 / 376.45 ms │ no change │ │ QQuery 10 │74.15 / 75.50 ±1.03 / 76.69 ms │ 74.09 / 75.69 ±1.55 / 78.37 ms │ no change │ │ QQuery 11 │86.10 / 87.12 ±1.01 / 89.03 ms │ 84.28 / 85.29 ±0.64 / 86.00 ms │ no change │ │ QQuery 12 │ 286.53 / 291.01 ±6.54 / 303.9
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460097439 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4459959221) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 362.80 / 369.28 ±4.14 / 374.74 ms │ 354.58 / 357.28 ±2.02 / 359.75 ms │ no change │ │ QQuery 2 │ 1030.32 / 1070.23 ±27.30 / 1105.30 ms │ 929.40 / 948.31 ±15.30 / 968.39 ms │ +1.13x faster │ │ QQuery 3 │ 1428.48 / 1510.12 ±60.82 / 1599.74 ms │ 1025.11 / 1082.40 ±42.05 / 1143.30 ms │ +1.40x faster │ │ QQuery 4 │679.55 / 707.00 ±26.06 / 753.87 ms │ 473.95 / 486.72 ±8.12 / 498.26 ms │ +1.45x faster │ │ QQuery 5 │ 1763.63 / 1777.14 ±18.22 / 1811.08 ms │ 1246.90 / 1266.17 ±21.99 / 1309.10 ms │ +1.40x faster │ │ QQuery 6 │ 135.98 / 138.11 ±2.17 / 141.38 ms │ 139.26 / 142.31 ±2.61 / 145.90 ms │ no change │ │ QQuery 7 │ 1994.20 / 2078.56 ±57.15 / 2146.21 ms │ 1495.18 / 1525.04 ±16.80 / 1541.29 ms │ +1.36x faster │ │ QQuery 8 │ 2315.76 / 2379.24 ±54.47 / 2471.65 ms │ 1774.01 / 1800.37 ±15.46 / 1818.12 ms │ +1.32x faster │ │ QQuery 9 │ 2500.49 / 2593.94 ±61.10 / 2672.57 ms │ 1664.36 / 1732.47 ±48.21 / 1786.31 ms │ +1.50x faster │ │ QQuery 10 │ 1117.29 / 1158.28 ±39.73 / 1227.75 ms │ 1033.69 / 1065.02 ±15.88 / 1077.76 ms │ +1.09x faster │ │ QQuery 11 │706.46 / 739.96 ±28.10 / 789.95 ms │ 609.80 / 643.39 ±24.13 / 671.04 ms │ +1.15x faster │ │ QQuery 12 │610.28 / 651.51 ±36.12 / 703.83 ms │ 538.98 / 581.34 ±32.49 / 621.83 ms │ +1.12x faster │ │ QQuery 13 │616.04 / 636.78 ±18.23 / 663.22 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460083925 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373801306) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpcds_sf1.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │12.48 / 13.17 ±0.86 / 14.83 ms │ 12.87 / 13.40 ±0.80 / 14.98 ms │ no change │ │ QQuery 2 │ 103.15 / 105.20 ±1.32 / 107.25 ms │ 103.71 / 104.72 ±0.92 / 106.28 ms │ no change │ │ QQuery 3 │33.97 / 34.28 ±0.26 / 34.60 ms │ 33.82 / 34.09 ±0.20 / 34.36 ms │ no change │ │ QQuery 4 │ 549.74 / 559.93 ±9.53 / 576.29 ms │ 556.43 / 563.49 ±6.41 / 574.96 ms │ no change │ │ QQuery 5 │86.84 / 90.00 ±3.73 / 96.44 ms │ 87.39 / 89.56 ±3.43 / 96.40 ms │ no change │ │ QQuery 6 │46.68 / 47.60 ±0.69 / 48.70 ms │ 46.55 / 47.31 ±0.68 / 48.30 ms │ no change │ │ QQuery 7 │ 158.31 / 160.69 ±1.66 / 162.52 ms │ 164.32 / 165.09 ±0.92 / 166.87 ms │ no change │ │ QQuery 8 │50.58 / 50.93 ±0.25 / 51.24 ms │ 50.85 / 51.06 ±0.23 / 51.47 ms │ no change │ │ QQuery 9 │81.07 / 86.49 ±2.94 / 89.18 ms │ 80.76 / 83.61 ±3.26 / 89.84 ms │ no change │ │ QQuery 10 │ 91.84 / 94.77 ±3.02 / 100.14 ms │ 91.76 / 93.92 ±1.72 / 96.78 ms │ no change │ │ QQuery 11 │ 351.81 / 358.28 ±4.53 / 364.86 ms │ 356.06 / 362.01 ±3.39 / 366.45 ms │ no change │ │ QQuery 12 │35.45 / 37.46 ±2.89 / 43.08 ms │ 35.54 / 36.14 ±0.48 / 36.88 ms │ no change │ │ QQuery 13 │ 146.78 / 149.15 ±1.60 / 151.64 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460095329 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4376856575) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 350.39 / 354.59 ±3.47 / 359.74 ms │ 350.92 / 353.79 ±2.68 / 358.82 ms │ no change │ │ QQuery 2 │ 1007.37 / 1039.49 ±27.31 / 1080.20 ms │ 949.30 / 964.07 ±9.51 / 977.80 ms │ +1.08x faster │ │ QQuery 3 │ 1434.73 / 1489.95 ±39.43 / 1555.39 ms │ 1062.65 / 1099.58 ±41.10 / 1177.58 ms │ +1.36x faster │ │ QQuery 4 │ 706.00 / 717.79 ±9.82 / 735.43 ms │ 461.41 / 472.96 ±8.35 / 486.55 ms │ +1.52x faster │ │ QQuery 5 │ 1804.39 / 1814.13 ±8.75 / 1830.10 ms │ 1235.67 / 1271.40 ±34.61 / 1318.44 ms │ +1.43x faster │ │ QQuery 6 │ 136.28 / 139.32 ±3.28 / 144.95 ms │ 133.39 / 137.35 ±3.50 / 142.67 ms │ no change │ │ QQuery 7 │ 1968.71 / 2049.00 ±59.23 / 2136.35 ms │ 1498.64 / 1525.24 ±25.53 / 1564.15 ms │ +1.34x faster │ │ QQuery 8 │ 2271.36 / 2303.70 ±31.39 / 2355.69 ms │ 1766.42 / 1794.94 ±23.21 / 1833.48 ms │ +1.28x faster │ │ QQuery 9 │ 2486.73 / 2491.23 ±5.19 / 2501.01 ms │ 1698.33 / 1728.84 ±20.19 / 1754.60 ms │ +1.44x faster │ │ QQuery 10 │ 1138.96 / 1187.00 ±31.56 / 1232.04 ms │ 1047.84 / 1093.66 ±30.76 / 1141.89 ms │ +1.09x faster │ │ QQuery 11 │724.16 / 750.71 ±29.62 / 804.20 ms │ 630.57 / 650.01 ±14.29 / 668.60 ms │ +1.15x faster │ │ QQuery 12 │ 603.85 / 614.57 ±7.41 / 623.99 ms │ 558.23 / 584.82 ±19.25 / 613.98 ms │ no change │ │ QQuery 13 │590.15 / 622.78 ±24.33 / 663.98 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460088972 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373980023-135-tmxg5 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460080631 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4372819392) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpcds_sf1.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 6.34 / 6.91 ±0.86 / 8.62 ms │ 6.39 / 6.88 ±0.82 / 8.51 ms │ no change │ │ QQuery 2 │82.80 / 83.35 ±0.29 / 83.58 ms │ 81.66 / 82.33 ±0.39 / 82.70 ms │ no change │ │ QQuery 3 │29.21 / 29.57 ±0.34 / 30.22 ms │ 30.01 / 30.59 ±0.35 / 31.06 ms │ no change │ │ QQuery 4 │515.41 / 534.29 ±14.97 / 551.00 ms │ 516.80 / 530.23 ±12.75 / 548.23 ms │ no change │ │ QQuery 5 │52.69 / 53.07 ±0.30 / 53.53 ms │ 53.09 / 53.62 ±0.76 / 55.11 ms │ no change │ │ QQuery 6 │35.64 / 36.07 ±0.35 / 36.68 ms │ 35.50 / 37.08 ±1.49 / 39.12 ms │ no change │ │ QQuery 7 │ 110.78 / 113.17 ±1.91 / 115.11 ms │ 112.53 / 114.19 ±1.68 / 116.47 ms │ no change │ │ QQuery 8 │39.70 / 39.79 ±0.06 / 39.87 ms │ 39.24 / 39.43 ±0.24 / 39.89 ms │ no change │ │ QQuery 9 │53.79 / 55.61 ±1.53 / 58.29 ms │ 54.56 / 56.27 ±1.63 / 59.05 ms │ no change │ │ QQuery 10 │81.39 / 83.80 ±2.32 / 87.81 ms │ 81.65 / 82.71 ±1.27 / 84.99 ms │ no change │ │ QQuery 11 │ 322.92 / 332.24 ±6.46 / 342.87 ms │ 321.49 / 331.38 ±10.12 / 348.85 ms │ no change │ │ QQuery 12 │29.80 / 31.24 ±0.89 / 32.24 ms │ 28.81 / 29.04 ±0.13 / 29.20 ms │ +1.08x faster │ │ QQuery 13 │ 128.28 / 129.49 ±1.01 / 131.22 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460079126 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373980023-134-g8f9z 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3248382963
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +153,120 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// Output channel with its associated memory reservation and spill writer.
+///
+/// `coalescer` is `None` for preserve-order mode, where downstream
+/// [`StreamingMergeBuilder`] performs the batching; otherwise it's a
+/// [`SharedCoalescer`] cloned from the per-partition one held by
+/// [`PartitionChannels`].
struct OutputChannel {
sender: DistributionSender,
reservation: SharedMemoryReservation,
spill_writer: SpillPoolWriter,
+shared_coalescer: Option,
+}
+
+impl OutputChannel {
+fn coalesce(&mut self, batch: RecordBatch) -> Result> {
+match &self.shared_coalescer {
+Some(shared) => Ok(shared.push_and_drain(batch)?),
+None => Ok(vec![batch]),
+}
+}
+
+/// Send a single batch through the channel for `partition`, applying
Review Comment:
Maybe worth pointing out in the comment that this is used *after* coalescing
(and doesn't internally coalesce)
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -797,6 +934,34 @@ impl BatchPartitioner {
/// arbitrary interleaving (and thus unordered) unless
/// [`Self::with_preserve_order`] specifies otherwise.
///
+/// # Batch coalescing
+///
+/// Repartitioning one [`RecordBatch`] implies creating multiple smaller
batches, potentially
+/// as many as the number of output partitions. [`RepartitionExec`] makes sure
that the returned
+/// batches adhere to the configured `datafusion.execution.batch_size` for
efficient operations,
+/// and for that, it will automatically coalesce batches right after
repartitioning.
+///
+/// For this, one shared [`LimitedBatchCoalescer`] per output partition is
used:
Review Comment:
love it 😍
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460078409 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373980023-133-fl7sb 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460066038 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4372819392) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf1.json ┏━━━┳┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇╇━╇━━━┩ │ QQuery 1 │ 39.43 / 40.50 ±1.18 / 42.16 ms │ 39.18 / 40.37 ±1.38 / 42.21 ms │ no change │ │ QQuery 2 │ 20.88 / 21.61 ±0.89 / 23.32 ms │ 21.01 / 21.19 ±0.12 / 21.35 ms │ no change │ │ QQuery 3 │ 35.57 / 37.56 ±1.27 / 38.84 ms │ 34.25 / 36.57 ±1.19 / 37.55 ms │ no change │ │ QQuery 4 │ 17.63 / 17.89 ±0.15 / 18.06 ms │ 18.13 / 18.21 ±0.06 / 18.29 ms │ no change │ │ QQuery 5 │ 46.04 / 47.08 ±0.70 / 48.24 ms │ 42.66 / 44.73 ±1.45 / 46.44 ms │ no change │ │ QQuery 6 │ 17.10 / 17.42 ±0.53 / 18.48 ms │ 16.84 / 17.28 ±0.43 / 18.06 ms │ no change │ │ QQuery 7 │ 51.20 / 52.54 ±1.56 / 55.50 ms │ 50.01 / 51.74 ±1.33 / 53.75 ms │ no change │ │ QQuery 8 │ 46.63 / 46.91 ±0.19 / 47.13 ms │ 45.95 / 46.59 ±0.40 / 47.08 ms │ no change │ │ QQuery 9 │ 51.81 / 53.19 ±1.01 / 54.17 ms │ 52.37 / 53.41 ±0.94 / 55.18 ms │ no change │ │ QQuery 10 │ 65.21 / 65.82 ±0.48 / 66.57 ms │ 65.37 / 65.73 ±0.43 / 66.58 ms │ no change │ │ QQuery 11 │ 14.17 / 14.45 ±0.18 / 14.71 ms │ 14.29 / 15.12 ±1.04 / 16.95 ms │ no change │ │ QQuery 12 │ 25.53 / 25.76 ±0.16 / 26.02 ms │ 24.79 / 25.22 ±0.42 / 25.78 ms │ no change │ │ QQuery 13 │ 36.27 / 36.96 ±0.58 / 37.64 ms │ 35.41 / 36.29 ±0.64 / 37.01 ms │ no change │ │ QQuery 14 │ 26.47 / 26.94 ±0.60 / 28.13
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4460071439 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373801306) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf1.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │42.02 / 43.09 ±1.03 / 45.06 ms │ 41.88 / 43.14 ±1.29 / 45.50 ms │ no change │ │ QQuery 2 │29.20 / 29.93 ±0.57 / 30.85 ms │ 28.57 / 29.98 ±0.95 / 31.13 ms │ no change │ │ QQuery 3 │62.50 / 64.00 ±1.54 / 66.30 ms │ 54.01 / 55.60 ±0.94 / 56.71 ms │ +1.15x faster │ │ QQuery 4 │24.24 / 24.54 ±0.28 / 24.98 ms │ 24.16 / 24.69 ±0.38 / 25.17 ms │ no change │ │ QQuery 5 │ 90.96 / 94.40 ±3.24 / 100.48 ms │ 80.63 / 82.75 ±1.63 / 84.73 ms │ +1.14x faster │ │ QQuery 6 │18.56 / 18.93 ±0.34 / 19.56 ms │ 18.63 / 18.70 ±0.08 / 18.82 ms │ no change │ │ QQuery 7 │ 98.95 / 100.45 ±1.11 / 102.15 ms │ 91.26 / 93.07 ±1.59 / 95.64 ms │ +1.08x faster │ │ QQuery 8 │85.77 / 88.52 ±1.76 / 90.44 ms │ 82.57 / 86.41 ±2.05 / 88.31 ms │ no change │ │ QQuery 9 │93.11 / 95.46 ±1.77 / 97.54 ms │ 89.31 / 91.54 ±1.72 / 93.97 ms │ no change │ │ QQuery 10 │71.08 / 71.34 ±0.21 / 71.68 ms │ 69.75 / 71.35 ±1.00 / 72.83 ms │ no change │ │ QQuery 11 │22.55 / 23.32 ±0.59 / 24.11 ms │ 22.64 / 23.58 ±0.65 / 24.54 ms │ no change │ │ QQuery 12 │47.75 / 48.85 ±0.79 / 50.12 ms │ 44.97 / 45.76 ±0.71 / 46.70 ms │ +1.07x faster │ │ QQuery 13 │49.21 / 51.38 ±1.46 / 53.24 ms │ 48.52 / 49.31 ±0.64 / 50.08 ms │ no chang
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459984918 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4459959221) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4459959221-126-t7kg9 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459984922 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4372819392) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4372819392-128-d2ss9 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459984039 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373801306) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373801306-132-ccmhc 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459984462 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4382478164) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4382478164-137-dtcx6 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459984754 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373801306) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373801306-131-p7g7s 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459983529 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4381836993) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4381836993-138-8kltt 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459982785 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4376856575) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4376856575-136-7grmr 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459982731 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4385430919) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4385430919-139-x9tds 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459982753 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4372819392) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4372819392-129-h2x5d 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459982386 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4372819392) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4372819392-127-dmj2k 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459981369 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4386511709) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4386511709-141-fclmw 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459981431 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4385685216) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4385685216-140-vpzvf 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459981413 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373801306) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373801306-130-6dvrj 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459981584 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4459957969) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4459957969-125-mbtts 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (83c31ec07d96061412268cdfff81596cd39b5b7f) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..83c31ec07d96061412268cdfff81596cd39b5b7f) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459959221 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 256 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4459957969 run benchmark clickbench_partitioned -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3244079268
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatc
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010: URL: https://github.com/apache/datafusion/pull/22010#discussion_r3244065194 ## Cargo.lock: ## @@ -1528,6 +1528,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] Review Comment: Reverted the commit for 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010: URL: https://github.com/apache/datafusion/pull/22010#discussion_r3244059070 ## Cargo.lock: ## @@ -1528,6 +1528,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] Review Comment: I don't think we can no, that package contains a work stealing queue, which is not really what we need here. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on code in PR #22010: URL: https://github.com/apache/datafusion/pull/22010#discussion_r3243893739 ## Cargo.lock: ## @@ -1528,6 +1528,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] Review Comment: not a huge fan of this new dependency --given we already have https://crates.io/crates/crossbeam-deque (a few lines above) maybe we can just that one so there are no net new dependencies? -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3243879370
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatchCoa
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3243602400
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatc
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3242587386
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatc
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4451670111 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4451463424) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 358.90 / 361.46 ±2.95 / 365.76 ms │ 356.63 / 360.12 ±2.89 / 364.76 ms │ no change │ │ QQuery 2 │ 989.31 / 1039.41 ±36.99 / 1085.24 ms │ 905.77 / 942.25 ±21.92 / 970.61 ms │ +1.10x faster │ │ QQuery 3 │ 1445.20 / 1486.99 ±47.26 / 1579.03 ms │ 1082.15 / 1100.26 ±11.03 / 1115.34 ms │ +1.35x faster │ │ QQuery 4 │ 714.31 / 725.62 ±9.99 / 738.10 ms │ 481.91 / 486.67 ±3.13 / 490.66 ms │ +1.49x faster │ │ QQuery 5 │ 1794.56 / 1818.08 ±18.23 / 1845.93 ms │ 1273.82 / 1299.79 ±29.54 / 1355.64 ms │ +1.40x faster │ │ QQuery 6 │ 137.19 / 144.45 ±9.40 / 162.93 ms │ 136.72 / 138.40 ±1.36 / 140.87 ms │ no change │ │ QQuery 7 │ 2003.39 / 2056.18 ±28.11 / 2084.59 ms │ 1509.25 / 1551.20 ±25.59 / 1589.70 ms │ +1.33x faster │ │ QQuery 8 │ 2316.71 / 2356.48 ±39.96 / 2427.18 ms │ 1790.74 / 1810.09 ±25.17 / 1858.29 ms │ +1.30x faster │ │ QQuery 9 │ 2526.10 / 2560.80 ±35.88 / 2624.32 ms │ 1791.95 / 1809.58 ±16.56 / 1837.60 ms │ +1.42x faster │ │ QQuery 10 │ 1148.14 / 1225.32 ±68.13 / 1351.64 ms │ 1048.13 / 1072.64 ±22.65 / .66 ms │ +1.14x faster │ │ QQuery 11 │692.31 / 736.06 ±45.23 / 791.83 ms │ 599.96 / 625.28 ±17.75 / 642.33 ms │ +1.18x faster │ │ QQuery 12 │624.38 / 638.64 ±15.16 / 664.10 ms │ 561.23 / 576.42 ±13.27 / 596.57 ms │ +1.11x faster │ │ QQuery 13 │614.37 / 637.91 ±16.36 / 658.38 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4451491866 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4451463424) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4451463424-88-fbs47 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (bbd10de9f01161639957ae1c090752158404116d) to 4fac70d (merge-base) [diff](https://github.com/apache/datafusion/compare/4fac70d9b274138b89fef8fce80dcafa378f5891..bbd10de9f01161639957ae1c090752158404116d) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4451463424 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 256 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3241695978
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatc
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4450871828 Moving it temporarily to draft, I want to improve it a bit more before it's ready for another review. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3237516331
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
Review Comment:
Manage to come up with
https://github.com/apache/datafusion/pull/22010/changes/bab26b7c584b4fe4109ee4442904643a25804f33.
There's several places in this file where partition->channel maps are
represented with `HashMap` instead of `Vec<_>`. Rather than moving
only this one to be a `Vec<_>` I tried to be consistent in the other places and
now they are all `Vec<_>`s.
This cascaded into a bunch of restructures that I think have a net positive
outcome, as it allowed to remove yet another abstraction (`struct
OutputChannels`) and simplify a bit the code, and now the diff is even smaller.
Let me know what you think.
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3237494012
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -1055,6 +1247,7 @@ impl ExecutionPlan for RepartitionExec {
let name = self.name().to_owned();
let schema = self.schema();
let schema_captured = Arc::clone(&schema);
+let fetch = self.fetch();
Review Comment:
Done in
https://github.com/apache/datafusion/pull/22010/changes/675cc4b6c2edc4174715e539a5425675cdd18461
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3237444229
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -1055,6 +1247,7 @@ impl ExecutionPlan for RepartitionExec {
let name = self.name().to_owned();
let schema = self.schema();
let schema_captured = Arc::clone(&schema);
+let fetch = self.fetch();
Review Comment:
🤔 Interesting, you are completely right, I was getting confused because of
the fact that `LimitedBatchCoalescer` is used here, but it looks like `None` is
passed in every case, so we might be better without it at all.
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3237422234
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
Review Comment:
🤔 thinking about it, it's probably possible to structure things so that this
never happens. I'll try it out.
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3237098214
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatc
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3237020075
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
Review Comment:
I think this is how it worked before:
https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/repartition/mod.rs#L1778-L1814
Before, the batches were getting coalesced later in the `RepartitionExec`
pipeline, and there was no memory accounting happening there, so I'd expect the
memory reservation to behave the same as `main`.
Not implying that's the best approach though... as I invest time in
improving this operator, I find it a bit hard to deal with, as it seems to have
accumulated several stacked abstractions (structs representing things) over the
years, and it's a bit hard to maintain.
I'd not be surprised if a full rewrite could yield a more maintainable
outcome.
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3236951720
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatc
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3236281180
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
Review Comment:
The reason why this is a `HashMap` is mainly because
that's what it was before this PR. Note how that did not change:
https://github.com/apache/datafusion/blob/dcf648255b92a34798871139aeba12d95f8f3032/datafusion/physical-plan/src/repartition/mod.rs#L559-L576
I'll try refactoring further and using `Vec` instead
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4442338874 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4442066583) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark clickbench_partitioned.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 0 │ 1.23 / 4.65 ±6.73 / 18.12 ms │ 1.21 / 4.65 ±6.76 / 18.17 ms │ no change │ │ QQuery 1 │13.24 / 13.57 ±0.19 / 13.76 ms │ 12.25 / 12.78 ±0.29 / 13.09 ms │ +1.06x faster │ │ QQuery 2 │37.23 / 37.55 ±0.27 / 37.88 ms │ 36.91 / 37.27 ±0.27 / 37.72 ms │ no change │ │ QQuery 3 │32.01 / 33.18 ±1.25 / 35.60 ms │ 31.92 / 32.09 ±0.15 / 32.38 ms │ no change │ │ QQuery 4 │ 238.77 / 240.39 ±1.07 / 241.50 ms │ 224.84 / 232.44 ±4.83 / 239.02 ms │ no change │ │ QQuery 5 │ 283.27 / 286.11 ±1.84 / 288.51 ms │ 277.81 / 281.77 ±2.86 / 286.57 ms │ no change │ │ QQuery 6 │ 7.18 / 7.56 ±0.22 / 7.83 ms │ 7.28 / 7.65 ±0.34 / 8.30 ms │ no change │ │ QQuery 7 │14.81 / 15.71 ±1.60 / 18.91 ms │ 13.61 / 13.70 ±0.06 / 13.79 ms │ +1.15x faster │ │ QQuery 8 │ 324.07 / 330.32 ±4.09 / 336.51 ms │ 313.10 / 316.33 ±3.51 / 322.47 ms │ no change │ │ QQuery 9 │ 449.61 / 466.99 ±9.36 / 476.87 ms │ 454.36 / 467.18 ±11.16 / 483.83 ms │ no change │ │ QQuery 10 │74.53 / 75.36 ±0.86 / 76.94 ms │ 74.76 / 75.47 ±0.63 / 76.63 ms │ no change │ │ QQuery 11 │86.60 / 87.23 ±0.38 / 87.65 ms │ 85.41 / 86.85 ±1.22 / 88.87 ms │ no change │ │ QQuery 12 │ 277.64 / 282.47 ±4.52 / 291.0
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3235263825
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
+for partition in partitions {
+let Some(channel) = self.inner.get(&partition) else {
+continue;
+};
+let Some(shared) = channel.shared_coalescer.clone() else {
+continue;
+};
+for batch in shared.finalize()? {
+self.send_to_channel(partition, batch).await;
+}
+}
+Ok(())
+}
+
+/// Send a single batch through the channel for `partition`, applying
+/// the memory reservation / spill-writer fallback. Removes the channel
+/// from `self.inner` if the receiver has hung up.
+async fn send_to_channel(&mut self, partition: usize, batch: RecordBatch) {
+let size = batch.get_array_memory_size();
+let timer = self.metrics.send_time[partition].timer();
+
+// Decide the payload outside of any await: never hold a MutexGuard
+// across an await point.
+let (payload, is_memory_batch) = {
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+match channel.reservation.try_grow(size) {
+Ok(_) => (Ok(RepartitionBatch::Memory(batch)), true),
+Err(_) => match channel.spill_writer.push_batch(&batch) {
+Ok(()) => (Ok(RepartitionBatch::Spilled), false),
+Err(err) => (Err(err), false),
+},
+}
+};
+
+let Some(channel) = self.inner.get(&partition) else {
+timer.done();
+return;
+};
+let send_err = channel.sender.send(Some(payload)).await.is_err();
+if send_err {
+if is_memory_batch && let Some(channel) =
self.inner.get(&partition) {
+channel.reservation.shrink(size);
+}
+self.inner.remove(&partition);
+}
+timer.done();
+}
+}
+
+/// A producer-side coalescer shared across all input tasks targeting a
+/// single output partition.
+///
+/// Bundles the [`LimitedBatchCoalescer`] (behind a [`Mutex`]) with the
+/// active-sender counter that tracks how many input tasks may still push
+/// into it. The last task to call [`Self::finalize`] is the one that
+/// finalizes the coalescer and ships the residual batch.
+///
+/// Cheap to [`Clone`]: both fields are [`Arc`]s.
+#[derive(Clone)]
+struct SharedCoalescer {
+inner: Arc>,
+active_senders: Arc,
+}
+
+impl SharedCoalescer {
+fn new(
+schema: SchemaRef,
+target_batch_size: usize,
+fetch: Option,
+num_senders: usize,
+) -> Self {
+Self {
+inner: Arc::new(Mutex::new(LimitedBatchCoa
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3235171117
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+let partitions: Vec = self.inner.keys().copied().collect();
Review Comment:
I wonder if we could avoid this copy/ allocation and just consume the inner
```rust
async fn finalize(&mut self) -> Result<()> {
for (partition, channel) in self.inner.drain() {
let Some(shared) = channel.shared_coalescer.clone() else {
continue;
};
for batch in shared.finalize()? {
self.send_to_channel(partition, batch).await;
}
}
Ok(())
}
```
it would probably require rejiggering how `self.send_to_channel` worked to
get a passed channel (rather than looking it up again)
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
+metrics: RepartitionMetrics,
+}
+
+impl OutputChannels {
+fn new(inner: HashMap, metrics: RepartitionMetrics)
-> Self {
+Self { inner, metrics }
+}
+
+fn is_empty(&self) -> bool {
+self.inner.is_empty()
+}
+
+fn metrics(&self) -> &RepartitionMetrics {
+&self.metrics
+}
+
+/// Push `batch` for `partition` through its shared coalescer (if any)
+/// and ship any newly completed batches through the channel.
+async fn coalesce_and_send(
+&mut self,
+partition: usize,
+batch: RecordBatch,
+) -> Result<()> {
+let Some(channel) = self.inner.get(&partition) else {
+return Ok(());
+};
+let to_send = match &channel.shared_coalescer {
+Some(shared) => shared.push_and_drain(batch)?,
+None => vec![batch],
+};
+for batch in to_send {
+self.send_to_channel(partition, batch).await;
+}
+Ok(())
+}
+
+/// For each output partition this task still has, decrement the shared
+/// active-senders counter. The last task to do so calls
+/// [`SharedCoalescer::finalize`] and ships the residual.
+async fn finalize(&mut self) -> Result<()> {
+
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4442114895 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4442066583) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4442066583-33-jr8fx 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4442066583 run benchmark clickbench_partitioned -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
alamb commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4401153807 Checking this one out -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4386672734 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4386511709) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 367.06 / 376.49 ±5.20 / 382.52 ms │ 372.31 / 375.69 ±3.13 / 380.99 ms │ no change │ │ QQuery 2 │ 1260.81 / 1297.89 ±24.76 / 1328.98 ms │ 1088.61 / 1148.21 ±35.65 / 1185.95 ms │ +1.13x faster │ │ QQuery 3 │ 1744.15 / 1806.38 ±43.72 / 1852.32 ms │ 1251.02 / 1289.81 ±32.00 / 1324.73 ms │ +1.40x faster │ │ QQuery 4 │888.76 / 906.97 ±12.79 / 922.98 ms │ 553.87 / 568.43 ±18.86 / 605.40 ms │ +1.60x faster │ │ QQuery 5 │ 2132.02 / 2158.75 ±18.28 / 2182.62 ms │ 1446.96 / 1511.02 ±46.98 / 1585.61 ms │ +1.43x faster │ │ QQuery 6 │ 144.88 / 147.65 ±1.69 / 149.89 ms │ 144.11 / 145.79 ±1.26 / 147.54 ms │ no change │ │ QQuery 7 │ 2409.54 / 2450.25 ±26.23 / 2490.30 ms │ 1798.17 / 1803.17 ±4.45 / 1811.21 ms │ +1.36x faster │ │ QQuery 8 │ 2691.89 / 2740.83 ±40.78 / 2802.86 ms │ 2044.77 / 2084.19 ±43.95 / 2163.09 ms │ +1.32x faster │ │ QQuery 9 │ 2885.59 / 2926.05 ±54.13 / 3031.23 ms │ 1965.90 / 2044.98 ±64.14 / 2160.93 ms │ +1.43x faster │ │ QQuery 10 │ 1453.06 / 1496.26 ±32.16 / 1545.21 ms │ 1319.83 / 1351.71 ±19.46 / 1378.89 ms │ +1.11x faster │ │ QQuery 11 │ 960.02 / 976.94 ±24.66 / 1025.09 ms │ 811.60 / 870.93 ±45.87 / 946.23 ms │ +1.12x faster │ │ QQuery 12 │751.15 / 787.85 ±39.31 / 845.59 ms │ 671.76 / 777.48 ±55.25 / 823.24 ms │ no change │ │ QQuery 13 │735.67 / 764.15 ±24.27 / 796.35 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4386547989 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4386511709) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4386511709-2040-qx5k4 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4386511709 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 300 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4386367667 > @gabotechs I notice peak memory is quite a bit higher here? I can imagine how this can happen in cases where the fanout is very big, it boils down to the gating mechanism implemented in `RepartitionExec` today: https://github.com/apache/datafusion/blob/dcf648255b92a34798871139aeba12d95f8f3032/datafusion/physical-plan/src/repartition/distributor_channels.rs#L21-L36 Before this PR, the batches that were flowing through there where of size `batch_size / output_partitions`, but with this PR, they are of size `batch_size`. The memory reporting there seems quite unstable though, for example, this other runs show the same peak memory usage: - https://github.com/apache/datafusion/pull/22010#issuecomment-4374065644 - https://github.com/apache/datafusion/pull/22010#issuecomment-4374088542 - https://github.com/apache/datafusion/pull/22010#issuecomment-4374117175 -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4386270831 https://github.com/user-attachments/assets/334249df-0f2b-437e-aecc-dbd0b5f389e4"; /> @gabotechs I notice peak memory is quite a bit higher here? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4386266755 https://github.com/user-attachments/assets/b2e90d85-8981-49cf-8d60-81eaa50c9cd1"; /> From: Apache Arrow DataFusion: a Fast, Embeddable, Modular Analytic Query Engine https://andrew.nerdnetworks.org/pdf/SIGMOD-2024-lamb.pdf I think there is some overlap of these 2024 figures at high core counts (>32 cores) and the reported improvements here. @alamb @gabotechs perhaps it would be good to redo this experiment once more. With morsel scan + optimization like this I expect this picture to look much better for the higher core counts :) Perhaps we could also test clickbench just on higher core counts and see if it improves. -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385792780 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4385685216) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 374.98 / 377.69 ±1.62 / 379.48 ms │ 373.97 / 376.74 ±1.91 / 379.20 ms │ no change │ │ QQuery 2 │ 1267.60 / 1325.63 ±39.16 / 1373.05 ms │ 1131.50 / 1173.97 ±25.78 / 1200.53 ms │ +1.13x faster │ │ QQuery 3 │ 1797.16 / 1825.13 ±19.25 / 1853.48 ms │ 1288.02 / 1318.33 ±18.11 / 1340.33 ms │ +1.38x faster │ │ QQuery 4 │863.53 / 886.46 ±20.46 / 916.70 ms │ 558.94 / 598.42 ±32.75 / 637.74 ms │ +1.48x faster │ │ QQuery 5 │ 2114.70 / 2133.66 ±14.97 / 2157.96 ms │ 1426.81 / 1496.98 ±48.79 / 1576.25 ms │ +1.43x faster │ │ QQuery 6 │ 144.35 / 150.47 ±5.81 / 160.89 ms │ 141.61 / 146.18 ±6.00 / 158.02 ms │ no change │ │ QQuery 7 │ 2426.03 / 2473.60 ±34.41 / 2518.15 ms │ 1798.66 / 1829.02 ±27.81 / 1878.56 ms │ +1.35x faster │ │ QQuery 8 │ 2711.55 / 2777.09 ±65.27 / 2889.80 ms │ 2062.51 / 2130.00 ±46.43 / 2182.96 ms │ +1.30x faster │ │ QQuery 9 │ 2896.73 / 2960.38 ±43.61 / 3032.65 ms │ 1984.24 / 2041.10 ±34.27 / 2074.79 ms │ +1.45x faster │ │ QQuery 10 │ 1477.16 / 1501.84 ±26.36 / 1552.01 ms │ 1285.89 / 1341.81 ±35.92 / 1397.85 ms │ +1.12x faster │ │ QQuery 11 │ 1001.36 / 1032.84 ±29.06 / 1077.74 ms │ 804.14 / 841.48 ±33.50 / 903.26 ms │ +1.23x faster │ │ QQuery 12 │777.00 / 797.55 ±27.06 / 850.98 ms │ 758.30 / 795.39 ±27.36 / 839.63 ms │ no change │ │ QQuery 13 │749.25 / 775.51 ±30.41 / 829.85 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385698232 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4385685216) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4385685216-2038-cjdr7 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385691293 Hi @Dandandan, you asked to view the benchmark queue (https://github.com/apache/datafusion/pull/22010#issuecomment-4385691249). | Comment | Repo | PR | User | Benchmarks | Status | | --- | --- | --- | --- | --- | --- | | [#4385685216](https://github.com/apache/datafusion/pull/22010#issuecomment-4385685216) | apache/datafusion | #22010 | Dandandan | ["tpch10"] | running | --- [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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385691249 show benchmark queue -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385690610 I am wondering why the benchmarks aren't completing anymore with the high target partition. Is it a deadlock or something else? -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385685216 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 300 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385441576 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4385430919) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4385430919-2037-6jqdl 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4385430919 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 512 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4382498372 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4382478164) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4382478164-2031-w9s9p 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4382478164 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 512 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4381857196 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4381836993) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4381836993-2030-67qvv 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4381836993 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 2048 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4377007472 Ok, as it looks like it does what was intended, I'll open this for review and improve the PR description. -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Samyak2 commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4377001173 This should fix https://github.com/apache/datafusion/issues/20491 too! -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4376934736 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4376856575) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf10.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │ 367.67 / 370.56 ±2.75 / 374.78 ms │ 365.18 / 373.67 ±4.77 / 379.99 ms │ no change │ │ QQuery 2 │ 1004.79 / 1044.73 ±25.91 / 1077.76 ms │ 911.89 / 929.35 ±13.91 / 949.82 ms │ +1.12x faster │ │ QQuery 3 │ 1449.50 / 1482.28 ±26.02 / 1515.45 ms │ 1020.76 / 1037.48 ±17.38 / 1068.95 ms │ +1.43x faster │ │ QQuery 4 │723.52 / 737.25 ±10.02 / 751.24 ms │ 434.60 / 469.21 ±19.98 / 496.75 ms │ +1.57x faster │ │ QQuery 5 │ 1761.69 / 1789.09 ±27.18 / 1828.80 ms │ 1226.71 / 1252.80 ±14.01 / 1265.94 ms │ +1.43x faster │ │ QQuery 6 │ 144.57 / 145.98 ±2.30 / 150.51 ms │ 143.31 / 143.63 ±0.25 / 143.98 ms │ no change │ │ QQuery 7 │ 1974.80 / 2092.05 ±68.11 / 2169.52 ms │ 1449.65 / 1485.49 ±25.19 / 1515.01 ms │ +1.41x faster │ │ QQuery 8 │ 2244.63 / 2306.15 ±35.70 / 2345.41 ms │ 1728.81 / 1790.63 ±34.56 / 1828.12 ms │ +1.29x faster │ │ QQuery 9 │ 2464.34 / 2498.28 ±19.34 / 2518.71 ms │ 1684.52 / 1727.57 ±52.17 / 1821.02 ms │ +1.45x faster │ │ QQuery 10 │ 1178.65 / 1188.35 ±7.41 / 1199.86 ms │ 1032.39 / 1063.70 ±28.55 / 1108.34 ms │ +1.12x faster │ │ QQuery 11 │702.30 / 724.13 ±17.12 / 746.29 ms │ 592.85 / 617.01 ±24.95 / 657.93 ms │ +1.17x faster │ │ QQuery 12 │625.06 / 645.32 ±16.86 / 664.50 ms │ 570.42 / 586.23 ±19.29 / 621.02 ms │ +1.10x faster │ │ QQuery 13 │603.10 / 632.75 ±15.98 / 651.15 ms │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4376873421 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4376856575) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4376856575-2023-d9wvg 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) using: tpch10 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on code in PR #22010:
URL: https://github.com/apache/datafusion/pull/22010#discussion_r3186325722
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -143,11 +144,183 @@ type MaybeBatch = Option>;
type InputPartitionsToCurrentPartitionSender =
Vec>;
type InputPartitionsToCurrentPartitionReceiver =
Vec>;
-/// Output channel with its associated memory reservation and spill writer
+/// One input task's collection of output channels (its send-side view of
+/// every output partition). Owns the per-call helpers for coalescing,
+/// finalizing, and sending.
+///
+/// This complements [`PartitionChannels`] (the per-output-partition,
+/// authoritative struct that owns `rx`, `spill_readers`, and the underlying
+/// `Mutex` / `AtomicUsize`). Each [`OutputChannel`]
+/// inside `inner` holds cloned `Arc`s pointing back at those shared
+/// resources.
+struct OutputChannels {
+inner: HashMap,
Review Comment:
Probably no perf change but I think `Vec>` is better
--
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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
Dandandan commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4376856575 run benchmark tpch10 ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 256 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4374117175 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpcds_sf1.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │59.89 / 60.88 ±0.72 / 62.06 ms │ 57.67 / 58.72 ±0.65 / 59.66 ms │ no change │ │ QQuery 2 │ 287.90 / 294.59 ±7.63 / 309.50 ms │ 293.62 / 298.18 ±3.69 / 302.38 ms │ no change │ │ QQuery 3 │ 153.80 / 159.23 ±4.70 / 166.20 ms │ 152.32 / 158.94 ±4.31 / 163.80 ms │ no change │ │ QQuery 4 │ 1584.03 / 1634.24 ±32.13 / 1675.55 ms │ 1577.42 / 1608.11 ±39.11 / 1679.48 ms │ no change │ │ QQuery 5 │629.67 / 645.17 ±12.01 / 664.54 ms │ 622.07 / 671.39 ±29.87 / 705.61 ms │ no change │ │ QQuery 6 │270.48 / 279.48 ±11.42 / 301.90 ms │ 270.68 / 278.75 ±5.46 / 284.94 ms │ no change │ │ QQuery 7 │ 1226.87 / 1237.03 ±9.32 / 1252.20 ms │ 1221.31 / 1254.15 ±25.33 / 1294.88 ms │ no change │ │ QQuery 8 │ 221.33 / 228.10 ±4.98 / 236.47 ms │ 208.70 / 220.59 ±6.84 / 229.31 ms │ no change │ │ QQuery 9 │276.35 / 289.40 ±12.16 / 310.43 ms │ 268.77 / 288.58 ±21.61 / 330.69 ms │ no change │ │ QQuery 10 │ 170.87 / 172.98 ±2.50 / 177.70 ms │ 158.84 / 170.55 ±6.10 / 176.61 ms │ no change │ │ QQuery 11 │ .47 / 1144.32 ±38.99 / 1197.94 ms │ 1089.10 / 1132.32 ±40.07 / 1188.62 ms │ no change │ │ QQuery 12 │ 114.87 / 117.21 ±1.80 / 119.88 ms │ 115.88 / 120.01 ±4.10 / 126.97 ms │ no change │
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4374088542 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark clickbench_partitioned.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 0 │ 1.56 / 5.11 ±6.88 / 18.87 ms │ 1.68 / 5.20 ±6.84 / 18.88 ms │ no change │ │ QQuery 1 │23.35 / 25.87 ±1.81 / 28.40 ms │ 23.59 / 25.94 ±1.43 / 27.78 ms │ no change │ │ QQuery 2 │49.25 / 51.11 ±1.24 / 52.88 ms │ 49.29 / 49.91 ±0.58 / 50.75 ms │ no change │ │ QQuery 3 │39.42 / 41.33 ±1.66 / 43.82 ms │ 40.68 / 42.23 ±1.93 / 46.00 ms │ no change │ │ QQuery 4 │ 365.86 / 373.34 ±3.93 / 377.23 ms │ 259.15 / 274.15 ±9.58 / 285.94 ms │ +1.36x faster │ │ QQuery 5 │ 397.96 / 403.97 ±7.19 / 417.98 ms │ 335.46 / 338.66 ±2.94 / 342.29 ms │ +1.19x faster │ │ QQuery 6 │22.21 / 23.65 ±0.83 / 24.65 ms │ 22.59 / 23.81 ±1.30 / 26.25 ms │ no change │ │ QQuery 7 │55.15 / 55.83 ±0.70 / 57.07 ms │ 53.34 / 54.65 ±0.78 / 55.40 ms │ no change │ │ QQuery 8 │ 521.18 / 523.30 ±2.43 / 526.99 ms │ 361.02 / 367.67 ±8.02 / 383.16 ms │ +1.42x faster │ │ QQuery 9 │ 407.52 / 412.56 ±5.12 / 422.37 ms │ 414.74 / 434.76 ±14.19 / 453.97 ms │ 1.05x slower │ │ QQuery 10 │ 149.19 / 154.10 ±4.25 / 159.92 ms │ 147.57 / 149.96 ±1.52 / 151.64 ms │ no change │ │ QQuery 11 │ 167.96 / 171.41 ±2.75 / 175.08 ms │ 161.24 / 165.93 ±4.23 / 172.86 ms │ no change │ │ QQuery 12 │ 405.92 / 408.13 ±2.45 / 412.7
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4374065644 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpch_sf1.json ┏━━━┳┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇╇━╇━━━┩ │ QQuery 1 │ 81.08 / 84.06 ±1.65 / 85.99 ms │ 80.95 / 83.92 ±1.91 / 86.82 ms │ no change │ │ QQuery 2 │ 77.40 / 83.16 ±3.41 / 86.29 ms │ 82.28 / 84.42 ±1.29 / 85.65 ms │ no change │ │ QQuery 3 │ 380.72 / 404.84 ±21.00 / 439.20 ms │ 339.05 / 361.13 ±19.01 / 395.26 ms │ +1.12x faster │ │ QQuery 4 │ 62.13 / 69.48 ±4.55 / 74.12 ms │ 67.85 / 68.46 ±0.54 / 69.47 ms │ no change │ │ QQuery 5 │ 641.00 / 692.54 ±34.21 / 734.60 ms │ 601.13 / 625.35 ±16.84 / 645.13 ms │ +1.11x faster │ │ QQuery 6 │ 28.95 / 33.50 ±3.18 / 37.56 ms │ 28.30 / 29.87 ±1.28 / 32.15 ms │ +1.12x faster │ │ QQuery 7 │ 656.49 / 677.65 ±16.14 / 703.51 ms │ 616.89 / 648.76 ±25.71 / 685.81 ms │ no change │ │ QQuery 8 │ 633.58 / 665.74 ±23.16 / 699.52 ms │ 621.61 / 658.45 ±31.18 / 697.94 ms │ no change │ │ QQuery 9 │ 608.61 / 638.65 ±20.63 / 669.90 ms │ 577.22 / 594.91 ±16.49 / 626.07 ms │ +1.07x faster │ │ QQuery 10 │ 119.28 / 128.39 ±6.38 / 138.74 ms │ 112.93 / 120.19 ±4.76 / 125.52 ms │ +1.07x faster │ │ QQuery 11 │ 82.81 / 86.38 ±2.59 / 89.26 ms │ 78.55 / 85.89 ±3.91 / 89.20 ms │ no change │ │ QQuery 12 │ 661.09 / 685.12 ±16.92 / 711.09 ms │ 667.52 / 689.34 ±15.81 / 711.68 ms │ no change │ │ QQuery 13 │ 291.07 / 328.78 ±23.90 / 350.88 ms │ 287.39 / 315.76 ±19.41 / 343.61 m
Re: [PR] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4373999348 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373980023-2021-bkfjb 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4373997621 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373980023-2020-4g4qq 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4373993504 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4373980023-2019-qhxhm 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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 gabotechs/repartition-coalesce-batches-berfore-channels (dcf648255b92a34798871139aeba12d95f8f3032) to b2fd2d3 (merge-base) [diff](https://github.com/apache/datafusion/compare/b2fd2d32b053f15632b9fa6941b363d67fd6844e..dcf648255b92a34798871139aeba12d95f8f3032) 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
gabotechs commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4373980023 run benchmarks ``` env: DATAFUSION_EXECUTION_TARGET_PARTITIONS: 256 ``` -- 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] perf: coalesce batches before sending to distributor channels in RepartitionExec [datafusion]
adriangbot commented on PR #22010: URL: https://github.com/apache/datafusion/pull/22010#issuecomment-4373927206 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22010#issuecomment-4373801306) **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 gabotechs_repartition-coalesce-batches-berfore-channels Benchmark tpcds_sf1.json ┏━━━┳━━━┳━┳━━━┓ ┃ Query ┃ HEAD ┃ gabotechs_repartition-coalesce-batches-berfore-channels ┃Change ┃ ┡━━━╇━━━╇━╇━━━┩ │ QQuery 1 │13.79 / 14.18 ±0.65 / 15.46 ms │ 13.72 / 14.19 ±0.69 / 15.57 ms │ no change │ │ QQuery 2 │ 159.22 / 160.09 ±0.61 / 160.74 ms │ 160.90 / 162.32 ±0.82 / 163.14 ms │ no change │ │ QQuery 3 │ 117.84 / 119.51 ±1.38 / 121.35 ms │ 117.74 / 118.88 ±0.88 / 120.37 ms │ no change │ │ QQuery 4 │ 1318.19 / 1375.95 ±62.45 / 1477.81 ms │ 1342.78 / 1411.19 ±50.60 / 1498.56 ms │ no change │ │ QQuery 5 │ 202.08 / 203.76 ±1.18 / 205.00 ms │ 204.12 / 205.77 ±1.54 / 207.72 ms │ no change │ │ QQuery 6 │ 139.79 / 141.55 ±2.63 / 146.77 ms │ 144.42 / 145.12 ±1.05 / 147.20 ms │ no change │ │ QQuery 7 │ 414.19 / 418.61 ±4.50 / 427.00 ms │ 430.53 / 433.52 ±2.82 / 438.10 ms │ no change │ │ QQuery 8 │ 120.58 / 121.74 ±0.81 / 122.95 ms │ 122.07 / 123.38 ±1.53 / 126.34 ms │ no change │ │ QQuery 9 │ 124.62 / 136.64 ±8.09 / 148.23 ms │ 126.56 / 135.82 ±7.00 / 143.82 ms │ no change │ │ QQuery 10 │ 112.02 / 112.72 ±0.57 / 113.50 ms │ 110.68 / 113.28 ±2.98 / 118.65 ms │ no change │ │ QQuery 11 │ 882.55 / 890.18 ±7.52 / 903.32 ms │ 905.46 / 984.28 ±68.02 / 1092.79 ms │ 1.11x slower │ │ QQuery 12 │48.04 / 49.41 ±1.08 / 51.34 ms │ 48.95 / 49.24 ±0.22 / 49.58 ms │ no change │
