/*
 * 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.hadoop.hive.ql.optimizer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Stack;

import java.util.stream.Collectors;
import org.apache.hadoop.hive.conf.Constants;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.ql.exec.ColumnInfo;
import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
import org.apache.hadoop.hive.ql.exec.MemoryInfo;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorFactory;
import org.apache.hadoop.hive.ql.exec.OperatorUtils;
import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.exec.RowSchema;
import org.apache.hadoop.hive.ql.exec.SelectOperator;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.Utilities.ReduceField;
import org.apache.hadoop.hive.ql.io.AcidUtils;
import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
import org.apache.hadoop.hive.ql.lib.SemanticDispatcher;
import org.apache.hadoop.hive.ql.lib.SemanticGraphWalker;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.lib.SemanticNodeProcessor;
import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
import org.apache.hadoop.hive.ql.lib.SemanticRule;
import org.apache.hadoop.hive.ql.lib.RuleRegExp;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
import org.apache.hadoop.hive.ql.parse.ParseContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.parse.type.*;
import org.apache.hadoop.hive.ql.plan.ColStatistics;
import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDescUtils;
import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.ListBucketingCtx;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
import org.apache.hadoop.hive.ql.plan.PlanUtils;
import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc;
import org.apache.hadoop.hive.ql.plan.SelectDesc;
import org.apache.hadoop.hive.ql.plan.Statistics;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.orc.OrcConf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

/**
 * 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.
 */
public class SortedDynPartitionOptimizer extends Transform {

  private static final String BUCKET_NUMBER_COL_NAME = "_bucket_number";

  @Override
  public ParseContext transform(ParseContext pCtx) throws SemanticException {

    // create a walker which walks the tree in a DFS manner while maintaining the
    // operator stack. The dispatcher generates the plan from the operator tree
    Map<SemanticRule, SemanticNodeProcessor> opRules = new LinkedHashMap<SemanticRule, SemanticNodeProcessor>();

    String FS = FileSinkOperator.getOperatorName() + "%";

    opRules.put(new RuleRegExp("Sorted Dynamic Partition", FS), getSortDynPartProc(pCtx));

    SemanticDispatcher disp = new DefaultRuleDispatcher(null, opRules, null);
    SemanticGraphWalker ogw = new DefaultGraphWalker(disp);

    ArrayList<Node> topNodes = new ArrayList<Node>();
    topNodes.addAll(pCtx.getTopOps().values());
    ogw.startWalking(topNodes, null);

    return pCtx;
  }

  private SemanticNodeProcessor getSortDynPartProc(ParseContext pCtx) {
    return new SortedDynamicPartitionProc(pCtx);
  }

  class SortedDynamicPartitionProc implements SemanticNodeProcessor {

    private final Logger LOG = LoggerFactory.getLogger(SortedDynPartitionOptimizer.class);
    protected ParseContext parseCtx;

    public SortedDynamicPartitionProc(ParseContext pCtx) {
      this.parseCtx = pCtx;
    }

    @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);
      // if all dp columns got constant folded then disable this optimization
      if (allStaticPartitions(fsParent, fsOp.getConf().getDynPartCtx())) {
        LOG.debug("Bailing out of sorted dynamic partition optimizer as all dynamic partition" +
            " columns got constant folded (static partitioning)");
        return null;
      }

      DynamicPartitionCtx dpCtx = fsOp.getConf().getDynPartCtx();
      List<Integer> partitionPositions = getPartitionPositions(dpCtx, fsParent.getSchema());

      if (!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<Integer>();
      for (int order : sortOrder) {
        sortNullOrder.add(order == 1 ? 0 : 1); // for asc, nulls first; for desc, nulls last
      }
      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);

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

      // Create ReduceSink operator
      ReduceSinkOperator rsOp = getReduceSinkOp(partitionPositions, sortPositions, sortOrder, sortNullOrder,
          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()) {
        descs.add(new ExprNodeColumnDesc(TypeInfoFactory.stringTypeInfo,
            ReduceField.KEY.toString()+"."+BUCKET_NUMBER_COL_NAME, null, false));
        colNames.add(BUCKET_NUMBER_COL_NAME);
        ColumnInfo ci = new ColumnInfo(BUCKET_NUMBER_COL_NAME, TypeInfoFactory.stringTypeInfo,
            selRS.getSignature().get(0).getTabAlias(), true, true);
        selRS.getSignature().add(ci);
        fsParent.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;
    }

