add support for partition evolution

Author: deniskuzZ

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

@@ -447,45 +447,56 @@ public static PartitionData toPartitionData(StructLike sourceKey, Types.StructTy
   public static Expression generateExpressionFromPartitionSpec(Table table, Map<String, String> partitionSpec,
       boolean latestSpecOnly) throws SemanticException {
 
-    Map<String, PartitionField> partitionFieldsByName = getPartitionFields(table, latestSpecOnly).stream()
-        .collect(Collectors.toMap(
-            partitionField -> table.schema().findColumnName(partitionField.sourceId()),
-            Function.identity())
+    // Group partition fields by source column name to handle partition evolution
+    // where the same source column may have multiple transforms across different specs
+    Map<String, List<PartitionField>> partitionFieldsBySourceName = getPartitionFields(table, latestSpecOnly).stream()
+        .collect(Collectors.groupingBy(
+            partitionField -> table.schema().findColumnName(partitionField.sourceId()))
         );
 
     Expression predicate = Expressions.alwaysTrue();
 
     for (Map.Entry<String, String> entry : partitionSpec.entrySet()) {
       String partitionColumn = entry.getKey();
-      PartitionField partitionField = partitionFieldsByName.get(partitionColumn);
+      List<PartitionField> partitionFields = partitionFieldsBySourceName.get(partitionColumn);
 
-      if (partitionField == null) {
+      if (partitionFields == null || partitionFields.isEmpty()) {
         throw new SemanticException(String.format("No partition column by the name: %s", partitionColumn));
       }
-      Types.NestedField sourceField = table.schema().findField(partitionField.sourceId());
-      Object sourceValue = Conversions.fromPartitionString(sourceField.type(), entry.getValue());
-      // Apply the transform to the source value
-      @SuppressWarnings("unchecked")
-      Transform<Object, Object> transform = (Transform<Object, Object>) partitionField.transform();
-      Object transformedValue = transform.bind(sourceField.type()).apply(sourceValue);
-
-      TransformSpec transformSpec = TransformSpec.fromString(transform.toString().toUpperCase(), sourceField.name());
-      UnboundTerm<Object> term = SchemaUtils.toTerm(transformSpec);
-
-      predicate = Expressions.and(
-          predicate, Expressions.equal(term, transformedValue));
+
+      // When there are multiple partition fields for the same source column (due to partition evolution),
+      // create an OR expression that matches any of the transforms
+      Expression columnPredicate = Expressions.alwaysFalse();
+
+      for (PartitionField partitionField : partitionFields) {
+        Types.NestedField sourceField = table.schema().findField(partitionField.sourceId());
+        Object sourceValue = Conversions.fromPartitionString(sourceField.type(), entry.getValue());
+        // Apply the transform to the source value
+        @SuppressWarnings("unchecked")
+        Transform<Object, Object> transform = (Transform<Object, Object>) partitionField.transform();
+        Object transformedValue = transform.bind(sourceField.type()).apply(sourceValue);
+
+        TransformSpec transformSpec = TransformSpec.fromString(transform.toString().toUpperCase(), sourceField.name());
+        UnboundTerm<Object> term = SchemaUtils.toTerm(transformSpec);
+
+        columnPredicate = Expressions.or(
+            columnPredicate, Expressions.equal(term, transformedValue));
+      }
+
+      predicate = Expressions.and(predicate, columnPredicate);
     }
 
     return predicate;
   }
 
   public static List<PartitionField> getPartitionFields(Table table, boolean latestSpecOnly) {
-    return latestSpecOnly ? table.spec().fields() :
-      table.specs().values().stream()
-        .flatMap(spec -> spec.fields().stream()
-            .filter(f -> !f.transform().isVoid()))
-        .distinct()
-        .collect(Collectors.toList());
+    if (latestSpecOnly) {
+      return table.spec().fields();
+    }
+    return table.specs().values().stream()
+        .flatMap(spec -> spec.fields().stream())
+        .filter(f -> !f.transform().isVoid())
+        .toList();
   }
 
   public static Partition getPartition(Configuration conf,
@@ -499,6 +510,7 @@ public static Partition getPartition(Configuration conf,
     }
 
     try {
+      // Use the actual partition path (with transforms) as the partition name
       String partitionName = partitionNames.getFirst();
       if (partitionSpec.size() != Warehouse.makeSpecFromName(partitionName).size()) {
         return null;

ql/src/java/org/apache/hadoop/hive/ql/ddl/table/info/desc/formatter/TextDescTableFormatter.java

@@ -174,8 +174,8 @@ private void addPartitionData(DataOutputStream out, HiveConf conf, String column
       List<FieldSchema> partitionColumns = null;
       if (table.isPartitioned()) {
         partitionColumns = table.hasNonNativePartitionSupport() ?
-                table.getStorageHandler().getPartitionKeys(table) :
-                table.getPartCols();
+            table.getStorageHandler().getPartitionKeys(table) :
+            table.getPartCols();
       }
       if (CollectionUtils.isNotEmpty(partitionColumns) &&
           conf.getBoolVar(ConfVars.HIVE_DISPLAY_PARTITION_COLUMNS_SEPARATELY)) {

ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java

@@ -84,9 +84,9 @@ public List<String> getValues() {
     Table table = this.getTable();
     List<String> values = new ArrayList<String>();
     for (FieldSchema fs :
-            table.hasNonNativePartitionSupport() ?
-                    table.getStorageHandler().getPartitionKeys(table) :
-                    table.getPartCols()) {
+        table.hasNonNativePartitionSupport() ?
+            table.getStorageHandler().getPartitionKeys(table) :
+            table.getPartCols()) {
       values.add(partSpec.get(fs.getName()));
     }
     return values;

ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java

@@ -1348,14 +1348,7 @@ public TableSpec(Hive db, HiveConf conf, ASTNode ast, boolean allowDynamicPartit
               if (partHandle == null) {
                 // if partSpec doesn't exists in DB, return a delegate one
                 // and the actual partition is created in MoveTask
-                try {
-                  partHandle = tableHandle.hasNonNativePartitionSupport() ?
-                          new DummyPartition(tableHandle, Warehouse.makePartName(partSpec, false), partSpec) :
-                          new Partition(tableHandle, partSpec, null);
-                } catch (MetaException e) {
-                  throw new SemanticException("Unable to construct name for dummy partition due to: ", e);
-                }
-
+                partHandle = createDummyPartitionHandle(tableHandle, partSpec);
               } else {
                 partitions.add(partHandle);
               }
@@ -1382,6 +1375,28 @@ public TableSpec(Hive db, HiveConf conf, ASTNode ast, boolean allowDynamicPartit
       }
     }
 
+    /**
+     * Creates a dummy partition handle for the given table and partition spec.
+     * The partition doesn't exist in DB yet - it will be created in MoveTask.
+     *
+     * @param table the table
+     * @param partSpec partition specification
+     * @return dummy partition handle
+     * @throws SemanticException if unable to construct partition name
+     */
+    private Partition createDummyPartitionHandle(Table table, Map<String, String> partSpec)
+        throws HiveException {
+      try {
+        if (table.hasNonNativePartitionSupport()) {
+          return new DummyPartition(table, Warehouse.makePartName(partSpec, false), partSpec);
+        } else {
+          return new Partition(table, partSpec, null);
+        }
+      } catch (MetaException e) {
+        throw new SemanticException("Unable to construct name for dummy partition due to: ", e);
+      }
+    }
+
     private boolean checkUseNativeApi(HiveConf conf, ASTNode ast) {
       boolean isLoad = ast.getParent() != null && ast.getParent().getType() == HiveParser.TOK_LOAD;
       return isLoad && tableHandle.isNonNative() && conf.getBoolVar(HIVE_LOAD_DATA_USE_NATIVE_API);