comphead commented on code in PR #7606:
URL: https://github.com/apache/datafusion-comet/pull/7606#discussion_r4554845809
##########
native/shuffle/src/writers/shuffle_block_writer.rs:
##########
@@ -748,6 +738,70 @@ mod tests {
assert_eq!(read_ipc_compressed(&local_out[16..]).unwrap(), batch);
}
+ /// The LZ4 frame encoder follows the zstd context's contract, and local
and remote encodes
+ /// write the same frame.
+ #[test]
+ fn rss_write_releases_lz4_encoder_local_write_retains_it() {
Review Comment:
This is nearly the same test as
`rss_write_releases_zstd_context_local_write_retains_it` above. Could the two
become one test that loops over `[CompressionCodec::Zstd(1),
CompressionCodec::Lz4Frame]`, with a single `holds_workspaces()` hook in
`ShuffleCodecContext` instead of `holds_zstd_cctx` plus `holds_lz4_encoder`?
The failing-sink case would then cover the zstd release on error too.
##########
native/shuffle/src/lz4_frame.rs:
##########
@@ -0,0 +1,308 @@
+// 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.
+
+//! LZ4 frames written with buffers and a hash table kept from one frame to
the next.
+//!
+//! Every compressed shuffle block is one LZ4 frame.
`lz4_flex::frame::FrameEncoder` owns the
+//! writer it compresses into, and blocks go to different outputs, so
`ShuffleBlockWriter` built
+//! one per block, and with it a zeroed 16 KiB hash table, a 64 KiB input
buffer and a 70 KiB
+//! output buffer. [`Lz4FrameEncoder`] keeps those and writes the frame format
itself around
+//! lz4_flex's block compressor, compressing whole blocks straight from the
caller's buffer.
+//!
+//! The frames have independent blocks of at most 64 KiB, no checksums, no
content size and an
+//! end mark, as remote shuffle's frames had. Local frames used to take their
block size from
+//! their first write, the schema message, which gave the same 64 KiB unless
that message was
+//! larger. A block that does not get smaller is stored uncompressed, as
`FrameEncoder` does.
Review Comment:
The sentence about how local frames used to pick their block size describes
the code this PR replaces. It fits the PR description better than the module
docs, where it will read as stale once nobody remembers the old path. The same
applies to the "Before, a local block took its block size..." sentence on
`lz4_frames_use_64_kib_blocks`. The reason for a fixed 64 KiB, that every
reader decodes it with the least memory, is already on `BLOCK_SIZE`.
##########
native/shuffle/src/lz4_frame.rs:
##########
@@ -0,0 +1,308 @@
+// 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.
+
+//! LZ4 frames written with buffers and a hash table kept from one frame to
the next.
+//!
+//! Every compressed shuffle block is one LZ4 frame.
`lz4_flex::frame::FrameEncoder` owns the
+//! writer it compresses into, and blocks go to different outputs, so
`ShuffleBlockWriter` built
+//! one per block, and with it a zeroed 16 KiB hash table, a 64 KiB input
buffer and a 70 KiB
+//! output buffer. [`Lz4FrameEncoder`] keeps those and writes the frame format
itself around
+//! lz4_flex's block compressor, compressing whole blocks straight from the
caller's buffer.
Review Comment:
lz4_flex's `FrameEncoder` can already write several frames in a row. After
`try_finish()`, the next `write` opens a new frame, and `begin_frame` resets
the table and buffers (lz4_flex 0.14 `frame/compress.rs`). Did you compare
against one long-lived `FrameEncoder<Vec<u8>>` per context, drained into
`output` through `get_mut()` after each `try_finish()`? It would need
`BlockSize::Max64KB` set explicitly, because `Auto` is fixed by the first
frame. It keeps the copy into `src` and adds a copy of the compressed bytes, so
I would expect it to give back part of the gain, but it leaves the frame format
to lz4_flex. If the custom writer is worth the extra speed, a sentence here on
why reuse was not enough would help the next reader.
--
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]
