From 55c42a02aed9b58aa97bfbdee6c4366152064b9f Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 29 Sep 2026 04:58:43 +0800 Subject: [PATCH 1/5] fix(metadata): detect gzip codec in .gz.metadata.json file names Codec::FromFileName used find_last_of with the multi-character suffix '.metadata.json', which matches the last single character of the set (the final 'n') instead of the suffix. For gzip-compressed metadata written by Iceberg Java as '.gz.metadata.json', the resulting file_name_without_suffix kept the wrong tail, the '.gz' check failed, and the file was misclassified as uncompressed kNone. Reading such a file then fails to decompress. Use rfind so the whole suffix matches. The legacy '.metadata.json.gz' compatibility branch is unchanged. Round-trip reading of Java-written gzip metadata files is covered by new unit tests. --- src/iceberg/table_metadata.cc | 2 +- src/iceberg/test/metadata_io_test.cc | 25 +++++++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc index 0763c4fe6c..37ea89638f 100644 --- a/src/iceberg/table_metadata.cc +++ b/src/iceberg/table_metadata.cc @@ -425,7 +425,7 @@ Result TableMetadataUtil::Codec::FromString( Result TableMetadataUtil::Codec::FromFileName( std::string_view file_name) { - auto pos = file_name.find_last_of(kTableMetadataFileSuffix); + auto pos = file_name.rfind(kTableMetadataFileSuffix); if (pos == std::string::npos) { return InvalidArgument("{} is not a valid metadata file", file_name); } diff --git a/src/iceberg/test/metadata_io_test.cc b/src/iceberg/test/metadata_io_test.cc index 03d9b0e1ae..a535074b0e 100644 --- a/src/iceberg/test/metadata_io_test.cc +++ b/src/iceberg/test/metadata_io_test.cc @@ -138,6 +138,31 @@ TEST_F(MetadataIOTest, ReadWriteCompressedMetadata) { EXPECT_EQ(*metadata_read, metadata); } +TEST_F(MetadataIOTest, CodecFromFileGzMetadataJson) { + // Java writers emit gzip-compressed metadata as ".gz.metadata.json"; + // the codec must be detected from the ".gz" before the metadata suffix. + auto result = TableMetadataUtil::Codec::FromFileName("00001-abc.gz.metadata.json"); + ASSERT_TRUE(result.has_value()); + EXPECT_EQ(*result, MetadataFileCodecType::kGzip); +} + +TEST_F(MetadataIOTest, CodecFromFilePlainAndLegacyGz) { + auto plain = TableMetadataUtil::Codec::FromFileName("00001-abc.metadata.json"); + ASSERT_TRUE(plain.has_value()); + EXPECT_EQ(*plain, MetadataFileCodecType::kNone); + + auto legacy_gz = TableMetadataUtil::Codec::FromFileName("00001-abc.metadata.json.gz"); + ASSERT_TRUE(legacy_gz.has_value()); + EXPECT_EQ(*legacy_gz, MetadataFileCodecType::kGzip); +} + +TEST_F(MetadataIOTest, CodecFromFileInvalidName) { + // Matches Iceberg Java fromFileName: a name without the '.metadata.json' + // substring is rejected as not a valid metadata file. + EXPECT_THAT(TableMetadataUtil::Codec::FromFileName("foo.txt"), + IsError(ErrorKind::kInvalidArgument)); +} + TEST_F(MetadataIOTest, WriteMetadataWithBase) { TableMetadata base = PrepareMetadata(); From 47d88fea0248b87934ccf7dde0fc116dcc9e0074 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 13 Oct 2026 16:22:32 +0800 Subject: [PATCH 2/5] fix(metadata): compress metadata when the gzip codec is selected NewTableMetadataFilePath names the file '-.gz.metadata.json' when write.metadata.compression-codec is gzip, but Write always emitted plain JSON, so the file contradicted its own name and no reader could inflate it. Add a GZipCompressor next to the existing GZipDecompressor and compress in Write when the location's codec is gzip, completing the round trip. --- src/iceberg/result.h | 2 + src/iceberg/table_metadata.cc | 8 ++++ src/iceberg/test/metadata_io_test.cc | 22 ++++++++++ src/iceberg/util/gzip_internal.cc | 61 ++++++++++++++++++++++++++++ src/iceberg/util/gzip_internal.h | 16 ++++++++ 5 files changed, 109 insertions(+) diff --git a/src/iceberg/result.h b/src/iceberg/result.h index a6666e1b1a..7398bcd6f4 100644 --- a/src/iceberg/result.h +++ b/src/iceberg/result.h @@ -39,6 +39,7 @@ enum class ErrorKind { kBadRequest, kCommitFailed, kCommitStateUnknown, + kCompressError, kDecompressError, kForbidden, kInternalServerError, @@ -106,6 +107,7 @@ DEFINE_ERROR_FUNCTION(AuthenticationFailed) DEFINE_ERROR_FUNCTION(BadRequest) DEFINE_ERROR_FUNCTION(CommitFailed) DEFINE_ERROR_FUNCTION(CommitStateUnknown) +DEFINE_ERROR_FUNCTION(CompressError) DEFINE_ERROR_FUNCTION(DecompressError) DEFINE_ERROR_FUNCTION(Forbidden) DEFINE_ERROR_FUNCTION(InternalServerError) diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc index 37ea89638f..629e50f493 100644 --- a/src/iceberg/table_metadata.cc +++ b/src/iceberg/table_metadata.cc @@ -489,8 +489,16 @@ Result TableMetadataUtil::Write(FileIO& io, const TableMetadata* ba Status TableMetadataUtil::Write(FileIO& io, const std::string& location, const TableMetadata& metadata) { + ICEBERG_ASSIGN_OR_RAISE(auto codec_type, Codec::FromFileName(location)); ICEBERG_ASSIGN_OR_RAISE(auto json, ToJson(metadata)); ICEBERG_ASSIGN_OR_RAISE(auto json_string, ToJsonString(json)); + + if (codec_type == MetadataFileCodecType::kGzip) { + auto gzip_compressor = std::make_unique(); + ICEBERG_RETURN_UNEXPECTED(gzip_compressor->Init()); + ICEBERG_ASSIGN_OR_RAISE(json_string, gzip_compressor->Compress(json_string)); + } + return io.WriteFile(location, json_string); } diff --git a/src/iceberg/test/metadata_io_test.cc b/src/iceberg/test/metadata_io_test.cc index a535074b0e..ea66c7f27f 100644 --- a/src/iceberg/test/metadata_io_test.cc +++ b/src/iceberg/test/metadata_io_test.cc @@ -163,6 +163,28 @@ TEST_F(MetadataIOTest, CodecFromFileInvalidName) { IsError(ErrorKind::kInvalidArgument)); } +TEST_F(MetadataIOTest, RoundTripGzipCompressedMetadata) { + // With metadata compression enabled the writer names the file + // "-.gz.metadata.json", the same layout Iceberg Java uses. + // The payload must really be gzip, and reading it back must decompress it. + TableMetadata metadata = PrepareMetadata(); + metadata.properties.Set(TableProperties::kMetadataCompression, std::string("gzip")); + + ICEBERG_UNWRAP_OR_FAIL(auto written_location, + TableMetadataUtil::Write(*io_, nullptr, "", metadata)); + EXPECT_THAT(written_location, testing::HasSubstr(".gz.metadata.json")); + + // The bytes on disk carry the gzip magic, so another engine can read them. + ICEBERG_UNWRAP_OR_FAIL(auto raw, io_->ReadFile(written_location, std::nullopt)); + ASSERT_GE(raw.size(), 2u); + EXPECT_EQ(static_cast(raw[0]), 0x1f); + EXPECT_EQ(static_cast(raw[1]), 0x8b); + + ICEBERG_UNWRAP_OR_FAIL(auto metadata_read, + TableMetadataUtil::Read(*io_, written_location)); + EXPECT_EQ(*metadata_read, metadata); +} + TEST_F(MetadataIOTest, WriteMetadataWithBase) { TableMetadata base = PrepareMetadata(); diff --git a/src/iceberg/util/gzip_internal.cc b/src/iceberg/util/gzip_internal.cc index 63e5acb2a4..123daaf0da 100644 --- a/src/iceberg/util/gzip_internal.cc +++ b/src/iceberg/util/gzip_internal.cc @@ -86,6 +86,57 @@ class ZlibImpl { z_stream stream_; }; +class ZlibDeflateImpl { + public: + ZlibDeflateImpl() { memset(&stream_, 0, sizeof(stream_)); } + + ~ZlibDeflateImpl() { + if (initialized_) { + deflateEnd(&stream_); + } + } + + Status Init() { + // Maximum window size, plus the offset that selects a gzip wrapper over + // the raw zlib one, so GZipDecompressor's auto-detection accepts it. + constexpr int kWindowBits = 15; + constexpr int kGzipWrapper = 16; + int ret = deflateInit2(&stream_, Z_DEFAULT_COMPRESSION, Z_DEFLATED, + kWindowBits | kGzipWrapper, 8, Z_DEFAULT_STRATEGY); + if (ret != Z_OK) { + return CompressError("deflateInit2 failed, result:{}", ret); + } + initialized_ = true; + return {}; + } + + Result Compress(const std::string& data) { + if (!initialized_) { + ICEBERG_RETURN_UNEXPECTED(Init()); + } + stream_.avail_in = static_cast(data.size()); + stream_.next_in = reinterpret_cast(const_cast(data.data())); + + std::vector out_buffer(32 * 1024); + std::string result; + int ret = 0; + do { + stream_.avail_out = static_cast(out_buffer.size()); + stream_.next_out = reinterpret_cast(out_buffer.data()); + ret = deflate(&stream_, Z_FINISH); + if (ret != Z_OK && ret != Z_STREAM_END && ret != Z_BUF_ERROR) { + return CompressError("deflate failed, result:{}", ret); + } + result.append(out_buffer.data(), out_buffer.size() - stream_.avail_out); + } while (ret != Z_STREAM_END); + return result; + } + + private: + bool initialized_ = false; + z_stream stream_; +}; + GZipDecompressor::GZipDecompressor() : zlib_impl_(std::make_unique()) {} GZipDecompressor::~GZipDecompressor() = default; @@ -96,4 +147,14 @@ Result GZipDecompressor::Decompress(const std::string& compressed_d return zlib_impl_->Decompress(compressed_data); } +GZipCompressor::GZipCompressor() : zlib_impl_(std::make_unique()) {} + +GZipCompressor::~GZipCompressor() = default; + +Status GZipCompressor::Init() { return zlib_impl_->Init(); } + +Result GZipCompressor::Compress(const std::string& data) { + return zlib_impl_->Compress(data); +} + } // namespace iceberg diff --git a/src/iceberg/util/gzip_internal.h b/src/iceberg/util/gzip_internal.h index 61cdc00b3d..b2cf07734b 100644 --- a/src/iceberg/util/gzip_internal.h +++ b/src/iceberg/util/gzip_internal.h @@ -27,6 +27,7 @@ namespace iceberg { class ZlibImpl; +class ZlibDeflateImpl; class GZipDecompressor { public: @@ -42,4 +43,19 @@ class GZipDecompressor { std::unique_ptr zlib_impl_; }; +/// \brief Compresses data into the gzip format read by GZipDecompressor. +class GZipCompressor { + public: + GZipCompressor(); + + ~GZipCompressor(); + + Status Init(); + + Result Compress(const std::string& data); + + private: + std::unique_ptr zlib_impl_; +}; + } // namespace iceberg From 13e014c84da129d0baae1a6968dd67041e440101 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 13 Oct 2026 21:14:59 +0800 Subject: [PATCH 3/5] test(util): cover GZipCompressor round trip and empty input --- src/iceberg/test/gzip_decompress_test.cc | 30 ++++++++++++++++++++++++ src/iceberg/util/gzip_internal.cc | 2 +- 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/src/iceberg/test/gzip_decompress_test.cc b/src/iceberg/test/gzip_decompress_test.cc index b855e4a06a..b380b4bceb 100644 --- a/src/iceberg/test/gzip_decompress_test.cc +++ b/src/iceberg/test/gzip_decompress_test.cc @@ -84,4 +84,34 @@ TEST_F(GZipTest, GZipDecompressedString) { ASSERT_EQ(read_data.value(), test_string); } +TEST_F(GZipTest, GZipCompressorRoundTrip) { + // Larger than the 32 KiB output buffer, so compression needs several passes. + auto test_string = GenerateRandomString(100 * 1024); + + GZipCompressor compressor; + ASSERT_THAT(compressor.Init(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto compressed, compressor.Compress(test_string)); + ASSERT_GE(compressed.size(), 2u); + EXPECT_EQ(static_cast(compressed[0]), 0x1f); + EXPECT_EQ(static_cast(compressed[1]), 0x8b); + + GZipDecompressor decompressor; + ASSERT_THAT(decompressor.Init(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto decompressed, decompressor.Decompress(compressed)); + EXPECT_EQ(decompressed, test_string); +} + +TEST_F(GZipTest, GZipCompressorEmptyInput) { + GZipCompressor compressor; + ASSERT_THAT(compressor.Init(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto compressed, compressor.Compress("")); + // An empty payload still yields a complete gzip stream (header and trailer). + EXPECT_FALSE(compressed.empty()); + + GZipDecompressor decompressor; + ASSERT_THAT(decompressor.Init(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto decompressed, decompressor.Decompress(compressed)); + EXPECT_TRUE(decompressed.empty()); +} + } // namespace iceberg diff --git a/src/iceberg/util/gzip_internal.cc b/src/iceberg/util/gzip_internal.cc index 123daaf0da..dd86c1b2c8 100644 --- a/src/iceberg/util/gzip_internal.cc +++ b/src/iceberg/util/gzip_internal.cc @@ -124,7 +124,7 @@ class ZlibDeflateImpl { stream_.avail_out = static_cast(out_buffer.size()); stream_.next_out = reinterpret_cast(out_buffer.data()); ret = deflate(&stream_, Z_FINISH); - if (ret != Z_OK && ret != Z_STREAM_END && ret != Z_BUF_ERROR) { + if (ret != Z_OK && ret != Z_STREAM_END) { return CompressError("deflate failed, result:{}", ret); } result.append(out_buffer.data(), out_buffer.size() - stream_.avail_out); From 32d5e335cb75a21396ef501873fe67de551e603d Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Wed, 14 Oct 2026 11:21:40 +0800 Subject: [PATCH 4/5] fix(util): reject GZipCompressor input larger than a single zlib pass --- src/iceberg/util/gzip_internal.cc | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/iceberg/util/gzip_internal.cc b/src/iceberg/util/gzip_internal.cc index dd86c1b2c8..29bbdc491f 100644 --- a/src/iceberg/util/gzip_internal.cc +++ b/src/iceberg/util/gzip_internal.cc @@ -23,6 +23,7 @@ #include #include +#include #include #include #include @@ -111,6 +112,9 @@ class ZlibDeflateImpl { } Result Compress(const std::string& data) { + if (data.size() > std::numeric_limits::max()) { + return CompressError("Cannot compress {} bytes in a single gzip pass", data.size()); + } if (!initialized_) { ICEBERG_RETURN_UNEXPECTED(Init()); } From 68c573a7a39e9b869f62ce313dfb0ea2c79ec947 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Fri, 16 Oct 2026 19:57:52 +0800 Subject: [PATCH 5/5] fix(util): reset GZipCompressor after each Compress call --- src/iceberg/test/gzip_decompress_test.cc | 19 +++++++++++++++++++ src/iceberg/util/gzip_internal.cc | 5 +++++ 2 files changed, 24 insertions(+) diff --git a/src/iceberg/test/gzip_decompress_test.cc b/src/iceberg/test/gzip_decompress_test.cc index b380b4bceb..6670a74565 100644 --- a/src/iceberg/test/gzip_decompress_test.cc +++ b/src/iceberg/test/gzip_decompress_test.cc @@ -114,4 +114,23 @@ TEST_F(GZipTest, GZipCompressorEmptyInput) { EXPECT_TRUE(decompressed.empty()); } +TEST_F(GZipTest, GZipCompressorReused) { + // One compressor can compress several payloads; each call yields a complete, + // independent gzip stream. + GZipCompressor compressor; + ASSERT_THAT(compressor.Init(), IsOk()); + + for (const auto& payload : {GenerateRandomString(1024), GenerateRandomString(2048)}) { + ICEBERG_UNWRAP_OR_FAIL(auto compressed, compressor.Compress(payload)); + ASSERT_GE(compressed.size(), 2u); + EXPECT_EQ(static_cast(compressed[0]), 0x1f); + EXPECT_EQ(static_cast(compressed[1]), 0x8b); + + GZipDecompressor decompressor; + ASSERT_THAT(decompressor.Init(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto decompressed, decompressor.Decompress(compressed)); + EXPECT_EQ(decompressed, payload); + } +} + } // namespace iceberg diff --git a/src/iceberg/util/gzip_internal.cc b/src/iceberg/util/gzip_internal.cc index 29bbdc491f..f5500333c9 100644 --- a/src/iceberg/util/gzip_internal.cc +++ b/src/iceberg/util/gzip_internal.cc @@ -133,6 +133,11 @@ class ZlibDeflateImpl { } result.append(out_buffer.data(), out_buffer.size() - stream_.avail_out); } while (ret != Z_STREAM_END); + // Rewind the stream so the next Compress() starts a fresh gzip member. + ret = deflateReset(&stream_); + if (ret != Z_OK) { + return CompressError("deflateReset failed, result:{}", ret); + } return result; }