    private boolean allStaticPartitions(Operator<? extends OperatorDesc> op,
        final DynamicPartitionCtx dynPartCtx) {
      int numDpCols = dynPartCtx.getNumDPCols();
      int numCols = op.getSchema().getColumnNames().size();
      List<String> dpCols = op.getSchema().getColumnNames().subList(numCols - numDpCols, numCols);
      if (op.getColumnExprMap() == null) {
        // find first operator upstream with valid (non-null) column expression map
        for(Operator<? extends OperatorDesc> parent : op.getParentOperators()) {
          if (parent.getColumnExprMap() != null) {
            op = parent;
            break;
          }
        }
      }
      if (op.getColumnExprMap() != null) {
        for(String dpCol : dpCols) {
          ExprNodeDesc end = ExprNodeDescUtils.findConstantExprOrigin(dpCol, op);
          if (!(end instanceof ExprNodeConstantDesc)) {
            return false;
          }
        }
      } else {
        return false;
      }
      return true;
    }

    // Remove RS and SEL introduced by enforce bucketing/sorting config
    // Convert PARENT -> RS -> SEL -> FS to PARENT -> FS
    private boolean removeRSInsertedByEnforceBucketing(FileSinkOperator fsOp) {

      Set<ReduceSinkOperator> reduceSinks = OperatorUtils.findOperatorsUpstream(fsOp,
          ReduceSinkOperator.class);
      Operator<? extends OperatorDesc> rsToRemove = null;
      List<ReduceSinkOperator> rsOps = parseCtx
          .getReduceSinkOperatorsAddedByEnforceBucketingSorting();
      boolean found = false;

      // iterate through all RS and locate the one introduce by enforce bucketing
      for (ReduceSinkOperator reduceSink : reduceSinks) {
        for (ReduceSinkOperator rsOp : rsOps) {
          if (reduceSink.equals(rsOp)) {
            rsToRemove = reduceSink;
            found = true;
            break;
          }
        }

        if (found) {
          break;
        }
      }

      // iF RS is found remove it and its child (EX) and connect its parent
      // and grand child
      if (found) {
        Operator<? extends OperatorDesc> rsParent = rsToRemove.getParentOperators().get(0);

        // RS is expected to have exactly ONE child
        assert(rsToRemove.getChildOperators().size() == 1);

        Operator<? extends OperatorDesc> rsChildToRemove = rsToRemove.getChildOperators().get(0);

        if (!(rsChildToRemove instanceof SelectOperator) || rsParent.getSchema().getSignature().size()
            != rsChildToRemove.getSchema().getSignature().size()) {
          // if schema size cannot be matched, then it could be because of constant folding
          // converting partition column expression to constant expression. The constant
          // expression will then get pruned by column pruner since it will not reference to
          // any columns.
          return false;
        }

        // if child is select and contains expression which isn't column it shouldn't
        // be removed because otherwise we will end up with different types/schema later
        // while introducing select for RS
        for(ExprNodeDesc expr: rsChildToRemove.getColumnExprMap().values()){
          if(!(expr instanceof ExprNodeColumnDesc)){
            return false;
          }
        }

        List<Operator<? extends OperatorDesc>> rsGrandChildren = rsChildToRemove.getChildOperators();

        rsParent.getChildOperators().remove(rsToRemove);
        rsParent.getChildOperators().addAll(rsGrandChildren);


        // fix grandchildren
        for (Operator<? extends OperatorDesc> rsGrandChild: rsGrandChildren) {
          rsGrandChild.getParentOperators().clear();
          rsGrandChild.getParentOperators().add(rsParent);
        }
        LOG.info("Removed " + rsToRemove.getOperatorId() + " and " + rsChildToRemove.getOperatorId()
            + " as it was introduced by enforce bucketing/sorting.");
      }
      return true;
    }

    private List<Integer> getPartitionPositions(DynamicPartitionCtx dpCtx, RowSchema schema) {
      int numPartCols = dpCtx.getNumDPCols();
      int numCols = schema.getSignature().size();
      List<Integer> partPos = Lists.newArrayList();

      // partition columns will always at the last
      for (int i = numCols - numPartCols; i < numCols; i++) {
        partPos.add(i);
      }
      return partPos;
    }

