HIVE-29808: Iceberg: Add guardrails to compaction with schema evolution

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java

@@ -2209,6 +2209,20 @@ Boolean hasAppendsOnly(Iterable<Snapshot> snapshots, SnapshotContext since) {
     return null;
   }
 
+  @Override
+  public void validateCompactionPartition(org.apache.hadoop.hive.ql.metadata.Table hmsTable, String partitionName)
+      throws HiveException {
+    Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
+    if (!IcebergTableUtil.hasUndergonePartitionEvolution(table)) {
+      return;
+    }
+    try {
+      IcebergTableUtil.getPartitionSpec(table, partitionName);
+    } catch (MetaException e) {
+      throw new HiveException(e);
+    }
+  }
+
   @Override
   public void validatePartSpec(org.apache.hadoop.hive.ql.metadata.Table hmsTable, Map<String, String> partitionSpec,
       RewritePolicy policy) throws SemanticException {

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java

@@ -775,15 +775,25 @@ public static PartitionSpec getPartitionSpec(Table icebergTable, String partitio
     // Extract field names from the path: "field1=val1/field2=val2" → [field1, field2]
     List<String> fieldNames = Lists.newArrayList(Warehouse.makeSpecFromName(partitionPath).keySet());
 
-    return icebergTable.specs().values().stream()
+    List<PartitionSpec> matches = icebergTable.specs().values().stream()
         .filter(spec -> {
           List<String> specFieldNames = spec.fields().stream()
               .map(PartitionField::name)
               .toList();
           return specFieldNames.equals(fieldNames);
         })
-        .findFirst() // Supposed to be only one matching spec
-        .orElseThrow(() -> new HiveException("No matching partition spec found for partition path: " + partitionPath));
+        .toList();
+
+    if (matches.size() > 1) {
+      throw new HiveException(String.format(
+          "Ambiguous partition spec for partition path %s: matched spec ids %s",
+          partitionPath,
+          matches.stream().map(PartitionSpec::specId).map(String::valueOf).collect(Collectors.joining(", "))));
+    }
+    if (matches.isEmpty()) {
+      throw new HiveException("No matching partition spec found for partition path: " + partitionPath);
+    }
+    return matches.get(0);
   }
 
   public static TransformSpec getTransformSpec(Table table, String transformName, int sourceId) {

iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestIcebergTableUtil.java

@@ -0,0 +1,117 @@
+/*
+ * 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.
+ */
+
+package org.apache.iceberg.mr.hive;
+
+import java.io.File;
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.UpdatePartitionSpec;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestIcebergTableUtil {
+
+  private static final Schema SCHEMA = new Schema(
+      Types.NestedField.optional(1, "first_name", Types.StringType.get()),
+      Types.NestedField.optional(2, "dept_id", Types.LongType.get())
+  );
+
+  @Rule
+  public TemporaryFolder tmp = new TemporaryFolder();
+
+  @Test
+  public void testGetPartitionSpecAmbiguousAfterPartitionEvolution() throws IOException {
+    Table table = createV1Table();
+    setPartitionSpec(table, "dept_id");
+    setPartitionSpec(table);
+    setPartitionSpec(table, "dept_id");
+
+    // v1 evolution: spec 1 is identity(dept_id), spec 2 is void(dept_id); both share the same field name.
+    assertEquals(4, table.specs().size());
+    assertEquals("identity", table.specs().get(1).fields().get(0).transform().toString());
+    assertEquals("void", table.specs().get(2).fields().get(0).transform().toString());
+
+    AssertHelpers.assertThrows(
+        "Should reject ambiguous partition spec resolution",
+        HiveException.class,
+        "Ambiguous partition spec for partition path dept_id=1: matched spec ids 1, 2",
+        () -> IcebergTableUtil.getPartitionSpec(table, "dept_id=1"));
+  }
+
+  @Test
+  public void testGetPartitionSpecReturnsUniqueMatch() throws Exception {
+    Table table = createV1Table();
+    setPartitionSpec(table, "dept_id");
+
+    PartitionSpec result = IcebergTableUtil.getPartitionSpec(table, "dept_id=1");
+    assertEquals(1, result.specId());
+  }
+
+  @Test
+  public void testGetPartitionSpecNoMatchingSpec() throws IOException {
+    Table table = createV1Table();
+
+    AssertHelpers.assertThrows(
+        "Should fail when no spec matches partition path fields",
+        HiveException.class,
+        "No matching partition spec found for partition path: dept_id=1",
+        () -> IcebergTableUtil.getPartitionSpec(table, "dept_id=1"));
+  }
+
+  private Table createV1Table() throws IOException {
+    File location = tmp.newFolder();
+    Configuration conf = new Configuration();
+    HadoopTables tables = new HadoopTables(conf);
+    return tables.create(
+        SCHEMA,
+        PartitionSpec.unpartitioned(),
+        ImmutableMap.of(
+            TableProperties.FORMAT_VERSION, "1",
+            TableProperties.DEFAULT_FILE_FORMAT, FileFormat.PARQUET.name()),
+        location.getAbsolutePath());
+  }
+
+  /**
+   * Mirrors Hive's {@code ALTER TABLE ... SET PARTITION SPEC (...)}: drop all current partition
+   * fields, then add identity fields for the requested spec (empty varargs → unpartitioned).
+   */
+  private void setPartitionSpec(Table table, String... identityFields) {
+    UpdatePartitionSpec update = table.updateSpec().caseSensitive(false);
+    table.spec().fields().forEach(field -> update.removeField(field.name()));
+    for (String field : identityFields) {
+      update.addField(field);
+    }
+    update.commit();
+    table.refresh();
+  }
+}

iceberg/iceberg-handler/src/test/queries/positive/iceberg_major_compaction_partition_evolution2.q

@@ -58,3 +58,17 @@ select * from ice_orc;
 describe formatted ice_orc;
 show compactions order by 'partition';
 
+alter table ice_orc set partition spec();
+
+insert into ice_orc VALUES ('fn9','ln9', 3);
+insert into ice_orc VALUES ('fn10','ln10', 3);
+
+select * from ice_orc;
+describe formatted ice_orc;
+
+alter table ice_orc COMPACT 'major' and wait;
+
+select * from ice_orc;
+describe formatted ice_orc;
+show compactions order by 'partition';
+

iceberg/iceberg-handler/src/test/queries/positive/iceberg_minor_compaction_partition_evolution.q

@@ -1,3 +1,4 @@
+-- SORT_QUERY_RESULTS
 -- Mask neededVirtualColumns due to non-strict order
 --! qt:replace:/(\s+neededVirtualColumns:\s)(.*)/$1#Masked#/
 -- Mask random uuid
@@ -67,3 +68,18 @@ select * from ice_orc where dept_id=1;
 select * from ice_orc where dept_id=2;
 describe formatted ice_orc;
 show compactions order by 'partition';
+
+alter table ice_orc set partition spec();
+
+insert into ice_orc VALUES ('fn9','ln9', 3), ('fn10','ln10', 3);
+insert into ice_orc VALUES ('fn11','ln11', 3), ('fn12','ln12', 3);
+
+select * from ice_orc;
+describe formatted ice_orc;
+
+alter table ice_orc set tblproperties ('compactor.threshold.target.size'='4000');
+alter table ice_orc COMPACT 'minor' and wait pool 'iceberg';
+
+select * from ice_orc;
+describe formatted ice_orc;
+show compactions order by 'partition';

iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_major_compaction_partition_evolution2.q.out

@@ -282,3 +282,174 @@ POSTHOOK: type: SHOW COMPACTIONS
 CompactionId	Database	Table	Partition	Type	State	Worker host	Worker	Enqueue Time	Start Time	Duration(ms)	HadoopJobId	Error message	Initiator host	Initiator	Pool name	TxnId	Next TxnId	Commit Time	Highest WriteId
 #Masked#	ice_comp	ice_orc	dept_id=2	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 --- 
 #Masked#	ice_comp	ice_orc	 --- 	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 --- 
+PREHOOK: query: alter table ice_orc set partition spec()
+PREHOOK: type: ALTERTABLE_SETPARTSPEC
+PREHOOK: Input: ice_comp@ice_orc
+POSTHOOK: query: alter table ice_orc set partition spec()
+POSTHOOK: type: ALTERTABLE_SETPARTSPEC
+POSTHOOK: Input: ice_comp@ice_orc
+POSTHOOK: Output: ice_comp@ice_orc
+PREHOOK: query: insert into ice_orc VALUES ('fn9','ln9', 3)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: ice_comp@ice_orc
+POSTHOOK: query: insert into ice_orc VALUES ('fn9','ln9', 3)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: ice_comp@ice_orc
+PREHOOK: query: insert into ice_orc VALUES ('fn10','ln10', 3)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: ice_comp@ice_orc
+POSTHOOK: query: insert into ice_orc VALUES ('fn10','ln10', 3)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: ice_comp@ice_orc
+PREHOOK: query: select * from ice_orc
+PREHOOK: type: QUERY
+PREHOOK: Input: ice_comp@ice_orc
+#### A masked pattern was here ####
+POSTHOOK: query: select * from ice_orc
+POSTHOOK: type: QUERY
+POSTHOOK: Input: ice_comp@ice_orc
+#### A masked pattern was here ####
+fn1	ln1	1
+fn10	ln10	3
+fn2	ln2	1
+fn5	ln5	2
+fn6	ln6	2
+fn9	ln9	3
+PREHOOK: query: describe formatted ice_orc
+PREHOOK: type: DESCTABLE
+PREHOOK: Input: ice_comp@ice_orc
+POSTHOOK: query: describe formatted ice_orc
+POSTHOOK: type: DESCTABLE
+POSTHOOK: Input: ice_comp@ice_orc
+# col_name            	data_type           	comment             
+first_name          	string              	                    
+last_name           	string              	                    
+dept_id             	bigint              	                    
+
+# Detailed Table Information	 	 
+Database:           	ice_comp            	 
+#### A masked pattern was here ####
+Retention:          	0                   	 
+#### A masked pattern was here ####
+Table Type:         	EXTERNAL_TABLE      	 
+Table Parameters:	 	 
+	COLUMN_STATS_ACCURATE	{\"BASIC_STATS\":\"true\",\"COLUMN_STATS\":{\"dept_id\":\"true\",\"first_name\":\"true\",\"last_name\":\"true\"}}
+	EXTERNAL            	TRUE                
+	bucketing_version   	2                   
+	compactor.threshold.target.size	1500                
+	current-schema      	{\"type\":\"struct\",\"schema-id\":0,\"fields\":[{\"id\":1,\"name\":\"first_name\",\"required\":false,\"type\":\"string\"},{\"id\":2,\"name\":\"last_name\",\"required\":false,\"type\":\"string\"},{\"id\":3,\"name\":\"dept_id\",\"required\":false,\"type\":\"long\"}]}
+	current-snapshot-id 	#Masked#
+	current-snapshot-summary	{\"manifests-created\":\"1\",\"manifests-kept\":\"2\",\"manifests-replaced\":\"0\",\"added-data-files\":\"1\",\"added-records\":\"1\",\"added-files-size\":\"#Masked#\",\"changed-partition-count\":\"1\",\"total-records\":\"6\",\"total-files-size\":\"#Masked#\",\"total-data-files\":\"4\",\"total-delete-files\":\"0\",\"total-position-deletes\":\"0\",\"total-equality-deletes\":\"0\",\"iceberg-version\":\"#Masked#\"}
+	current-snapshot-timestamp-ms	#Masked#       
+	format-version      	2                   
+#### A masked pattern was here ####
+	numFiles            	4                   
+	numRows             	6                   
+	parquet.compression 	zstd                
+#### A masked pattern was here ####
+	rawDataSize         	0                   
+	serialization.format	1                   
+	snapshot-count      	14                  
+	storage_handler     	org.apache.iceberg.mr.hive.HiveIcebergStorageHandler
+	table_type          	ICEBERG             
+	totalSize           	#Masked#
+#### A masked pattern was here ####
+	uuid                	#Masked#
+	write.delete.mode   	merge-on-read       
+	write.format.default	orc                 
+	write.merge.mode    	merge-on-read       
+	write.metadata.delete-after-commit.enabled	true                
+	write.update.mode   	merge-on-read       
+
+# Storage Information	 	 
+SerDe Library:      	org.apache.iceberg.mr.hive.HiveIcebergSerDe	 
+InputFormat:        	org.apache.iceberg.mr.hive.HiveIcebergInputFormat	 
+OutputFormat:       	org.apache.iceberg.mr.hive.HiveIcebergOutputFormat	 
+Compressed:         	No                  	 
+Sort Columns:       	[]                  	 
+PREHOOK: query: alter table ice_orc COMPACT 'major' and wait
+PREHOOK: type: ALTERTABLE_COMPACT
+PREHOOK: Input: ice_comp@ice_orc
+PREHOOK: Output: ice_comp@ice_orc
+POSTHOOK: query: alter table ice_orc COMPACT 'major' and wait
+POSTHOOK: type: ALTERTABLE_COMPACT
+POSTHOOK: Input: ice_comp@ice_orc
+POSTHOOK: Output: ice_comp@ice_orc
+PREHOOK: query: select * from ice_orc
+PREHOOK: type: QUERY
+PREHOOK: Input: ice_comp@ice_orc
+#### A masked pattern was here ####
+POSTHOOK: query: select * from ice_orc
+POSTHOOK: type: QUERY
+POSTHOOK: Input: ice_comp@ice_orc
+#### A masked pattern was here ####
+fn1	ln1	1
+fn10	ln10	3
+fn2	ln2	1
+fn5	ln5	2
+fn6	ln6	2
+fn9	ln9	3
+PREHOOK: query: describe formatted ice_orc
+PREHOOK: type: DESCTABLE
+PREHOOK: Input: ice_comp@ice_orc
+POSTHOOK: query: describe formatted ice_orc
+POSTHOOK: type: DESCTABLE
+POSTHOOK: Input: ice_comp@ice_orc
+# col_name            	data_type           	comment             
+first_name          	string              	                    
+last_name           	string              	                    
+dept_id             	bigint              	                    
+
+# Detailed Table Information	 	 
+Database:           	ice_comp            	 
+#### A masked pattern was here ####
+Retention:          	0                   	 
+#### A masked pattern was here ####
+Table Type:         	EXTERNAL_TABLE      	 
+Table Parameters:	 	 
+	COLUMN_STATS_ACCURATE	{\"BASIC_STATS\":\"true\",\"COLUMN_STATS\":{\"dept_id\":\"true\",\"first_name\":\"true\",\"last_name\":\"true\"}}
+	EXTERNAL            	TRUE                
+	bucketing_version   	2                   
+	compactor.threshold.target.size	1500                
+	current-schema      	{\"type\":\"struct\",\"schema-id\":0,\"fields\":[{\"id\":1,\"name\":\"first_name\",\"required\":false,\"type\":\"string\"},{\"id\":2,\"name\":\"last_name\",\"required\":false,\"type\":\"string\"},{\"id\":3,\"name\":\"dept_id\",\"required\":false,\"type\":\"long\"}]}
+	current-snapshot-id 	#Masked#
+	current-snapshot-summary	{\"manifests-created\":\"4\",\"manifests-kept\":\"0\",\"manifests-replaced\":\"3\",\"added-data-files\":\"1\",\"deleted-data-files\":\"4\",\"added-records\":\"6\",\"deleted-records\":\"6\",\"added-files-size\":\"#Masked#\",\"removed-files-size\":\"#Masked#\",\"changed-partition-count\":\"3\",\"total-records\":\"6\",\"total-files-size\":\"#Masked#\",\"total-data-files\":\"1\",\"total-delete-files\":\"0\",\"total-position-deletes\":\"0\",\"total-equality-deletes\":\"0\",\"iceberg-version\":\"#Masked#\"}
+	current-snapshot-timestamp-ms	#Masked#       
+	format-version      	2                   
+#### A masked pattern was here ####
+	numFiles            	1                   
+	numRows             	6                   
+	parquet.compression 	zstd                
+#### A masked pattern was here ####
+	rawDataSize         	0                   
+	serialization.format	1                   
+	snapshot-count      	15                  
+	storage_handler     	org.apache.iceberg.mr.hive.HiveIcebergStorageHandler
+	table_type          	ICEBERG             
+	totalSize           	#Masked#
+#### A masked pattern was here ####
+	uuid                	#Masked#
+	write.delete.mode   	merge-on-read       
+	write.format.default	orc                 
+	write.merge.mode    	merge-on-read       
+	write.metadata.delete-after-commit.enabled	true                
+	write.update.mode   	merge-on-read       
+
+# Storage Information	 	 
+SerDe Library:      	org.apache.iceberg.mr.hive.HiveIcebergSerDe	 
+InputFormat:        	org.apache.iceberg.mr.hive.HiveIcebergInputFormat	 
+OutputFormat:       	org.apache.iceberg.mr.hive.HiveIcebergOutputFormat	 
+Compressed:         	No                  	 
+Sort Columns:       	[]                  	 
+PREHOOK: query: show compactions order by 'partition'
+PREHOOK: type: SHOW COMPACTIONS
+POSTHOOK: query: show compactions order by 'partition'
+POSTHOOK: type: SHOW COMPACTIONS
+CompactionId	Database	Table	Partition	Type	State	Worker host	Worker	Enqueue Time	Start Time	Duration(ms)	HadoopJobId	Error message	Initiator host	Initiator	Pool name	TxnId	Next TxnId	Commit Time	Highest WriteId
+#Masked#	ice_comp	ice_orc	dept_id=2	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 --- 
+#Masked#	ice_comp	ice_orc	 --- 	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 --- 
+#Masked#	ice_comp	ice_orc	 --- 	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 ---