feat: add ExpireSnapshots following 3-goal update pattern

shangxinli · shangxinli · commit 36755e3dbfcb · 2025-11-29T06:50:27.000-08:00
Add ExpireSnapshots as a concrete implementation in iceberg/update/
directory, following the 3-goal architecture for table updates.

Goals accomplished:
1. TableMetadataBuilder::RemoveSnapshots() methods exist
2. table::RemoveSnapshots TableUpdate class exists
3. Expose via Table::ExpireSnapshots() and Transaction::ExpireSnapshots()

Implementation details:
- ExpireSnapshots extends PendingUpdateTyped&lt;vector&lt;shared_ptr&lt;Snapshot&gt;&gt;&gt;
- Fluent API with builder methods: ExpireSnapshotId, ExpireOlderThan,
  RetainLast, DeleteWith, SetCleanupLevel
- Table::ExpireSnapshots() factory method creates instances
- Transaction::ExpireSnapshots() interface method added
- Organized in src/iceberg/update/ subdirectory structure

Builder methods:
- ExpireSnapshotId: Mark specific snapshots for removal by ID
- ExpireOlderThan: Expire snapshots older than timestamp
- RetainLast: Keep N most recent snapshots
- DeleteWith: Custom file deletion callback
- SetCleanupLevel: Control cleanup scope (None/MetadataOnly/All)