    // Get the bucket positions for the table
    private List<Integer> getBucketPositions(List<String> tabBucketCols, List<FieldSchema> tabCols) {
      List<Integer> posns = new ArrayList<Integer>();
      for (String bucketCol : tabBucketCols) {
        int pos = 0;
        for (FieldSchema tabCol : tabCols) {
          if (bucketCol.equals(tabCol.getName())) {
            posns.add(pos);
            break;
          }
          pos++;
        }
      }
      return posns;
    }

    // Try to infer possible sort columns in the query
    // i.e. the sequence must be pRS-SEL*-fsParent
    // Returns true if columns could be inferred, false otherwise
    private void inferSortPositions(Operator<? extends OperatorDesc> fsParent,
        List<Integer> sortPositions, List<Integer> sortOrder) throws SemanticException {
      // If it is not a SEL operator, we bail out
      if (!(fsParent instanceof SelectOperator)) {
        return;
      }
      SelectOperator pSel = (SelectOperator) fsParent;
      Operator<? extends OperatorDesc> parent = pSel;
      while (!(parent instanceof ReduceSinkOperator)) {
        if (parent.getNumParent() != 1 ||
            !(parent instanceof SelectOperator)) {
          return;
        }
        parent = parent.getParentOperators().get(0);
      }
      // Backtrack SEL columns to pRS
      List<ExprNodeDesc> selColsInPRS =
          ExprNodeDescUtils.backtrack(pSel.getConf().getColList(), pSel, parent);
      ReduceSinkOperator pRS = (ReduceSinkOperator) parent;
      for (int i = 0; i < pRS.getConf().getKeyCols().size(); i++) {
        ExprNodeDesc col = pRS.getConf().getKeyCols().get(i);
        int pos = selColsInPRS.indexOf(col);
        if (pos == -1) {
          sortPositions.clear();
          sortOrder.clear();
          return;
        }
        sortPositions.add(pos);
        sortOrder.add(pRS.getConf().getOrder().charAt(i) == '+' ? 1 : 0); // 1 asc, 0 desc
      }
    }

