Re: [PR] feat(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
adriangbot commented on PR #9836: URL: https://github.com/apache/arrow-rs/pull/9836#issuecomment-4614050347 🤖 Arrow criterion benchmark completed (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/9836#issuecomment-4614005395) **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 ipc-writer-avoid-repetitive-allocs main - -- arrow_ipc_stream_writer/FileWriter/write_10 1.00163.1±1.46µs ? ?/sec1.13184.7±1.68µs? ?/sec arrow_ipc_stream_writer/StreamWriter/write_10 1.00159.2±1.92µs ? ?/sec1.17186.3±1.86µs? ?/sec arrow_ipc_stream_writer/StreamWriter/write_10/zstd1.00 7.2±0.03ms ? ?/sec1.01 7.3±0.11ms? ?/sec ``` Resource Usage **base (merge-base)** | Metric | Value | ||---| | Wall time | 30.0s | | Peak memory | 2.7 GiB | | Avg memory | 2.6 GiB | | CPU user | 27.6s | | CPU sys | 0.6s | | Peak spill | 0 B | **branch** | Metric | Value | ||---| | Wall time | 35.0s | | Peak memory | 2.7 GiB | | Avg memory | 2.6 GiB | | CPU user | 31.5s | | CPU sys | 0.1s | | Peak spill | 0 B | --- [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]
Re: [PR] feat(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
adriangbot commented on PR #9836: URL: https://github.com/apache/arrow-rs/pull/9836#issuecomment-4614034762 🤖 Arrow criterion benchmark running (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/9836#issuecomment-4614005395) **Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4614005395-418-8bhzd 6.12.68+ #1 SMP Wed Apr 1 02:23:28 UTC 2026 aarch64 GNU/Linux` CPU Details (lscpu) ``` Architecture:aarch64 CPU op-mode(s): 64-bit Byte Order: Little Endian CPU(s): 16 On-line CPU(s) list: 0-15 Vendor ID: ARM Model name: Neoverse-V2 Model: 1 Thread(s) per core: 1 Core(s) per cluster: 16 Socket(s): - Cluster(s): 1 Stepping:r0p1 BogoMIPS:2000.00 Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti L1d cache: 1 MiB (16 instances) L1i cache: 1 MiB (16 instances) L2 cache:32 MiB (16 instances) L3 cache:80 MiB (1 instance) NUMA node(s):1 NUMA node0 CPU(s): 0-15 Vulnerability Gather data sampling: Not affected Vulnerability Indirect target selection: Not affected Vulnerability Itlb multihit: Not affected Vulnerability L1tf: Not affected Vulnerability Mds: Not affected Vulnerability Meltdown: Not affected Vulnerability Mmio stale data: Not affected Vulnerability Reg file data sampling:Not affected Vulnerability Retbleed: Not affected Vulnerability Spec rstack overflow: Not affected Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl Vulnerability Spectre v1:Mitigation; __user pointer sanitization Vulnerability Spectre v2:Mitigation; CSV2, BHB Vulnerability Srbds: Not affected Vulnerability Tsa: Not affected Vulnerability Tsx async abort: Not affected Vulnerability Vmscape: Not affected ``` Comparing ipc-writer-avoid-repetitive-allocs (6ab2cbbc933a7573d0d227aa6b4b358febf78fba) to 4fa8d2f (merge-base) [diff](https://github.com/apache/arrow-rs/compare/4fa8d2ff5f18f2d773f9642631715509f844a062..6ab2cbbc933a7573d0d227aa6b4b358febf78fba) BENCH_NAME=ipc_writer BENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench ipc_writer BENCH_FILTER= 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]
Re: [PR] feat(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
alamb commented on PR #9836: URL: https://github.com/apache/arrow-rs/pull/9836#issuecomment-4614005395 run benchmark ipc_writer -- 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(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
pchintar commented on code in PR #9836:
URL: https://github.com/apache/arrow-rs/pull/9836#discussion_r3210219705
##
arrow-ipc/src/writer.rs:
##
@@ -1550,6 +1596,163 @@ impl RecordBatchWriter for StreamWriter {
}
}
+fn has_dictionary_batch(batch: &RecordBatch) -> bool {
+batch
+.schema()
+.fields()
+.iter()
+.any(|field| has_dictionary_type(field.data_type()))
+}
+
+fn has_dictionary_type(data_type: &DataType) -> bool {
+match data_type {
+DataType::Dictionary(_, _) => true,
+DataType::Struct(fields) => fields.iter().any(|f|
has_dictionary_type(f.data_type())),
+DataType::List(field)
+| DataType::LargeList(field)
+| DataType::ListView(field)
+| DataType::LargeListView(field)
+| DataType::FixedSizeList(field, _) =>
has_dictionary_type(field.data_type()),
+DataType::Map(field, _) => has_dictionary_type(field.data_type()),
+DataType::RunEndEncoded(_, field) =>
has_dictionary_type(field.data_type()),
+DataType::Union(fields, _) => fields
+.iter()
+.any(|(_, f)| has_dictionary_type(f.data_type())),
+_ => false,
+}
+}
+
+fn write_record_batch_fast(
+writer: &mut W,
+batch: &RecordBatch,
+write_options: &IpcWriteOptions,
+compression_context: &mut CompressionContext,
+scratch: &mut IpcWriterScratch,
+) -> Result<(usize, usize), ArrowError> {
+scratch.fbb.reset();
+scratch.nodes.clear();
+scratch.buffers.clear();
+scratch.arrow_data.clear();
+scratch.variadic_buffer_counts.clear();
+
+let batch_compression_type = write_options.batch_compression_type;
Review Comment:
Now with shared `encode_record_batch_into(...)` and the removal of
`write_record_batch_fast(...)` there is:
* ONE serialization implementation
* reused by both paths & no duplicated logic
So this is resolved now I guess?
--
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(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
pchintar commented on PR #9836: URL: https://github.com/apache/arrow-rs/pull/9836#issuecomment-4408532787 > Thansk for this @pchintar -- > > I am a little worried that this introduces a second almost-but-not-quite-tthe-same path for writing batches to IPC. I think we could do better if we did a more holistic rearranging of the code -- let me know what you think @alamb The dictionary path remains unchanged in this PR and should not require any additional copies, alignment requirements or regressions for that specific case. I also looked into threading the writer further into `IpcDataGenerator`, but that would require restructuring the recursive dictionary emission path as well, so I kept this change focused on the shared reusable encoding path for this PR. -- 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(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
pchintar commented on code in PR #9836:
URL: https://github.com/apache/arrow-rs/pull/9836#discussion_r3210219705
##
arrow-ipc/src/writer.rs:
##
@@ -1550,6 +1596,163 @@ impl RecordBatchWriter for StreamWriter {
}
}
+fn has_dictionary_batch(batch: &RecordBatch) -> bool {
+batch
+.schema()
+.fields()
+.iter()
+.any(|field| has_dictionary_type(field.data_type()))
+}
+
+fn has_dictionary_type(data_type: &DataType) -> bool {
+match data_type {
+DataType::Dictionary(_, _) => true,
+DataType::Struct(fields) => fields.iter().any(|f|
has_dictionary_type(f.data_type())),
+DataType::List(field)
+| DataType::LargeList(field)
+| DataType::ListView(field)
+| DataType::LargeListView(field)
+| DataType::FixedSizeList(field, _) =>
has_dictionary_type(field.data_type()),
+DataType::Map(field, _) => has_dictionary_type(field.data_type()),
+DataType::RunEndEncoded(_, field) =>
has_dictionary_type(field.data_type()),
+DataType::Union(fields, _) => fields
+.iter()
+.any(|(_, f)| has_dictionary_type(f.data_type())),
+_ => false,
+}
+}
+
+fn write_record_batch_fast(
+writer: &mut W,
+batch: &RecordBatch,
+write_options: &IpcWriteOptions,
+compression_context: &mut CompressionContext,
+scratch: &mut IpcWriterScratch,
+) -> Result<(usize, usize), ArrowError> {
+scratch.fbb.reset();
+scratch.nodes.clear();
+scratch.buffers.clear();
+scratch.arrow_data.clear();
+scratch.variadic_buffer_counts.clear();
+
+let batch_compression_type = write_options.batch_compression_type;
Review Comment:
Now with `shared encode_record_batch_into(...)` and the removal of
`write_record_batch_fast(...)` there is:
* ONE serialization implementation
* reused by both paths & no duplicated logic
So this is resolved now I guess?
--
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(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
pchintar commented on code in PR #9836:
URL: https://github.com/apache/arrow-rs/pull/9836#discussion_r3210172297
##
arrow-ipc/src/writer.rs:
##
@@ -194,6 +194,26 @@ impl Default for IpcWriteOptions {
/// [Arrow IPC Format]:
https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc
pub struct IpcDataGenerator {}
+struct IpcWriterScratch {
Review Comment:
I've added comments as mentioned
--
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(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
pchintar commented on code in PR #9836:
URL: https://github.com/apache/arrow-rs/pull/9836#discussion_r3210172297
##
arrow-ipc/src/writer.rs:
##
@@ -194,6 +194,26 @@ impl Default for IpcWriteOptions {
/// [Arrow IPC Format]:
https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc
pub struct IpcDataGenerator {}
+struct IpcWriterScratch {
Review Comment:
I've added comments as mentioned
--
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(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
pchintar commented on PR #9836: URL: https://github.com/apache/arrow-rs/pull/9836#issuecomment-4408255411 @alamb Thanks for the valuable feedback. I've re-worked this to: * remove the separate `write_record_batch_fast` path * share the reusable scratch-based encoding path instead * added documentation & comments for the scratch + encoding helpers * reuse the existing `write_message_parts` writer path * replaced the fully qualified `crate::FieldNode` / `crate::Buffer` usage with direct imports -- 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(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
alamb commented on code in PR #9836:
URL: https://github.com/apache/arrow-rs/pull/9836#discussion_r3204026164
##
arrow-ipc/src/writer.rs:
##
@@ -194,6 +194,26 @@ impl Default for IpcWriteOptions {
/// [Arrow IPC Format]:
https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc
pub struct IpcDataGenerator {}
+struct IpcWriterScratch {
Review Comment:
I think some comments here explaining the rationale for this structure would
be useful (avoid allocations during IPC read)
##
arrow-ipc/src/writer.rs:
##
@@ -194,6 +194,26 @@ impl Default for IpcWriteOptions {
/// [Arrow IPC Format]:
https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc
pub struct IpcDataGenerator {}
+struct IpcWriterScratch {
+fbb: FlatBufferBuilder<'static>,
+nodes: Vec,
Review Comment:
nit: can you please use `use crate::FieldNode` in the above and then use
`FieldNode` directly (same for Buffer below) to avoid fully qulaified names?
##
arrow-ipc/src/writer.rs:
##
@@ -1550,6 +1596,163 @@ impl RecordBatchWriter for StreamWriter {
}
}
+fn has_dictionary_batch(batch: &RecordBatch) -> bool {
+batch
+.schema()
+.fields()
+.iter()
+.any(|field| has_dictionary_type(field.data_type()))
+}
+
+fn has_dictionary_type(data_type: &DataType) -> bool {
+match data_type {
+DataType::Dictionary(_, _) => true,
+DataType::Struct(fields) => fields.iter().any(|f|
has_dictionary_type(f.data_type())),
+DataType::List(field)
+| DataType::LargeList(field)
+| DataType::ListView(field)
+| DataType::LargeListView(field)
+| DataType::FixedSizeList(field, _) =>
has_dictionary_type(field.data_type()),
+DataType::Map(field, _) => has_dictionary_type(field.data_type()),
+DataType::RunEndEncoded(_, field) =>
has_dictionary_type(field.data_type()),
+DataType::Union(fields, _) => fields
+.iter()
+.any(|(_, f)| has_dictionary_type(f.data_type())),
+_ => false,
+}
+}
+
+fn write_record_batch_fast(
+writer: &mut W,
+batch: &RecordBatch,
+write_options: &IpcWriteOptions,
+compression_context: &mut CompressionContext,
+scratch: &mut IpcWriterScratch,
+) -> Result<(usize, usize), ArrowError> {
+scratch.fbb.reset();
+scratch.nodes.clear();
+scratch.buffers.clear();
+scratch.arrow_data.clear();
+scratch.variadic_buffer_counts.clear();
+
+let batch_compression_type = write_options.batch_compression_type;
Review Comment:
this seems like a copy/paste/modify slight variation of
https://github.com/apache/arrow-rs/blob/8ce051e126cb09e4a688cb91d54bb3553780decc/arrow-ipc/src/writer.rs#L1764-L1763
##
arrow-ipc/src/writer.rs:
##
@@ -1550,6 +1596,163 @@ impl RecordBatchWriter for StreamWriter {
}
}
+fn has_dictionary_batch(batch: &RecordBatch) -> bool {
+batch
+.schema()
+.fields()
+.iter()
+.any(|field| has_dictionary_type(field.data_type()))
+}
+
+fn has_dictionary_type(data_type: &DataType) -> bool {
+match data_type {
+DataType::Dictionary(_, _) => true,
+DataType::Struct(fields) => fields.iter().any(|f|
has_dictionary_type(f.data_type())),
+DataType::List(field)
+| DataType::LargeList(field)
+| DataType::ListView(field)
+| DataType::LargeListView(field)
+| DataType::FixedSizeList(field, _) =>
has_dictionary_type(field.data_type()),
+DataType::Map(field, _) => has_dictionary_type(field.data_type()),
+DataType::RunEndEncoded(_, field) =>
has_dictionary_type(field.data_type()),
+DataType::Union(fields, _) => fields
+.iter()
+.any(|(_, f)| has_dictionary_type(f.data_type())),
+_ => false,
+}
+}
+
+fn write_record_batch_fast(
Review Comment:
Could you please add some documentation to this function explaining how it
is different than `write_record_batch` (I think the big change is that it
doesn't handle dictionaries)?
##
arrow-ipc/src/writer.rs:
##
@@ -1165,23 +1188,33 @@ impl FileWriter {
));
}
-let (encoded_dictionaries, encoded_message) = self.data_gen.encode(
-batch,
-&mut self.dictionary_tracker,
-&self.write_options,
-&mut self.compression_context,
-)?;
+let (meta, data) = if has_dictionary_batch(batch) {
+let (encoded_dictionaries, encoded_message) = self.data_gen.encode(
Review Comment:
If the goal of this PR is to avoid allocations in the IPC writer, I think we
should take a more holistic approach
Rather than trying to fast path only the non dictionary path, what do you
think about threading the writer somehow into `self.data_gen`?
It seems like the current code bu
Re: [PR] feat(ipc): Avoid repeated heap allocations and buffer copies for non-Dictionary types in IPC writer [arrow-rs]
pchintar commented on PR #9836: URL: https://github.com/apache/arrow-rs/pull/9836#issuecomment-4398935121 Hi @alamb I just wanted to clarify: so this PR is indeed ready to be reviewed for non-dictionary types (The `ipc_writer` benchmark uses only non-Dictionary type). I've updated the PR title by explicitly mentioning "non-Dictionary" to make that clear & so the Dictionary work will be a separate PR after this. So, could you kindly review this one when you get a chance? Thanks. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
