class BeamSqlDateFloorExpression extends BeamSqlExpression {

  @Override
  public BeamSqlPrimitive evaluate(
      Row inputRow, BoundedWindow window, ImmutableMap<Integer, Object> correlateEnv) {
    ReadableInstant date = opValueEvaluated(0, inputRow, window, correlateEnv);
    long time = date.getMillis();
    TimeUnitRange unit = ((BeamSqlPrimitive<TimeUnitRange>) op(1)).getValue();

    long newTime = DateTimeUtils.unixTimestampFloor(unit, time);

    DateTime newDate = new DateTime(newTime, date.getZone());
    return BeamSqlPrimitive.of(outputType, newDate);
  }

  public  BeamSqlDateFloorExpression(List<BeamSqlExpression> operands);

}