    public ReduceSinkOperator getReduceSinkOp(List<Integer> partitionPositions,
        List<Integer> sortPositions, List<Integer> sortOrder, List<Integer> sortNullOrder,
        ArrayList<ExprNodeDesc> allCols, ArrayList<ExprNodeDesc> bucketColumns, int numBuckets,
        Operator<? extends OperatorDesc> parent, AcidUtils.Operation writeType) throws SemanticException {

      // Order of KEY columns
      // 1) Partition columns
      // 2) Bucket number column
      // 3) Sort columns
      Set<Integer> keyColsPosInVal = Sets.newLinkedHashSet();
      ArrayList<ExprNodeDesc> keyCols = Lists.newArrayList();
      List<Integer> newSortOrder = Lists.newArrayList();
      List<Integer> newSortNullOrder = Lists.newArrayList();

      keyColsPosInVal.addAll(partitionPositions);
      if (bucketColumns != null && !bucketColumns.isEmpty()) {
        keyColsPosInVal.add(-1);
      }
      keyColsPosInVal.addAll(sortPositions);

      // by default partition and bucket columns are sorted in ascending order
      Integer order = 1;
      if (sortOrder != null && !sortOrder.isEmpty()) {
        if (sortOrder.get(0) == 0) {
          order = 0;
        }
      }
      for (int i = 0; i < keyColsPosInVal.size(); i++) {
        newSortOrder.add(order);
      }

      String orderStr = "";
      for (Integer i : newSortOrder) {
        if (i == 1) {
          orderStr += "+";
        } else {
          orderStr += "-";
        }
      }

      // if partition and bucket columns are sorted in ascending order, by default
      // nulls come first; otherwise nulls come last
      Integer nullOrder = order == 1 ? 0 : 1;
      if (sortNullOrder != null && !sortNullOrder.isEmpty()) {
        if (sortNullOrder.get(0) == 0) {
          nullOrder = 0;
        } else {
          nullOrder = 1;
        }
      }
      for (int i = 0; i < keyColsPosInVal.size(); i++) {
        newSortNullOrder.add(nullOrder);
      }

      String nullOrderStr = "";
      for (Integer i : newSortNullOrder) {
        if (i == 0) {
          nullOrderStr += "a";
        } else {
          nullOrderStr += "z";
        }
      }

      Map<String, ExprNodeDesc> colExprMap = Maps.newHashMap();
      ArrayList<ExprNodeDesc> partCols = Lists.newArrayList();

      // we will clone here as RS will update bucket column key with its
      // corresponding with bucket number and hence their OIs
      for (Integer idx : keyColsPosInVal) {
        if (idx < 0) {
          ExprNodeDesc bucketNumColUDF = ExprNodeGenericFuncDesc.newInstance(
              FunctionRegistry.getFunctionInfo("bucket_number").getGenericUDF(), new ArrayList<>());
          keyCols.add(bucketNumColUDF);
          colExprMap.put(Utilities.ReduceField.KEY + "." +BUCKET_NUMBER_COL_NAME, bucketNumColUDF);
        } else {
          keyCols.add(allCols.get(idx).clone());
        }
      }

      ArrayList<ExprNodeDesc> valCols = Lists.newArrayList();
      for (int i = 0; i < allCols.size(); i++) {
        if (!keyColsPosInVal.contains(i)) {
          valCols.add(allCols.get(i).clone());
        }
      }

      for (Integer idx : partitionPositions) {
        partCols.add(allCols.get(idx).clone());
      }

      // in the absence of SORTED BY clause, the sorted dynamic partition insert
      // should honor the ordering of records provided by ORDER BY in SELECT statement
      ReduceSinkOperator parentRSOp = OperatorUtils.findSingleOperatorUpstream(parent,
          ReduceSinkOperator.class);
      if (parentRSOp != null && parseCtx.getQueryProperties().hasOuterOrderBy()) {
        String parentRSOpOrder = parentRSOp.getConf().getOrder();
        String parentRSOpNullOrder = parentRSOp.getConf().getNullOrder();
        if (parentRSOpOrder != null && !parentRSOpOrder.isEmpty() && sortPositions.isEmpty()) {
          keyCols.addAll(parentRSOp.getConf().getKeyCols());
          orderStr += parentRSOpOrder;
          nullOrderStr += parentRSOpNullOrder;
        }
      }

      // map _col0 to KEY._col0, etc
      Map<String, String> nameMapping = new HashMap<>();
      ArrayList<String> keyColNames = Lists.newArrayList();
      for (ExprNodeDesc keyCol : keyCols) {
        String keyColName = keyCol.getExprString();
        keyColNames.add(keyColName);
        colExprMap.put(Utilities.ReduceField.KEY + "." +keyColName, keyCol);
        nameMapping.put(keyColName, Utilities.ReduceField.KEY + "." + keyColName);
      }
      ArrayList<String> valColNames = Lists.newArrayList();
      for (ExprNodeDesc valCol : valCols) {
        String colName = valCol.getExprString();
        valColNames.add(colName);
        colExprMap.put(Utilities.ReduceField.VALUE + "." + colName, valCol);
        nameMapping.put(colName, Utilities.ReduceField.VALUE + "." + colName);
      }

      // Create Key/Value TableDesc. When the operator plan is split into MR tasks,
      // the reduce operator will initialize Extract operator with information
      // from Key and Value TableDesc
      List<FieldSchema> fields = PlanUtils.getFieldSchemasFromColumnList(keyCols,
          keyColNames, 0, "");
      TableDesc keyTable = PlanUtils.getReduceKeyTableDesc(fields, orderStr, nullOrderStr);
      List<FieldSchema> valFields = PlanUtils.getFieldSchemasFromColumnList(valCols,
          valColNames, 0, "");
      TableDesc valueTable = PlanUtils.getReduceValueTableDesc(valFields);
      List<List<Integer>> distinctColumnIndices = Lists.newArrayList();

      // Number of reducers is set to default (-1)
      ReduceSinkDesc rsConf = new ReduceSinkDesc(keyCols, keyCols.size(), valCols,
          keyColNames, distinctColumnIndices, valColNames, -1, partCols, -1, keyTable,
          valueTable, writeType);
      rsConf.setBucketCols(bucketColumns);
      rsConf.setNumBuckets(numBuckets);

      ArrayList<ColumnInfo> signature = new ArrayList<>();
      for (int index = 0; index < parent.getSchema().getSignature().size(); index++) {
        ColumnInfo colInfo = new ColumnInfo(parent.getSchema().getSignature().get(index));
        colInfo.setInternalName(nameMapping.get(colInfo.getInternalName()));
        signature.add(colInfo);
      }
      ReduceSinkOperator op = (ReduceSinkOperator) OperatorFactory.getAndMakeChild(
          rsConf, new RowSchema(signature), parent);
      op.setColumnExprMap(colExprMap);
      return op;
    }

