feat: transactional UpdateProperties support

Author: shuxu.li

src/iceberg/CMakeLists.txt

@@ -19,6 +19,7 @@ set(ICEBERG_INCLUDES "$<BUILD_INTERFACE:${PROJECT_BINARY_DIR}/src>"
                      "$<BUILD_INTERFACE:${PROJECT_SOURCE_DIR}/src>")
 set(ICEBERG_SOURCES
     arrow_c_data_guard_internal.cc
+    base_transaction.cc
     catalog/memory/in_memory_catalog.cc
     expression/aggregate.cc
     expression/binder.cc
@@ -49,6 +50,7 @@ set(ICEBERG_SOURCES
     partition_field.cc
     partition_spec.cc
     partition_summary.cc
+    pending_update.cc
     row/arrow_array_wrapper.cc
     row/manifest_wrapper.cc
     row/partition_values.cc

src/iceberg/base_transaction.cc

@@ -0,0 +1,83 @@
+/*
+ * 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/base_transaction.h"
+
+#include <utility>
+
+#include "iceberg/catalog.h"
+#include "iceberg/pending_update.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/table_requirements.h"
+#include "iceberg/table_update.h"
+
+namespace iceberg {
+
+BaseTransaction::BaseTransaction(std::shared_ptr<Table> table,
+                                 std::shared_ptr<Catalog> catalog)
+    : table_(std::move(table)), catalog_(std::move(catalog)) {
+  ICEBERG_DCHECK(table_ != nullptr, "table must not be null");
+  ICEBERG_DCHECK(catalog_ != nullptr, "catalog must not be null");
+}
+
+const std::shared_ptr<Table>& BaseTransaction::table() const { return table_; }
+
+std::shared_ptr<PropertiesUpdate> BaseTransaction::UpdateProperties() {
+  return RegisterUpdate<PropertiesUpdate>();
+}
+
+std::shared_ptr<AppendFiles> BaseTransaction::NewAppend() {
+  throw NotImplemented("BaseTransaction::NewAppend not implemented");
+}
+
+Status BaseTransaction::CommitTransaction() {
+  const auto& metadata = table_->metadata();
+  if (!metadata) {
+    return InvalidArgument("Table metadata is null");
+  }
+
+  auto builder = TableMetadataBuilder::BuildFrom(metadata.get());
+  for (const auto& pending_update : pending_updates_) {
+    if (!pending_update) {
+      continue;
+    }
+    ICEBERG_RETURN_UNEXPECTED(pending_update->Apply(*builder));
+  }
+
+  auto table_updates = builder->GetChanges();
+  TableUpdateContext context(metadata.get(), /*is_replace=*/false);
+  for (const auto& update : table_updates) {
+    ICEBERG_RETURN_UNEXPECTED(update->GenerateRequirements(context));
+  }
+  ICEBERG_ASSIGN_OR_RAISE(auto table_requirements, context.Build());
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto updated_table,
+      catalog_->UpdateTable(table_->name(), table_requirements, table_updates));
+
+  if (updated_table) {
+    table_ = std::shared_ptr<Table>(std::move(updated_table));
+  }
+
+  pending_updates_.clear();
+  return {};
+}
+
+}  // namespace iceberg

src/iceberg/base_transaction.h

@@ -0,0 +1,56 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <vector>
+
+#include "iceberg/transaction.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Base class for transaction implementations
+class BaseTransaction : public Transaction {
+ public:
+  BaseTransaction(std::shared_ptr<Table> table, std::shared_ptr<Catalog> catalog);
+  ~BaseTransaction() override = default;
+
+  const std::shared_ptr<Table>& table() const override;
+
+  std::shared_ptr<PropertiesUpdate> UpdateProperties() override;
+
+  std::shared_ptr<AppendFiles> NewAppend() override;
+
+  Status CommitTransaction() override;
+
+ protected:
+  template <typename UpdateType, typename... Args>
+  std::shared_ptr<UpdateType> RegisterUpdate(Args&&... args) {
+    auto update = std::make_shared<UpdateType>(std::forward<Args>(args)...);
+    pending_updates_.push_back(update);
+    return update;
+  }
+
+  std::shared_ptr<Table> table_;
+  std::shared_ptr<Catalog> catalog_;
+  std::vector<std::shared_ptr<PendingUpdate>> pending_updates_;
+};
+
+}  // namespace iceberg

src/iceberg/pending_update.cc

