Merge branch 'main' into support_avro_read

Author: zjw1111

.gitattributes

@@ -3,4 +3,6 @@
 *.jar filter=lfs diff=lfs merge=lfs -text
 *.tar filter=lfs diff=lfs merge=lfs -text
 *.gz filter=lfs diff=lfs merge=lfs -text
-79d01717-8380-4504-86e1-387e6c058d0a filter=lfs diff=lfs merge=lfs -text
+test/test_data/sst/none/79d01717-8380-4504-86e1-387e6c058d0a filter=lfs diff=lfs merge=lfs -text
+test/test_data/sst/lz4/10540951-41d3-4216-aa2c-b15dfd25eb75 filter=lfs diff=lfs merge=lfs -text
+test/test_data/sst/zstd/83d05c53-2353-4160-b756-d50dd851b474 filter=lfs diff=lfs merge=lfs -text

include/paimon/utils/read_ahead_cache.h

@@ -42,33 +42,53 @@ class PAIMON_EXPORT CacheConfig {
  public:
     CacheConfig();
     CacheConfig(uint64_t buffer_size_limit, uint64_t range_size_limit, uint64_t hole_size_limit,
-                uint32_t pre_buffer_range_count);
+                uint64_t pre_buffer_limit);
 
     /// Returns the maximum total size (in bytes) of cached data.
     uint64_t GetBufferSizeLimit() const {
         return buffer_size_limit_;
     }
 
+    /// Sets the maximum total size (in bytes) of cached data.
+    void SetBufferSizeLimit(uint64_t buffer_size_limit) {
+        buffer_size_limit_ = buffer_size_limit;
+    }
+
     /// Returns the maximum allowed size (in bytes) for a single cached range.
     uint64_t GetRangeSizeLimit() const {
         return range_size_limit_;
     }
 
+    /// Sets the maximum allowed size (in bytes) for a single cached range.
+    void SetRangeSizeLimit(uint64_t range_size_limit) {
+        range_size_limit_ = range_size_limit;
+    }
+
     /// Returns the maximum gap size (in bytes) considered mergeable between adjacent ranges.
     uint64_t GetHoleSizeLimit() const {
         return hole_size_limit_;
     }
 
-    /// Returns the number of ranges to pre-buffer ahead of the current read position.
-    uint32_t GetPreBufferRangeCount() const {
-        return pre_buffer_range_count_;
+    /// Sets the maximum gap size (in bytes) considered mergeable between adjacent ranges.
+    void SetHoleSizeLimit(uint64_t hole_size_limit) {
+        hole_size_limit_ = hole_size_limit;
+    }
+
+    /// Returns the maximum size to pre-buffer ahead of the current read position.
+    uint64_t GetPreBufferLimit() const {
+        return pre_buffer_limit_;
+    }
+
+    /// Sets the maximum size to pre-buffer ahead of the current read position.
+    void SetPreBufferLimit(uint64_t pre_buffer_limit) {
+        pre_buffer_limit_ = pre_buffer_limit;
     }
 
  private:
     uint64_t buffer_size_limit_;
     uint64_t range_size_limit_;
     uint64_t hole_size_limit_;
-    uint32_t pre_buffer_range_count_;
+    uint64_t pre_buffer_limit_;
 };
 
 /// A byte range with offset and length.

src/paimon/CMakeLists.txt

