feat: implement ValidateRedundantPartitions for PartitionSpec (#442)

- Add ValidateRedundantPartitions method to detect redundant partition fields
- Use deduplication key (source_id + transform.DedupName()) to identify conflicts
- Integrate validation into PartitionSpec::Validate method

Author: gty404

src/iceberg/partition_spec.cc

@@ -23,9 +23,11 @@
 #include <cstddef>
 #include <cstdint>
 #include <format>
+#include <map>
 #include <memory>
 #include <ranges>
 #include <unordered_map>
+#include <utility>
 
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
@@ -132,6 +134,7 @@ bool PartitionSpec::Equals(const PartitionSpec& other) const {
 
 Status PartitionSpec::Validate(const Schema& schema, bool allow_missing_fields) const {
   ICEBERG_RETURN_UNEXPECTED(ValidatePartitionName(schema, *this));
+  ICEBERG_RETURN_UNEXPECTED(ValidateRedundantPartitions(*this));
 
   std::unordered_map<int32_t, int32_t> parents = IndexParents(schema);
   for (const auto& partition_field : fields_) {
@@ -184,11 +187,10 @@ Status PartitionSpec::ValidatePartitionName(const Schema& schema,
   std::unordered_set<std::string> partition_names;
   for (const auto& partition_field : spec.fields()) {
     auto name = std::string(partition_field.name());
-    ICEBERG_PRECHECK(!name.empty(), "Cannot use empty partition name: {}", name);
+    ICEBERG_CHECK(!name.empty(), "Cannot use empty partition name: {}", name);
 
-    if (partition_names.contains(name)) {
-      return InvalidArgument("Cannot use partition name more than once: {}", name);
-    }
+    ICEBERG_CHECK(!partition_names.contains(name),
+                  "Cannot use partition name more than once: {}", name);
     partition_names.insert(name);
 
     ICEBERG_ASSIGN_OR_RAISE(auto schema_field, schema.FindFieldByName(name));
@@ -199,17 +201,15 @@ Status PartitionSpec::ValidatePartitionName(const Schema& schema,
       // field name as long as they are sourced from the same schema field
       if (schema_field.has_value() &&
           schema_field.value().get().field_id() != partition_field.source_id()) {
-        return InvalidArgument(
+        return ValidationFailed(
             "Cannot create identity partition sourced from different field in schema: {}",
             name);
       }
     } else {
       // for all other transforms we don't allow conflicts between partition name and
       // schema field name
-      if (schema_field.has_value()) {
-        return InvalidArgument(
-            "Cannot create partition from name that exists in schema: {}", name);
-      }
+      ICEBERG_CHECK(!schema_field.has_value(),
+                    "Cannot create partition from name that exists in schema: {}", name);
     }
   }
 
@@ -261,4 +261,27 @@ bool PartitionSpec::HasSequentialFieldIds(const PartitionSpec& spec) {
   return true;
 }
 
+Status PartitionSpec::ValidateRedundantPartitions(const PartitionSpec& spec) {
+  // Use a map to track deduplication keys (source_id + transform dedup name)
+  std::map<std::pair<int32_t, std::string>, const PartitionField*> dedup_fields;
+
+  for (const auto& field : spec.fields()) {
+    // The dedup name is provided by the transform's DedupName() method
+    // which typically returns the transform's string representation
+    auto dedup_key = std::make_pair(field.source_id(), field.transform()->DedupName());
+
+    // Check if this dedup key already exists
+    // If it does, we have found a redundant partition field
+    auto existing_field_iter = dedup_fields.find(dedup_key);
+    ICEBERG_CHECK(existing_field_iter == dedup_fields.end(),
+                  "Cannot add redundant partition: {} conflicts with {}",
+                  field.ToString(), existing_field_iter->second->ToString());
+
+    // Add this field to the dedup map for future conflict detection
+    dedup_fields[dedup_key] = &field;
+  }
+
+  return {};
+}
+
 }  // namespace iceberg

src/iceberg/partition_spec.h

@@ -135,6 +135,9 @@ class ICEBERG_EXPORT PartitionSpec : public util::Formattable {
   /// \brief Compare two partition specs for equality.
   bool Equals(const PartitionSpec& other) const;
 
+  /// \brief Validates that there are no redundant partition fields in the spec.
+  static Status ValidateRedundantPartitions(const PartitionSpec& spec);
+
   using SourceIdToFieldsMap = std::unordered_map<int32_t, std::vector<PartitionFieldRef>>;
   static Result<SourceIdToFieldsMap> InitSourceIdToFieldsMap(const PartitionSpec&);
 

src/iceberg/test/partition_spec_test.cc

@@ -303,4 +303,126 @@ TEST(PartitionSpecTest, PartitionFieldInStructInMap) {
   EXPECT_THAT(result_value, HasErrorMessage("Invalid partition field parent"));
 }
 
+TEST(PartitionSpecTest, ValidateRedundantPartitionsExactDuplicates) {
+  // Create a schema with different field types
+  auto ts_field = SchemaField::MakeRequired(1, "ts", timestamp());
+  auto id_field = SchemaField::MakeRequired(2, "id", int64());
+  Schema schema({ts_field, id_field}, Schema::kInitialSchemaId);
+
+  // Test: exact duplicate transforms on same field (same dedup name)
+  {
+    PartitionField field1(1, 1000, "ts_day_1_0", Transform::Day());
+    PartitionField field2(1, 1001, "ts_day_1_1", Transform::Day());
+
+    auto result = PartitionSpec::Make(schema, 1, {field1, field2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+    EXPECT_THAT(result, HasErrorMessage("Cannot add redundant partition"));
+    EXPECT_THAT(result, HasErrorMessage("conflicts with"));
+  }
+
+  // Test: same bucket size on same field (redundant)
+  {
+    PartitionField bucket1(2, 1000, "id_bucket_16_2_0", Transform::Bucket(16));
+    PartitionField bucket2(2, 1001, "id_bucket_16_2_1", Transform::Bucket(16));
+
+    auto result = PartitionSpec::Make(schema, 1, {bucket1, bucket2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+    EXPECT_THAT(result, HasErrorMessage("Cannot add redundant partition"));
+  }
+
+  // Test: same truncate width on same field (redundant)
+  {
+    auto name_field = SchemaField::MakeRequired(3, "name", string());
+    Schema schema_with_string({name_field}, Schema::kInitialSchemaId);
+
+    PartitionField truncate1(3, 1000, "name_trunc_4_3_1", Transform::Truncate(4));
+    PartitionField truncate2(3, 1001, "name_trunc_4_3_2", Transform::Truncate(4));
+
+    auto result =
+        PartitionSpec::Make(schema_with_string, 1, {truncate1, truncate2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+    EXPECT_THAT(result, HasErrorMessage("Cannot add redundant partition"));
+  }
+}
+
+TEST(PartitionSpecTest, ValidateRedundantPartitionsAllowedCases) {
+  // Create a schema with different field types
+  auto ts_field = SchemaField::MakeRequired(1, "ts", timestamp());
+  auto id_field = SchemaField::MakeRequired(2, "id", int64());
+  auto name_field = SchemaField::MakeRequired(3, "name", string());
+  Schema schema({ts_field, id_field, name_field}, Schema::kInitialSchemaId);
+
+  // Test: different bucket sizes on same field (allowed - different dedup names)
+  {
+    PartitionField bucket16(2, 1000, "id_bucket_16_2", Transform::Bucket(16));
+    PartitionField bucket32(2, 1001, "id_bucket_32_2", Transform::Bucket(32));
+
+    auto result = PartitionSpec::Make(schema, 1, {bucket16, bucket32}, false);
+    EXPECT_THAT(result, IsOk());
+  }
+
+  // Test: different truncate widths on same field (allowed - different dedup names)
+  {
+    PartitionField truncate4(3, 1000, "name_trunc_4_3", Transform::Truncate(4));
+    PartitionField truncate8(3, 1001, "name_trunc_8_3", Transform::Truncate(8));
+
+    auto result = PartitionSpec::Make(schema, 1, {truncate4, truncate8}, false);
+    EXPECT_THAT(result, IsOk());
+  }
+
+  // Test: same transforms on different fields (allowed)
+  {
+    PartitionField ts_day(1, 1000, "ts_day_1", Transform::Day());
+    PartitionField id_bucket(2, 1001, "id_bucket_2", Transform::Bucket(16));
+
+    auto result = PartitionSpec::Make(schema, 1, {ts_day, id_bucket}, false);
+    EXPECT_THAT(result, IsOk());
+  }
+
+  // Test: different transforms on same field (allowed if dedup names differ)
+  {
+    PartitionField ts_day(1, 1000, "ts_day_1", Transform::Day());
+    PartitionField ts_month(1, 1001, "ts_month_1", Transform::Month());
+
+    // This should be allowed since Day and Month have different dedup names
+    // The Java logic only checks for exact dedup name matches
+    auto result = PartitionSpec::Make(schema, 1, {ts_day, ts_month}, false);
+    EXPECT_THAT(result, IsOk());
+  }
+
+  // Test: single partition field (no redundancy possible)
+  {
+    PartitionField single_field(1, 1000, "ts_year_1", Transform::Year());
+
+    auto result = PartitionSpec::Make(schema, 1, {single_field}, false);
+    EXPECT_THAT(result, IsOk());
+  }
+}
+
+TEST(PartitionSpecTest, ValidateRedundantPartitionsIdentityTransforms) {
+  // Create a schema with different field types
+  auto id_field = SchemaField::MakeRequired(1, "id", int64());
+  auto name_field = SchemaField::MakeRequired(2, "name", string());
+  Schema schema({id_field, name_field}, Schema::kInitialSchemaId);
+
+  // Test: multiple identity transforms on same field (redundant)
+  {
+    PartitionField identity1(1, 1000, "id_1_0", Transform::Identity());
+    PartitionField identity2(1, 1001, "id_1_1", Transform::Identity());
+
+    auto result = PartitionSpec::Make(schema, 1, {identity1, identity2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+    EXPECT_THAT(result, HasErrorMessage("Cannot add redundant partition"));
+  }
+
+  // Test: identity transforms on different fields (allowed)
+  {
+    PartitionField id_identity(1, 1000, "id_1", Transform::Identity());
+    PartitionField name_identity(2, 1001, "name_2", Transform::Identity());
+
+    auto result = PartitionSpec::Make(schema, 1, {id_identity, name_identity}, false);
+    EXPECT_THAT(result, IsOk());
+  }
+}
+
 }  // namespace iceberg