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/binder.cc
     expression/evaluator.cc
@@ -43,6 +44,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/struct_like.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,9 +22,14 @@
 /// \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/macros.h"
 
 namespace iceberg {
 
@@ -60,6 +65,17 @@ class ICEBERG_EXPORT PendingUpdate {
 
  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;
 };
 
 /// \brief Template class for type-safe table metadata changes using builder pattern
@@ -89,6 +105,80 @@ class ICEBERG_EXPORT PendingUpdateTyped : public PendingUpdate {
 
  protected:
   PendingUpdateTyped() = default;
+
+  /// \brief Apply the pending changes to a TableMetadataBuilder
+  ///
+  /// Default implementation: calls Apply() to get the result, then applies it
+  /// to the builder using ApplyResult().
+  ///
+  /// \param builder The TableMetadataBuilder to apply changes to
+  /// \return Status::OK if the changes were applied successfully, or an error
+  Status Apply(TableMetadataBuilder& builder) override {
+    auto result = Apply();
+    ICEBERG_RETURN_UNEXPECTED(result);
+
+    return ApplyResult(builder, std::move(result.value()));
+  }
+
+  /// \brief Apply the result to a TableMetadataBuilder
+  ///
+  /// Subclasses must implement this method to apply the result of Apply()
+  /// to the builder.
+  ///
+  /// \param builder The TableMetadataBuilder to apply the result to
+  /// \param result The result from Apply()
+  /// \return Status::OK if the result was applied successfully, or an error
+  virtual Status ApplyResult(TableMetadataBuilder& builder, T result) = 0;
+};
+
+/// \brief Builder for updating (set/remove) table properties
+///
+/// This class provides a fluent API for setting or removing table properties within a
+/// transaction. Mutations are accumulated and applied atomically when the transaction
+/// is committed.
+struct ICEBERG_EXPORT PropertiesUpdateChanges {
+  std::unordered_map<std::string, std::string> updates;
+  std::vector<std::string> removals;
+};
+
+class ICEBERG_EXPORT PropertiesUpdate
+    : public PendingUpdateTyped<PropertiesUpdateChanges> {
+ public:
+  PropertiesUpdate() = default;
+  ~PropertiesUpdate() override = default;
+
+  PropertiesUpdate(const PropertiesUpdate&) = delete;
+  PropertiesUpdate& operator=(const PropertiesUpdate&) = delete;
+
+  /// \brief Set a property key-value pair
+  ///
+  /// \param key The property key
+  /// \param value The property value
+  /// \return Reference to this builder for method chaining
+  PropertiesUpdate& Set(std::string const& key, std::string const& value);
+
+  /// \brief Remove a property key
+  ///
+  /// \param key The property key to remove
+  /// \return Reference to this builder for method chaining
+  PropertiesUpdate& Remove(std::string const& key);
+
+  /// \brief Apply the pending changes and return the uncommitted result
+  ///
+  /// \return The pending property updates/removals, or an error
+  Result<PropertiesUpdateChanges> Apply() override;
+
+  /// \brief Apply and commit the pending changes to the table
+  ///
+  /// \return Status::OK if the commit was successful, or an error
+  Status Commit() override;
+
+ private:
+  Status ApplyResult(TableMetadataBuilder& builder,
+                     PropertiesUpdateChanges result) override;
+
+  std::unordered_map<std::string, std::string> updates_;
+  std::vector<std::string> removals_;
 };
 
 }  // namespace iceberg

src/iceberg/table.h

@@ -118,6 +118,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_;