HIVE-29458: Iceberg: Sort expressions should not be added for distribution. Partition transforms are added for clustering (distribution and sorting)

Author: kokila-19

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

@@ -890,9 +890,9 @@ public Map<Integer, List<TransformSpec>> getPartitionTransformSpecs(
           Pair::first, Collectors.mapping(Pair::second, Collectors.toList())));
   }
 
-  private List<TransformSpec> getSortTransformSpec(Table table) {
-    return table.sortOrder().fields().stream().map(s ->
-            IcebergTableUtil.getTransformSpec(table, s.transform().toString(), s.sourceId()))
+  private List<TransformSpec> getWriteSortTransformSpecs(Table table) {
+    return table.sortOrder().fields().stream()
+        .map(s -> IcebergTableUtil.getTransformSpec(table, s.transform().toString(), s.sourceId()))
         .toList();
   }
 
@@ -913,11 +913,16 @@ public DynamicPartitionCtx createDPContext(
         hiveConf.getVar(ConfVars.DEFAULT_PARTITION_NAME),
         hiveConf.getIntVar(ConfVars.DYNAMIC_PARTITION_MAX_PARTS_PER_NODE));
 
+    // Add Iceberg partition transforms as custom partition expressions.
+    // These are used for both distribution (deciding reducers) and sorting within reducers.
     if (table.spec().isPartitioned() &&
-          hiveConf.getIntVar(ConfVars.HIVE_OPT_SORT_DYNAMIC_PARTITION_THRESHOLD) >= 0) {
-      addCustomSortExpr(table, hmsTable, writeOperation, dpCtx, getPartitionTransformSpec(hmsTable));
+        hiveConf.getIntVar(ConfVars.HIVE_OPT_SORT_DYNAMIC_PARTITION_THRESHOLD) >= 0) {
+      addCustomPartitionTransformExpressions(table, hmsTable, writeOperation, dpCtx,
+          getPartitionTransformSpec(hmsTable));
     }
 
+    // Add write sort order expressions as custom sort expressions.
+    // These are used ONLY for sorting within reducers, NOT for distribution.
     SortOrder sortOrder = table.sortOrder();
     if (sortOrder.isSorted()) {
       List<Integer> customSortPositions = Lists.newLinkedList();
@@ -943,7 +948,8 @@ public DynamicPartitionCtx createDPContext(
         }
       }
 
-      addCustomSortExpr(table, hmsTable, writeOperation, dpCtx, getSortTransformSpec(table));
+      addCustomWriteSortExpressions(table, hmsTable, writeOperation, dpCtx,
+          getWriteSortTransformSpecs(table));
     }
 
     // Even if table has no explicit sort order, honor z-order if configured
@@ -999,21 +1005,43 @@ private void addZOrderCustomExpr(Map<String, String> props, DynamicPartitionCtx
     }
   }
 
