feat: implement ValidateRedundantPartitions for PartitionSpec

- 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_) {
@@ -261,4 +264,25 @@ 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()) {
+    // Create dedup key: (source_id, transform_dedup_name)
+    auto dedup_key = std::make_pair(field.source_id(), field.transform()->DedupName());
+
+    // Check if this dedup key already exists
+    auto existing_field_iter = dedup_fields.find(dedup_key);
+    ICEBERG_PRECHECK(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
+    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,127 @@ TEST(PartitionSpecTest, PartitionFieldInStructInMap) {
   EXPECT_THAT(result_value, HasErrorMessage("Invalid partition field parent"));
 }
 
+// Test ValidateRedundantPartitions
+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_day1", Transform::Day());
+    PartitionField field2(1, 1001, "ts_day2", Transform::Day());
+
+    auto result = PartitionSpec::Make(schema, 1, {field1, field2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+    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_bucket1", Transform::Bucket(16));
+    PartitionField bucket2(2, 1001, "id_bucket2", Transform::Bucket(16));
+
+    auto result = PartitionSpec::Make(schema, 1, {bucket1, bucket2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+    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_trunc1", Transform::Truncate(4));
+    PartitionField truncate2(3, 1001, "name_trunc2", Transform::Truncate(4));
+
+    auto result =
+        PartitionSpec::Make(schema_with_string, 1, {truncate1, truncate2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+    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_bucket16", Transform::Bucket(16));
+    PartitionField bucket32(2, 1001, "id_bucket32", 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_trunc4", Transform::Truncate(4));
+    PartitionField truncate8(3, 1001, "name_trunc8", 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", Transform::Day());
+    PartitionField id_bucket(2, 1001, "id_bucket", 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", Transform::Day());
+    PartitionField ts_month(1, 1001, "ts_month", 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", 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, "id1", Transform::Identity());
+    PartitionField identity2(1, 1001, "id2", Transform::Identity());
+
+    auto result = PartitionSpec::Make(schema, 1, {identity1, identity2}, false);
+    EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
+    EXPECT_THAT(result, HasErrorMessage("Cannot add redundant partition"));
+  }
+
+  // Test: identity transforms on different fields (allowed)
+  {
+    PartitionField id_identity(1, 1000, "id", Transform::Identity());
+    PartitionField name_identity(2, 1001, "name", Transform::Identity());
+
+    auto result = PartitionSpec::Make(schema, 1, {id_identity, name_identity}, false);
+    EXPECT_THAT(result, IsOk());
+  }
+}
+
 }  // namespace iceberg