@@ -13,6 +13,9 @@
 # limitations under the License.
 
 set(PAIMON_COMMON_SRCS
+    common/compression/block_compression_factory.cpp
+    common/compression/block_compressor.cpp
+    common/compression/block_decompressor.cpp
     common/data/abstract_binary_writer.cpp
     common/data/binary_array.cpp
     common/data/binary_array_writer.cpp
@@ -116,6 +119,7 @@ set(PAIMON_COMMON_SRCS
     common/utils/bloom_filter.cpp
     common/utils/bloom_filter64.cpp
     common/utils/bucket_id_calculator.cpp
+    common/utils/crc32c.cpp
     common/utils/decimal_utils.cpp
     common/utils/delta_varint_compressor.cpp
     common/utils/path_util.cpp
@@ -441,7 +445,9 @@ if(PAIMON_BUILD_TESTS)
 
     add_paimon_test(common_sst_file_format_test
                     SOURCES
+                    common/compression/block_compression_factory_test.cpp
                     common/sst/sst_file_io_test.cpp
+                    common/utils/crc32c_test.cpp
                     STATIC_LINK_LIBS
                     paimon_shared
                     test_utils_static

src/paimon/common/compression/block_compression_factory.cpp

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#include "paimon/common/compression/block_compression_factory.h"
+
+#include "paimon/common/compression/lz4/lz4_block_compression_factory.h"
+#include "paimon/common/compression/none_block_compression_factory.h"
+#include "paimon/common/compression/zstd/zstd_block_compression_factory.h"
+
+namespace paimon {
+
+Result<std::shared_ptr<BlockCompressionFactory>> BlockCompressionFactory::Create(
+    BlockCompressionType compression) {
+    switch (compression) {
+        case BlockCompressionType::NONE:
+            return std::make_shared<NoneBlockCompressionFactory>();
+        case BlockCompressionType::LZ4:
+            return std::make_shared<Lz4BlockCompressionFactory>();
+        case BlockCompressionType::ZSTD:
+            return std::make_shared<ZstdBlockCompressionFactory>(ZSTD_COMPRESSION_LEVEL);
+        default:
+            // TODO(liangzi): LZO support
+            return Status::Invalid("Unsupported compression type: " +
+                                   std::to_string(static_cast<int8_t>(compression)));
+    }
+}
+
+}  // namespace paimon

src/paimon/common/compression/block_compression_factory.h

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "paimon/common/compression/block_compression_type.h"
+#include "paimon/common/compression/block_compressor.h"
+#include "paimon/common/compression/block_decompressor.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+/// Each compression codec has an implementation of {@link BlockCompressionFactory} to create
+/// compressors and decompressors.
+class BlockCompressionFactory {
+ public:
+    static Result<std::shared_ptr<BlockCompressionFactory>> Create(
+        BlockCompressionType compression);
+
+    BlockCompressionFactory() = default;
+    virtual ~BlockCompressionFactory() = default;
+
+ public:
+    virtual BlockCompressionType GetCompressionType() const = 0;
+
+    virtual std::shared_ptr<BlockCompressor> GetCompressor() = 0;
+
+    virtual std::shared_ptr<BlockDecompressor> GetDecompressor() = 0;
+
+ private:
+    // Align java implementation
+    static constexpr int32_t ZSTD_COMPRESSION_LEVEL = 1;
+};
+}  // namespace paimon

src/paimon/common/compression/block_compression_factory_test.cpp

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#include "paimon/common/compression/block_compression_factory.h"
+
+#include <cstdint>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/defs.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/status.h"
+#include "paimon/testing/mock/mock_file_batch_reader.h"
+#include "paimon/testing/utils/read_result_collector.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon {
+class Predicate;
+}  // namespace paimon
+
+namespace paimon::test {
+class CompressionFactoryTest : public ::testing::TestWithParam<BlockCompressionType> {};
+
+TEST_P(CompressionFactoryTest, TESTCompressThenDecompress) {
+    int32_t original_len = 16;
+    BlockCompressionType type = GetParam();
+
+    std::string data(original_len, '\0');
+    for (int32_t i = 0; i < original_len; i++) {
+        data[i] = static_cast<char>(i);
+    }
+
+    ASSERT_OK_AND_ASSIGN(auto factory, BlockCompressionFactory::Create(type));
+    ASSERT_EQ(type, factory->GetCompressionType());
+
+    // compress
+    auto compressor = factory->GetCompressor();
+    auto max_len = compressor->GetMaxCompressedSize(data.size());
+    std::string compressed_data(max_len, '\0');
+    auto compressed_size =
+        compressor->Compress(data.data(), data.size(), compressed_data.data(), max_len);
+    ASSERT_OK(compressed_size);
+    ASSERT_GT(compressed_size.value(), 0);
+    compressed_data.resize(compressed_size.value());
+
+    // decompress
+    auto decompressor = factory->GetDecompressor();
+    std::string decompressed_data(original_len, '\0');
+    auto decompressed_size =
+        decompressor->Decompress(compressed_data.data(), compressed_data.size(),
+                                 decompressed_data.data(), decompressed_data.size());
+    ASSERT_OK(decompressed_size);
+    ASSERT_GT(decompressed_size.value(), 0);
+    ASSERT_EQ(data, decompressed_data);
+
+    std::string read_write_le{4, '\0'};
+    compressor->WriteIntLE(123, read_write_le.data());
+    ASSERT_EQ(123, decompressor->ReadIntLE(read_write_le.data()));
+    compressor->WriteIntLE(100000, read_write_le.data());
+    ASSERT_EQ(100000, decompressor->ReadIntLE(read_write_le.data()));
+    compressor->WriteIntLE(-6555, read_write_le.data());
+    ASSERT_EQ(-6555, decompressor->ReadIntLE(read_write_le.data()));
+    compressor->WriteIntLE(0, read_write_le.data());
+    ASSERT_EQ(0, decompressor->ReadIntLE(read_write_le.data()));
+}
+
+INSTANTIATE_TEST_SUITE_P(BlockCompressionTypeGroup, CompressionFactoryTest,
+                         ::testing::Values(BlockCompressionType::LZ4, BlockCompressionType::ZSTD));
+
+}  // namespace paimon::test

src/paimon/common/compression/block_compression_type.h

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#pragma once
+
+#include "paimon/result.h"
+
+namespace paimon {
+
+/// Block Compression type.
+enum class BlockCompressionType { NONE = 0, ZSTD = 1, LZ4 = 2, LZO = 3 };
+
+}  // namespace paimon

src/paimon/common/compression/block_compressor.cpp

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#include "paimon/common/compression/block_compressor.h"
+
+namespace paimon {
+
+void BlockCompressor::WriteIntLE(int32_t val, char* buf) {
+    buf[0] = static_cast<char>(val);
+    buf[1] = static_cast<char>(val >> 8);
+    buf[2] = static_cast<char>(val >> 16);
+    buf[3] = static_cast<char>(val >> 24);
+}
+
+}  // namespace paimon

src/paimon/common/compression/block_compressor.h

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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.
+ */
+
+#pragma once
+
+#include "paimon/result.h"
+
+namespace paimon {
+
+/// A compressor which compresses a whole byte array each time.
+class BlockCompressor {
+ public:
+    static void WriteIntLE(int32_t val, char* buf);
+
+ public:
+    virtual ~BlockCompressor() = default;
+
+ public:
+    /**
+     * Get the max compressed size for a given original size.
+     * @param src_size The original size
+     * @return The max compressed size
+     */
+    virtual int32_t GetMaxCompressedSize(int32_t src_size) = 0;
+
+    /**
+     * Compress data read from src, and write the compressed data to dst.
+     *
+     * @param src Uncompressed data to read from
+     * @param src_length The length of data which want to be compressed
+     * @param dst The target to write compressed data
+     * @param dst_length The max length of data
+     * @return Length of compressed data
+     */
+    /// Compress data read from src, and write the compressed data to dst.
+    virtual Result<int32_t> Compress(const char* src, int32_t src_length, char* dst,
+                                     int32_t dst_length) = 0;
+
+ public:
+    /**
+     * We put two integers before each compressed block, the first integer represents the compressed
+     * length of the block, and the second one represents the original length of the block.
+     */
+    static constexpr int32_t HEADER_LENGTH = 8;
+};
+}  // namespace paimon