fix: LZ4 block compressor/decompressor safety issues (alibaba#344)

Author: lxy-9602

src/paimon/common/compression/block_compression_factory_test.cpp

@@ -79,6 +79,39 @@ TEST_P(CompressionFactoryTest, TESTCompressThenDecompress) {
     ASSERT_EQ(0, decompressor->ReadIntLE(read_write_le.data()));
 }
 
+TEST_P(CompressionFactoryTest, TestDecompressTruncatedHeader) {
+    BlockCompressionType type = GetParam();
+    ASSERT_OK_AND_ASSIGN(auto factory, BlockCompressionFactory::Create(type));
+    auto decompressor = factory->GetDecompressor();
+
+    // Source shorter than 8-byte header should return Invalid, not read out of bounds
+    char short_buf[] = {0x01, 0x02, 0x03};
+    char dst[64] = {};
+    ASSERT_NOK(decompressor->Decompress(short_buf, 3, dst, sizeof(dst)));
+}
+
+TEST_P(CompressionFactoryTest, TestCompressInsufficientOutputBuffer) {
+    BlockCompressionType type = GetParam();
+    ASSERT_OK_AND_ASSIGN(auto factory, BlockCompressionFactory::Create(type));
+    auto compressor = factory->GetCompressor();
+
+    // Incompressible data with a tiny output buffer should fail, not produce a corrupt block
+    std::string data(1024, '\0');
+    for (int32_t i = 0; i < static_cast<int32_t>(data.size()); i++) {
+        data[i] = static_cast<char>(i % 251);
+    }
+    // Output buffer too small: only HEADER_LENGTH bytes, no room for compressed payload
+    std::string tiny_dst(8, '\0');
+    ASSERT_NOK(compressor->Compress(data.data(), data.size(), tiny_dst.data(), tiny_dst.size()));
+
+    // Output buffer smaller than HEADER_LENGTH — must not trigger UB from negative capacity
+    char micro_dst[4] = {};
+    ASSERT_NOK(compressor->Compress(data.data(), data.size(), micro_dst, sizeof(micro_dst)));
+
+    // Zero-length output buffer
+    ASSERT_NOK(compressor->Compress(data.data(), data.size(), nullptr, 0));
+}
+
 INSTANTIATE_TEST_SUITE_P(BlockCompressionTypeGroup, CompressionFactoryTest,
                          ::testing::Values(BlockCompressionType::LZ4, BlockCompressionType::ZSTD));
 

src/paimon/common/compression/lz4/lz4_block_compressor.h

@@ -32,10 +32,15 @@ class Lz4BlockCompressor : public BlockCompressor {
 
     Result<int32_t> Compress(const char* src, int32_t src_length, char* dst,
                              int32_t dst_length) override {
+        if (dst_length < BlockCompressor::HEADER_LENGTH) {
+            return Status::Invalid(fmt::format(
+                "Output buffer too small for LZ4 block header, expected at least {} bytes, got {}",
+                BlockCompressor::HEADER_LENGTH, dst_length));
+        }
         int32_t compressed_size =
             LZ4_compress_default(src, dst + BlockCompressor::HEADER_LENGTH, src_length,
                                  dst_length - BlockCompressor::HEADER_LENGTH);
-        if (compressed_size < 0) {
+        if (compressed_size <= 0) {
             return Status::Invalid(fmt::format("Compression failed with code {}", compressed_size));
         }
         WriteIntLE(compressed_size, dst);

src/paimon/common/compression/lz4/lz4_block_decompressor.h

@@ -28,6 +28,11 @@ class Lz4BlockDecompressor : public BlockDecompressor {
  public:
     Result<int32_t> Decompress(const char* src, int32_t src_length, char* dst,
                                int32_t dst_length) override {
+        if (src_length < HEADER_LENGTH) {
+            return Status::Invalid(fmt::format(
+                "Source data too short for LZ4 block header, expected at least {} bytes, got {}",
+                HEADER_LENGTH, src_length));
+        }
         auto compressed_len = ReadIntLE(src);
         auto original_len = ReadIntLE(src + 4);
         PAIMON_RETURN_NOT_OK(ValidateLength(compressed_len, original_len));

src/paimon/format/parquet/predicate_pushdown_test.cpp

@@ -343,7 +343,7 @@ TEST_F(PredicatePushdownTest, TestStringData) {
         CheckResult(read_schema, predicate, /*expected_array=*/expected_array);
     }
     {
-        // f0 like 'me', no data
+        // f0 like 'me', has data
         ASSERT_OK_AND_ASSIGN(const auto predicate,
                              PredicateBuilder::Like(
                                  /*field_index=*/0, /*field_name=*/"f0", FieldType::STRING,