-  private void addCustomSortExpr(Table table,  org.apache.hadoop.hive.ql.metadata.Table hmsTable,
-      Operation writeOperation, DynamicPartitionCtx dpCtx,
-      List<TransformSpec> transformSpecs) {
+  private void addCustomPartitionTransformExpressions(Table table,
+      org.apache.hadoop.hive.ql.metadata.Table hmsTable, Operation writeOperation,
+      DynamicPartitionCtx dpCtx, List<TransformSpec> transformSpecs) {
+    Map<String, Integer> fieldOrderMap = buildFieldOrderMap(table);
+    int offset = getWriteRowOffset(hmsTable, writeOperation);
+
+    dpCtx.addCustomPartitionExpressions(transformSpecs.stream()
+        .map(spec -> IcebergTransformSortFunctionUtil.getCustomTransformExpr(
+            spec, fieldOrderMap.get(spec.getColumnName()) + offset))
+        .collect(Collectors.toList()));
+  }
+
+  private void addCustomWriteSortExpressions(Table table,
+      org.apache.hadoop.hive.ql.metadata.Table hmsTable, Operation writeOperation,
+      DynamicPartitionCtx dpCtx, List<TransformSpec> transformSpecs) {
+    Map<String, Integer> fieldOrderMap = buildFieldOrderMap(table);
+    int offset = getWriteRowOffset(hmsTable, writeOperation);
+
+    dpCtx.addCustomSortExpressions(transformSpecs.stream()
+        .map(spec -> IcebergTransformSortFunctionUtil.getCustomTransformExpr(
+            spec, fieldOrderMap.get(spec.getColumnName()) + offset))
+        .collect(Collectors.toList()));
+  }
+
+  private Map<String, Integer> buildFieldOrderMap(Table table) {
     List<Types.NestedField> fields = table.schema().columns();
     Map<String, Integer> fieldOrderMap = Maps.newHashMapWithExpectedSize(fields.size());
     for (int i = 0; i < fields.size(); ++i) {
       fieldOrderMap.put(fields.get(i).name(), i);
     }
+    return fieldOrderMap;
+  }
 
-    int offset = (shouldOverwrite(hmsTable, writeOperation) ?
-        ACID_VIRTUAL_COLS_AS_FIELD_SCHEMA : acidSelectColumns(hmsTable, writeOperation)).size();
-
-    dpCtx.addCustomSortExpressions(transformSpecs.stream().map(spec ->
-        IcebergTransformSortFunctionUtil.getCustomSortExprs(spec, fieldOrderMap.get(spec.getColumnName()) + offset)
-    ).collect(Collectors.toList()));
+  private int getWriteRowOffset(org.apache.hadoop.hive.ql.metadata.Table hmsTable, Operation writeOperation) {
+    return (shouldOverwrite(hmsTable, writeOperation) ?
+        ACID_VIRTUAL_COLS_AS_FIELD_SCHEMA :
+        acidSelectColumns(hmsTable, writeOperation)).size();
   }
 
   @Override

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

@@ -37,7 +37,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
 /**
- * A utility class which provides Iceberg transform sort functions.
+ * Utility for building Iceberg transform expressions used by both partitioning and sorting.
  */
 public final class IcebergTransformSortFunctionUtil {
 
@@ -136,7 +136,7 @@ private IcebergTransformSortFunctionUtil() {
             }
           };
 
-  public static Function<List<ExprNodeDesc>, ExprNodeDesc> getCustomSortExprs(TransformSpec spec, int index) {
+  public static Function<List<ExprNodeDesc>, ExprNodeDesc> getCustomTransformExpr(TransformSpec spec, int index) {
     switch (spec.getTransformType()) {
       case BUCKET:
         return BUCKET_SORT_EXPR.apply(index, spec.getTransformParam());

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

@@ -109,3 +109,28 @@ INSERT INTO default.zorder_props VALUES (3, 'B'),(1, 'A'),(7, 'C'),(2, 'A'),(9,
 DESCRIBE FORMATTED default.zorder_props;
 SELECT * FROM default.zorder_props;
 DROP TABLE default.zorder_props;
+
+CREATE TABLE default.zorder_tsdl_test (
+    ts timestamp,
+    dd double,
+    ll int)
+PARTITIONED BY SPEC (bucket(4, ll))
+WRITE ORDERED BY zorder (ts, dd)
+STORED BY iceberg
+STORED As orc;
+
+explain insert into default.zorder_tsdl_test  values  (TIMESTAMP '2022-01-01 00:00:00',  0.0,       0);
+
+drop table default.zorder_tsdl_test;
+
+
+CREATE TABLE default.zorder_tsdl_test1 (
+    ts timestamp,
+    dd double)
+PARTITIONED BY (ll int)
+WRITE ORDERED BY zorder (ts, dd)
+STORED BY iceberg
+STORED As orc;
+
+explain insert into default.zorder_tsdl_test1  values  (TIMESTAMP '2022-01-01 00:00:00',  0.0,       0);
+drop table default.zorder_tsdl_test1;