feat: update properties api

Author: WZhuo

src/iceberg/CMakeLists.txt

@@ -39,10 +39,12 @@ set(ICEBERG_SOURCES
     manifest_reader_internal.cc
     manifest_writer.cc
     metadata_columns.cc
+    metrics_config.cc
     name_mapping.cc
     partition_field.cc
     partition_spec.cc
     partition_summary.cc
+    properties_update.cc
     row/arrow_array_wrapper.cc
     row/manifest_wrapper.cc
     row/struct_like.cc

src/iceberg/meson.build

@@ -61,10 +61,12 @@ iceberg_sources = files(
     'manifest_reader_internal.cc',
     'manifest_writer.cc',
     'metadata_columns.cc',
+    'metrics_config.cc',
     'name_mapping.cc',
     'partition_field.cc',
     'partition_spec.cc',
     'partition_summary.cc',
+    'properties_update.cc',
     'row/arrow_array_wrapper.cc',
     'row/manifest_wrapper.cc',
     'row/struct_like.cc',
@@ -166,9 +168,12 @@ install_headers(
         'manifest_writer.h',
         'metadata_columns.h',
         'metrics.h',
+        'metrics_config.h',
         'name_mapping.h',
         'partition_field.h',
         'partition_spec.h',
+        'pending_update.h',
+        'properties_update.h',
         'result.h',
         'schema_field.h',
         'schema.h',

src/iceberg/metrics_config.cc

@@ -0,0 +1,49 @@
+/*
+ * 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/metrics_config.h"
+
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/schema.h"
+#include "iceberg/table_properties.h"
+
+namespace iceberg {
+
+Status MetricsConfig::VerifyReferencedColumns(
+    const std::unordered_map<std::string, std::string>& updates, const Schema& schema) {
+  for (const auto& [key, value] : updates) {
+    if (!key.starts_with(TableProperties::kMetricModeColumnConfPrefix)) {
+      continue;
+    }
+    auto field_name =
+        std::string_view(key).substr(TableProperties::kMetricModeColumnConfPrefix.size());
+    auto field = schema.FindFieldByName(field_name);
+    if (!field.has_value() || !field.value().has_value()) {
+      return InvalidArgument(
+          "Invalid metrics config, could not find column {} from table prop {} in "
+          "schema {}",
+          field_name, key, schema.ToString());
+    }
+  }
+  return {};
+}
+
+}  // namespace iceberg

src/iceberg/metrics_config.h

@@ -0,0 +1,45 @@
+/*
+ * 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
+
+/// \file iceberg/metrics_config.h
+/// \brief Metrics configuration for Iceberg tables
+
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Configuration utilities for table metrics
+class ICEBERG_EXPORT MetricsConfig {
+ public:
+  /// \brief Verify that all referenced columns are valid
+  /// \param updates The updates to verify
+  /// \param schema The schema to verify against
+  /// \return OK if all referenced columns are valid
+  static Status VerifyReferencedColumns(
+      const std::unordered_map<std::string, std::string>& updates, const Schema& schema);
+};
+
+}  // namespace iceberg

src/iceberg/properties_update.cc

@@ -0,0 +1,130 @@
+/*
+ * 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/properties_update.h"
+
+#include <cstdint>
+#include <memory>
+
+#include "iceberg/catalog.h"
+#include "iceberg/file_format.h"
+#include "iceberg/metrics_config.h"
+#include "iceberg/result.h"
+#include "iceberg/table.h"
+#include "iceberg/table_identifier.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/table_properties.h"
+#include "iceberg/table_requirements.h"
+#include "iceberg/table_update.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+PropertiesUpdate::PropertiesUpdate(TableIdentifier identifier,
+                                   std::shared_ptr<Catalog> catalog,
+                                   std::shared_ptr<TableMetadata> metadata)
+    : identifier_(std::move(identifier)),
+      catalog_(std::move(catalog)),
+      metadata_(std::move(metadata)) {}
+
+PropertiesUpdate& PropertiesUpdate::Set(std::string key, std::string value) {
+  if (std::ranges::find(removals_, key) != std::ranges::end(removals_)) {
+    return *this;
+  }
+
+  if (!TableProperties::reserved_properties().contains(key) ||
+      key == TableProperties::kFormatVersion.key()) {
+    updates_.emplace(std::move(key), std::move(value));
+  }
+
+  return *this;
+}
+
+PropertiesUpdate& PropertiesUpdate::Remove(std::string key) {
+  if (updates_.contains(key)) {
+    return *this;
+  }
+  if (std::ranges::find(removals_, key) == removals_.end()) {
+    removals_.push_back(std::move(key));
+  }
+  return *this;
+}
+
+PropertiesUpdate& PropertiesUpdate::DefaultFormat(FileFormatType format) {
+  updates_.emplace(TableProperties::kDefaultFileFormat.key(), ToString(format));
+  return *this;
+}
+
+Result<void> PropertiesUpdate::Apply() {
+  if (!metadata_) {
+    return InvalidArgument("Cannot commit an empty table");
+  }
+
+  auto iter = updates_.find(TableProperties::kFormatVersion.key());
+  if (iter != updates_.end()) {
+    try {
+      int parsed_version = std::stoi(iter->second);
+      if (parsed_version > TableMetadata::kSupportedTableFormatVersion) {
+        return InvalidArgument(
+            "Cannot upgrade table to unsupported format version: v{} (supported: v{})",
+            parsed_version, TableMetadata::kSupportedTableFormatVersion);
+      }
+      format_version_ = static_cast<int8_t>(parsed_version);
+    } catch (const std::invalid_argument& e) {
+      return InvalidArgument("Invalid format version '{}': not a valid integer",
+                             iter->second);
+    } catch (const std::out_of_range& e) {
+      return InvalidArgument("Format version '{}' is out of range", iter->second);
+    }
+
+    updates_.erase(iter);
+  }
+
+  if (metadata_->Schema().has_value()) {
+    ICEBERG_RETURN_UNEXPECTED(
+        MetricsConfig::VerifyReferencedColumns(updates_, *metadata_->Schema().value()));
+  }
+  return {};
+}
+
+Status PropertiesUpdate::Commit() {
+  ICEBERG_RETURN_UNEXPECTED(Apply());
+
+  std::vector<std::unique_ptr<TableUpdate>> updates;
+  if (!updates_.empty()) {
+    updates.emplace_back(std::make_unique<table::SetProperties>(std::move(updates_)));
+  }
+  if (!removals_.empty()) {
+    updates.emplace_back(std::make_unique<table::RemoveProperties>(std::move(removals_)));
+  }
+  if (format_version_.has_value()) {
+    updates.emplace_back(
+        std::make_unique<table::UpgradeFormatVersion>(format_version_.value()));
+  };
+
+  if (!updates.empty()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto requirements,
+                            TableRequirements::ForUpdateTable(*metadata_, updates));
+    ICEBERG_RETURN_UNEXPECTED(
+        catalog_->UpdateTable(identifier_, requirements, std::move(updates)));
+  }
+  return {};
+}
+
+}  // namespace iceberg

src/iceberg/properties_update.h

@@ -0,0 +1,97 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "iceberg/file_format.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/pending_update.h"
+#include "iceberg/table_identifier.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Updates table properties.
+class ICEBERG_EXPORT PropertiesUpdate : public PendingUpdateTyped<void> {
+ public:
+  /// \brief Constructs a PropertiesUpdate for the specified table.
+  ///
+  /// \param identifier The table identifier
+  /// \param catalog The catalog containing the table
+  /// \param metadata The current table metadata
+  PropertiesUpdate(TableIdentifier identifier, std::shared_ptr<Catalog> catalog,
+                   std::shared_ptr<TableMetadata> metadata);
+
+  /// \brief Sets a property key-value pair.
+  ///
+  /// If the key was previously marked for removal, this operation cancels the removal.
+  /// Reserved properties are handled according to TableProperties rules.
+  ///
+  /// \param key The property key
+  /// \param value The property value
+  /// \return Reference to this PropertiesUpdate for chaining
+  PropertiesUpdate& Set(std::string key, std::string value);
+
+  /// \brief Marks a property for removal.
+  ///
+  /// If the key was previously set for update, this operation cancels the update.
+  ///
+  /// \param key The property key to remove
+  /// \return Reference to this PropertiesUpdate for chaining
+  PropertiesUpdate& Remove(std::string key);
+
+  /// \brief Sets the default file format for the table.
+  ///
+  /// This is a convenience method for setting the "write.format.default" property.
+  ///
+  /// \param format The file format type to use as default
+  /// \return Reference to this PropertiesUpdate for chaining
+  PropertiesUpdate& DefaultFormat(FileFormatType format);
+
+  /// \brief Applies the property changes without committing them.
+  ///
+  /// Validates the pending property changes but does not commit them to the table.
+  /// This method can be used to validate changes before actually committing them.
+  ///
+  /// \return Status::OK if the changes are valid, or an error if validation fails
+  Result<void> Apply() override;
+
+  /// \brief Commits the property changes to the table.
+  ///
+  /// Validates the changes and applies them to the table through the catalog.
+  ///
+  /// \return OK if the changes are valid and committed successfully, or an error
+  Status Commit() override;
+
+ private:
+  TableIdentifier identifier_;
+  std::shared_ptr<Catalog> catalog_;
+  std::shared_ptr<TableMetadata> metadata_;
+
+  std::unordered_map<std::string, std::string> updates_;
+  std::vector<std::string> removals_;
+  std::optional<int8_t> format_version_;
+};
+
+}  // namespace iceberg

src/iceberg/table.cc

@@ -23,6 +23,7 @@
 
 #include "iceberg/catalog.h"
 #include "iceberg/partition_spec.h"
+#include "iceberg/properties_update.h"
 #include "iceberg/schema.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table_metadata.h"
@@ -110,6 +111,10 @@ const std::vector<SnapshotLogEntry>& Table::history() const {
   return metadata_->snapshot_log;
 }
 
+std::unique_ptr<PropertiesUpdate> Table::UpdateProperties() const {
+  return std::make_unique<PropertiesUpdate>(identifier_, catalog_, metadata_);
+}
+
 std::unique_ptr<Transaction> Table::NewTransaction() const {
   throw NotImplemented("Table::NewTransaction is not implemented");
 }

src/iceberg/table.h

@@ -104,6 +104,12 @@ class ICEBERG_EXPORT Table {
   /// \return a vector of history entries
   const std::vector<SnapshotLogEntry>& history() const;
 
+  /// \brief Create a new UpdateProperties to update table properties and commit the
+  /// changes
+  ///
+  /// \return a new UpdateProperties instance
+  virtual std::unique_ptr<PropertiesUpdate> UpdateProperties() const;
+
   /// \brief Create a new table scan builder for this table
   ///
   /// Once a table scan builder is created, it can be refined to project columns and