Apply() and Commit() have placeholder implementations that will be
completed when snapshot expiration logic is fully implemented.
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
@@ -66,6 +66,7 @@ set(ICEBERG_SOURCES
     transform.cc
     transform_function.cc
     type.cc
+    update/expire_snapshots.cc
     util/bucket_util.cc
     util/conversions.cc
     util/decimal.cc
@@ -132,6 +133,7 @@ iceberg_install_all_headers(iceberg)
 add_subdirectory(catalog)
 add_subdirectory(expression)
 add_subdirectory(row)
+add_subdirectory(update)
 add_subdirectory(util)
 
 if(ICEBERG_BUILD_BUNDLE)
diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build
@@ -88,6 +88,7 @@ iceberg_sources = files(
     'transform.cc',
     'transform_function.cc',
     'type.cc',
+    'update/expire_snapshots.cc',
     'util/bucket_util.cc',
     'util/conversions.cc',
     'util/decimal.cc',
@@ -201,6 +202,7 @@ install_headers(
 subdir('catalog')
 subdir('expression')
 subdir('row')
+subdir('update')
 subdir('util')
 
 if get_option('tests').enabled()
diff --git a/src/iceberg/table.cc b/src/iceberg/table.cc
@@ -28,6 +28,7 @@
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_properties.h"
 #include "iceberg/table_scan.h"
+#include "iceberg/update/expire_snapshots.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -114,6 +115,10 @@ std::unique_ptr<Transaction> Table::NewTransaction() const {
   throw NotImplemented("Table::NewTransaction is not implemented");
 }
 
+std::shared_ptr<iceberg::ExpireSnapshots> Table::NewExpireSnapshots() {
+  return std::make_shared<iceberg::ExpireSnapshots>(this);
+}
+
 const std::shared_ptr<FileIO>& Table::io() const { return io_; }
 
 std::unique_ptr<TableScanBuilder> Table::NewScan() const {
diff --git a/src/iceberg/table.h b/src/iceberg/table.h
@@ -115,6 +115,11 @@ class ICEBERG_EXPORT Table {
   /// \return a pointer to the new Transaction
   virtual std::unique_ptr<Transaction> NewTransaction() const;
 
+  /// \brief Create a new expire snapshots operation for this table
+  ///
+  /// \return a shared pointer to the new ExpireSnapshots operation
+  virtual std::shared_ptr<ExpireSnapshots> NewExpireSnapshots();
+
   /// \brief Returns a FileIO to read and write table data and metadata files
   const std::shared_ptr<FileIO>& io() const;
 
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
@@ -97,6 +97,8 @@ add_iceberg_test(expression_test
                  literal_test.cc
                  predicate_test.cc)
 
+add_iceberg_test(update_test SOURCES update/expire_snapshots_test.cc)
+
 add_iceberg_test(json_serde_test
                  SOURCES
                  test_common.cc
diff --git a/src/iceberg/test/update/expire_snapshots_test.cc b/src/iceberg/test/update/expire_snapshots_test.cc
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/update/expire_snapshots.h"
+
+#include <memory>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/result.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/test/matchers.h"
+
+namespace iceberg {
+
+// Basic API tests for ExpireSnapshots
+// Full functional tests will be added when the implementation is complete
+
+TEST(ExpireSnapshotsTest, FluentApiChaining) {
+  // Test that the fluent API works correctly with method chaining
+  ExpireSnapshots expire(nullptr);
+
+  auto& result =
+      expire.ExpireSnapshotId(123).ExpireOlderThan(1000000).RetainLast(5).SetCleanupLevel(
+          CleanupLevel::kMetadataOnly);
+
+  // Verify that chaining returns the same object
+  EXPECT_EQ(&result, &expire);
+}
+
+TEST(ExpireSnapshotsTest, ExpireSnapshotId) {
+  ExpireSnapshots expire(nullptr);
+  expire.ExpireSnapshotId(123);
+
+  // Currently returns NotImplemented - this test will be expanded
+  // when the actual implementation is added
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, ExpireMultipleSnapshotIds) {
+  ExpireSnapshots expire(nullptr);
+  expire.ExpireSnapshotId(100).ExpireSnapshotId(200).ExpireSnapshotId(300);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, ExpireOlderThan) {
+  ExpireSnapshots expire(nullptr);
+  int64_t timestamp = 1609459200000;  // 2021-01-01 00:00:00 UTC
+  expire.ExpireOlderThan(timestamp);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, RetainLast) {
+  ExpireSnapshots expire(nullptr);
+  expire.RetainLast(10);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, DeleteWithCallback) {
+  ExpireSnapshots expire(nullptr);
+  std::vector<std::string> deleted_files;
+
+  expire.DeleteWith(
+      [&deleted_files](std::string_view file) { deleted_files.emplace_back(file); });
+
+  // Currently returns NotImplemented
+  auto result = expire.Commit();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, CleanupLevelNone) {
+  ExpireSnapshots expire(nullptr);
+  expire.SetCleanupLevel(CleanupLevel::kNone);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, CleanupLevelMetadataOnly) {
+  ExpireSnapshots expire(nullptr);
+  expire.SetCleanupLevel(CleanupLevel::kMetadataOnly);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, CleanupLevelAll) {
+  ExpireSnapshots expire(nullptr);
+  expire.SetCleanupLevel(CleanupLevel::kAll);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, CombinedConfiguration) {
+  ExpireSnapshots expire(nullptr);
+  int64_t timestamp = 1609459200000;
+
+  expire.ExpireSnapshotId(100)
+      .ExpireSnapshotId(200)
+      .ExpireOlderThan(timestamp)
+      .RetainLast(5)
+      .SetCleanupLevel(CleanupLevel::kMetadataOnly);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, CommitNotImplemented) {
+  ExpireSnapshots expire(nullptr);
+  expire.ExpireSnapshotId(123);
+
+  // Currently returns NotImplemented
+  auto result = expire.Commit();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+TEST(ExpireSnapshotsTest, ApplyNotImplemented) {
+  ExpireSnapshots expire(nullptr);
+  expire.ExpireOlderThan(1000000);
+
+  // Currently returns NotImplemented
+  auto result = expire.Apply();
+  EXPECT_THAT(result, IsError(ErrorKind::kNotImplemented));
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/transaction.h b/src/iceberg/transaction.h
@@ -44,6 +44,11 @@ class ICEBERG_EXPORT Transaction {
   /// \return a new AppendFiles
   virtual std::shared_ptr<AppendFiles> NewAppend() = 0;
 
+  /// \brief Create a new expire snapshots operation for this transaction
+  ///
+  /// \return a shared pointer to the new ExpireSnapshots operation
+  virtual std::shared_ptr<ExpireSnapshots> NewExpireSnapshots() = 0;
+
   /// \brief Apply the pending changes from all actions and commit
   ///
   /// This method applies all pending data operations and metadata updates in the
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
@@ -161,6 +161,9 @@ class PendingUpdate;
 template <typename T>
 class PendingUpdateTyped;
 
+enum class CleanupLevel;
+class ExpireSnapshots;
+
 /// ----------------------------------------------------------------------------
 /// TODO: Forward declarations below are not added yet.
 /// ----------------------------------------------------------------------------
diff --git a/src/iceberg/update/CMakeLists.txt b/src/iceberg/update/CMakeLists.txt
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+iceberg_install_all_headers(iceberg/update)
diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/update/expire_snapshots.h"
+
+#include "iceberg/result.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+
+namespace iceberg {
+
+ExpireSnapshots::ExpireSnapshots(Table* table) : table_(table) {}
+
+ExpireSnapshots& ExpireSnapshots::ExpireSnapshotId(int64_t snapshot_id) {
+  snapshot_ids_to_expire_.push_back(snapshot_id);
+  return *this;
+}
+
+ExpireSnapshots& ExpireSnapshots::ExpireOlderThan(int64_t timestamp_millis) {
+  expire_older_than_ms_ = timestamp_millis;
+  return *this;
+}
+
+ExpireSnapshots& ExpireSnapshots::RetainLast(int num_snapshots) {
+  retain_last_ = num_snapshots;
+  return *this;
+}
+
+ExpireSnapshots& ExpireSnapshots::DeleteWith(
+    std::function<void(std::string_view)> delete_func) {
+  delete_func_ = std::move(delete_func);
+  return *this;
+}
+
+ExpireSnapshots& ExpireSnapshots::SetCleanupLevel(CleanupLevel level) {
+  cleanup_level_ = level;
+  return *this;
+}
+
+Result<std::vector<std::shared_ptr<Snapshot>>> ExpireSnapshots::Apply() {
+  // Placeholder implementation - full snapshot expiration logic to be implemented
+  return NotImplemented("ExpireSnapshots::Apply() is not yet implemented");
+}
+
+Status ExpireSnapshots::Commit() {
+  // Placeholder implementation - full commit logic to be implemented
+  return NotImplemented("ExpireSnapshots::Commit() is not yet implemented");
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/update/expire_snapshots.h b/src/iceberg/update/expire_snapshots.h
diff --git a/src/iceberg/update/meson.build b/src/iceberg/update/meson.build