    /**
     * Get the sort positions for the sort columns.
     *
     * @param tabSortCols
     * @param tabCols
     * @return
     */
    private List<Integer> getSortPositions(List<Order> tabSortCols,
        List<FieldSchema> tabCols) {
      List<Integer> sortPositions = Lists.newArrayList();
      for (Order sortCol : tabSortCols) {
        int pos = 0;
        for (FieldSchema tabCol : tabCols) {
          if (sortCol.getCol().equals(tabCol.getName())) {
            sortPositions.add(pos);
            break;
          }
          pos++;
        }
      }
      return sortPositions;
    }

    /**
     * Get the sort order for the sort columns.
     *
     * @param tabSortCols
     * @param tabCols
     * @return
     */
    private List<Integer> getSortOrders(List<Order> tabSortCols,
        List<FieldSchema> tabCols) {
      List<Integer> sortOrders = Lists.newArrayList();
      for (Order sortCol : tabSortCols) {
        for (FieldSchema tabCol : tabCols) {
          if (sortCol.getCol().equals(tabCol.getName())) {
            sortOrders.add(sortCol.getOrder());
            break;
          }
        }
      }
      return sortOrders;
    }

    private ArrayList<ExprNodeDesc> getPositionsToExprNodes(List<Integer> pos,
        List<ColumnInfo> colInfos) {
      ArrayList<ExprNodeDesc> cols = Lists.newArrayList();

      for (Integer idx : pos) {
        ColumnInfo ci = colInfos.get(idx);
        ExprNodeColumnDesc encd = new ExprNodeColumnDesc(ci);
        cols.add(encd);
      }

      return cols;
    }

    // the idea is to estimate how many number of writers this insert can spun up.
    // Writers are proportional to number of partitions being inserted i.e cardinality of the partition columns
    //  if these writers are less than number of writers allowed within the memory pool (estimated) we go ahead with
    //  adding extra RS
    //  The way max number of writers allowed are computed based on
    //  (executor/container memory) * (percentage of memory taken by orc)
    //  and dividing that by max memory (stripe size) taken by a single writer.
    private boolean shouldDo(List<Integer> partitionPos, Operator<? extends OperatorDesc> fsParent) {

      int threshold = HiveConf.getIntVar(this.parseCtx.getConf(),
          HiveConf.ConfVars.HIVEOPTSORTDYNAMICPARTITIONTHRESHOLD);
      long MAX_WRITERS = -1;

      switch (threshold) {
      case -1:
        return false;
      case 0:
        break;
      case 1:
        return true;
      default:
        MAX_WRITERS = threshold;
        break;
      }

      Statistics tStats = fsParent.getStatistics();
      if (tStats == null) {
        return true;
      }

      List<ColStatistics> colStats = tStats.getColumnStats();
      if (colStats == null || colStats.isEmpty()) {
        return true;
      }
      long partCardinality = 1;

      // compute cardinality for partition columns
      for (Integer idx : partitionPos) {
        ColumnInfo ci = fsParent.getSchema().getSignature().get(idx);
        ColStatistics partStats = fsParent.getStatistics().getColumnStatisticsFromColName(ci.getInternalName());
        if (partStats == null) {
          // statistics for this partition are for some reason not available
          return true;
        }
        partCardinality = partCardinality * partStats.getCountDistint();
      }

      if (MAX_WRITERS < 0) {
        double orcMemPool = this.parseCtx.getConf().getDouble(OrcConf.MEMORY_POOL.getHiveConfName(),
            (Double) OrcConf.MEMORY_POOL.getDefaultValue());
        long orcStripSize = this.parseCtx.getConf().getLong(OrcConf.STRIPE_SIZE.getHiveConfName(),
            (Long) OrcConf.STRIPE_SIZE.getDefaultValue());
        MemoryInfo memoryInfo = new MemoryInfo(this.parseCtx.getConf());
        LOG.debug("Memory info during SDPO opt: {}", memoryInfo);
        long executorMem = memoryInfo.getMaxExecutorMemory();
        MAX_WRITERS = (long) (executorMem * orcMemPool) / orcStripSize;

      }
      if (partCardinality <= MAX_WRITERS) {
        return false;
      }
      return true;
    }
  }
}