apache / apache/arrow

[C++][Parquet] lz4-hadoop doesn't support block compression

Open
#43,745 7 comments 0 reactions 0 assignees View on GitHub
Component: C++ Type: bug
Dominant language
C++
Stars
17.1k
Forks
4.3k
Avg merge
3d 13h
Merged PRs (30d)
88

Description

### Describe the bug, including details regarding any error messages, version, and platform.

platform: Ubuntu 22.04, x86_64
arrow: release-17.0.0-rc2

According to the
https://github.com/apache/hadoop/blob/release-3.4.1-RC1/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/BlockCompressorStream.java#L82
and
https://github.com/apache/hadoop/blob/release-3.4.1-RC1/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/Lz4Codec.java#L92
the lz4-hadoop should be implemented with block stream, which means that the input may be split into blocks, each block will be compressed with lz4.
The outputs will be like that
`- 4-byte big endian uncompressed_size of all blocks`
`- 4-byte big endian compressed_size of the flowing block`
`< lz4 compressed block >`
`- 4-byte big endian compressed_size of the flowing block`
`< lz4 compressed block >`
`- 4-byte big endian compressed_size of the flowing block`
`< lz4 compressed block >`
`... repeated until uncompressed_size from outer block is consumed ...`

The implement of lz4-hadoop decompression in arrow seems only accept one block, as it will return `kNotHadoop` immediately if the `maybe_decompressed_size` of the first block is not equal to `expected_decompressed_size`(actually, it is the size of all blocks's decompressed_size).
https://github.com/apache/arrow/blob/release-17.0.0-rc2/cpp/src/arrow/util/compression_lz4.cc#L509

### Code Example
Java write
```
import org.apache.hadoop.fs.Path;
import org.apache.parquet.column.ParquetProperties;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.GroupFactory;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.hadoop.ParquetFileWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.example.ExampleParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.MessageTypeParser;

import java.io.IOException;

public class WriteParquetFileExample {
public static void main(String[] args) {
MessageType schema = MessageTypeParser.parseMessageType(
"message Pair {\n" +
" required binary field1;\n" +
" required int32 field2;\n" +
"}"
);

GroupFactory factory = new SimpleGroupFactory(schema);
ParquetWriter writer;
try {
writer = ExampleParquetWriter.builder(new Path("/path/to/example.parquet"))
.withWriteMode(ParquetFileWriter.Mode.CREATE)
.withWriterVersion(ParquetProperties.WriterVersion.PARQUET_1_0)
.withCompressionCodec(CompressionCodecName.LZ4)
.config("io.compression.codec.lz4.buffersize", "65536")
.withType(schema)
.build();
for (int i = 0; i < 65536; i++) {
Group group = factory.newGroup();
group.add("field1", "abcdefghijklmn___________" + i);
group.add("field2", i);
writer.write(group);
}
writer.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}

```

C++ read
```
#include
#include
#include
#include

int main(int argc, char** argv) {
parquet::ReaderProperties reader_props;
std::string file_path = "/path/to/example.parquet";
std::unique_ptr file_reader = parquet::ParquetFileReader::OpenFile(file_path, false, reader_props);
std::unique_ptr<::parquet::ParquetFileReader> reader = ::parquet::ParquetFileReader::OpenFile(file_path);
std::unique_ptr<::parquet::arrow::FileReader> out;
auto st = parquet::arrow::FileReader::Make(arrow::default_memory_pool(), std::move(reader), &out);
assert(st.ok());

for (int i = 0; i < out->num_row_groups(); ++i) {
auto row_group = out->RowGroup(i);
std::unique_ptr<::arrow::RecordBatchReader> batch_reader;
auto status = out->GetRecordBatchReader(&batch_reader);
std::cout << status.message() << std::endl;
assert(status.ok());
std::shared_ptr _batch;
status = batch_reader->ReadNext(&_batch);
std::cout << status.message() << std::endl;
assert(status.ok());
arrow::PrettyPrint(*_batch, 0, &std::cout);
}

return 0;
}
```
Create the parquet file with java and read the file with c++ will get the error `Corrupt Lz4 compressed data.` from arrow status message.

### Component(s)

C++

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.