private Operator<? extends OperatorDesc> getGranularitySelOp(
        Operator<? extends OperatorDesc> fsParent,
        String segmentGranularity
    ) throws SemanticException {
      final ArrayList<ColumnInfo> parentCols =
          Lists.newArrayList(fsParent.getSchema().getSignature());
      final ArrayList<ExprNodeDesc> descs = Lists.newArrayList();
      final List<String> colNames = Lists.newArrayList();
      PrimitiveCategory timestampType = null;
      int timestampPos = -1;
      for (int i = 0; i < parentCols.size(); i++) {
        ColumnInfo ci = parentCols.get(i);
        ExprNodeColumnDesc columnDesc = new ExprNodeColumnDesc(ci);
        descs.add(columnDesc);
        colNames.add(columnDesc.getExprString());
        if (columnDesc.getTypeInfo().getCategory() == ObjectInspector.Category.PRIMITIVE
                && (((PrimitiveTypeInfo) columnDesc.getTypeInfo()).getPrimitiveCategory() == PrimitiveCategory.TIMESTAMP ||
            ((PrimitiveTypeInfo) columnDesc.getTypeInfo()).getPrimitiveCategory() == PrimitiveCategory.TIMESTAMPLOCALTZ)) {
          if (timestampPos != -1) {
            throw new SemanticException("Multiple columns with timestamp/timestamp with local time-zone type on query result; "
                    + "could not resolve which one is the right column");
          }
          timestampType = ((PrimitiveTypeInfo) columnDesc.getTypeInfo()).getPrimitiveCategory();
          timestampPos = i;
        }
      }
      if (timestampPos == -1) {
        throw new SemanticException("No column with timestamp with local time-zone type on query result; "
                + "one column should be of timestamp with local time-zone type");
      }
      final RowSchema selRS = new RowSchema(fsParent.getSchema());
      // Granularity (partition) column
      final String udfName;
      Class<? extends UDF> udfClass;
      switch (segmentGranularity) {
        case "YEAR":
          udfName = "floor_year";
          udfClass = UDFDateFloorYear.class;
          break;
        case "MONTH":
          udfName = "floor_month";
          udfClass = UDFDateFloorMonth.class;
          break;
        case "WEEK":
          udfName = "floor_week";
          udfClass = UDFDateFloorWeek.class;
          break;
        case "DAY":
          udfName = "floor_day";
          udfClass = UDFDateFloorDay.class;
          break;
        case "HOUR":
          udfName = "floor_hour";
          udfClass = UDFDateFloorHour.class;
          break;
        case "MINUTE":
          udfName = "floor_minute";
          udfClass = UDFDateFloorMinute.class;
          break;
        case "SECOND":
          udfName = "floor_second";
          udfClass = UDFDateFloorSecond.class;
          break;
        default:
          throw new SemanticException(String.format(Locale.ENGLISH,
              "Unknown Druid Granularity [%s], Accepted values are [YEAR, MONTH, WEEK, DAY, HOUR, MINUTE, SECOND]",
              segmentGranularity
          ));
      }


      // Timestamp column type in Druid is either timestamp or timestamp with local time-zone, i.e.,
      // a specific instant in time. Thus, for the latest, we have this value and we need to extract the
      // granularity to split the data when we are storing it in Druid. However, Druid stores
      // the data in UTC. Thus, we need to apply the following logic on the data to extract
      // the granularity correctly:
      // 1) Read the timestamp with local time-zone value.
      // 2) Extract UTC epoch (millis) from timestamp with local time-zone.
      // 3) Cast the long to a timestamp.
      // 4) Apply the granularity function on the timestamp value.
      // That way, '2010-01-01 00:00:00 UTC' and '2009-12-31 16:00:00 PST' (same instant)
      // will end up in the same Druid segment.

      // #1 - Read the column value
      ExprNodeDesc expr = new ExprNodeColumnDesc(parentCols.get(timestampPos));
      if (timestampType == PrimitiveCategory.TIMESTAMPLOCALTZ) {
        // #2 - UTC epoch for instant
        expr = new ExprNodeGenericFuncDesc(
            TypeInfoFactory.longTypeInfo, new GenericUDFEpochMilli(), Lists.newArrayList(expr));
        // #3 - Cast to timestamp
        expr = new ExprNodeGenericFuncDesc(
            TypeInfoFactory.timestampTypeInfo, new GenericUDFTimestamp(), Lists.newArrayList(expr));
      }
      // #4 - We apply the granularity function
      expr = new ExprNodeGenericFuncDesc(
          TypeInfoFactory.timestampTypeInfo,
          new GenericUDFBridge(udfName, false, udfClass.getName()),
          Lists.newArrayList(expr));
      descs.add(expr);
      colNames.add(Constants.DRUID_TIMESTAMP_GRANULARITY_COL_NAME);
      // Add granularity to the row schema
      final ColumnInfo ci = new ColumnInfo(Constants.DRUID_TIMESTAMP_GRANULARITY_COL_NAME, TypeInfoFactory.timestampTypeInfo,
              selRS.getSignature().get(0).getTabAlias(), false, false);
      selRS.getSignature().add(ci);
      if (targetShardsPerGranularity > 0 ) {
        // add another partitioning key based on floor(1/rand) % targetShardsPerGranularity
        final ColumnInfo partitionKeyCi =
            new ColumnInfo(Constants.DRUID_SHARD_KEY_COL_NAME, TypeInfoFactory.longTypeInfo,
                selRS.getSignature().get(0).getTabAlias(), false, false
            );
        final ExprNodeDesc targetNumShardDescNode =
            new ExprNodeConstantDesc(TypeInfoFactory.intTypeInfo, targetShardsPerGranularity);
        final ExprNodeGenericFuncDesc randomFn = ExprNodeGenericFuncDesc
            .newInstance(new GenericUDFBridge("rand", false, UDFRand.class.getName()),
                Lists.newArrayList()
            );

        final ExprNodeGenericFuncDesc random = ExprNodeGenericFuncDesc.newInstance(
            new GenericUDFFloor(), Lists.newArrayList(ExprNodeGenericFuncDesc
                .newInstance(new GenericUDFOPDivide(),
                    Lists.newArrayList(new ExprNodeConstantDesc(TypeInfoFactory.doubleTypeInfo, 1.0), randomFn)
                )));
        final ExprNodeGenericFuncDesc randModMax = ExprNodeGenericFuncDesc
            .newInstance(new GenericUDFOPMod(),
                Lists.newArrayList(random, targetNumShardDescNode)
            );
        descs.add(randModMax);
        colNames.add(Constants.DRUID_SHARD_KEY_COL_NAME);
        selRS.getSignature().add(partitionKeyCi);
      }
      // Create SelectDesc
      final SelectDesc selConf = new SelectDesc(descs, colNames);
      // Create Select Operator
      final SelectOperator selOp = (SelectOperator) OperatorFactory.getAndMakeChild(
              selConf, selRS, fsParent);

      return selOp;
    }
Introduces a RS before FS to partition data by configuration specified
time granularity.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionTimeGranularityOptimizer.java