This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 5107eda0 fix(metadata): write and detect gzip-compressed table
metadata (#1014)
5107eda0 is described below
commit 5107eda08bdea434ede4efee66a8e5bc7732cc22
Author: YangJie <[email protected]>
AuthorDate: Wed Oct 21 12:11:00 2026 -0400
fix(metadata): write and detect gzip-compressed table metadata (#1014)
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 '<name>.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.
NewTableMetadataFilePath names the file '<version>-<uuid>.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 | 10 ++++-
src/iceberg/test/gzip_decompress_test.cc | 49 ++++++++++++++++++++++
src/iceberg/test/metadata_io_test.cc | 47 +++++++++++++++++++++
src/iceberg/util/gzip_internal.cc | 70 ++++++++++++++++++++++++++++++++
src/iceberg/util/gzip_internal.h | 16 ++++++++
6 files changed, 193 insertions(+), 1 deletion(-)
diff --git a/src/iceberg/result.h b/src/iceberg/result.h
index a6666e1b..7398bcd6 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 0d24782a..b67ab32f 100644
--- a/src/iceberg/table_metadata.cc
+++ b/src/iceberg/table_metadata.cc
@@ -425,7 +425,7 @@ Result<MetadataFileCodecType>
TableMetadataUtil::Codec::FromString(
Result<MetadataFileCodecType> 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);
}
@@ -489,8 +489,16 @@ Result<std::string> 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<GZipCompressor>();
+ 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/gzip_decompress_test.cc
b/src/iceberg/test/gzip_decompress_test.cc
index b855e4a0..6670a745 100644
--- a/src/iceberg/test/gzip_decompress_test.cc
+++ b/src/iceberg/test/gzip_decompress_test.cc
@@ -84,4 +84,53 @@ 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<unsigned char>(compressed[0]), 0x1f);
+ EXPECT_EQ(static_cast<unsigned char>(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());
+}
+
+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<unsigned char>(compressed[0]), 0x1f);
+ EXPECT_EQ(static_cast<unsigned char>(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/test/metadata_io_test.cc
b/src/iceberg/test/metadata_io_test.cc
index 03d9b0e1..ea66c7f2 100644
--- a/src/iceberg/test/metadata_io_test.cc
+++ b/src/iceberg/test/metadata_io_test.cc
@@ -138,6 +138,53 @@ TEST_F(MetadataIOTest, ReadWriteCompressedMetadata) {
EXPECT_EQ(*metadata_read, metadata);
}
+TEST_F(MetadataIOTest, CodecFromFileGzMetadataJson) {
+ // Java writers emit gzip-compressed metadata as "<name>.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, RoundTripGzipCompressedMetadata) {
+ // With metadata compression enabled the writer names the file
+ // "<version>-<uuid>.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<unsigned char>(raw[0]), 0x1f);
+ EXPECT_EQ(static_cast<unsigned char>(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 63e5acb2..f5500333 100644
--- a/src/iceberg/util/gzip_internal.cc
+++ b/src/iceberg/util/gzip_internal.cc
@@ -23,6 +23,7 @@
#include <zlib.h>
#include <cstring>
+#include <limits>
#include <memory>
#include <string>
#include <vector>
@@ -86,6 +87,65 @@ 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<std::string> Compress(const std::string& data) {
+ if (data.size() > std::numeric_limits<uInt>::max()) {
+ return CompressError("Cannot compress {} bytes in a single gzip pass",
data.size());
+ }
+ if (!initialized_) {
+ ICEBERG_RETURN_UNEXPECTED(Init());
+ }
+ stream_.avail_in = static_cast<uInt>(data.size());
+ stream_.next_in = reinterpret_cast<Bytef*>(const_cast<char*>(data.data()));
+
+ std::vector<char> out_buffer(32 * 1024);
+ std::string result;
+ int ret = 0;
+ do {
+ stream_.avail_out = static_cast<uInt>(out_buffer.size());
+ stream_.next_out = reinterpret_cast<Bytef*>(out_buffer.data());
+ ret = deflate(&stream_, Z_FINISH);
+ 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);
+ } 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;
+ }
+
+ private:
+ bool initialized_ = false;
+ z_stream stream_;
+};
+
GZipDecompressor::GZipDecompressor() :
zlib_impl_(std::make_unique<ZlibImpl>()) {}
GZipDecompressor::~GZipDecompressor() = default;
@@ -96,4 +156,14 @@ Result<std::string> GZipDecompressor::Decompress(const
std::string& compressed_d
return zlib_impl_->Decompress(compressed_data);
}
+GZipCompressor::GZipCompressor() :
zlib_impl_(std::make_unique<ZlibDeflateImpl>()) {}
+
+GZipCompressor::~GZipCompressor() = default;
+
+Status GZipCompressor::Init() { return zlib_impl_->Init(); }
+
+Result<std::string> 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 61cdc00b..b2cf0773 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<ZlibImpl> zlib_impl_;
};
+/// \brief Compresses data into the gzip format read by GZipDecompressor.
+class GZipCompressor {
+ public:
+ GZipCompressor();
+
+ ~GZipCompressor();
+
+ Status Init();
+
+ Result<std::string> Compress(const std::string& data);
+
+ private:
+ std::unique_ptr<ZlibDeflateImpl> zlib_impl_;
+};
+
} // namespace iceberg
