ARROW-8676: [Rust] IPC RecordBatch body compression

[nevi-me]

Adds Recordbatch body compression, which compresses the buffers that make up arrays (e.g. offsets, null buffer).
I've restricted the write side to only work with v5 of the metadata. We can expand on this later, as I think the non-legacy v4 supports the BodyCompression method implemented here. Reading should be fine if the compression info is specified.

This PR is built on top of ARROW-10299 (#9122).

I have not yet implemented ZSTD compression, but I expect it shouldn't be too much work, so can still be done as part of this PR.

[nevi-me]

For anyone who understands how LZ4 works, I need some help.
I'm able to read LZ4 compressed data written in Python from Rust, but I get an error when trying to read a file written by the Rust writer from pyarrow.

The error that I'm getting is:

OSError: Lz4 compressed input contains more than one frame

To reproduce the error, please:

1. Run the arrow unit tests, so that a compressed file is created.
2. Run the below to read the file in pyarrow

import pyarrow as pa
file = pa.ipc.open_file("$ARROW_ROOT/rust/arrow/target/debug/testdata/primitive_lz4.arrow_file")
batches = file.read_all()

Thanks

[mqy]

The error can be found at https://github.com/apache/arrow/blob/master/cpp/src/arrow/util/compression_lz4.cc line 283-285:

if (input_len != 0) {
      return Status::IOError("Lz4 compressed input contains more than one frame");
}

Seems that input_len remains non-zero while the decompression completed.

[github-actions]

https://issues.apache.org/jira/browse/ARROW-8676

[nevi-me]

Seems that input_len remains non-zero while the decompression completed.

the logic of how the buffers are compressed isn't explained in a beginner-friendly way, so I likely am implementing the compression incorrectly. I used the Java implementation in #8949 as a reference. I couldn't quite understand the C++ implementation.

[mqy]

This loop can be omitted.

[mqy]

@nevi-me would you please have a look at BodyCompressionBuilder from ipc/gen/Message.rs?

[nevi-me]

@nevi-me would you please have a look at BodyCompressionBuilder from ipc/gen/Message.rs?

Did you notice something with it? I might need a bit more context :)

[mqy]

Did you notice something with it? I might need a bit more context :)

pub struct BodyCompressionArgs {
    pub codec: CompressionType,
    pub method: BodyCompressionMethod,
}

Perhaps you missed RecordBatchBuilder::add_compression()?

[nevi-me]

Perhaps you missed RecordBatchBuilder::add_compression()?

I missed it at the dictionary write, but not when writing a plain recordbatch; so it's not that. If I hadn't written the compression details completely, the C++ implementation wouldn't have known that the message is compressed or with LZ4.
I'll debug when I have sufficient time

[mqy]

Perhaps you missed RecordBatchBuilder::add_compression()?

I missed it at the dictionary write, but not when writing a plain recordbatch; so it's not that. If I hadn't written the compression details completely, the C++ implementation wouldn't have known that the message is compressed or with LZ4.
I'll debug when I have sufficient time

@nevi-me I pulled your branch locally, found add_compression in writer.rs. I'm sorry that I failed to search add_compression from https://github.com/apache/arrow/pull/9137/fies because writer.rs is not loaded.

I'm installing pyarrow, after run test test_write_file_v5_compressed.

[mqy]

@nevi-me no luck to install pyarrow 2.0.0 due to various dependency errors, but found something:
Message.rs requires LZ4 frame format, not the block format.

  // LZ4 frame format, for portability, as provided by lz4frame.h or wrappers
  // thereof. Not to be confused with "raw" (also called "block") format
  // provided by lz4.h

It's a bit complicate to implement with c bindings as what frameCompress.c does. Also found another crate named lzzzz for lz4 frame:
https://docs.rs/lzzzz/0.8.0/lzzzz/lz4f/index.html, where both compress_to_vec and decompress_to_vec looks quite simple, I had made the unit test passed with these APIs.

FYI:

• [lz4frame.h] mentioned in Message.rs: https://github.com/lz4/lz4/blob/fdf2ef5809ca875c454510610764d9125ef2ebbd/lib/lz4frame.h
• [lz4 frame example c code] https://github.com/lz4/lz4/blob/fdf2ef5809ca875c454510610764d9125ef2ebbd/examples/frameCompress.c
• [lz4-rs lz4-sys APIs] https://github.com/10XGenomics/lz4-rs/blob/master/lz4-sys/src/lib.rs
• [c++ uses LZ4F_xxx APIs] https://github.com/apache/arrow/blob/3694794bdfd0677b95b8c95681e392512f1c9237/cpp/src/arrow/util/compression_lz4.cc

[mqy]

C++ has flags for codec. Perhaps we should also add feature for this, because:

1. Compression may not be a very common feature, I saw ARROW-11188: [Rust] Support crypto functions from PostgreSQL dialect #9139 also has similar discussions.
2. lz4 library has unsafe codes that depend on liblz4

[mqy]

Perhaps the arg total_len of buffers.push(ipc::Buffer::new(...) should not account for pad_len.

[mqy]

@nevi-me perhaps the arg total_len of buffers.push(ipc::Buffer::new(...) should not account for pad_len. After fixing this possible bug, my test on pyarrow passed and no failure with cargo test, but I'm doubting if this is the root cause, because:

• I changed quite a lot based on your PR.
• This line was last updated in PR ARROW-518 a year ago.

I took quite some time to compare with the C++ code, it looks c++ set buffer size as actual memory size.
The IPC spec recordbatch-message states that:

The size field of Buffer is not required to account for padding bytes. Since this metadata can be used to communicate in-memory pointer addresses between libraries, it is recommended to set size to the actual memory size rather than the padded size.

FYI: 68a558b at master...mqy:nevi-me_ARROW-8676 based on your PR.

[nevi-me]

I'll come back to this in the coming weeks