@Override
    public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
        Object... nodeOutputs) throws SemanticException {

      // introduce RS and EX before FS. If the operator tree already contains
      // RS then ReduceSinkDeDuplication optimization should merge them
      FileSinkOperator fsOp = (FileSinkOperator) nd;

      LOG.info("Sorted dynamic partitioning optimization kicked in..");

      // if not dynamic partitioning then bail out
      if (fsOp.getConf().getDynPartCtx() == null) {
        LOG.debug("Bailing out of sort dynamic partition optimization as dynamic partitioning context is null");
        return null;
      }

      // if list bucketing then bail out
      ListBucketingCtx lbCtx = fsOp.getConf().getLbCtx();
      if (lbCtx != null && !lbCtx.getSkewedColNames().isEmpty()
          && !lbCtx.getSkewedColValues().isEmpty()) {
        LOG.debug("Bailing out of sort dynamic partition optimization as list bucketing is enabled");
        return null;
      }

      Table destTable = fsOp.getConf().getTable();
      if (destTable == null) {
        LOG.debug("Bailing out of sort dynamic partition optimization as destination table is null");
        return null;
      }

      if (destTable.isMaterializedView() &&
          (destTable.getProperty(Constants.MATERIALIZED_VIEW_SORT_COLUMNS) != null ||
              destTable.getProperty(Constants.MATERIALIZED_VIEW_DISTRIBUTE_COLUMNS) != null)) {
        LOG.debug("Bailing out of sort dynamic partition optimization as destination is a materialized view"
            + "with CLUSTER/SORT/DISTRIBUTE spec");
        return null;
      }

      // unlink connection between FS and its parent
      Operator<? extends OperatorDesc> fsParent = fsOp.getParentOperators().get(0);
      DynamicPartitionCtx dpCtx = fsOp.getConf().getDynPartCtx();

      ArrayList<ColumnInfo> parentCols = Lists.newArrayList(fsParent.getSchema().getSignature());
      ArrayList<ExprNodeDesc> allRSCols = Lists.newArrayList();
      for (ColumnInfo ci : parentCols) {
        allRSCols.add(new ExprNodeColumnDesc(ci));
      }

      // if all dp columns / custom sort expressions got constant folded then disable this optimization
      if (allStaticPartitions(fsParent, allRSCols, dpCtx)) {
        LOG.debug("Bailing out of sorted dynamic partition optimizer as all dynamic partition" +
            " columns got constant folded (static partitioning)");
        return null;
      }

      List<Integer> partitionPositions = getPartitionPositions(dpCtx, fsParent.getSchema());
      LinkedList<Function<List<ExprNodeDesc>, ExprNodeDesc>> customSortExprs =
          new LinkedList<>(dpCtx.getCustomSortExpressions());
      LinkedList<Integer> customSortOrder = new LinkedList<>(dpCtx.getCustomSortOrder());
      LinkedList<Integer> customNullOrder = new LinkedList<>(dpCtx.getCustomSortNullOrder());

      // If custom sort expressions are present, there is an explicit requirement to do sorting
      if (customSortExprs.isEmpty() && !shouldDo(partitionPositions, fsParent)) {
        return null;
      }
      // 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
      // reduce sink key whereas RS inserted by this optimization will have
      // partition columns followed by bucket number followed by sort columns in
      // the reduce sink key. Since both key columns are not prefix subset
      // ReduceSinkDeDuplication will not merge them together resulting in 2 MR jobs.
      // To avoid that we will remove the RS (and EX) inserted by enforce bucketing/sorting.
      if (!removeRSInsertedByEnforceBucketing(fsOp)) {
        LOG.debug("Bailing out of sort dynamic partition optimization as some partition columns " +
            "got constant folded.");
        return null;
      }

      // unlink connection between FS and its parent
      fsParent = fsOp.getParentOperators().get(0);
      // store the index of the file sink operator to later insert the modified operator with RS at the same position
      int fsOpIndex = fsParent.getChildOperators().indexOf(fsOp);
      fsParent.getChildOperators().remove(fsOp);

      // if enforce bucketing/sorting is disabled numBuckets will not be set.
      // set the number of buckets here to ensure creation of empty buckets
      int numBuckets = destTable.getNumBuckets();
      dpCtx.setNumBuckets(numBuckets);

      // Get the positions for partition, bucket and sort columns
      List<Integer> bucketPositions = getBucketPositions(destTable.getBucketCols(),
          destTable.getCols());
      List<Integer> sortPositions = null;
      List<Integer> sortOrder = null;
      ArrayList<ExprNodeDesc> bucketColumns = null;
      if (fsOp.getConf().getWriteType() == AcidUtils.Operation.UPDATE ||
          fsOp.getConf().getWriteType() == AcidUtils.Operation.DELETE) {
        // When doing updates and deletes we always want to sort on the rowid because the ACID
        // reader will expect this sort order when doing reads.  So
        // ignore whatever comes from the table and enforce this sort order instead.
        sortPositions = Collections.singletonList(0);
        sortOrder = Collections.singletonList(1); // 1 means asc, could really use enum here in the thrift if
        /**
         * ROW__ID is always the 1st column of Insert representing Update/Delete operation
         * (set up in {@link org.apache.hadoop.hive.ql.parse.UpdateDeleteSemanticAnalyzer})
         * and we wrap it in UDFToInteger
         * (in {@link org.apache.hadoop.hive.ql.parse.SemanticAnalyzer#getPartitionColsFromBucketColsForUpdateDelete(Operator, boolean)})
         * which extracts bucketId from it
         * see {@link org.apache.hadoop.hive.ql.udf.UDFToInteger#evaluate(RecordIdentifier)}*/
        ColumnInfo ci = fsParent.getSchema().getSignature().get(0);
        if (!VirtualColumn.ROWID.getTypeInfo().equals(ci.getType())) {
          throw new IllegalStateException("expected 1st column to be ROW__ID but got wrong type: " + ci.toString());
        }

        if (numBuckets > 0) {
          bucketColumns = new ArrayList<>();
          //add a cast(ROW__ID as int) to wrap in UDFToInteger()
          bucketColumns.add(ExprNodeTypeCheck.getExprNodeDefaultExprProcessor()
              .createConversionCast(new ExprNodeColumnDesc(ci), TypeInfoFactory.intTypeInfo));
        }
      } else {
        if (!destTable.getSortCols().isEmpty()) {
          // Sort columns specified by table
          sortPositions = getSortPositions(destTable.getSortCols(), destTable.getCols());
          sortOrder = getSortOrders(destTable.getSortCols(), destTable.getCols());
        } else if (HiveConf.getBoolVar(this.parseCtx.getConf(), HiveConf.ConfVars.HIVE_SORT_WHEN_BUCKETING) &&
            !bucketPositions.isEmpty()) {
          // We use clustered columns as sort columns
          sortPositions = new ArrayList<>(bucketPositions);
          sortOrder = sortPositions.stream().map(e -> 1).collect(Collectors.toList());
        } else {
          // Infer sort columns from operator tree
          sortPositions = Lists.newArrayList();
          sortOrder = Lists.newArrayList();
          inferSortPositions(fsParent, sortPositions, sortOrder);
        }
        List<ColumnInfo> colInfos = fsParent.getSchema().getSignature();
        bucketColumns = getPositionsToExprNodes(bucketPositions, colInfos);
      }
      List<Integer> sortNullOrder = new ArrayList<>();
      for (int order : sortOrder) {
        sortNullOrder.add(NullOrdering.defaultNullOrder(order, parseCtx.getConf()).getCode());
      }
      LOG.debug("Got sort order");
      for (int i : sortPositions) {
        LOG.debug("sort position " + i);
      }
      for (int i : sortOrder) {
        LOG.debug("sort order " + i);
      }
      for (int i : sortNullOrder) {
        LOG.debug("sort null order " + i);
      }

      // update file sink descriptor
      fsOp.getConf().setMultiFileSpray(false);
      fsOp.getConf().setNumFiles(1);
      fsOp.getConf().setTotalFiles(1);

      // Create ReduceSink operator
      ReduceSinkOperator rsOp = getReduceSinkOp(partitionPositions, sortPositions, sortOrder,
          sortNullOrder, customSortExprs, customSortOrder, customNullOrder, allRSCols, bucketColumns, numBuckets,
          fsParent, fsOp.getConf().getWriteType());
      // we have to make sure not to reorder the child operators as it might cause weird behavior in the tasks at
      // the same level. when there is auto stats gather at the same level as another operation then it might
      // cause unnecessary preemption. Maintaining the order here to avoid such preemption and possible errors
      // Ref TEZ-3296
      fsParent.getChildOperators().remove(rsOp);
      fsParent.getChildOperators().add(fsOpIndex, rsOp);
      rsOp.getConf().setBucketingVersion(fsOp.getConf().getBucketingVersion());

      List<ExprNodeDesc> descs = new ArrayList<ExprNodeDesc>(allRSCols.size());
      List<String> colNames = new ArrayList<String>();
      String colName;
      final List<ColumnInfo> fileSinkSchema = fsOp.getSchema().getSignature();
      for (int i = 0; i < allRSCols.size(); i++) {
        ExprNodeDesc col = allRSCols.get(i);
        ExprNodeDesc newColumnExpr = null;
        colName = col.getExprString();
        colNames.add(colName);
        if (partitionPositions.contains(i) || sortPositions.contains(i)) {
          newColumnExpr = (new ExprNodeColumnDesc(col.getTypeInfo(), ReduceField.KEY.toString()+"."+colName, null, false));
        } else {
          newColumnExpr = (new ExprNodeColumnDesc(col.getTypeInfo(), ReduceField.VALUE.toString()+"."+colName, null, false));
        }

        // make sure column type matches with expected types in FS op
        if(i < fileSinkSchema.size()) {
          final ColumnInfo fsColInfo = fileSinkSchema.get(i);
          if (!newColumnExpr.getTypeInfo().equals(fsColInfo.getType())) {
            newColumnExpr = ExprNodeTypeCheck.getExprNodeDefaultExprProcessor()
                .createConversionCast(newColumnExpr, (PrimitiveTypeInfo) fsColInfo.getType());
          }
        }
        descs.add(newColumnExpr);
      }
      RowSchema selRS = new RowSchema(fsParent.getSchema());

      if (bucketColumns != null && !bucketColumns.isEmpty()) {
        customSortExprs.add(BUCKET_SORT_EXPRESSION);
      }

      for (Function<List<ExprNodeDesc>, ExprNodeDesc> customSortExpr : customSortExprs) {
        ExprNodeDesc colExpr = customSortExpr.apply(allRSCols);
        String customSortColName = colExpr.getExprString();
        TypeInfo customSortColTypeInfo = colExpr.getTypeInfo();

        descs.add(new ExprNodeColumnDesc(customSortColTypeInfo, ReduceField.KEY + "." + customSortColName,
            null, false));
        colNames.add(customSortColName);
        ColumnInfo ci = new ColumnInfo(
            customSortColName, customSortColTypeInfo, selRS.getSignature().get(0).getTabAlias(), true, true);
        selRS.getSignature().add(ci);
        rsOp.getSchema().getSignature().add(ci);
      }

      // Create SelectDesc
      SelectDesc selConf = new SelectDesc(descs, colNames);

      // Create Select Operator
      SelectOperator selOp = (SelectOperator) OperatorFactory.getAndMakeChild(
          selConf, selRS, rsOp);

      // link SEL to FS
      fsOp.getParentOperators().clear();
      fsOp.getParentOperators().add(selOp);
      selOp.getChildOperators().add(fsOp);

      // Set if partition sorted or partition bucket sorted
      fsOp.getConf().setDpSortState(FileSinkDesc.DPSortState.PARTITION_SORTED);
      if (bucketColumns!=null && !bucketColumns.isEmpty()) {
        fsOp.getConf().setDpSortState(FileSinkDesc.DPSortState.PARTITION_BUCKET_SORTED);
      }

      // update partition column info in FS descriptor
      fsOp.getConf().setPartitionCols(rsOp.getConf().getPartitionCols());

      rsOp.setStatistics(rsOp.getParentOperators().get(0).getStatistics());

      LOG.info("Inserted " + rsOp.getOperatorId() + " and " + selOp.getOperatorId()
          + " as parent of " + fsOp.getOperatorId() + " and child of " + fsParent.getOperatorId());

      parseCtx.setReduceSinkAddedBySortedDynPartition(true);
      return null;
    }
When dynamic partitioning (with or without bucketing and sorting) is enabled, this optimization
sorts the records on partition, bucket and sort columns respectively before inserting records
into the destination table. This enables reducers to keep only one record writer all the time
thereby reducing the the memory pressure on the reducers.
Sorting is based on the Dynamic Partitioning context that is already created in the file sink operator.
If that contains instructions for custom expression sorting, then this optimizer will disregard any partitioning or
bucketing information of the Hive (table format) table, and will arrange the plan solely as per the custom exprs.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionOptimizer.java