fix: review comments

Author: zhjwpku

src/iceberg/transform.cc

@@ -388,6 +388,28 @@ std::string Transform::ToString() const {
   std::unreachable();
 }
 
+std::string Transform::GeneratePartitionName(std::string_view source_name) const {
+  switch (transform_type_) {
+    case TransformType::kIdentity:
+      return std::string(source_name);
+    case TransformType::kBucket:
+      // Format: sourceName_bucket_N (matching Java: sourceName + "_bucket_" + numBuckets)
+      return std::format("{}_bucket_{}", source_name, std::get<int32_t>(param_));
+    case TransformType::kTruncate:
+      // Format: sourceName_trunc_N (matching Java: sourceName + "_trunc_" + width)
+      return std::format("{}_trunc_{}", source_name, std::get<int32_t>(param_));
+    case TransformType::kYear:
+    case TransformType::kMonth:
+    case TransformType::kDay:
+    case TransformType::kHour:
+    case TransformType::kUnknown:
+      return std::format("{}_{}", source_name, TransformTypeToString(transform_type_));
+    case TransformType::kVoid:
+      return std::format("{}_null", source_name);
+  }
+  std::unreachable();
+}
+
 TransformFunction::TransformFunction(TransformType transform_type,
                                      std::shared_ptr<Type> source_type)
     : transform_type_(transform_type), source_type_(std::move(source_type)) {}

src/iceberg/transform.h

