feat: add snapshot util (#420)

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -83,6 +83,7 @@ set(ICEBERG_SOURCES
     util/decimal.cc
     util/gzip_internal.cc
     util/murmurhash3_internal.cc
+    util/snapshot_util.cc
     util/temporal_util.cc
     util/timepoint.cc
     util/truncate_util.cc

src/iceberg/avro/avro_stream_internal.cc

@@ -66,8 +66,9 @@ bool AvroInputStream::next(const uint8_t** data, size_t* len) {
 }
 
 void AvroInputStream::backup(size_t len) {
-  ICEBERG_CHECK(len <= buffer_pos_, "Cannot backup {} bytes, only {} bytes available",
-                len, buffer_pos_);
+  ICEBERG_CHECK_OR_DIE(len <= buffer_pos_,
+                       "Cannot backup {} bytes, only {} bytes available", len,
+                       buffer_pos_);
 
   buffer_pos_ -= len;
   byte_count_ -= len;
@@ -88,7 +89,8 @@ size_t AvroInputStream::byteCount() const { return byte_count_; }
 
 void AvroInputStream::seek(int64_t position) {
   auto status = input_stream_->Seek(position);
-  ICEBERG_CHECK(status.ok(), "Failed to seek to {}, got {}", position, status.ToString());
+  ICEBERG_CHECK_OR_DIE(status.ok(), "Failed to seek to {}, got {}", position,
+                       status.ToString());
 
   buffer_pos_ = 0;
   available_bytes_ = 0;
@@ -116,8 +118,9 @@ bool AvroOutputStream::next(uint8_t** data, size_t* len) {
 }
 
 void AvroOutputStream::backup(size_t len) {
-  ICEBERG_CHECK(len <= buffer_pos_, "Cannot backup {} bytes, only {} bytes available",
-                len, buffer_pos_);
+  ICEBERG_CHECK_OR_DIE(len <= buffer_pos_,
+                       "Cannot backup {} bytes, only {} bytes available", len,
+                       buffer_pos_);
   buffer_pos_ -= len;
 }
 
@@ -126,12 +129,12 @@ uint64_t AvroOutputStream::byteCount() const { return flushed_bytes_ + buffer_po
 void AvroOutputStream::flush() {
   if (buffer_pos_ > 0) {
     auto status = output_stream_->Write(buffer_.data(), buffer_pos_);
-    ICEBERG_CHECK(status.ok(), "Write failed {}", status.ToString());
+    ICEBERG_CHECK_OR_DIE(status.ok(), "Write failed {}", status.ToString());
     flushed_bytes_ += buffer_pos_;
     buffer_pos_ = 0;
   }
   auto status = output_stream_->Flush();
-  ICEBERG_CHECK(status.ok(), "Flush failed {}", status.ToString());
+  ICEBERG_CHECK_OR_DIE(status.ok(), "Flush failed {}", status.ToString());
 }
 
 const std::shared_ptr<::arrow::io::OutputStream>& AvroOutputStream::arrow_output_stream()

src/iceberg/exception.h

@@ -44,7 +44,7 @@ class ICEBERG_EXPORT ExpressionError : public IcebergError {
   explicit ExpressionError(const std::string& what) : IcebergError(what) {}
 };
 
-#define ICEBERG_CHECK(condition, ...)                        \
+#define ICEBERG_CHECK_OR_DIE(condition, ...)                 \
   do {                                                       \
     if (!(condition)) [[unlikely]] {                         \
       throw iceberg::IcebergError(std::format(__VA_ARGS__)); \

src/iceberg/meson.build

@@ -105,6 +105,7 @@ iceberg_sources = files(
     'util/decimal.cc',
     'util/gzip_internal.cc',
     'util/murmurhash3_internal.cc',
+    'util/snapshot_util.cc',
     'util/temporal_util.cc',
     'util/timepoint.cc',
     'util/truncate_util.cc',

src/iceberg/table_metadata.cc

@@ -70,9 +70,9 @@ Result<std::shared_ptr<Schema>> TableMetadata::Schema() const {
 }
 
 Result<std::shared_ptr<Schema>> TableMetadata::SchemaById(
-    const std::optional<int32_t>& schema_id) const {
+    std::optional<int32_t> schema_id) const {
   auto iter = std::ranges::find_if(schemas, [schema_id](const auto& schema) {
-    return schema->schema_id() == schema_id;
+    return schema != nullptr && schema->schema_id() == schema_id;
   });
   if (iter == schemas.end()) {
     return NotFound("Schema with ID {} is not found", schema_id.value_or(-1));
@@ -82,7 +82,7 @@ Result<std::shared_ptr<Schema>> TableMetadata::SchemaById(
 
 Result<std::shared_ptr<PartitionSpec>> TableMetadata::PartitionSpec() const {
   auto iter = std::ranges::find_if(partition_specs, [this](const auto& spec) {
-    return spec->spec_id() == default_spec_id;
+    return spec != nullptr && spec->spec_id() == default_spec_id;
   });
   if (iter == partition_specs.end()) {
     return NotFound("Default partition spec is not found");
@@ -92,7 +92,7 @@ Result<std::shared_ptr<PartitionSpec>> TableMetadata::PartitionSpec() const {
 
 Result<std::shared_ptr<SortOrder>> TableMetadata::SortOrder() const {
   auto iter = std::ranges::find_if(sort_orders, [this](const auto& order) {
-    return order->order_id() == default_sort_order_id;
+    return order != nullptr && order->order_id() == default_sort_order_id;
   });
   if (iter == sort_orders.end()) {
     return NotFound("Default sort order is not found");
@@ -106,7 +106,7 @@ Result<std::shared_ptr<Snapshot>> TableMetadata::Snapshot() const {
 
 Result<std::shared_ptr<Snapshot>> TableMetadata::SnapshotById(int64_t snapshot_id) const {
   auto iter = std::ranges::find_if(snapshots, [snapshot_id](const auto& snapshot) {
-    return snapshot->snapshot_id == snapshot_id;
+    return snapshot != nullptr && snapshot->snapshot_id == snapshot_id;
   });
   if (iter == snapshots.end()) {
     return NotFound("Snapshot with ID {} is not found", snapshot_id);

src/iceberg/table_metadata.h

@@ -128,7 +128,7 @@ struct ICEBERG_EXPORT TableMetadata {
   Result<std::shared_ptr<iceberg::Schema>> Schema() const;
   /// \brief Get the current schema by ID, return NotFoundError if not found
   Result<std::shared_ptr<iceberg::Schema>> SchemaById(
-      const std::optional<int32_t>& schema_id) const;
+      std::optional<int32_t> schema_id) const;
   /// \brief Get the current partition spec, return NotFoundError if not found
   Result<std::shared_ptr<iceberg::PartitionSpec>> PartitionSpec() const;
   /// \brief Get the current sort order, return NotFoundError if not found

src/iceberg/test/CMakeLists.txt

@@ -70,6 +70,7 @@ add_iceberg_test(table_test
                  SOURCES
                  metrics_config_test.cc
                  snapshot_test.cc
+                 snapshot_util_test.cc
                  table_metadata_builder_test.cc
                  table_requirement_test.cc
                  table_requirements_test.cc

src/iceberg/test/meson.build

@@ -47,6 +47,7 @@ iceberg_tests = {
         'sources': files(
             'metrics_config_test.cc',
             'snapshot_test.cc',
+            'snapshot_util_test.cc',
             'table_metadata_builder_test.cc',
             'table_requirement_test.cc',
             'table_requirements_test.cc',