From 5b343b82c3b75d9578e07d50f4b82a03028f85e0 Mon Sep 17 00:00:00 2001
From: Simbarashe Dzinamarira <simbadzina@gmail.com>
Date: Tue, 24 Mar 2026 17:56:41 -0700
Subject: [PATCH] [Coral-Schema] Fix duplicate field error when partition
 column already exists in schema

When a Hive view projects a partition column as a regular field in its schema,
addPartitionColsToSchema() would attempt to add it again, causing
AvroRuntimeException: "Duplicate field X in record". The fix skips partition
columns that already exist in the schema by name.
---
 .../coral/schema/avro/SchemaUtilities.java    |  7 ++-
 .../schema/avro/SchemaUtilitiesTests.java     | 50 +++++++++++++++++++
 2 files changed, 56 insertions(+), 1 deletion(-)

diff --git a/coral-schema/src/main/java/com/linkedin/coral/schema/avro/SchemaUtilities.java b/coral-schema/src/main/java/com/linkedin/coral/schema/avro/SchemaUtilities.java
index 8b5bf292b..1f23e71a8 100644
--- a/coral-schema/src/main/java/com/linkedin/coral/schema/avro/SchemaUtilities.java
+++ b/coral-schema/src/main/java/com/linkedin/coral/schema/avro/SchemaUtilities.java
@@ -448,7 +448,12 @@ static Schema addPartitionColsToSchema(@Nonnull Schema schema, @Nonnull Table ta
         convertFieldSchemaToAvroSchema("partitionCols", "partitionCols", true, tableOrView.getPartitionKeys());
 
     List<Schema.Field> fieldsWithPartitionColumns = cloneFieldList(schema.getFields());
-    fieldsWithPartitionColumns.addAll(cloneFieldList(partitionColumnsSchema.getFields(), true));
+    Set<String> existingFieldNames = schema.getFields().stream().map(Schema.Field::name).collect(Collectors.toSet());
+    for (Schema.Field partitionField : cloneFieldList(partitionColumnsSchema.getFields(), true)) {
+      if (!existingFieldNames.contains(partitionField.name())) {
+        fieldsWithPartitionColumns.add(partitionField);
+      }
+    }
 
     Schema schemaWithPartitionColumns =
         Schema.createRecord(schema.getName(), schema.getDoc(), schema.getNamespace(), schema.isError());
diff --git a/coral-schema/src/test/java/com/linkedin/coral/schema/avro/SchemaUtilitiesTests.java b/coral-schema/src/test/java/com/linkedin/coral/schema/avro/SchemaUtilitiesTests.java
index 0dc5d94ca..2b39e1201 100644
--- a/coral-schema/src/test/java/com/linkedin/coral/schema/avro/SchemaUtilitiesTests.java
+++ b/coral-schema/src/test/java/com/linkedin/coral/schema/avro/SchemaUtilitiesTests.java
@@ -6,6 +6,7 @@
 package com.linkedin.coral.schema.avro;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -14,6 +15,9 @@
 
 import org.apache.avro.Schema;
 import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -346,4 +350,50 @@ public void testSetupNameAndNamespaceDetectsDeeplyNestedCollisions() {
     Assert.assertTrue(metadataNamespace.contains("IntermediateRecord"),
         "Metadata namespace should follow hierarchical naming. Got: " + metadataNamespace);
   }
+
+  /**
+   * Test that addPartitionColsToSchema does not add a duplicate field when the schema
+   * already contains a field with the same name as a partition column.
+   * This can happen when a Hive view projects a partition column as a regular column.
+   */
+  @Test
+  public void testAddPartitionColsToSchemaSkipsDuplicates() {
+    Schema schemaWithPartCol = SchemaBuilder.record("testRecord").namespace("com.test").fields().name("id").type()
+        .nullable().intType().noDefault().name("date_col").type().nullable().stringType().noDefault().endRecord();
+
+    Table table = new Table();
+    table.setDbName("default");
+    table.setTableName("test_table");
+    table.setPartitionKeys(Collections.singletonList(new FieldSchema("date_col", "string", null)));
+    table.setSd(new StorageDescriptor());
+
+    // Before the fix, this would throw: "Duplicate field date_col in record testRecord"
+    Schema result = SchemaUtilities.addPartitionColsToSchema(schemaWithPartCol, table);
+
+    Assert.assertEquals(result.getFields().size(), 2, "Should have 2 fields: id and date_col (no duplicate)");
+    Assert.assertNotNull(result.getField("id"));
+    Assert.assertNotNull(result.getField("date_col"));
+  }
+
+  /**
+   * Test that addPartitionColsToSchema still adds partition columns when they don't already exist.
+   */
+  @Test
+  public void testAddPartitionColsToSchemaAddsNewPartitionCol() {
+    Schema schemaWithoutPartCol = SchemaBuilder.record("testRecord").namespace("com.test").fields().name("id").type()
+        .nullable().intType().noDefault().name("value").type().nullable().stringType().noDefault().endRecord();
+
+    Table table = new Table();
+    table.setDbName("default");
+    table.setTableName("test_table");
+    table.setPartitionKeys(Collections.singletonList(new FieldSchema("date_col", "string", null)));
+    table.setSd(new StorageDescriptor());
+
+    Schema result = SchemaUtilities.addPartitionColsToSchema(schemaWithoutPartCol, table);
+
+    Assert.assertEquals(result.getFields().size(), 3, "Should have 3 fields: id, value, and date_col");
+    Assert.assertNotNull(result.getField("id"));
+    Assert.assertNotNull(result.getField("value"));
+    Assert.assertNotNull(result.getField("date_col"));
+  }
 }