fix

Author: deniskuzZ

ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionOptimizer.java

@@ -215,6 +215,12 @@ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
       if (customSortExprs.isEmpty() && !shouldDo(partitionPositions, customPartitionExprs, fsParent, allRSCols)) {
         return null;
       }
+
+      // Mark that sorting will be applied with custom partition expressions, so the writer layer
+      // (e.g. Iceberg) knows the input is ordered and can use a clustered writer.
+      if (!customPartitionExprs.isEmpty()) {
+        dpCtx.setHasCustomPartitionOrSortExpression(true);
+      }
       // if RS is inserted by enforce bucketing or sorting, we need to remove it
       // since ReduceSinkDeDuplication will not merge them to single RS.
       // RS inserted by enforce bucketing/sorting will have bucketing column in

ql/src/java/org/apache/hadoop/hive/ql/plan/DynamicPartitionCtx.java

@@ -252,7 +252,6 @@ public List<Function<List<ExprNodeDesc>, ExprNodeDesc>> getCustomPartitionExpres
   public void addCustomPartitionExpressions(
           List<Function<List<ExprNodeDesc>, ExprNodeDesc>> customPartitionExpressions) {
     if (!org.apache.commons.collections.CollectionUtils.isEmpty(customPartitionExpressions)) {
-      this.hasCustomPartitionOrSortExpr = true;
       this.customPartitionExpressions.addAll(customPartitionExpressions);
     }
   }
@@ -290,4 +289,8 @@ public void setCustomSortNullOrder(List<Integer> customSortNullOrder) {
   public boolean hasCustomPartitionOrSortExpression() {
     return hasCustomPartitionOrSortExpr;
   }
+
+  public void setHasCustomPartitionOrSortExpression(boolean hasCustomPartitionOrSortExpr) {
+    this.hasCustomPartitionOrSortExpr = hasCustomPartitionOrSortExpr;
+  }
 }