@@ -197,6 +197,11 @@ class ICEBERG_EXPORT Transform : public util::Formattable {
   /// \brief Returns a string representation of this transform (e.g., "bucket[16]").
   std::string ToString() const override;
 
+  /// \brief Generates a partition name for the transform.
+  /// \param source_name The name of the source column.
+  /// \return A string representation of the partition name.
+  std::string GeneratePartitionName(std::string_view source_name) const;
+
   /// \brief Equality comparison.
   friend bool operator==(const Transform& lhs, const Transform& rhs) {
     return lhs.Equals(rhs);

src/iceberg/transform_function.h

@@ -59,6 +59,9 @@ class ICEBERG_EXPORT BucketTransform : public TransformFunction {
   /// \brief Returns INT32 as the output type.
   std::shared_ptr<Type> ResultType() const override;
 
+  /// \brief Returns the number of buckets.
+  int32_t num_buckets() const { return num_buckets_; }
+
   /// \brief Create a BucketTransform.
   /// \param source_type Type of the input data.
   /// \param num_buckets Number of buckets to hash into.

src/iceberg/update/update_partition_spec.cc

@@ -43,9 +43,16 @@ UpdatePartitionSpec::UpdatePartitionSpec(TableIdentifier identifier,
     : identifier_(std::move(identifier)),
       catalog_(std::move(catalog)),
       base_metadata_(std::move(base)) {
-  ICEBERG_DCHECK(catalog_, "Catalog is required to construct UpdatePartitionSpec");
-  ICEBERG_DCHECK(base_metadata_,
-                 "Base table metadata is required to construct UpdatePartitionSpec");
+  if (catalog_ == nullptr) [[unlikely]] {
+    AddError(ErrorKind::kInvalidArgument,
+             "Catalog is required to construct UpdatePartitionSpec");
+    return;
+  }
+  if (base_metadata_ == nullptr) [[unlikely]] {
+    AddError(ErrorKind::kInvalidArgument,
+             "Base table metadata is required to construct UpdatePartitionSpec");
+    return;
+  }
   format_version_ = base_metadata_->format_version;
 
   // Get the current/default partition spec
@@ -94,17 +101,12 @@ UpdatePartitionSpec& UpdatePartitionSpec::AddNonDefaultSpec() {
 UpdatePartitionSpec& UpdatePartitionSpec::AddField(const std::string& source_name) {
   // Find the source field in the schema
   auto field_result = schema_->FindFieldByName(source_name, case_sensitive_);
-  if (!field_result.has_value()) {
-    AddError(ErrorKind::kInvalidArgument,
-             std::format("Cannot find source field: {}", source_name));
-    return *this;
-  }
+  BUILDER_RETURN_IF_ERROR(field_result);
 
   auto field_opt = field_result.value();
   if (!field_opt.has_value()) {
-    AddError(ErrorKind::kInvalidArgument,
-             std::format("Cannot find source field: {}", source_name));
-    return *this;
+    return AddError(ErrorKind::kInvalidArgument,
+                    std::format("Cannot find source field: {}", source_name));
   }
 
   int32_t source_id = field_opt.value().get().field_id();
@@ -160,8 +162,8 @@ UpdatePartitionSpec& UpdatePartitionSpec::AddFieldInternal(
     const std::string* name, int32_t source_id, std::shared_ptr<Transform> transform) {
   // Check for duplicate name in added fields
   if (name != nullptr) {
-    auto it = name_to_added_field_.find(*name);
-    if (it != name_to_added_field_.end()) {
+    auto it = added_field_names_.find(*name);
+    if (it != added_field_names_.end()) {
       AddError(ErrorKind::kInvalidArgument,
                std::format("Cannot add duplicate partition field: {}", *name));
       return *this;
@@ -174,30 +176,32 @@ UpdatePartitionSpec& UpdatePartitionSpec::AddFieldInternal(
   auto existing_it = transform_to_field_.find(validation_key);
   if (existing_it != transform_to_field_.end()) {
     const auto& existing = existing_it->second;
-    if (deletes_.contains(existing.field_id()) && *existing.transform() == *transform) {
+    if (deletes_.contains(existing->field_id()) && *existing->transform() == *transform) {
       // If the field was deleted and we're re-adding the same one, just undo the delete
-      return RewriteDeleteAndAddField(existing, name);
+      return RewriteDeleteAndAddField(*existing, name);
     }
 
-    if (deletes_.find(existing.field_id()) == deletes_.end()) {
+    if (deletes_.find(existing->field_id()) == deletes_.end()) {
       AddError(
           ErrorKind::kInvalidArgument,
           std::format(
-              "Cannot add duplicate partition field for source {} with transform {}, "
+              "Cannot add duplicate partition field {} for source {} with transform {}, "
               "conflicts with {}",
-              source_id, transform->ToString(), existing.ToString()));
+              name ? *name : "unknown", source_id, transform->ToString(),
+              existing->ToString()));
       return *this;
     }
   }
 
   // Check if already being added
-  auto added_it = transform_to_added_field_.find(validation_key);
-  if (added_it != transform_to_added_field_.end()) {
-    AddError(ErrorKind::kInvalidArgument,
-             std::format(
-                 "Cannot add duplicate partition field for source {} with transform {}, "
-                 "already added: {}",
-                 source_id, transform->ToString(), added_it->second.ToString()));
+  if (transform_to_added_field_.contains(validation_key)) {
+    AddError(
+        ErrorKind::kInvalidArgument,
+        std::format(
+            "Cannot add duplicate partition field {} for source {} with transform {}, "
+            "already added: {}",
+            name ? *name : "unknown", source_id, transform->ToString(),
+            transform_to_added_field_.at(validation_key)));
     return *this;
   }
 
@@ -218,19 +222,18 @@ UpdatePartitionSpec& UpdatePartitionSpec::AddFieldInternal(
 
   // Check for redundant time-based partitions
   CheckForRedundantAddedPartitions(new_field);
-
-  transform_to_added_field_.emplace(validation_key, new_field);
+  transform_to_added_field_.emplace(validation_key, field_name);
 
   // Handle name conflicts with existing fields
   auto existing_name_it = name_to_field_.find(field_name);
   if (existing_name_it != name_to_field_.end()) {
     const auto& existing_field = existing_name_it->second;
-    if (!deletes_.contains(existing_field.field_id())) {
-      if (IsVoidTransform(existing_field)) {
+    if (!deletes_.contains(existing_field->field_id())) {
+      if (IsVoidTransform(*existing_field)) {
         // Rename the old deleted field
         std::string renamed =
-            std::format("{}_{}", existing_field.name(), existing_field.field_id());
-        renames_[std::string(existing_field.name())] = renamed;
+            std::format("{}_{}", existing_field->name(), existing_field->field_id());
+        RenameField(std::string(existing_field->name()), renamed);
       } else {
         AddError(
             ErrorKind::kInvalidArgument,
@@ -240,13 +243,13 @@ UpdatePartitionSpec& UpdatePartitionSpec::AddFieldInternal(
     } else {
       // Field is being deleted, rename it to avoid conflict
       std::string renamed =
-          std::format("{}_{}", existing_field.name(), existing_field.field_id());
-      renames_[std::string(existing_field.name())] = renamed;
+          std::format("{}_{}", existing_field->name(), existing_field->field_id());
+      renames_[std::string(existing_field->name())] = renamed;
     }
   }
 
-  name_to_added_field_.emplace(field_name, new_field);
   adds_.push_back(new_field);
+  added_field_names_.emplace(field_name);
 
   return *this;
 }
@@ -262,8 +265,7 @@ UpdatePartitionSpec& UpdatePartitionSpec::RewriteDeleteAndAddField(
 
 UpdatePartitionSpec& UpdatePartitionSpec::RemoveField(const std::string& name) {
   // Cannot delete newly added fields
-  auto added_it = name_to_added_field_.find(name);
-  if (added_it != name_to_added_field_.end()) {
+  if (added_field_names_.contains(name)) {
     AddError(ErrorKind::kInvalidArgument,
              std::format("Cannot delete newly added field: {}", name));
     return *this;
@@ -283,7 +285,7 @@ UpdatePartitionSpec& UpdatePartitionSpec::RemoveField(const std::string& name) {
     return *this;
   }
 
-  deletes_.insert(field_it->second.field_id());
+  deletes_.insert(field_it->second->field_id());
   return *this;
 }
 
@@ -326,8 +328,7 @@ UpdatePartitionSpec& UpdatePartitionSpec::RemoveField(
 UpdatePartitionSpec& UpdatePartitionSpec::RemoveFieldByTransform(
     const TransformKey& key, const std::string& term_str) {
   // Cannot delete newly added fields
-  auto added_it = transform_to_added_field_.find(key);
-  if (added_it != transform_to_added_field_.end()) {
+  if (transform_to_added_field_.contains(key)) {
     AddError(ErrorKind::kInvalidArgument,
              std::format("Cannot delete newly added field: {}", term_str));
     return *this;
@@ -342,29 +343,28 @@ UpdatePartitionSpec& UpdatePartitionSpec::RemoveFieldByTransform(
 
   const auto& field = field_it->second;
   // Cannot rename and delete
-  if (renames_.find(std::string(field.name())) != renames_.end()) {
+  if (renames_.find(std::string(field->name())) != renames_.end()) {
     AddError(ErrorKind::kInvalidArgument,
-             std::format("Cannot rename and delete partition field: {}", field.name()));
+             std::format("Cannot rename and delete partition field: {}", field->name()));
     return *this;
   }
 
-  deletes_.insert(field.field_id());
+  deletes_.insert(field->field_id());
   return *this;
 }
 
 UpdatePartitionSpec& UpdatePartitionSpec::RenameField(const std::string& name,
                                                       const std::string& new_name) {
   // Handle existing void field with the new name
   auto existing_it = name_to_field_.find(new_name);
-  if (existing_it != name_to_field_.end() && IsVoidTransform(existing_it->second)) {
-    std::string renamed =
-        std::format("{}_{}", existing_it->second.name(), existing_it->second.field_id());
-    renames_[new_name] = renamed;
+  if (existing_it != name_to_field_.end() && IsVoidTransform(*existing_it->second)) {
+    std::string renamed = std::format("{}_{}", existing_it->second->name(),
+                                      existing_it->second->field_id());
+    RenameField(std::string(existing_it->second->name()), renamed);
   }
 
   // Cannot rename newly added fields
-  auto added_it = name_to_added_field_.find(name);
-  if (added_it != name_to_added_field_.end()) {
+  if (added_field_names_.contains(name)) {
     AddError(ErrorKind::kInvalidArgument,
              std::format("Cannot rename newly added partition field: {}", name));
     return *this;
@@ -378,7 +378,7 @@ UpdatePartitionSpec& UpdatePartitionSpec::RenameField(const std::string& name,
   }
 
   // Cannot delete and rename
-  if (deletes_.contains(field_it->second.field_id())) {
+  if (deletes_.contains(field_it->second->field_id())) {
     AddError(ErrorKind::kInvalidArgument,
              std::format("Cannot delete and rename partition field: {}", name));
     return *this;
@@ -507,51 +507,7 @@ std::string UpdatePartitionSpec::GeneratePartitionName(
     source_name = std::string(field_result.value().value().get().name());
   }
 
-  // Extract parameter from transform string for bucket and truncate
-  // Transform::ToString() returns "bucket[16]" or "truncate[4]" format
-  std::string transform_str = transform->ToString();
-
-  switch (transform->transform_type()) {
-    case TransformType::kIdentity:
-      return source_name;
-    case TransformType::kBucket: {
-      // Parse "bucket[N]" to extract N
-      // Format: sourceName_bucket_N (matching Java: sourceName + "_bucket_" + numBuckets)
-      size_t open_bracket = transform_str.find('[');
-      size_t close_bracket = transform_str.find(']');
-      if (open_bracket != std::string::npos && close_bracket != std::string::npos) {
-        std::string param_str =
-            transform_str.substr(open_bracket + 1, close_bracket - open_bracket - 1);
-        return std::format("{}_{}_{}", source_name, "bucket", param_str);
-      }
-      return std::format("{}_bucket", source_name);
-    }
-    case TransformType::kTruncate: {
-      // Parse "truncate[N]" to extract N
-      // Format: sourceName_trunc_N (matching Java: sourceName + "_trunc_" + width)
-      size_t open_bracket = transform_str.find('[');
-      size_t close_bracket = transform_str.find(']');
-      if (open_bracket != std::string::npos && close_bracket != std::string::npos) {
-        std::string param_str =
-            transform_str.substr(open_bracket + 1, close_bracket - open_bracket - 1);
-        return std::format("{}_{}_{}", source_name, "trunc", param_str);
-      }
-      return std::format("{}_trunc", source_name);
-    }
-    case TransformType::kYear:
-      return std::format("{}_year", source_name);
-    case TransformType::kMonth:
-      return std::format("{}_month", source_name);
-    case TransformType::kDay:
-      return std::format("{}_day", source_name);
-    case TransformType::kHour:
-      return std::format("{}_hour", source_name);
-    case TransformType::kVoid:
-      return std::format("{}_null", source_name);
-    case TransformType::kUnknown:
-      return std::format("{}_unknown", source_name);
-  }
-  std::unreachable();
+  return transform->GeneratePartitionName(source_name);
 }
 
 bool UpdatePartitionSpec::IsTimeTransform(const std::shared_ptr<Transform>& transform) {
@@ -571,36 +527,33 @@ bool UpdatePartitionSpec::IsVoidTransform(const PartitionField& field) {
 }
 
 void UpdatePartitionSpec::CheckForRedundantAddedPartitions(const PartitionField& field) {
-  if (HasErrors()) return;
-
   if (IsTimeTransform(field.transform())) {
-    auto it = added_time_fields_.find(field.source_id());
-    if (it != added_time_fields_.end()) {
+    if (added_time_fields_.contains(field.source_id())) {
       AddError(ErrorKind::kInvalidArgument,
                std::format("Cannot add redundant partition field: {} conflicts with {}",
-                           field.ToString(), it->second.ToString()));
+                           field.ToString(), added_time_fields_.at(field.source_id())));
       return;
     }
-    added_time_fields_.emplace(field.source_id(), field);
+    added_time_fields_.emplace(field.source_id(), field.ToString());
   }
 }
 
-std::unordered_map<std::string, PartitionField> UpdatePartitionSpec::IndexSpecByName(
-    const PartitionSpec& spec) {
-  std::unordered_map<std::string, PartitionField> index;
+std::unordered_map<std::string, const PartitionField*>
+UpdatePartitionSpec::IndexSpecByName(const PartitionSpec& spec) {
+  std::unordered_map<std::string, const PartitionField*> index;
   for (const auto& field : spec.fields()) {
-    index.emplace(std::string(field.name()), field);
+    index.emplace(std::string(field.name()), &field);
   }
   return index;
 }
 
-std::unordered_map<UpdatePartitionSpec::TransformKey, PartitionField,
+std::unordered_map<UpdatePartitionSpec::TransformKey, const PartitionField*,
                    UpdatePartitionSpec::TransformKeyHash>
 UpdatePartitionSpec::IndexSpecByTransform(const PartitionSpec& spec) {
-  std::unordered_map<TransformKey, PartitionField, TransformKeyHash> index;
+  std::unordered_map<TransformKey, const PartitionField*, TransformKeyHash> index;
   for (const auto& field : spec.fields()) {
     TransformKey key{field.source_id(), field.transform()->ToString()};
-    index.emplace(key, field);
+    index.emplace(key, &field);
   }
   return index;
 }