class BeamSqlExtractExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, ImmutableMap<Integer, Object> correlateEnv) {
    ReadableInstant time = opValueEvaluated(1, inputRow, window, correlateEnv);

    TimeUnitRange unit = ((BeamSqlPrimitive<TimeUnitRange>) op(0)).getValue();

    switch (unit) {
      case YEAR:
      case QUARTER:
      case MONTH:
      case DAY:
      case DOW:
      case WEEK:
      case DOY:
      case CENTURY:
      case MILLENNIUM:
        Long timeByDay = time.getMillis() / DateTimeUtils.MILLIS_PER_DAY;
        Long extracted = DateTimeUtils.unixDateExtract(unit, timeByDay);
        return BeamSqlPrimitive.of(outputType, extracted);

      case HOUR:
      case MINUTE:
      case SECOND:
        int timeInDay = (int) (time.getMillis() % DateTimeUtils.MILLIS_PER_DAY);
        extracted = (long) DateTimeUtils.unixTimeExtract(unit, timeInDay);
        return BeamSqlPrimitive.of(outputType, extracted);

      default:
        throw new UnsupportedOperationException(
            "Extract for time unit: " + unit + " not supported!");
    }
  }

  public  BeamSqlExtractExpression(List<BeamSqlExpression> operands);

}