diff --git a/src/iceberg/result.h b/src/iceberg/result.h index a6666e1b1..7398bcd6f 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 0763c4fe6..629e50f49 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); } @@ -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/gzip_decompress_test.cc b/src/iceberg/test/gzip_decompress_test.cc index b855e4a06..6670a7456 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(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()); +} + +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/test/metadata_io_test.cc b/src/iceberg/test/metadata_io_test.cc index 03d9b0e1a..ea66c7f27 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 ".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 + // "-.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 63e5acb2a..f5500333c 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 @@ -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 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()); + } + 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) { + 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()) {} GZipDecompressor::~GZipDecompressor() = default; @@ -96,4 +156,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 61cdc00b3..b2cf07734 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