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

      // introduce RS and EX before FS
      FileSinkOperator fsOp = (FileSinkOperator) nd;
      final String sh = fsOp.getConf().getTableInfo().getOutputFileFormatClassName();
      if (parseCtx.getQueryProperties().isQuery() || sh == null || !sh
              .equals(Constants.DRUID_HIVE_OUTPUT_FORMAT)) {
        // Bail out, nothing to do
        return null;
      }
      String segmentGranularity;
      final String targetShardsProperty;
      final Table table = fsOp.getConf().getTable();
      if (table != null) {
        // case the statement is an INSERT
        segmentGranularity = table.getParameters().get(Constants.DRUID_SEGMENT_GRANULARITY);
        targetShardsProperty =
            table.getParameters().getOrDefault(Constants.DRUID_TARGET_SHARDS_PER_GRANULARITY, "0");

      } else if (parseCtx.getCreateViewDesc() != null) {
        // case the statement is a CREATE MATERIALIZED VIEW AS
        segmentGranularity = parseCtx.getCreateViewDesc().getTblProps()
                .get(Constants.DRUID_SEGMENT_GRANULARITY);
        targetShardsProperty = parseCtx.getCreateViewDesc().getTblProps()
            .getOrDefault(Constants.DRUID_TARGET_SHARDS_PER_GRANULARITY, "0");
      } else if (parseCtx.getCreateTable() != null) {
        // case the statement is a CREATE TABLE AS
        segmentGranularity = parseCtx.getCreateTable().getTblProps()
                .get(Constants.DRUID_SEGMENT_GRANULARITY);
        targetShardsProperty = parseCtx.getCreateTable().getTblProps()
            .getOrDefault(Constants.DRUID_TARGET_SHARDS_PER_GRANULARITY, "0");
      } else {
        throw new SemanticException("Druid storage handler used but not an INSERT, "
                + "CMVAS or CTAS statement");
      }
      segmentGranularity = Strings.isNullOrEmpty(segmentGranularity) ? HiveConf
          .getVar(parseCtx.getConf(),
              HiveConf.ConfVars.HIVE_DRUID_INDEXING_GRANULARITY
          ) : segmentGranularity;
      targetShardsPerGranularity = Integer.parseInt(targetShardsProperty);

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

      // unlink connection between FS and its parent
      final Operator<? extends OperatorDesc> fsParent = fsOp.getParentOperators().get(0);
      fsParent.getChildOperators().clear();

      if (targetShardsPerGranularity > 0) {
        partitionKeyPos = fsParent.getSchema().getSignature().size() + 1;
      }
      granularityKeyPos = fsParent.getSchema().getSignature().size();
      // Create SelectOp with granularity column
      final Operator<? extends OperatorDesc> granularitySelOp = getGranularitySelOp(fsParent,
              segmentGranularity
      );

      // Create ReduceSinkOp operator
      final ArrayList<ColumnInfo> parentCols =
          Lists.newArrayList(granularitySelOp.getSchema().getSignature());
      final ArrayList<ExprNodeDesc> allRSCols = Lists.newArrayList();
      for (ColumnInfo ci : parentCols) {
        allRSCols.add(new ExprNodeColumnDesc(ci));
      }
      // Get the key positions
      final List<Integer> keyPositions;
      final List<Integer> sortOrder;
      final List<Integer> sortNullOrder;
      //Order matters, assuming later that __time_granularity comes first then __druidPartitionKey
      if (targetShardsPerGranularity > 0) {
        keyPositions = Lists.newArrayList(granularityKeyPos, partitionKeyPos);
        sortOrder = Lists.newArrayList(1, 1); // asc
        sortNullOrder = Lists.newArrayList(0, 0); // nulls first
      } else {
        keyPositions = Lists.newArrayList(granularityKeyPos);
        sortOrder = Lists.newArrayList(1); // asc
        sortNullOrder = Lists.newArrayList(0); // nulls first
      }
      ReduceSinkOperator rsOp = getReduceSinkOp(keyPositions, sortOrder,
          sortNullOrder, allRSCols, granularitySelOp, fsOp.getConf().getWriteType());
      rsOp.getConf().setBucketingVersion(fsOp.getConf().getBucketingVersion());

      // Create backtrack SelectOp
      final List<ExprNodeDesc> descs = new ArrayList<>(allRSCols.size());
      final List<String> colNames = new ArrayList<>();
      for (int i = 0; i < allRSCols.size(); i++) {
        ExprNodeDesc col = allRSCols.get(i);
        final String colName = col.getExprString();
        colNames.add(colName);
        if (keyPositions.contains(i)) {
          descs.add(
              new ExprNodeColumnDesc(col.getTypeInfo(), ReduceField.KEY.toString() + "." + colName,
                  null, false
              ));
        } else {
          descs.add(new ExprNodeColumnDesc(col.getTypeInfo(),
              ReduceField.VALUE.toString() + "." + colName, null, false
          ));
        }
      }
      RowSchema selRS = new RowSchema(granularitySelOp.getSchema());
      SelectDesc selConf = new SelectDesc(descs, colNames);
      SelectOperator backtrackSelOp = (SelectOperator) OperatorFactory.getAndMakeChild(
          selConf, selRS, rsOp);

      // Link backtrack SelectOp to FileSinkOp
      fsOp.getParentOperators().clear();
      fsOp.getParentOperators().add(backtrackSelOp);
      backtrackSelOp.getChildOperators().add(fsOp);

      // Update file sink descriptor
      fsOp.getConf().setDpSortState(FileSinkDesc.DPSortState.PARTITION_SORTED);
      fsOp.getConf().setPartitionCols(rsOp.getConf().getPartitionCols());
      final ColumnInfo granularityColumnInfo =
          new ColumnInfo(granularitySelOp.getSchema().getSignature().get(granularityKeyPos));
      fsOp.getSchema().getSignature().add(granularityColumnInfo);
      if (targetShardsPerGranularity > 0) {
        final ColumnInfo partitionKeyColumnInfo =
            new ColumnInfo(granularitySelOp.getSchema().getSignature().get(partitionKeyPos));
        fsOp.getSchema().getSignature().add(partitionKeyColumnInfo);
      }

      LOG.info("Inserted " + granularitySelOp.getOperatorId() + ", " + rsOp.getOperatorId() + " and "
          + backtrackSelOp.getOperatorId() + " as parent of " + fsOp.getOperatorId()
          + " and child of " + fsParent.getOperatorId());
      parseCtx.setReduceSinkAddedBySortedDynPartition(true);
      return null;
    }
Introduces a RS before FS to partition data by configuration specified
time granularity.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionTimeGranularityOptimizer.java