yihua commented on code in PR #19917: URL: https://github.com/apache/hudi/pull/19917#discussion_r4010277903
########## hudi-io/src/main/java/org/apache/hudi/io/compress/airlift/HoodieAirliftSnappyCompressor.java: ########## @@ -0,0 +1,72 @@ +/* + * 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. + */ + +package org.apache.hudi.io.compress.airlift; + +import org.apache.hudi.io.compress.CompressionCodec; +import org.apache.hudi.io.compress.HoodieCompressor; + +import io.airlift.compress.hadoop.HadoopInputStream; +import io.airlift.compress.hadoop.HadoopOutputStream; +import io.airlift.compress.snappy.SnappyHadoopStreams; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; + +import static org.apache.hudi.io.util.IOUtils.readFully; + +/** + * Implementation of {@link HoodieCompressor} for {@link CompressionCodec#SNAPPY} compression + * codec using airlift aircompressor's Snappy compressor and decompressor. + */ +public class HoodieAirliftSnappyCompressor implements HoodieCompressor { + private final SnappyHadoopStreams snappyStreams; + + public HoodieAirliftSnappyCompressor() { + snappyStreams = new SnappyHadoopStreams(); + } + + @Override + public int decompress(InputStream compressedInput, + byte[] targetByteArray, + int offset, + int length) throws IOException { + try (HadoopInputStream stream = snappyStreams.createInputStream(compressedInput)) { + return readFully(stream, targetByteArray, offset, length); Review Comment: Tracing this from `HFileBlock.unpack()`: the input stream is bounded by `onDiskSizeWithoutHeader`, which includes the trailing checksum bytes, and the requested length is `uncompressedSizeWithoutHeader + sizeCheckSum`, so after the last Snappy chunk `readFully` keeps reading and `SnappyHadoopInputStream` parses the checksum words as another block-length/chunk-length pair. Files from our native writer survive only because the NULL checksum is all zeros, but an HBase-written SNAPPY HFile with CRC32C and more than one checksum word per block (any block over 16KB on disk, with a 1MB default block size) fails with `EOFException: encountered EOF while reading block data` or `IOException: Chunk uncompressed size is greater than block size`; GZIP only gets away with the same layout because `GZIPInputStream` ignores trailing garbage. Could we bound the stream in `unpack()` to the compressed payload (`onDiskDataSizeWithHeader - HFILEBLOCK_HEADER_SIZE`) and request exactly `uncompressedSizeW ithoutHeader`, and cover it with an HBase-generated SNAPPY fixture as suggested above? ########## hudi-io/src/main/java/org/apache/hudi/io/compress/airlift/HoodieAirliftSnappyCompressor.java: ########## @@ -0,0 +1,72 @@ +/* + * 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. + */ + +package org.apache.hudi.io.compress.airlift; + +import org.apache.hudi.io.compress.CompressionCodec; +import org.apache.hudi.io.compress.HoodieCompressor; + +import io.airlift.compress.hadoop.HadoopInputStream; +import io.airlift.compress.hadoop.HadoopOutputStream; +import io.airlift.compress.snappy.SnappyHadoopStreams; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; + +import static org.apache.hudi.io.util.IOUtils.readFully; + +/** + * Implementation of {@link HoodieCompressor} for {@link CompressionCodec#SNAPPY} compression + * codec using airlift aircompressor's Snappy compressor and decompressor. + */ +public class HoodieAirliftSnappyCompressor implements HoodieCompressor { Review Comment: non-blocking: This is a line-for-line copy of `HoodieAirliftGzipCompressor` with `JdkGzipHadoopStreams` swapped for `SnappyHadoopStreams`. It might be worth pulling the three methods into an abstract `HoodieAirliftCompressor` that takes a `HadoopStreams` in its constructor, so the next codec (LZ4 and ZSTD are in the same jar) is a one-line subclass. -- 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]