@@ -0,0 +1,64 @@
+/*
+ * 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/pending_update.h"
+
+#include "iceberg/catalog.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/table_update.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+// ============================================================================
+// UpdateProperties implementation
+// ============================================================================
+
+PropertiesUpdate& PropertiesUpdate::Set(std::string const& key,
+                                        std::string const& value) {
+  updates_[key] = value;
+  return *this;
+}
+
+PropertiesUpdate& PropertiesUpdate::Remove(std::string const& key) {
+  removals_.push_back(key);
+  return *this;
+}
+
+Result<PropertiesUpdateChanges> PropertiesUpdate::Apply() {
+  return PropertiesUpdateChanges{updates_, removals_};
+}
+
+Status PropertiesUpdate::ApplyResult(TableMetadataBuilder& builder,
+                                     PropertiesUpdateChanges result) {
+  if (!result.updates.empty()) {
+    builder.SetProperties(result.updates);
+  }
+  if (!result.removals.empty()) {
+    builder.RemoveProperties(result.removals);
+  }
+  return {};
+}
+
+Status PropertiesUpdate::Commit() {
+  return NotImplemented("UpdateProperties::Commit() not implemented");
+}
+
+}  // namespace iceberg

src/iceberg/pending_update.h

@@ -22,10 +22,15 @@
 /// \file iceberg/pending_update.h
 /// API for table changes using builder pattern
 
+#include <string>
+#include <unordered_map>
+#include <vector>
+
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
 #include "iceberg/util/error_collector.h"
+#include "iceberg/util/macros.h"
 
 namespace iceberg {
 
@@ -67,6 +72,17 @@ class ICEBERG_EXPORT PendingUpdate : public ErrorCollector {
 
  protected:
   PendingUpdate() = default;
+
+  /// \brief Apply the pending changes to a TableMetadataBuilder
+  ///
+  /// This method applies the changes by calling builder's specific methods.
+  /// The builder will automatically record corresponding TableUpdate objects.
+  ///
+  /// \param builder The TableMetadataBuilder to apply changes to
+  /// \return Status::OK if the changes were applied successfully, or an error
+  virtual Status Apply(TableMetadataBuilder& builder) = 0;
+
+  friend class BaseTransaction;
 };
 
 }  // namespace iceberg

src/iceberg/table.h

@@ -124,6 +124,9 @@ class ICEBERG_EXPORT Table {
   /// \brief Returns a FileIO to read and write table data and metadata files
   const std::shared_ptr<FileIO>& io() const;
 
+  /// \brief Return the underlying table metadata
+  const std::shared_ptr<TableMetadata>& metadata() const { return metadata_; }
+
  private:
   const TableIdentifier identifier_;
   std::shared_ptr<TableMetadata> metadata_;

src/iceberg/table_metadata.cc

@@ -566,4 +566,17 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataBuilder::Build() {
   return result;
 }
 
+std::vector<std::unique_ptr<TableUpdate>> TableMetadataBuilder::GetChanges() {
+  return std::move(impl_->changes);
+}
+
+std::vector<const TableUpdate*> TableMetadataBuilder::GetChangesView() const {
+  std::vector<const TableUpdate*> result;
+  result.reserve(impl_->changes.size());
+  for (const auto& change : impl_->changes) {
+    result.push_back(change.get());
+  }
+  return result;
+}
+
 }  // namespace iceberg

src/iceberg/table_metadata.h

@@ -421,6 +421,23 @@ class ICEBERG_EXPORT TableMetadataBuilder : public ErrorCollector {
   /// \brief Destructor
   ~TableMetadataBuilder();
 
+  /// \brief Get all recorded TableUpdate changes
+  ///
+  /// This method extracts all TableUpdate objects that were automatically
+  /// recorded when modification methods were called. The changes are moved
+  /// out of the builder, so this method should only be called once.
+  ///
+  /// \return A vector of TableUpdate objects representing all changes
+  std::vector<std::unique_ptr<TableUpdate>> GetChanges();
+
+  /// \brief Get all recorded TableUpdate changes (const version for inspection)
+  ///
+  /// This method returns a view of all TableUpdate objects without moving them.
+  /// Useful for inspection before committing.
+  ///
+  /// \return A vector of const pointers to TableUpdate objects
+  std::vector<const TableUpdate*> GetChangesView() const;
+
   // Delete copy operations (use BuildFrom to create a new builder)
   TableMetadataBuilder(const TableMetadataBuilder&) = delete;
   TableMetadataBuilder& operator=(const TableMetadataBuilder&) = delete;

src/iceberg/table_update.cc

@@ -180,8 +180,9 @@ void SetProperties::ApplyTo(TableMetadataBuilder& builder) const {
   builder.SetProperties(updated_);
 }
 
-void SetProperties::GenerateRequirements(TableUpdateContext& context) const {
-  // SetProperties doesn't generate any requirements
+Status SetProperties::GenerateRequirements(TableUpdateContext& context) const {
+  // No requirements
+  return {};
 }
 
 // RemoveProperties
@@ -190,8 +191,9 @@ void RemoveProperties::ApplyTo(TableMetadataBuilder& builder) const {
   builder.RemoveProperties(removed_);
 }
 
-void RemoveProperties::GenerateRequirements(TableUpdateContext& context) const {
-  // RemoveProperties doesn't generate any requirements
+Status RemoveProperties::GenerateRequirements(TableUpdateContext& context) const {
+  // No requirements
+  return {};
 }
 
 // SetLocation

src/iceberg/test/CMakeLists.txt

@@ -77,6 +77,7 @@ add_iceberg_test(table_test
                  table_requirement_test.cc
                  table_requirements_test.cc
                  table_update_test.cc
+                 transaction_pending_update_test.cc
                  update_properties_test.cc)
 
 add_iceberg_test(expression_test