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.

Author: LuciferYang

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)

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);
 }
 

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

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();
 

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

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