Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4836974480 > @alamb Thank you so much for all your time and guidance on this one! I learned a ton working through the architecture with you. I really appreciate your help throughout the whole process and for getting this merged! Thank you for bearing with my @pantShrey -- this is a tricky one as it involved API design and making sure we didn't mess up existing performance. I think it is a nice addition to the DataFusion APUs @philippemnoel 🥂 Any chance that one of you can help review an example I made using these new APIs: - https://github.com/apache/datafusion/pull/23170/changes -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3494591175
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
Review Comment:
Filed to track: https://github.com/apache/datafusion/issues/23247
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
Review Comment:
- Filed to track: https://github.com/apache/datafusion/issues/23247
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
philippemnoel commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4835783950 Thank you to the both of you!! 🙏 -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb merged PR #21882: URL: https://github.com/apache/datafusion/pull/21882 -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4835667468 @alamb Thank you so much for all your time and guidance on this one! I learned a ton working through the architecture with you. I really appreciate your help throughout the whole process and for getting this merged! -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4835478290 Al right -- I think this is looking good and I put it in th emerge 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4835262343 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834955270) **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 abstract-spill-file Benchmark external_aggr.json ┏━━┳┳━━━┳━━━┓ ┃ Query┃ HEAD ┃ abstract-spill-file ┃Change ┃ ┡━━╇╇━━━╇━━━┩ │ Q1(64.0 MB) │ 56.78 / 59.80 ±3.68 / 67.02 ms │55.29 / 57.89 ±2.47 / 61.73 ms │ no change │ │ Q1(32.0 MB) │ 53.78 / 56.67 ±1.67 / 58.29 ms │50.23 / 52.54 ±1.74 / 54.98 ms │ +1.08x faster │ │ Q1(16.0 MB) │ 52.48 / 55.14 ±2.37 / 58.50 ms │53.83 / 55.24 ±1.57 / 58.10 ms │ no change │ │ Q2(512.0 MB) │ 287.06 / 301.86 ±19.14 / 335.82 ms │ 267.01 / 282.20 ±9.85 / 295.77 ms │ +1.07x faster │ │ Q2(256.0 MB) │ 247.80 / 259.58 ±18.58 / 296.58 ms │ 244.64 / 247.82 ±2.49 / 251.51 ms │ no change │ │ Q2(128.0 MB) │ 249.87 / 277.59 ±42.37 / 359.92 ms │ 243.18 / 248.16 ±3.21 / 252.38 ms │ +1.12x faster │ │ Q2(64.0 MB) │ 250.91 / 264.10 ±20.55 / 305.06 ms │ 247.45 / 250.98 ±3.38 / 257.01 ms │ no change │ │ Q2(32.0 MB) │ 320.96 / 324.18 ±1.97 / 326.12 ms │ 316.89 / 320.98 ±3.33 / 326.99 ms │ no change │ └──┴┴───┴───┘ ┏┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡╇━━━┩ │ Total Time (HEAD) │ 1598.91ms │ │ Total Time (abstract-spill-file) │ 1515.80ms │ │ Average Time (HEAD)│ 199.86ms │ │ Average Time (abstract-spill-file) │ 189.47ms │ │ Queries Faster │ 3 │ │ Queries Slower │ 0 │ │ Queries with No Change │ 5 │ │ Queries with Failure │ 0 │ └┴───┘ ``` Resource Usage **external_aggr — base (merge-base)** | Metric | Value |
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4835124860 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834955270) **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 abstract-spill-file Benchmark sort_tpch1.json ┏━━━┳┳┳━━━┓ ┃ Query ┃ HEAD ┃ abstract-spill-file ┃Change ┃ ┡━━━╇╇╇━━━┩ │ Q1│ 110.91 / 111.86 ±0.85 / 113.26 ms │ 111.92 / 112.72 ±0.83 / 113.80 ms │ no change │ │ Q2│ 100.79 / 101.50 ±0.84 / 103.14 ms │ 99.56 / 100.30 ±0.65 / 101.49 ms │ no change │ │ Q3│ 644.30 / 650.02 ±4.19 / 656.33 ms │ 642.78 / 646.66 ±3.44 / 652.21 ms │ no change │ │ Q4│ 178.26 / 183.66 ±6.86 / 196.88 ms │ 179.31 / 180.62 ±1.19 / 182.80 ms │ no change │ │ Q5│ 260.34 / 261.52 ±1.53 / 264.46 ms │ 259.96 / 260.76 ±0.76 / 261.84 ms │ no change │ │ Q6│ 274.19 / 274.50 ±0.16 / 274.65 ms │ 273.69 / 274.75 ±1.01 / 276.56 ms │ no change │ │ Q7│ 446.89 / 449.24 ±2.71 / 454.46 ms │ 445.51 / 446.94 ±1.69 / 450.23 ms │ no change │ │ Q8│ 310.90 / 321.85 ±11.45 / 342.19 ms │ 310.48 / 319.68 ±10.01 / 337.86 ms │ no change │ │ Q9│ 332.15 / 349.14 ±14.24 / 366.84 ms │ 330.84 / 340.22 ±5.68 / 346.53 ms │ no change │ │ Q10 │ 466.39 / 473.45 ±4.67 / 479.97 ms │ 470.20 / 474.78 ±2.83 / 478.38 ms │ no change │ │ Q11 │ 229.22 / 247.24 ±17.39 / 271.41 ms │ 228.15 / 239.37 ±13.22 / 262.59 ms │ no change │ └───┴┴┴───┘ ┏┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡╇━━━┩ │ Total Time (HEAD) │ 3423.97ms │ │ Total Time (abstract-spill-file) │ 3396.80ms │ │ Average Time (HEAD)│ 311.27ms │ │ Average Time (abstract-spill-file) │ 308.80ms │ │ Queries Faster │ 0 │ │ Queries Slower │ 0 │ │ Queries with No Change │11 │ │ Queries with Failure
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4835064051 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834930095) **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 ``` group HEAD abstract-spill-file - --- spill_compression/q16/lz4_frame1.12 39.4±2.14ms? ?/sec 1.00 35.3±0.39ms? ?/sec spill_compression/q16/uncompressed 1.28 28.8±4.65ms? ?/sec 1.00 22.5±1.21ms? ?/sec spill_compression/q16/zstd 1.00 64.4±2.44ms? ?/sec 1.02 65.5±0.44ms? ?/sec spill_compression/q2/lz4_frame 1.08 18.5±2.08ms? ?/sec 1.00 17.2±0.14ms? ?/sec spill_compression/q2/uncompressed 1.77 14.0±1.21ms? ?/sec 1.00 7.9±0.42ms? ?/sec spill_compression/q2/zstd 1.02 29.7±0.29ms? ?/sec 1.00 29.2±0.38ms? ?/sec spill_compression/q20/lz4_frame1.05 26.0±3.18ms? ?/sec 1.00 24.8±0.70ms? ?/sec spill_compression/q20/uncompressed 1.65 19.0±4.11ms? ?/sec 1.00 11.5±0.62ms? ?/sec spill_compression/q20/zstd 1.01 46.2±1.00ms? ?/sec 1.00 45.8±0.59ms? ?/sec spill_compression/wide/lz4_frame 1.00 85.0±4.16ms? ?/sec 1.05 89.4±2.21ms? ?/sec spill_compression/wide/uncompressed1.4283.9±41.36ms? ?/sec 1.00 59.2±2.86ms? ?/sec spill_compression/wide/zstd1.00162.9±6.30ms? ?/sec 1.02165.9±0.72ms? ?/sec spill_io/StreamReader/read_100/1.50 36.6±4.89ms? ?/sec 1.00 24.4±0.77ms? ?/sec ``` Resource Usage **spill_io — base (merge-base)** | Metric | Value | ||---| | Wall time | 325.1s | | Peak memory | 263.4 MiB | | Avg memory | 56.3 MiB | | CPU user | 123.1s | | CPU sys | 47.1s | | Peak spill | 0 B | **spill_io — branch** | Metric | Value | ||---| | Wall time | 325.1s | | Peak memory | 304.1 MiB | | Avg memory | 80.4 MiB | | CPU user | 1
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4835064785 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834955270) **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 ``` group HEAD abstract-spill-file - --- spill_compression/q16/lz4_frame1.17 40.3±2.82ms? ?/sec 1.00 34.4±0.58ms? ?/sec spill_compression/q16/uncompressed 1.45 28.8±4.72ms? ?/sec 1.00 19.8±1.24ms? ?/sec spill_compression/q16/zstd 1.00 64.7±2.92ms? ?/sec 1.00 65.0±0.71ms? ?/sec spill_compression/q2/lz4_frame 1.09 18.4±2.12ms? ?/sec 1.00 16.9±0.21ms? ?/sec spill_compression/q2/uncompressed 1.98 14.8±4.53ms? ?/sec 1.00 7.5±0.37ms? ?/sec spill_compression/q2/zstd 1.05 30.5±3.25ms? ?/sec 1.00 29.2±0.38ms? ?/sec spill_compression/q20/lz4_frame1.05 25.6±2.96ms? ?/sec 1.00 24.3±0.72ms? ?/sec spill_compression/q20/uncompressed 1.97 19.5±5.44ms? ?/sec 1.00 9.9±0.35ms? ?/sec spill_compression/q20/zstd 1.02 46.3±0.90ms? ?/sec 1.00 45.3±0.41ms? ?/sec spill_compression/wide/lz4_frame 1.0592.4±16.09ms? ?/sec 1.00 87.6±1.33ms? ?/sec spill_compression/wide/uncompressed1.3980.4±10.30ms? ?/sec 1.00 57.9±3.06ms? ?/sec spill_compression/wide/zstd1.00162.8±7.24ms? ?/sec 1.01165.1±0.56ms? ?/sec spill_io/StreamReader/read_100/1.62 37.6±6.18ms? ?/sec 1.00 23.2±0.80ms? ?/sec ``` Resource Usage **spill_io — base (merge-base)** | Metric | Value | ||---| | Wall time | 235.1s | | Peak memory | 243.2 MiB | | Avg memory | 75.1 MiB | | CPU user | 119.7s | | CPU sys | 47.0s | | Peak spill | 0 B | **spill_io — branch** | Metric | Value | ||---| | Wall time | 280.1s | | Peak memory | 361.5 MiB | | Avg memory | 118.3 MiB | | CPU user |
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4834988099 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834955270) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4834955270-737-5gqsv 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (4b930402326bcf43d2d895ee541fcd1800b67abf) to 367f08e (merge-base) [diff](https://github.com/apache/datafusion/compare/367f08e9c315b41ed1f481af839f49b918c2a5cc..4b930402326bcf43d2d895ee541fcd1800b67abf) using: external_aggr 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4834988222 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834955270) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4834955270-738-55th8 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (4b930402326bcf43d2d895ee541fcd1800b67abf) to 367f08e (merge-base) [diff](https://github.com/apache/datafusion/compare/367f08e9c315b41ed1f481af839f49b918c2a5cc..4b930402326bcf43d2d895ee541fcd1800b67abf) using: sort_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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4834983537 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834955270) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4834955270-739-6fvlr 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (4b930402326bcf43d2d895ee541fcd1800b67abf) to 367f08e (merge-base) [diff](https://github.com/apache/datafusion/compare/367f08e9c315b41ed1f481af839f49b918c2a5cc..4b930402326bcf43d2d895ee541fcd1800b67abf) using: spill_io 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4834956265 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4834930095) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4834930095-736-2qkg5 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (4b930402326bcf43d2d895ee541fcd1800b67abf) to 367f08e (merge-base) [diff](https://github.com/apache/datafusion/compare/367f08e9c315b41ed1f481af839f49b918c2a5cc..4b930402326bcf43d2d895ee541fcd1800b67abf) using: spill_io 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4834955270 run benchmark external_aggr sort_tpch spill_io -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3493339272
##
datafusion/execution/src/disk_manager.rs:
##
@@ -492,8 +492,13 @@ impl SpillFile for RefCountedTempFile {
> {
match open_result {
Ok(file) => Box::pin(
-tokio_util::io::ReaderStream::new(file)
-.map(|r| r.map_err(DataFusionError::IoError)),
+// Use a 1MB read buffer. The default 8KB causes
excessive async
Review Comment:
👍
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4834930095 run benchmarks spill_io -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4834387047 run benchark spill_io -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4804303942 @alamb I spent some time looking into the `spill_io` regression. From what I can tell, increasing the `ReaderStream` capacity somewhat restores the benchmark performance locally. I initially tried going up to **1 MB**, but capacities around **256 KB** started failing the SQL logic tests, so for now I've pushed **128 KB**. My current understanding is that the previous `StreamReader` + `BufReader` implementation, despite having an 8 KB buffer, would typically read an entire IPC frame without yielding. With the current Tokio async stream, once the initial 8 KB is consumed, the task yields repeatedly while reading the remainder of the frame. Combined with the "copy" into the decoder's scratch buffer, this seems to add noticeable overhead for multi-MB frames/batches. ```rust tokio_util::io::ReaderStream::with_capacity(file, 128 * 1024) ``` I've pushed the change mainly so the CI benchmarks can run and to get your thoughts. If this direction makes sense, or should I make it configurable 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r346549
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
+/// Intended for close/sync/commit operations.
+fn finish(&mut self) -> Result<()>;
+}
+
+/// Factory for creating spill files.
+pub trait TempFileFactory:
+Send + Sync + std::panic::UnwindSafe + std::panic::RefUnwindSafe
Review Comment:
sure, I have removed the trait bounds
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3476470033
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
+/// Intended for close/sync/commit operations.
+fn finish(&mut self) -> Result<()>;
+}
+
+/// Factory for creating spill files.
+pub trait TempFileFactory:
+Send + Sync + std::panic::UnwindSafe + std::panic::RefUnwindSafe
Review Comment:
I think we should accept the "semver" break as this is a breaking API change
anyways
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3473897803
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
+/// Intended for close/sync/commit operations.
+fn finish(&mut self) -> Result<()>;
+}
+
+/// Factory for creating spill files.
+pub trait TempFileFactory:
+Send + Sync + std::panic::UnwindSafe + std::panic::RefUnwindSafe
Review Comment:
IIRC, it was a semver failure that forced me to add it.
Because `DiskManager` and `DiskManagerBuilder` now hold an `Arc`, they lost their auto-implemented `UnwindSafe` and
`RefUnwindSafe` traits. This triggered a `cargo-semver-checks` failure because
existing downstream code relies on `DiskManager` being unwind-safe:
```text
Checking datafusion-execution v53.1.0 -> v53.1.0 (no change; assume patch)
--- failure auto_trait_impl_removed: auto trait no longer implemented ---
type DiskManagerBuilder is no longer UnwindSafe
type DiskManagerBuilder is no longer RefUnwindSafe
type DiskManager is no longer UnwindSafe
...
```
I didn't have any other architectural reason for adding the bounds beyond
satisfying the checker. Since this PR already requires an upgrade guide, would
you prefer I remove these bounds and we just accept the semver break, or should
I keep them and add a doc comment explaining why they are there?
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3470735459
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
Review Comment:
It was also strange that the SpillWriter is a sync API, but the read stream
API is `async`
```rust
fn read_stream(&self) -> Result>
+ Send>>>;
```
I found this while working on an example showing how to write to a remote
object store
- https://github.com/apache/datafusion/pull/23170
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3470711669
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
+/// Intended for close/sync/commit operations.
+fn finish(&mut self) -> Result<()>;
+}
+
+/// Factory for creating spill files.
+pub trait TempFileFactory:
+Send + Sync + std::panic::UnwindSafe + std::panic::RefUnwindSafe
Review Comment:
Why does this need to be `std::panic::UnwindSafe +
std::panic::RefUnwindSafe`? If there is a good reason it should be documented
on the trait as well I think
I hit this while working on an example in
- https://github.com/apache/datafusion/pull/23170
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+}
+
+/// Writer for spill file backends.
+pub trait SpillWriter: std::io::Write + Send {
+/// Intended for close/sync/commit operations.
+fn finish(&mut self) -> Result<()>;
+}
+
+/// Factory for creating spill files.
+pub trait TempFileFactory:
+Send + Sync + std::panic::UnwindSafe + std::panic::RefUnwindSafe
Review Comment:
Why does this need to be `std::panic::UnwindSafe +
std::panic::RefUnwindSafe`? If there is a good reason it should be documented
on the trait as well I think
I hit this while working on an example of using this API in
- https://github.com/apache/datafusion/pull/23170
--
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]
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4794087315 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **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 abstract-spill-file Benchmark external_aggr.json ┏━━┳┳┳━━┓ ┃ Query┃ HEAD ┃ abstract-spill-file ┃ Change ┃ ┡━━╇╇╇━━┩ │ Q1(64.0 MB) │ 56.10 / 59.52 ±2.89 / 64.35 ms │ 58.65 / 62.52 ±3.23 / 68.39 ms │ 1.05x slower │ │ Q1(32.0 MB) │ 50.06 / 53.68 ±2.49 / 57.65 ms │ 56.73 / 58.42 ±1.71 / 61.63 ms │ 1.09x slower │ │ Q1(16.0 MB) │ 51.58 / 55.33 ±2.55 / 58.88 ms │ 60.85 / 62.24 ±0.72 / 62.82 ms │ 1.12x slower │ │ Q2(512.0 MB) │ 282.03 / 304.71 ±22.38 / 333.45 ms │ 269.78 / 299.49 ±27.33 / 347.85 ms │no change │ │ Q2(256.0 MB) │ 248.60 / 263.29 ±16.62 / 293.08 ms │ 260.72 / 267.83 ±6.19 / 277.68 ms │no change │ │ Q2(128.0 MB) │ 245.61 / 269.83 ±40.40 / 350.46 ms │ 264.78 / 269.21 ±4.26 / 277.30 ms │no change │ │ Q2(64.0 MB) │ 248.06 / 251.73 ±2.85 / 255.35 ms │ 281.25 / 284.44 ±2.43 / 286.85 ms │ 1.13x slower │ │ Q2(32.0 MB) │ 319.21 / 365.31 ±71.36 / 504.54 ms │ 365.67 / 371.74 ±6.17 / 383.59 ms │no change │ └──┴┴┴──┘ ┏┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡╇━━━┩ │ Total Time (HEAD) │ 1623.40ms │ │ Total Time (abstract-spill-file) │ 1675.90ms │ │ Average Time (HEAD)│ 202.93ms │ │ Average Time (abstract-spill-file) │ 209.49ms │ │ Queries Faster │ 0 │ │ Queries Slower │ 4 │ │ Queries with No Change │ 4 │ │ Queries with Failure │ 0 │ └┴───┘ ``` Resource Usage **external_aggr — base (merge-base)** | Metric | Value |
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4794007951 🤔 hmm the spill_io benchmark looks pretty bad: https://github.com/apache/datafusion/pull/21882#issuecomment-4793993789 -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793993789 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **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 ``` group HEAD abstract-spill-file - --- spill_compression/q16/lz4_frame1.00 39.0±3.66ms? ?/sec 1.47 57.5±0.89ms? ?/sec spill_compression/q16/uncompressed 1.00 36.3±7.70ms? ?/sec 2.21 80.2±1.54ms? ?/sec spill_compression/q16/zstd 1.00 63.7±3.56ms? ?/sec 1.24 78.7±2.59ms? ?/sec spill_compression/q2/lz4_frame 1.00 18.9±3.68ms? ?/sec 1.40 26.5±0.43ms? ?/sec spill_compression/q2/uncompressed 1.0019.1±11.60ms? ?/sec 1.91 36.5±0.73ms? ?/sec spill_compression/q2/zstd 1.0032.4±10.13ms? ?/sec 1.04 33.8±0.47ms? ?/sec spill_compression/q20/lz4_frame1.00 25.6±2.86ms? ?/sec 1.49 38.2±0.70ms? ?/sec spill_compression/q20/uncompressed 1.00 23.5±3.94ms? ?/sec 2.06 48.4±0.85ms? ?/sec spill_compression/q20/zstd 1.00 45.9±0.66ms? ?/sec 1.15 53.0±1.45ms? ?/sec spill_compression/wide/lz4_frame 1.00 92.4±6.29ms? ?/sec 1.58145.7±1.73ms? ?/sec spill_compression/wide/uncompressed1.00 99.8±5.27ms? ?/sec 2.17216.7±3.17ms? ?/sec spill_compression/wide/zstd1.00162.5±5.29ms? ?/sec 1.23199.7±2.65ms? ?/sec spill_io/StreamReader/read_100/1.00 51.2±9.19ms? ?/sec 2.88147.2±3.32ms? ?/sec ``` Resource Usage **spill_io — base (merge-base)** | Metric | Value | ||---| | Wall time | 475.1s | | Peak memory | 243.3 MiB | | Avg memory | 35.7 MiB | | CPU user | 119.4s | | CPU sys | 44.8s | | Peak spill | 0 B | **spill_io — branch** | Metric | Value | ||---| | Wall time | 505.1s | | Peak memory | 335.5 MiB | | Avg memory | 65.3 MiB | | CPU user | 1
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793992499 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **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 abstract-spill-file Benchmark smj.json ┏━━━┳━━┳━━┳━━━┓ ┃ Query ┃ HEAD ┃ abstract-spill-file ┃Change ┃ ┡━━━╇━━╇━━╇━━━┩ │ QQuery 1 │ 8.73 / 8.87 ±0.11 / 9.03 ms │ 8.48 / 8.65 ±0.12 / 8.81 ms │ no change │ │ QQuery 2 │173.51 / 175.16 ±1.62 / 177.82 ms │172.88 / 177.16 ±4.98 / 183.72 ms │ no change │ │ QQuery 3 │105.93 / 106.85 ±0.64 / 107.59 ms │106.49 / 107.72 ±0.93 / 109.25 ms │ no change │ │ QQuery 4 │ 28.39 / 28.58 ±0.13 / 28.75 ms │ 28.65 / 28.85 ±0.17 / 29.12 ms │ no change │ │ QQuery 5 │ 21.67 / 21.84 ±0.20 / 22.20 ms │ 21.72 / 21.99 ±0.26 / 22.44 ms │ no change │ │ QQuery 6 │169.58 / 175.41 ±5.59 / 185.10 ms │175.53 / 178.79 ±3.86 / 185.76 ms │ no change │ │ QQuery 7 │211.12 / 216.14 ±9.21 / 234.55 ms │212.57 / 214.02 ±0.86 / 215.20 ms │ no change │ │ QQuery 8 │ 20.17 / 20.42 ±0.16 / 20.56 ms │ 20.36 / 20.63 ±0.21 / 20.94 ms │ no change │ │ QQuery 9 │214.68 / 222.24 ±9.45 / 240.74 ms │215.47 / 218.22 ±2.04 / 221.07 ms │ no change │ │ QQuery 10 │ 69.59 / 75.10 ±4.96 / 84.24 ms │ 72.36 / 75.92 ±3.20 / 80.88 ms │ no change │ │ QQuery 11 │ 27.25 / 27.42 ±0.12 / 27.56 ms │ 27.35 / 27.86 ±0.29 / 28.23 ms │ no change │ │ QQuery 12 │ 70.02 / 72.16 ±1.55 / 73.84 ms │ 65.35 / 69.86 ±2.64 / 73.45 ms │ no change │ │ QQuery 13 │109.20 / 112.69 ±4.81 / 122.15 ms │101.47 / 108.63 ±5.19 / 117.24 ms │ no change │ │ QQuery 14 │ 70.41 / 72.19 ±1.18 / 73.75 ms │ 70.46 / 71.34 ±0.95 / 73.05 ms │ no change │ │ QQuery 15 │ 70.53 / 77.58 ±8.14 / 93.44 ms │ 71.41 / 72.53 ±1.31 / 75.09 ms │ +1.07x faster │ │ QQuery 16 │ 12.81 / 13.02 ±0.17 / 13.24 ms │ 12.97 /
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793990828 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **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 abstract-spill-file Benchmark sort_tpch1.json ┏━━━┳┳┳━━━┓ ┃ Query ┃ HEAD ┃ abstract-spill-file ┃Change ┃ ┡━━━╇╇╇━━━┩ │ Q1│ 164.84 / 165.66 ±0.59 / 166.56 ms │ 165.50 / 166.07 ±0.59 / 167.15 ms │ no change │ │ Q2│ 140.97 / 141.78 ±0.82 / 143.24 ms │ 141.99 / 143.41 ±1.15 / 145.17 ms │ no change │ │ Q3│ 653.09 / 655.34 ±1.69 / 658.07 ms │ 651.11 / 655.03 ±2.16 / 656.93 ms │ no change │ │ Q4│ 195.33 / 197.64 ±3.57 / 204.73 ms │ 196.02 / 197.48 ±1.33 / 199.45 ms │ no change │ │ Q5│ 279.10 / 280.15 ±0.94 / 281.71 ms │ 279.14 / 280.18 ±0.83 / 281.11 ms │ no change │ │ Q6│ 293.39 / 295.03 ±0.90 / 295.85 ms │ 292.47 / 297.87 ±6.15 / 308.42 ms │ no change │ │ Q7│ 472.58 / 475.57 ±2.16 / 477.88 ms │ 466.57 / 468.88 ±1.57 / 470.91 ms │ no change │ │ Q8│ 333.33 / 341.98 ±7.37 / 352.17 ms │ 327.11 / 331.25 ±2.63 / 335.19 ms │ no change │ │ Q9│ 347.86 / 362.05 ±11.58 / 375.08 ms │ 342.58 / 350.16 ±10.85 / 371.25 ms │ no change │ │ Q10 │ 500.05 / 519.53 ±20.74 / 546.77 ms │ 484.92 / 493.79 ±6.00 / 501.49 ms │ no change │ │ Q11 │ 246.90 / 256.73 ±9.84 / 274.39 ms │ 245.79 / 252.65 ±12.38 / 277.40 ms │ no change │ └───┴┴┴───┘ ┏┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡╇━━━┩ │ Total Time (HEAD) │ 3691.47ms │ │ Total Time (abstract-spill-file) │ 3636.79ms │ │ Average Time (HEAD)│ 335.59ms │ │ Average Time (abstract-spill-file) │ 330.62ms │ │ Queries Faster │ 0 │ │ Queries Slower │ 0 │ │ Queries with No Change │11 │ │ Queries with Failure
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793942224 The other thing I think that would be helpful in crafting this API is an example of how to provide a user defined spill manager. For example it would be interesting to try and implement one that is backed by an ObjectStore to show how a user could use a remote source as a spill location I will try and make such an example using this API and report back -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793893402 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4793874108-671-fcdf7 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (4904cd774f509a29ccb661e84a80cbe2e362ef50) to 5fcc550 (merge-base) [diff](https://github.com/apache/datafusion/compare/5fcc5507f4a683eaa46cde435a5200f2d588f01b..4904cd774f509a29ccb661e84a80cbe2e362ef50) using: sort_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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3470390324
##
datafusion/physical-plan/src/spill/spill_manager.rs:
##
@@ -25,8 +25,8 @@ use arrow::datatypes::{ByteViewType, SchemaRef};
use arrow::record_batch::RecordBatch;
use datafusion_common::{DataFusionError, Result, config::SpillCompression};
use datafusion_execution::SendableRecordBatchStream;
-use datafusion_execution::disk_manager::RefCountedTempFile;
use datafusion_execution::runtime_env::RuntimeEnv;
+use datafusion_execution::spill_file::SpillFile;
Review Comment:
it is nice that this function basically just changes`RefCountedTempFile` to
`Arc` 👍
##
datafusion/physical-plan/src/spill/mod.rs:
##
@@ -39,254 +36,216 @@ use arrow::array::{
Array, ArrayRef, BinaryViewArray, BufferSpec, GenericByteViewArray,
StringViewArray,
layout, make_array,
};
+use arrow::buffer::Buffer;
use arrow::datatypes::DataType;
use arrow::datatypes::{ByteViewType, Schema, SchemaRef};
use arrow::ipc::{
MetadataVersion,
-reader::StreamReader,
+reader::StreamDecoder,
writer::{IpcWriteOptions, StreamWriter},
};
use arrow::record_batch::RecordBatch;
use arrow_data::ArrayDataBuilder;
use arrow_ipc::CompressionType;
+use datafusion_common::Result;
use datafusion_common::config::SpillCompression;
-use datafusion_common::{DataFusionError, Result, exec_datafusion_err,
exec_err};
-use datafusion_common_runtime::SpawnedTask;
use datafusion_execution::RecordBatchStream;
-use datafusion_execution::disk_manager::RefCountedTempFile;
-use futures::{FutureExt as _, Stream};
+use datafusion_execution::spill_file::SpillFile;
+use futures::Stream;
use log::debug;
-/// Stream that reads spill files from disk where each batch is read in a
spawned blocking task
-/// It will read one batch at a time and will not do any buffering, to buffer
data use [`crate::common::spawn_buffered`]
-///
-/// A simpler solution would be spawning a long-running blocking task for each
-/// file read (instead of each batch). This approach does not work because when
-/// the number of concurrent reads exceeds the Tokio thread pool limit,
-/// deadlocks can occur and block progress.
+/// Stream that reads spill files from a [`SpillFile`] backend as a stream of
[`RecordBatch`]es.
+/// Uses [`StreamDecoder`] to decode IPC bytes received from the backend's
async byte stream.
+/// Backends handle their own threading concerns internally - OS files use
+/// `tokio::fs::File` which performs blocking IO per-syscall without holding a
thread
+/// for the file's lifetime, avoiding deadlocks when concurrent reads exceed
thread pool limits.
struct SpillReaderStream {
schema: SchemaRef,
-state: SpillReaderStreamState,
+decoder: StreamDecoder,
+byte_stream: Pin> + Send>>,
+is_done: bool,
+
/// Maximum memory size observed among spilling sorted record batches.
/// This is used for validation purposes during reading each RecordBatch
from spill.
/// For context on why this value is recorded and validated,
/// see `physical_plan/sort/multi_level_merge.rs`.
max_record_batch_memory: Option,
-}
-
-// Small margin allowed to accommodate slight memory accounting variation
-const SPILL_BATCH_MEMORY_MARGIN: usize = 4096;
-
-/// When we poll for the next batch, we will get back both the batch and the
reader,
-/// so we can call `next` again.
-type NextRecordBatchResult = Result<(StreamReader>,
Option)>;
-
-enum SpillReaderStreamState {
-/// Initial state: the stream was not initialized yet
-/// and the file was not opened
-Uninitialized(RefCountedTempFile),
-/// A read is in progress in a spawned blocking task for which we hold the
handle.
-ReadInProgress(SpawnedTask),
+/// Holds leftover bytes from a chunk when a batch is yielded early
+current_buffer: Buffer,
-/// A read has finished and we wait for being polled again in order to
start reading the next batch.
-Waiting(StreamReader>),
+/// Keeps the file alive until the stream is dropped
+_spill_file: Arc,
-/// The stream has finished, successfully or not.
-Done,
+schema_validated: bool,
}
+// Small margin allowed to accommodate slight memory accounting variation
+const SPILL_BATCH_MEMORY_MARGIN: usize = 4096;
+
impl SpillReaderStream {
fn new(
schema: SchemaRef,
-spill_file: RefCountedTempFile,
+spill_file: Arc,
max_record_batch_memory: Option,
-) -> Self {
-Self {
+) -> Result {
+let byte_stream = spill_file.read_stream()?;
+// DataFusion controls what it writes so it can trust its own IPC
output,
+// matching the behavior of the previous StreamReader-based
implementation.
+let decoder = unsafe { StreamDecoder::new().with_skip_validation(true)
};
+Ok(Self {
schema,
-state: SpillReaderStreamState::Uninitia
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793892211 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4793874108-672-kd9hg 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (4904cd774f509a29ccb661e84a80cbe2e362ef50) to 5fcc550 (merge-base) [diff](https://github.com/apache/datafusion/compare/5fcc5507f4a683eaa46cde435a5200f2d588f01b..4904cd774f509a29ccb661e84a80cbe2e362ef50) using: spill_io 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793889106 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4793874108-670-f6jmk 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (024ec98290b4be1f82919d694133e076b4125ba3) to 5fcc550 (merge-base) [diff](https://github.com/apache/datafusion/compare/5fcc5507f4a683eaa46cde435a5200f2d588f01b..024ec98290b4be1f82919d694133e076b4125ba3) using: smj 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793881205 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4793874108-669-jrcrv 6.12.85+ #1 SMP Mon May 11 08:17:35 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 abstract-spill-file (024ec98290b4be1f82919d694133e076b4125ba3) to 5fcc550 (merge-base) [diff](https://github.com/apache/datafusion/compare/5fcc5507f4a683eaa46cde435a5200f2d588f01b..024ec98290b4be1f82919d694133e076b4125ba3) using: external_aggr 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793874108 run benchmark external_aggr smj sort_tpch spill_io -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793870463 > From what I can see, ResourcesExhausted is exclusively caught by operators to handle memory limits (which triggers the fallback to start spilling). Conversely, hitting a disk limit during an active spill has no fallback -- it is a fatal error that simply aborts the query. Because of this, I removed the adapter's error-stashing workaround and allowed the disk quota failures to just bubble up as standard std::io::Errors with the descriptive text, since the end-user UX and control flow remain identical. i suppose we could also just define something that looked like std::io::write but had a different error type 🤮 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4793740333 > My current compromise is a very small tracking wrapper around SpillWriter that simply forwards std::io::Write calls while counting the exact serialized bytes written by Arrow. This keeps the metrics accurate for compressed spills and also allows backend-local quota tracking to remain at the write boundary rather than requiring every backend to implement current_disk_usage()-style APIs. This makes sense to me -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4787103055 @alamb, extremely sorry for the delay in pushing the latest changes. While waiting on your guidance regarding the RecordBatch abstraction, I started working through the other review comments, but I ended up getting fairly stuck on a design decision around the writer adapter and spent more time on it than I expected. I did make the writer-side changes you suggested: `SpillWriter` now extends `std::io::Write`, which allowed me to remove the original `SpillWriteAdapter` entirely. However, removing the adapter exposed two issues. 1) The first is metrics tracking. The adapter was previously the place where I could observe the exact number of compressed IPC bytes written to the backend and update both the global disk usage accounting and the `spilled_bytes` metrics. This also removed the need for `InProgressSpillFile` to repeatedly call `current_disk_usage()` / `update_disk_usage()`. Without that interception point, I'm struggling to see how to accurately track spill metrics. Today the only estimate available at the RecordBatch level is derived from `get_array_memory_size()`, which reflects the in-memory Arrow representation rather than the final serialized IPC payload. Once IPC compression (LZ4/ZSTD) is enabled, that value can differ substantially from the actual bytes written to the backend. Additionally, Arrow's `StreamWriter` doesn't expose the number of bytes written per batch, so there doesn't seem to be a way to observe the final serialized size other than wrapping the `std::io::Write` boundary itself. My current compromise is a very small tracking wrapper around `SpillWriter` that simply forwards `std::io::Write` calls while counting the exact serialized bytes written by Arrow. This keeps the metrics accurate for compressed spills and also allows backend-local quota tracking to remain at the write boundary rather than requiring every backend to implement `current_disk_usage()`-style APIs. Do you think this approach is acceptable, or would you prefer that I revert to the explicit disk-usage tracking approach instead? My concern with the latter is that it becomes difficult for non-filesystem backends to implement efficiently and increases the backend API surface, while still only providing an approximation of the actual spill size. 2) The second issue is error propagation. Before this change, quota enforcement could return `DataFusionError::ResourcesExhausted` directly through the `datafusion_common::Result` path(`drain()` allowed skipping the io boundary). After moving the interface to `std::io::Write`, quota failures have to cross the `std::io::Error` boundary. As far as I can tell, that makes it difficult to preserve `ResourcesExhausted` semantics all the way up to callers such as the spill operators, which currently check for that specific error type. However, I audited the codebase (e.g., `row_hash.rs`, `nestedloopjoin.rs`) to see if losing this specific enum variant actually breaks any control flow. From what I can see, `ResourcesExhausted` is exclusively caught by operators to handle memory limits (which triggers the fallback to start spilling). Conversely, hitting a disk limit during an active spill has no fallback -- it is a fatal error that simply aborts the query. Because of this, I removed the ad apter's error-stashing workaround and allowed the disk quota failures to just bubble up as standard `std::io::Error`s with the descriptive text, since the end-user UX and control flow remain identical. On the read path, I'd also like to gently push back on returning `Box`, if that's okay. The original implementation used `StreamReader>`, which required `spawn_blocking` to avoid blocking the async executor. Avoiding that thread-pool dependency was the original motivation for the state machine. I replaced that path with Arrow's `StreamDecoder` fed by `tokio::fs::File` + `ReaderStream`, which keeps the read side fully async without requiring `spawn_blocking`. You're absolutely right that this introduces an extra buffering step compared to `StreamReader`, so there is a real tradeoff there. My thinking was that the async behaviour was worth that cost, but I'm happy to discuss further if you feel differently. I've also started experimenting locally with the RecordBatch-level abstraction you suggested, but I didn't want to go too far without first getting your direction. Given the above, would you prefer that I continue down the RecordBatch route and revisit these APIs as part of that refactor, or should I finish the current approach first and keep the RecordBatch abstraction as a follow-up? -- 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.
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4753642262 @alamb Thank you so much for going through this! The `RecordBatch`-level abstraction idea is really interesting, and I'll go through the implications more comprehensively , but two things come to mind right away that are worth aligning on: 1. **Dependency direction:** `SpillFile` lives in `datafusion-execution`, which doesn't currently depend on `arrow-ipc`. Moving IPC encoding/decoding into the OS backend would add that dependency. 2. **Schema:** If `read_stream()` returns a `SendableRecordBatchStream`, the backend needs a schema to initialize the decoder either stored at creation time or passed as a `SchemaRef` parameter. I do think this is the right long-term direction, and it would let us clean up `SpillWriteAdapter` entirely. That said, since this PR is currently blocking ParadeDB's Postgres integration, I'm a little worried that expanding the scope here might stall them further. Happy to defer to your judgement on whether to tackle it in this PR or as a follow-up! -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3443817678
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
Review Comment:
The current path doesn't use `StreamReader` at all , it uses
`StreamDecoder`, Arrow's push-based API where you feed Bytes chunks and it
emits `RecordBatch`es without a `Read` impl. Returning `Box` would
reintroduce the thread pool exhaustion and the old state machine that existed
to solve it. The `StreamDecoder` path keeps the read fully async via
`tokio::fs::File` + `ReaderStream`
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3443817678
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
Review Comment:
The current path doesn't use `StreamReader` at all , it uses
`StreamDecoder`, Arrow's push-based API where you feed Bytes chunks and it
emits `RecordBatch`es without a `Read` impl. Returning `Box` would
reintroduce the thread pool exhaustion and the old state machine existed to
solve. The `StreamDecoder` path keeps the read fully async via
`tokio::fs::File` + `ReaderStream`
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3443817678
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
Review Comment:
The current path doesn't use `StreamReader` at all , it uses
`StreamDecoder`, Arrow's push-based API where you feed Bytes chunks and it
emits `RecordBatch`es without a `Read` impl. Returning `Box would
reintroduce the thread pool exhaustion and the old state machine existed to
solve. The `StreamDecoder` path keeps the read fully async via
`tokio::fs::File` + `ReaderStream`
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882: URL: https://github.com/apache/datafusion/pull/21882#discussion_r3443799909 ## datafusion/physical-plan/src/spill/mod.rs: ## Review Comment: Thank you!! I will remove the function and the related struct. -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3443797123
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -3098,9 +3098,9 @@ mod test {
let input_partitions = vec![partition1, partition2];
// Set up context with tight memory limit to force spilling
-// Sorting needs some non-spillable memory, so 64 bytes should force
spilling while still allowing the query to complete
+// Sorting needs some non-spillable memory, so 608 bytes should force
spilling while still allowing the query to complete
Review Comment:
I touched on this earlier in the PR (see the thread with @adriangb above),
but happy to revisit! The test is extremely sensitive to memory thresholds ,the
`SpillWriteAdapter` and the async read infrastructure together appear to push
the baseline up enough that the original 64B limit becomes too tight for the
operator to initialize, but increasing the limit caused `RepartitionExec` and
`RepartitionMerge` to compete for the pool in a way that didn't trigger
reliable spilling. I'll be honest that I'm not fully confident in exactly how
much each piece contributes to the memory shift, which is part of why I
preserved the semantic intent of the test (a spill occurs and output is
correctly sorted) rather than trying to nail down a precise new threshold. I'd
really appreciate your guidance on the best approach here, happy to go
whichever direction you think is cleanest.
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3443798942
##
datafusion/physical-plan/src/spill/mod.rs:
##
@@ -274,19 +258,73 @@ pub fn spill_record_batch_by_size(
Ok(())
}
-/// Write in Arrow IPC Stream format to a file.
-///
-/// Stream format is used for spill because it supports dictionary
replacement, and the random
-/// access of IPC File format is not needed (IPC File format doesn't support
dictionary replacement).
+/// An adapter that implements `std::io::Write` to bridge Arrow's synchronous
Review Comment:
Agreed, happy to make this change! The adapter was a bit unfortunate, it
ended up there because in the issue discussion we'd moved away from using
std::io::Write on the main trait, which led me toward Bytes to give backends
owned data. But as you point out, that just pushed the intermediate copy one
level up into the adapter itself. Switching to &[u8] directly lets us delete
the adapter entirely, which is a much cleaner outcome.
##
datafusion/execution/src/disk_manager.rs:
##
@@ -402,59 +452,7 @@ impl RefCountedTempFile {
self.tempfile.as_ref()
}
-/// Updates the global disk usage counter after modifications to the
underlying file.
-///
-/// # Errors
-/// - Returns an error if the global disk usage exceeds the configured
limit.
-pub fn update_disk_usage(&mut self) -> Result<()> {
-// Get new file size from OS
-let metadata = self.tempfile.as_file().metadata()?;
-let new_disk_usage = metadata.len();
-
-// Get the old disk usage
Review Comment:
I moved the tracking to inside `FileSpillWriter::write()`, updating the
counter incrementally per write, so the callers don't need calling
`update_disk_usage()` after each write
##
datafusion/physical-plan/src/spill/mod.rs:
##
@@ -39,212 +39,187 @@ use arrow::array::{
Array, ArrayRef, BinaryViewArray, BufferSpec, GenericByteViewArray,
StringViewArray,
layout, make_array,
};
+use arrow::buffer::Buffer;
use arrow::datatypes::DataType;
use arrow::datatypes::{ByteViewType, Schema, SchemaRef};
use arrow::ipc::{
MetadataVersion,
-reader::StreamReader,
+reader::StreamDecoder,
writer::{IpcWriteOptions, StreamWriter},
};
use arrow::record_batch::RecordBatch;
use arrow_data::ArrayDataBuilder;
use arrow_ipc::CompressionType;
use datafusion_common::config::SpillCompression;
-use datafusion_common::{DataFusionError, Result, exec_datafusion_err,
exec_err};
-use datafusion_common_runtime::SpawnedTask;
+use datafusion_common::{DataFusionError, Result, exec_datafusion_err};
use datafusion_execution::RecordBatchStream;
-use datafusion_execution::disk_manager::RefCountedTempFile;
-use futures::{FutureExt as _, Stream};
+use datafusion_execution::spill_file::SpillFile;
+use futures::Stream;
use log::debug;
-/// Stream that reads spill files from disk where each batch is read in a
spawned blocking task
-/// It will read one batch at a time and will not do any buffering, to buffer
data use [`crate::common::spawn_buffered`]
-///
-/// A simpler solution would be spawning a long-running blocking task for each
-/// file read (instead of each batch). This approach does not work because when
-/// the number of concurrent reads exceeds the Tokio thread pool limit,
-/// deadlocks can occur and block progress.
+/// Stream that reads spill files from a [`SpillFile`] backend as a stream of
[`RecordBatch`]es.
+/// Uses [`StreamDecoder`] to decode IPC bytes received from the backend's
async byte stream.
+/// Backends handle their own threading concerns internally - OS files use
+/// `tokio::fs::File` which performs blocking IO per-syscall without holding a
thread
+/// for the file's lifetime, avoiding deadlocks when concurrent reads exceed
thread pool limits.
struct SpillReaderStream {
schema: SchemaRef,
-state: SpillReaderStreamState,
+decoder: StreamDecoder,
+byte_stream: Pin> + Send>>,
Review Comment:
You're right that `StreamReader` reads directly into its parse buffer
without an intermediate Bytes object, so it genuinely avoids that step. The
tradeoff is that driving it requires `spawn_blocking`, which reintroduces the
thread pool exhaustion problem the old state machine existed to solve. The
`StreamDecoder` path accepts that slight extra buffering in exchange for
staying fully async.
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3443798268
##
datafusion/physical-plan/src/spill/mod.rs:
##
@@ -39,212 +39,187 @@ use arrow::array::{
Array, ArrayRef, BinaryViewArray, BufferSpec, GenericByteViewArray,
StringViewArray,
layout, make_array,
};
+use arrow::buffer::Buffer;
use arrow::datatypes::DataType;
use arrow::datatypes::{ByteViewType, Schema, SchemaRef};
use arrow::ipc::{
MetadataVersion,
-reader::StreamReader,
+reader::StreamDecoder,
writer::{IpcWriteOptions, StreamWriter},
};
use arrow::record_batch::RecordBatch;
use arrow_data::ArrayDataBuilder;
use arrow_ipc::CompressionType;
use datafusion_common::config::SpillCompression;
-use datafusion_common::{DataFusionError, Result, exec_datafusion_err,
exec_err};
-use datafusion_common_runtime::SpawnedTask;
+use datafusion_common::{DataFusionError, Result, exec_datafusion_err};
use datafusion_execution::RecordBatchStream;
-use datafusion_execution::disk_manager::RefCountedTempFile;
-use futures::{FutureExt as _, Stream};
+use datafusion_execution::spill_file::SpillFile;
+use futures::Stream;
use log::debug;
-/// Stream that reads spill files from disk where each batch is read in a
spawned blocking task
-/// It will read one batch at a time and will not do any buffering, to buffer
data use [`crate::common::spawn_buffered`]
-///
-/// A simpler solution would be spawning a long-running blocking task for each
-/// file read (instead of each batch). This approach does not work because when
-/// the number of concurrent reads exceeds the Tokio thread pool limit,
-/// deadlocks can occur and block progress.
+/// Stream that reads spill files from a [`SpillFile`] backend as a stream of
[`RecordBatch`]es.
+/// Uses [`StreamDecoder`] to decode IPC bytes received from the backend's
async byte stream.
+/// Backends handle their own threading concerns internally - OS files use
+/// `tokio::fs::File` which performs blocking IO per-syscall without holding a
thread
+/// for the file's lifetime, avoiding deadlocks when concurrent reads exceed
thread pool limits.
struct SpillReaderStream {
schema: SchemaRef,
-state: SpillReaderStreamState,
+decoder: StreamDecoder,
+byte_stream: Pin> + Send>>,
+is_done: bool,
+
/// Maximum memory size observed among spilling sorted record batches.
/// This is used for validation purposes during reading each RecordBatch
from spill.
/// For context on why this value is recorded and validated,
/// see `physical_plan/sort/multi_level_merge.rs`.
max_record_batch_memory: Option,
-}
-// Small margin allowed to accommodate slight memory accounting variation
-const SPILL_BATCH_MEMORY_MARGIN: usize = 4096;
+/// Holds leftover bytes from a chunk when a batch is yielded early
+current_buffer: Buffer,
-/// When we poll for the next batch, we will get back both the batch and the
reader,
-/// so we can call `next` again.
-type NextRecordBatchResult = Result<(StreamReader>,
Option)>;
+/// Keeps the file alive until the stream is dropped
+_spill_file: Arc,
-enum SpillReaderStreamState {
-/// Initial state: the stream was not initialized yet
-/// and the file was not opened
-Uninitialized(RefCountedTempFile),
-
-/// A read is in progress in a spawned blocking task for which we hold the
handle.
-ReadInProgress(SpawnedTask),
-
-/// A read has finished and we wait for being polled again in order to
start reading the next batch.
-Waiting(StreamReader>),
-
-/// The stream has finished, successfully or not.
-Done,
+schema_validated: bool,
}
+// Small margin allowed to accommodate slight memory accounting variation
+const SPILL_BATCH_MEMORY_MARGIN: usize = 4096;
+
impl SpillReaderStream {
fn new(
schema: SchemaRef,
-spill_file: RefCountedTempFile,
+spill_file: Arc,
max_record_batch_memory: Option,
-) -> Self {
-Self {
+) -> Result {
+let byte_stream = spill_file.read_stream()?;
+// DataFusion controls what it writes so it can trust its own IPC
output,
+// matching the behavior of the previous StreamReader-based
implementation.
+let decoder = unsafe { StreamDecoder::new().with_skip_validation(true)
};
+Ok(Self {
schema,
-state: SpillReaderStreamState::Uninitialized(spill_file),
+decoder,
+byte_stream,
max_record_batch_memory,
-}
+is_done: false,
+current_buffer: Buffer::from(&[]),
+_spill_file: spill_file,
+schema_validated: false,
+})
}
+}
-fn poll_next_inner(
-&mut self,
-cx: &mut Context<'_>,
-) -> Poll>> {
-match &mut self.state {
-SpillReaderStreamState::Uninitialized(_) => {
-// Temporarily replace with `Done` to be abl
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4746103576 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4745858585) **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 abstract-spill-file Benchmark external_aggr.json ┏━━┳┳━━━┳━━┓ ┃ Query┃ HEAD ┃ abstract-spill-file ┃ Change ┃ ┡━━╇╇━━━╇━━┩ │ Q1(64.0 MB) │ 56.47 / 60.31 ±3.77 / 67.38 ms │58.56 / 61.46 ±3.36 / 67.88 ms │no change │ │ Q1(32.0 MB) │ 51.15 / 52.79 ±1.29 / 54.17 ms │55.60 / 57.68 ±2.10 / 61.25 ms │ 1.09x slower │ │ Q1(16.0 MB) │ 54.94 / 56.52 ±1.05 / 58.09 ms │59.16 / 61.43 ±1.51 / 63.06 ms │ 1.09x slower │ │ Q2(512.0 MB) │ 262.47 / 290.74 ±17.74 / 316.12 ms │ 281.48 / 295.73 ±8.47 / 307.19 ms │no change │ │ Q2(256.0 MB) │ 253.53 / 269.51 ±21.23 / 309.45 ms │ 263.20 / 264.93 ±1.21 / 266.78 ms │no change │ │ Q2(128.0 MB) │ 253.51 / 284.35 ±40.83 / 356.51 ms │ 269.65 / 273.67 ±3.64 / 280.30 ms │no change │ │ Q2(64.0 MB) │ 251.51 / 275.61 ±39.20 / 353.66 ms │ 287.38 / 289.66 ±1.31 / 291.31 ms │ 1.05x slower │ │ Q2(32.0 MB) │ 321.56 / 368.19 ±53.49 / 449.30 ms │ 380.82 / 384.33 ±2.27 / 387.78 ms │no change │ └──┴┴───┴──┘ ┏┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡╇━━━┩ │ Total Time (HEAD) │ 1658.02ms │ │ Total Time (abstract-spill-file) │ 1688.88ms │ │ Average Time (HEAD)│ 207.25ms │ │ Average Time (abstract-spill-file) │ 211.11ms │ │ Queries Faster │ 0 │ │ Queries Slower │ 3 │ │ Queries with No Change │ 5 │ │ Queries with Failure │ 0 │ └┴───┘ ``` Resource Usage **external_aggr — base (merge-base)** | Metric | Value | |---
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3438657558
##
datafusion/physical-plan/src/repartition/mod.rs:
##
@@ -3098,9 +3098,9 @@ mod test {
let input_partitions = vec![partition1, partition2];
// Set up context with tight memory limit to force spilling
-// Sorting needs some non-spillable memory, so 64 bytes should force
spilling while still allowing the query to complete
+// Sorting needs some non-spillable memory, so 608 bytes should force
spilling while still allowing the query to complete
Review Comment:
Why did this test need to change? It seems to me like we shouldn't have to
change existing tests for a new pluggable backend -- shouldn't the new code
only be exercised if you have explicitly added a new spilling backend?
##
datafusion/physical-plan/src/spill/mod.rs:
##
@@ -274,19 +258,73 @@ pub fn spill_record_batch_by_size(
Ok(())
}
-/// Write in Arrow IPC Stream format to a file.
-///
-/// Stream format is used for spill because it supports dictionary
replacement, and the random
-/// access of IPC File format is not needed (IPC File format doesn't support
dictionary replacement).
+/// An adapter that implements `std::io::Write` to bridge Arrow's synchronous
Review Comment:
This looks a lot like a `BufferWrtier` -- and it seems like this require a
separate copy.
Wouldn't a better API just be to write directly to the inner spill writer?
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,71 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+
+/// Opens a synchronous reader for this file.
+/// Used by legacy operators (like SortMergeJoin) that haven't been fully
migrated to async.
+///
+/// Backends that only support async reads should leave this default
implementation,
+/// which will safely return a NotImplemented error if used in synchronous
contexts.
+fn open_sync_reader(&self) -> Result> {
+datafusion_common::exec_err!(
+"Synchronous reads are not supported by this spill backend. \
+This backend cannot be used with synchronous operators like
SortMergeJoin \
+until they are refactored to be fully asynchronous."
+)
+}
+}
+
+/// Writer for spill file backends.
+/// Receives zero-copy `Bytes` payloads from the IPCStreamWriter adapter.
+pub trait SpillWriter: Send {
+fn write(&mut self, data: Bytes) -> Result<()>;
Review Comment:
This is all true -- however, I think that since the underlying IPC writer
takes a std::io::Write, forcing all backends to use `Bytes` will likely require
an extra unecessary copy (see comments below on SpillWriterAdapter) anways.
If you use a std::io::write like interface here, backends that want to queue
chunks can do so (by copying into Bytes buffers themselves)
Thus what i suggest is:
1. Change this to look more like std::io::wrote:
```rust
fn write(&mut self, data: &[u8]) -> Result<()>;
```
Which will allow you to get rid of the write adapter
##
datafusion/physical-plan/src/spill/mod.rs:
##
Review Comment:
Update here is the PR:
- https://github.com/apache/datafusion/pull/23029
##
datafusion/physical-plan/src/spill/mod.rs:
##
@@ -39,212 +39,187 @@ use arrow::array::{
Array, ArrayRef, BinaryViewArray, BufferSpec, GenericBy
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745993584 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4745858955) **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 abstract-spill-file Benchmark smj.json ┏━━━┳━━━┳━━┳━━━┓ ┃ Query ┃ HEAD ┃ abstract-spill-file ┃Change ┃ ┡━━━╇━━━╇━━╇━━━┩ │ QQuery 1 │ 8.52 / 9.04 ±0.35 / 9.44 ms │ 8.45 / 8.73 ±0.25 / 9.17 ms │ no change │ │ QQuery 2 │ 170.82 / 173.15 ±1.56 / 175.20 ms │171.88 / 175.55 ±3.78 / 182.56 ms │ no change │ │ QQuery 3 │ 106.61 / 108.88 ±2.56 / 113.80 ms │105.91 / 108.46 ±1.69 / 110.44 ms │ no change │ │ QQuery 4 │27.97 / 28.23 ±0.33 / 28.88 ms │ 28.22 / 28.43 ±0.21 / 28.83 ms │ no change │ │ QQuery 5 │21.59 / 21.81 ±0.17 / 22.02 ms │ 21.45 / 21.84 ±0.31 / 22.29 ms │ no change │ │ QQuery 6 │ 171.61 / 177.06 ±3.72 / 182.16 ms │167.35 / 170.44 ±1.70 / 172.23 ms │ no change │ │ QQuery 7 │ 208.67 / 213.11 ±4.00 / 220.22 ms │207.53 / 212.78 ±4.55 / 220.96 ms │ no change │ │ QQuery 8 │20.05 / 20.51 ±0.36 / 21.16 ms │ 20.40 / 21.66 ±1.80 / 25.24 ms │ 1.06x slower │ │ QQuery 9 │217.50 / 225.69 ±10.71 / 246.82 ms │215.03 / 218.85 ±5.17 / 229.06 ms │ no change │ │ QQuery 10 │71.33 / 77.74 ±9.49 / 96.50 ms │ 70.97 / 78.85 ±6.84 / 91.59 ms │ no change │ │ QQuery 11 │27.42 / 27.63 ±0.16 / 27.83 ms │ 27.08 / 27.29 ±0.22 / 27.64 ms │ no change │ │ QQuery 12 │65.77 / 70.60 ±4.12 / 75.71 ms │ 67.77 / 72.26 ±2.37 / 74.50 ms │ no change │ │ QQuery 13 │103.14 / 113.35 ±12.17 / 136.14 ms │ 97.98 / 105.90 ±4.19 / 109.89 ms │ +1.07x faster │ │ QQuery 14 │70.06 / 71.23 ±0.93 / 72.82 ms │ 68.64 / 70.87 ±1.79 / 74.03 ms │ no change │ │ QQuery 15 │69.28 / 73.78 ±4.36 / 81.31 ms │ 69.94 / 72.40 ±3.04 / 78.39 ms │ no change │ │ QQuery 16 │13.06 / 13.58 ±0.29 / 13.87
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745931057 🤖 Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4745820494) **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 ``` group HEAD abstract-spill-file - --- spill_compression/q16/lz4_frame1.00 34.7±8.23ms? ?/sec 1.69 58.5±0.95ms? ?/sec spill_compression/q16/uncompressed 1.00 36.0±4.52ms? ?/sec 2.33 83.9±1.97ms? ?/sec spill_compression/q16/zstd 1.00 63.3±2.66ms? ?/sec 1.23 78.1±1.11ms? ?/sec spill_compression/q2/lz4_frame 1.00 19.2±4.03ms? ?/sec 1.42 27.2±0.50ms? ?/sec spill_compression/q2/uncompressed 1.00 18.2±4.51ms? ?/sec 2.11 38.5±1.14ms? ?/sec spill_compression/q2/zstd 1.0032.0±10.07ms? ?/sec 1.05 33.7±0.55ms? ?/sec spill_compression/q20/lz4_frame1.00 25.8±3.47ms? ?/sec 1.51 39.0±0.53ms? ?/sec spill_compression/q20/uncompressed 1.00 23.7±4.39ms? ?/sec 2.19 51.8±1.06ms? ?/sec spill_compression/q20/zstd 1.00 46.5±1.73ms? ?/sec 1.13 52.6±0.55ms? ?/sec spill_compression/wide/lz4_frame 1.00 89.9±5.61ms? ?/sec 1.69151.9±2.76ms? ?/sec spill_compression/wide/uncompressed1.00100.1±7.76ms? ?/sec 2.36236.5±4.43ms? ?/sec spill_compression/wide/zstd1.00160.8±4.77ms? ?/sec 1.26202.4±3.04ms? ?/sec spill_io/StreamReader/read_100/1.00 50.7±9.84ms? ?/sec 3.11157.9±2.85ms? ?/sec ``` Resource Usage **spill_io — base (merge-base)** | Metric | Value | ||---| | Wall time | 375.1s | | Peak memory | 203.8 MiB | | Avg memory | 47.6 MiB | | CPU user | 121.5s | | CPU sys | 41.5s | | Peak spill | 0 B | **spill_io — branch** | Metric | Value | ||---| | Wall time | 395.1s | | Peak memory | 294.5 MiB | | Avg memory | 77.8 MiB | | CPU user | 1
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745880244 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4745858585) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4745858585-594-ltzkc 6.12.68+ #1 SMP Sat May 2 07:49:07 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 abstract-spill-file (d71a415c9f212389705a6d96898db2783891380d) to 6176a6d (merge-base) [diff](https://github.com/apache/datafusion/compare/6176a6dae1f36e30c83bc880bb8b6cb1d1a175a8..d71a415c9f212389705a6d96898db2783891380d) using: external_aggr 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745880205 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4745858955) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4745858955-595-vw4pm 6.12.68+ #1 SMP Sat May 2 07:49:07 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 abstract-spill-file (d71a415c9f212389705a6d96898db2783891380d) to 6176a6d (merge-base) [diff](https://github.com/apache/datafusion/compare/6176a6dae1f36e30c83bc880bb8b6cb1d1a175a8..d71a415c9f212389705a6d96898db2783891380d) using: smj 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745858955 run benchmark smj -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745858585 run benchmark external_aggr -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangbot commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745843173 🤖 Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/21882#issuecomment-4745820494) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4745820494-593-2hs7c 6.12.68+ #1 SMP Sat May 2 07:49:07 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 abstract-spill-file (d71a415c9f212389705a6d96898db2783891380d) to 6176a6d (merge-base) [diff](https://github.com/apache/datafusion/compare/6176a6dae1f36e30c83bc880bb8b6cb1d1a175a8..d71a415c9f212389705a6d96898db2783891380d) using: spill_io 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4745820494 run benchmark spill_io -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4723168898 @alamb #22230 Has been merged! I've also rebased this PR on top of latest main, so it is ready for your review whenever you have the time. -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4488665457 Hey @alamb, quick update! I've reworked both PRs to make them easier to review independently: - **#22230 (SMJ refactor)** no longer depends on this PR, it now migrates SortMergeJoin to async spill abstractions while keeping the concrete `RefCountedTempFile` type, so it can be reviewed and merged standalone. - **This PR** now only does one focused thing: introduces the `SpillFile` trait + `TempFileFactory` and swaps the internal type from `RefCountedTempFile` → `Arc` and internal changes in the spill module. I've also removed the `open_sync` tech debt that was here before, and have added the skip validation in streamdecoder The plan is for #22230 to merge first, then I'll rebase this on top of it. Would be grateful if you could take a look whenever you get the chance! -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4463623482 Done @alamb, I've split it out into PR #22230 . It won't compile standalone since it depends on this PR, but the SMJ logic is self-contained for review. I've noted the dependency clearly in the description. Once this merges I'll rebase and mark it ready. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4461783339 Hey @alamb, I looked into splitting the SMJ refactor out but it uses Arc from this PR, so it can't compile against main standalone. I can open it as a draft for an SMJ expert to review the logic, with a clear note it depends on this one. Would that work? -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4460722869 > Rather than opening a second stacked PR and temporarily introducing the open_sync tech debt to main, I went ahead and rolled the refactor directly into this PR to keep things clean. I hope this approach is okay with you! Grerat-- can you please make a PR for just the SMJ refactor and then stack this PR on it? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4460725905 That will make it easier / faster to review (I am not a SMJ expert so I can't really review that part effiicently) -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4454323289 Hey @alamb, quick update! While working on the `SortMergeJoin` async migration in parallel, I realised the changes were actually quite contained (~260 insertions, ~170 deletions). Rather than opening a second stacked PR and temporarily introducing the `open_sync` tech debt to main, I went ahead and rolled the refactor directly into this PR to keep things clean. I hope this approach is okay with you! I believe the PR is now ready for review, so I've marked it as such. I'd appreciate another look whenever you have the time. Thank you! -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4445120833 `cargo-semver-checks` flagged `DiskManagerMode::Custom` as a breaking change since the enum isn't `#[non_exhaustive]`. Happy to add it if preferred, but wanted to check first since it would affect downstream users matching on this enum. -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
github-actions[bot] commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4443969061 Thank you for opening this pull request! Reviewer note: [cargo-semver-checks](https://github.com/obi1kenobi/cargo-semver-checks) reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details ``` Cloning apache/main Building datafusion-execution v53.1.0 (current) Built [ 31.198s] (current) Parsing datafusion-execution v53.1.0 (current) Parsed [ 0.028s] (current) Building datafusion-execution v53.1.0 (baseline) Built [ 29.854s] (baseline) Parsing datafusion-execution v53.1.0 (baseline) Parsed [ 0.027s] (baseline) Checking datafusion-execution v53.1.0 -> v53.1.0 (no change; assume patch) Checked [ 0.288s] 222 checks: 220 pass, 2 fail, 0 warn, 30 skip --- failure auto_trait_impl_removed: auto trait no longer implemented --- Description: A public type has stopped implementing one or more auto traits. This can break downstream code that depends on the traits being implemented. ref: https://doc.rust-lang.org/reference/special-types-and-traits.html#auto-traits impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.47.0/src/lints/auto_trait_impl_removed.ron Failed in: type DiskManagerBuilder is no longer UnwindSafe, in /home/runner/work/datafusion/datafusion/datafusion/execution/src/disk_manager.rs:40 type DiskManagerBuilder is no longer RefUnwindSafe, in /home/runner/work/datafusion/datafusion/datafusion/execution/src/disk_manager.rs:40 type DiskManager is no longer UnwindSafe, in /home/runner/work/datafusion/datafusion/datafusion/execution/src/disk_manager.rs:196 type DiskManager is no longer UnwindSafe, in /home/runner/work/datafusion/datafusion/datafusion/execution/src/disk_manager.rs:196 type DiskManagerMode is no longer UnwindSafe, in /home/runner/work/datafusion/datafusion/datafusion/execution/src/disk_manager.rs:125 type DiskManagerMode is no longer RefUnwindSafe, in /home/runner/work/datafusion/datafusion/datafusion/execution/src/disk_manager.rs:125 --- failure enum_variant_added: enum variant added on exhaustive enum --- Description: A publicly-visible enum without #[non_exhaustive] has a new variant. ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.47.0/src/lints/enum_variant_added.ron Failed in: variant DiskManagerMode:Custom in /home/runner/work/datafusion/datafusion/datafusion/execution/src/disk_manager.rs:137 Summary semver requires new major version: 2 major and 0 minor checks failed Finished [ 62.833s] datafusion-execution Building datafusion-physical-plan v53.1.0 (current) Built [ 32.534s] (current) Parsing datafusion-physical-plan v53.1.0 (current) Parsed [ 0.132s] (current) Building datafusion-physical-plan v53.1.0 (baseline) Built [ 33.547s] (baseline) Parsing datafusion-physical-plan v53.1.0 (baseline) Parsed [ 0.133s] (baseline) Checking datafusion-physical-plan v53.1.0 -> v53.1.0 (no change; assume patch) Checked [ 0.788s] 222 checks: 222 pass, 30 skip Summary no semver update required Finished [ 68.824s] datafusion-physical-plan ``` -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4443232799 > @adriangb Thank you so much for the guidance! I updated the test to simply assert that a spill does occur (`spill_count > 0`) and that the batch output order remains perfectly sorted, rather than trying to force every single batch to spill. I hope this aligns with the semantic purpose you originally envisioned for the test. I really appreciate your help getting me unstuck here! That makes sense to me. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4443168784 @adriangb Thank you so much for the guidance! I updated the test to simply assert that a spill does occur (`spill_count > 0`) and that the batch output order remains perfectly sorted, rather than trying to force every single batch to spill. I hope this aligns with the semantic purpose you originally envisioned for the test. I really appreciate your help getting me unstuck 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4443161188 @alamb I’ve addressed the nits and force-pushed the updates. Could you please trigger the CI and take another look when you have a moment? In the meantime, I am working on migrating `SortMergeJoin` to the new spill abstractions in parallel so that both can be reviewed quickly. Thank you again for your time! -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
adriangb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4437114961 IIRC that test was added when we added spilling to RepartitionExec. Conceptually the test is simple: if RepartitionExec is configured to preserve order and it spills we need to make sure that spilling did not shuffle the data. The orchestration however is difficult: forcing a RepartitionExec to spill usually requires skewed upstream partition consumption rates. You could try to change the test to eg use a GroupBy or maybe we can use a RepartitionExec in isolation if we pull from the streams in the right way. I think the structure can be changed quite a bit as long as we preserve the semantic meaning of the test, I am not surprised that it is pretty fragile to changes. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4435286903 Hey @adriangb, Andrew suggested I reach out to you since you originally authored `repartition::test::test_preserve_order_with_spilling`. I'm currently hitting a wall with it while migrating the spilling architecture to async streams. The test is currently stuck in a memory-accounting deadlock. Here’s what is happening: * If I set the memory pool limit tight enough to force a spill, `RepartitionMerge` panics during initialization. It needs to reserve some memory to set up its streams, but exhausts the pool before completing its unspillable setup. * However, if I increase the pool limit to give Merge enough headroom to initialize safely and then scale up the data volume to force overflow, the `RepartitionExec` producers greedily consume the additional memory first. This either ends up starving Merge again or allows the query to complete entirely in memory without triggering a spill. I was able to trigger a spill once by setting the test memory limit to 608 B, but even that was not sufficient for the test to pass reliably. Is there a correct or idiomatic way to configure this test (batch sizes, data volume, memory pool limits, etc.) to reliably force a `RepartitionExec` spill without violating the Merge operator’s baseline initialization overhead? Or am I approaching this incorrectly and missing something obvious? I would really appreciate any guidance you could provide. -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4434692591 > test_preserve_order_with_spilling Sadly I am not familar with this test so I don't have a lot to offer you Maybe you can look at git history and see who introduced the test and maybe they might have some ideas -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4432242479 > > I am happy to defer to your judgment if you feel the tech debt must be addressed first! > > How about we try it in parallel? @alamb sure, i have already started to work on that locally while waiting for the response also i am actually still stuck on the `test repartition::test::test_preserve_order_with_spilling` If I keep the memory limit tight, the test panics immediately because the StreamingMerge operators and producer channels require a baseline of unspillable memory just to initialize their internal buffers. However, if I increase the limit to accommodate that initialization overhead and scale up the data size, the operator either fills new internal buffers (triggering OOM again) or processes everything in RAM without ever triggering a spill. I have tried shrinking the initial memory footprint by overriding the `SessionConfig` batch size, but it still runs into similar problem I would really appreciate any guidance on this, am I missing something obvious 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4431851670 > I am happy to defer to your judgment if you feel the tech debt must be addressed first! How about we try it in parallel? -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4415353823 @alamb Thank you so much for the review! I scoped out the SortMergeJoin migration today, specifically looking at bitwise_stream.rs and process_key_match_with_filter, to see what it would take. Because SortMergeJoin currently reads from the spill file via a synchronous for loop inside a hand-rolled poll state machine, making the read path truly async requires a major rewrite. We can't just .await the stream, so we may need to store the SendableRecordBatchStream in the execution state and manually persist variables like matched_count across Poll::Pending yields. Because ParadeDB is hoping to unblock their Postgres integration next week, I'm worried a state machine rewrite of this scale will stall them. Would you be open to merging this core abstraction first (with open_sync_reader marked as #[deprecated])? I can open a dedicated tracking issue for the SortMergeJoin async migration and tackle it as a fast follow-up PR. I am happy to defer to your judgment if you feel the tech debt must be addressed first! -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3214836201
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,71 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+
+/// Opens a synchronous reader for this file.
+/// Used by legacy operators (like SortMergeJoin) that haven't been fully
migrated to async.
+///
+/// Backends that only support async reads should leave this default
implementation,
+/// which will safely return a NotImplemented error if used in synchronous
contexts.
+fn open_sync_reader(&self) -> Result> {
+datafusion_common::exec_err!(
+"Synchronous reads are not supported by this spill backend. \
+This backend cannot be used with synchronous operators like
SortMergeJoin \
+until they are refactored to be fully asynchronous."
+)
+}
+}
+
+/// Writer for spill file backends.
+/// Receives zero-copy `Bytes` payloads from the IPCStreamWriter adapter.
+pub trait SpillWriter: Send {
+fn write(&mut self, data: Bytes) -> Result<()>;
Review Comment:
Yes, you are right. The reason I didn't use Write trait which uses &[u8]
was for ownership reasons. Some backends might queue chunks to a background
task (e.g., S3 multipart via a channel) and need to hold the data past the
write() call's return. &[u8] can't express that, and it would force a second
copy between the SpillWriteAdapter and the SpillWriter.
Also, the custom SpillWriter trait contains finish(), which maps perfectly
to complete_multipart_upload for S3 and resource owner cleanup for Postgres.
--
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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on code in PR #21882:
URL: https://github.com/apache/datafusion/pull/21882#discussion_r3204748777
##
datafusion/execution/src/disk_manager.rs:
##
@@ -89,19 +100,28 @@ impl DiskManagerBuilder {
max_temp_directory_size: self.max_temp_directory_size,
used_disk_space: Arc::new(AtomicU64::new(0)),
active_files_count: Arc::new(AtomicUsize::new(0)),
+custom_factory: None,
})
}
DiskManagerMode::Disabled => Ok(DiskManager {
local_dirs: Mutex::new(None),
max_temp_directory_size: self.max_temp_directory_size,
used_disk_space: Arc::new(AtomicU64::new(0)),
active_files_count: Arc::new(AtomicUsize::new(0)),
+custom_factory: None,
+}),
+DiskManagerMode::Custom(factory) => Ok(DiskManager {
+local_dirs: Mutex::new(None),
+max_temp_directory_size: self.max_temp_directory_size,
+used_disk_space: Arc::new(AtomicU64::new(0)),
+active_files_count: Arc::new(AtomicUsize::new(0)),
+custom_factory: Some(factory),
}),
}
}
}
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Default)]
pub enum DiskManagerMode {
Review Comment:
If we are going to permit a TempFileFactory here, I wonder if it would make
sense (maybe as a follow on PR) here to move everything over
So like
```rust
pub enum DiskManagerMode {
Custom(Arc),
}
```
And then implement a basic `DirectoriesTempFileFactor` and a
`NoTempFilesFactory` that are provided along with datafusion
That might simplify the code, and it would also ensure the `TempFileFactory`
API is rich enough for the existing temp file strategies
##
datafusion/execution/src/spill_file.rs:
##
@@ -0,0 +1,71 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use datafusion_common::Result;
+use futures::Stream;
+use std::path::Path;
+use std::pin::Pin;
+use std::sync::Arc;
+
+/// Abstraction over a spill file backend.
+/// Implementations handle their own quota enforcement and blocking concerns.
+pub trait SpillFile: Send + Sync {
+/// Returns the OS path if this is a local file, None otherwise.
+fn path(&self) -> Option<&Path> {
+None
+}
+
+/// Returns current size in bytes if cheaply available.
+fn size(&self) -> Option;
+
+/// Returns file contents as an async stream of byte chunks.
+fn read_stream(&self) -> Result> +
Send>>>;
+
+/// Opens a writer for appending data to this file.
+fn open_writer(&self) -> Result>;
+
+/// Opens a synchronous reader for this file.
+/// Used by legacy operators (like SortMergeJoin) that haven't been fully
migrated to async.
+///
+/// Backends that only support async reads should leave this default
implementation,
+/// which will safely return a NotImplemented error if used in synchronous
contexts.
+fn open_sync_reader(&self) -> Result> {
+datafusion_common::exec_err!(
+"Synchronous reads are not supported by this spill backend. \
+This backend cannot be used with synchronous operators like
SortMergeJoin \
+until they are refactored to be fully asynchronous."
+)
+}
+}
+
+/// Writer for spill file backends.
+/// Receives zero-copy `Bytes` payloads from the IPCStreamWriter adapter.
+pub trait SpillWriter: Send {
+fn write(&mut self, data: Bytes) -> Result<()>;
Review Comment:
This is pretty similar to https://doc.rust-lang.org/std/io/trait.Write.html
🤔
##
datafusion/execution/src/disk_manager.rs:
##
@@ -473,6 +528,120 @@ fn create_local_dirs(local_dirs: &[PathBuf]) ->
Result>> {
.collect()
}
+pub struct OsSpillWriter {
Review Comment:
maybe "file spill writer"?
##
datafusion/execution/src/disk_manager.rs:
##
@@ -174,8 +207,20 @@ pub struct DiskManager {
used_disk_space: Arc,
/// Number of active temporary files created by this disk manager
active_files_count
Re: [PR] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4412430851 > @alamb I opened this draft PR to get early feedback on the architecture. > > 1. The first point is around the sync read path. I introduced `open_sync_reader` because `SortMergeJoin` currently has synchronous, blocking code paths that directly open files using paths and `BufReader`, instead of going through the spill abstractions. Converting this to fully async would significantly increase the scope of this PR. > >* Does it make sense to keep this escape hatch for now and handle making these operators async in a follow-up PR? Kind of, though it seems like accumulating technical debt as we'll have APIs that will not be needed once we complete the work for SortMergeJoin What do you think about making a first PR to migrate SortMergeJoin to use the spill abstraction? > 2. The second point is regarding test failures. I have not modified the original 64B limit in the tests because I wanted guidance here. Currently, the `repartition` test in `mod.rs` is failing, and it seems related to spilling not being triggered correctly, the new `SpillWriteAdapter` adds slight allocation overhead which makes the original 64-byte memory limit too tight for the merge heap to initialize (~296 bytes needed), bumping up the memory limit causes the test to not spill anymore, I believe increasing the test data size might solve the issue, but am not sure. Makes sense to me -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
alamb commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4398467700 Thanks -- will try and look at this shortly -- 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] feat: introduce pluggable SpillFile trait and TempFileFactory for custom spill backends [datafusion]
pantShrey commented on PR #21882: URL: https://github.com/apache/datafusion/pull/21882#issuecomment-4329695321 @alamb I opened this draft PR to get early feedback on the architecture. 1. The first point is around the sync read path. I introduced `open_sync_reader` because `SortMergeJoin` currently has synchronous, blocking code paths that directly open files using paths and `BufReader`, instead of going through the spill abstractions. Converting this to fully async would significantly increase the scope of this PR. - Does it make sense to keep this escape hatch for now and handle making these operators async in a follow-up PR? 2. The second point is regarding test failures. I have not modified the original 64 MB limit in the tests because I wanted guidance here. Currently, the `repartition` test in `mod.rs` is failing, and it seems related to spilling not being triggered correctly, but I have not been able to fully identify the root cause. I might be missing something here, so would really appreciate your guidance. -- 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]
