leonardBang commented on a change in pull request #14084: URL: https://github.com/apache/flink/pull/14084#discussion_r547050151
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
##########
@@ -480,15 +480,17 @@ class AggFunctionFactory(
val valueType = argTypes(0)
if (needRetraction(index)) {
valueType.getTypeRoot match {
- case TINYINT | SMALLINT | INTEGER | BIGINT | FLOAT | DOUBLE | BOOLEAN
| VARCHAR | DECIMAL =>
+ case TINYINT | SMALLINT | INTEGER | BIGINT | FLOAT | DOUBLE | BOOLEAN
| VARCHAR | DECIMAL |
+ TIME_WITHOUT_TIME_ZONE | DATE | TIMESTAMP_WITHOUT_TIME_ZONE =>
new FirstValueWithRetractAggFunction(valueType)
case t =>
throw new TableException(s"FIRST_VALUE with retract aggregate
function does not " +
s"support type: ''$t''.\nPlease re-check the data type.")
}
} else {
valueType.getTypeRoot match {
- case TINYINT | SMALLINT | INTEGER | BIGINT | FLOAT | DOUBLE | BOOLEAN
| VARCHAR | DECIMAL =>
+ case TINYINT | SMALLINT | INTEGER | BIGINT | FLOAT | DOUBLE | BOOLEAN
| VARCHAR | DECIMAL |
+ TIME_WITHOUT_TIME_ZONE | DATE | TIMESTAMP_WITHOUT_TIME_ZONE =>
Review comment:
we can also support `TIMESTAMP_WITH_LOCAL_TIME_ZONE`
##########
File path:
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunctionWithOrderTest.java
##########
@@ -405,6 +412,280 @@ protected Double getValue(String v) {
}
}
+ /**
+ * Test for {@link DateType}.
+ */
+ public static final class DateFirstValueAggFunctionWithOrderTest
+ extends FirstValueAggFunctionWithOrderTestBase<LocalDate> {
+
+ @Override
+ protected List<List<LocalDate>> getInputValueSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ LocalDate.parse("2020-11-11"),
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-13")
+ ),
+ Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-13"),
+ LocalDate.parse("2020-11-14")
+ ),
+ Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-11"),
+ null,
+ LocalDate.parse("2020-11-15"),
+ LocalDate.parse("2020-11-10"),
+ LocalDate.parse("2020-11-09"),
+ null
+ ),
+ Arrays.asList(
+ null,
+ null,
+ null
+ ),
+ Arrays.asList(
+ null,
+ LocalDate.parse("2020-11-12")
+ ));
+ }
+
+ @Override
+ protected List<List<Long>> getInputOrderSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ 6L,
+ 2L,
+ 3L
+ ),
+ Arrays.asList(
+ 1L,
+ 2L,
+ 3L
+ ),
+ Arrays.asList(
+ 10L,
+ 2L,
+ 5L,
+ 3L,
+ 11L,
+ 7L,
+ 5L
+ ),
+ Arrays.asList(
+ 6L,
+ 9L,
+ 5L
+ ),
+ Arrays.asList(
+ 4L,
+ 3L
+ )
+ );
+ }
+
+ @Override
+ protected List<LocalDate> getExpectedResults() {
+ return Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-11"),
+ null,
+ LocalDate.parse("2020-11-12")
+ );
+ }
+
+ @Override
+ protected AggregateFunction<LocalDate, RowData> getAggregator()
{
+ return new
FirstValueAggFunction<>(DataTypes.DATE().getLogicalType());
+ }
+ }
+
+ /**
+ * Test for {@link TimeType}.
+ */
+ public static final class TimeFirstValueAggFunctionWithOrderTest
+ extends FirstValueAggFunctionWithOrderTestBase<LocalTime> {
+
+ private int precision = 3;
+
+ @Override
+ protected List<List<LocalTime>> getInputValueSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ LocalTime.parse("12:00:00.123"),
+ LocalTime.parse("12:45:00.345"),
+ LocalTime.parse("18:30:15.678")
+ ),
+ Arrays.asList(
+ LocalTime.parse("18:00:00.123"),
+ LocalTime.parse("18:45:00.345"),
+ LocalTime.parse("20:30:15.678")
+ ),
+ Arrays.asList(
+ LocalTime.parse("12:00:00.123"),
+ LocalTime.parse("12:45:00.345"),
+ null,
+ LocalTime.parse("18:00:00.123"),
+ LocalTime.parse("18:45:00.345"),
+ LocalTime.parse("20:30:15.678"),
+ null
+ ),
+ Arrays.asList(
+ null,
+ null,
+ null
+ ),
+ Arrays.asList(
+ null,
+ LocalTime.parse("18:00:00.345")
+ ));
Review comment:
ditto, that's great if we can keep same style, please check other places.
##########
File path:
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunctionWithOrderTest.java
##########
@@ -405,6 +412,280 @@ protected Double getValue(String v) {
}
}
+ /**
+ * Test for {@link DateType}.
+ */
+ public static final class DateFirstValueAggFunctionWithOrderTest
+ extends FirstValueAggFunctionWithOrderTestBase<LocalDate> {
+
+ @Override
+ protected List<List<LocalDate>> getInputValueSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ LocalDate.parse("2020-11-11"),
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-13")
+ ),
+ Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-13"),
+ LocalDate.parse("2020-11-14")
+ ),
+ Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-11"),
+ null,
+ LocalDate.parse("2020-11-15"),
+ LocalDate.parse("2020-11-10"),
+ LocalDate.parse("2020-11-09"),
+ null
+ ),
+ Arrays.asList(
+ null,
+ null,
+ null
+ ),
+ Arrays.asList(
+ null,
+ LocalDate.parse("2020-11-12")
+ ));
Review comment:
minor:
```
)
);
```
##########
File path:
flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/aggfunctions/FirstValueAggFunctionWithOrderTest.java
##########
@@ -405,6 +412,280 @@ protected Double getValue(String v) {
}
}
+ /**
+ * Test for {@link DateType}.
+ */
+ public static final class DateFirstValueAggFunctionWithOrderTest
+ extends FirstValueAggFunctionWithOrderTestBase<LocalDate> {
+
+ @Override
+ protected List<List<LocalDate>> getInputValueSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ LocalDate.parse("2020-11-11"),
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-13")
+ ),
+ Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-13"),
+ LocalDate.parse("2020-11-14")
+ ),
+ Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-11"),
+ null,
+ LocalDate.parse("2020-11-15"),
+ LocalDate.parse("2020-11-10"),
+ LocalDate.parse("2020-11-09"),
+ null
+ ),
+ Arrays.asList(
+ null,
+ null,
+ null
+ ),
+ Arrays.asList(
+ null,
+ LocalDate.parse("2020-11-12")
+ ));
+ }
+
+ @Override
+ protected List<List<Long>> getInputOrderSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ 6L,
+ 2L,
+ 3L
+ ),
+ Arrays.asList(
+ 1L,
+ 2L,
+ 3L
+ ),
+ Arrays.asList(
+ 10L,
+ 2L,
+ 5L,
+ 3L,
+ 11L,
+ 7L,
+ 5L
+ ),
+ Arrays.asList(
+ 6L,
+ 9L,
+ 5L
+ ),
+ Arrays.asList(
+ 4L,
+ 3L
+ )
+ );
+ }
+
+ @Override
+ protected List<LocalDate> getExpectedResults() {
+ return Arrays.asList(
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-12"),
+ LocalDate.parse("2020-11-11"),
+ null,
+ LocalDate.parse("2020-11-12")
+ );
+ }
+
+ @Override
+ protected AggregateFunction<LocalDate, RowData> getAggregator()
{
+ return new
FirstValueAggFunction<>(DataTypes.DATE().getLogicalType());
+ }
+ }
+
+ /**
+ * Test for {@link TimeType}.
+ */
+ public static final class TimeFirstValueAggFunctionWithOrderTest
+ extends FirstValueAggFunctionWithOrderTestBase<LocalTime> {
+
+ private int precision = 3;
+
+ @Override
+ protected List<List<LocalTime>> getInputValueSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ LocalTime.parse("12:00:00.123"),
+ LocalTime.parse("12:45:00.345"),
+ LocalTime.parse("18:30:15.678")
+ ),
+ Arrays.asList(
+ LocalTime.parse("18:00:00.123"),
+ LocalTime.parse("18:45:00.345"),
+ LocalTime.parse("20:30:15.678")
+ ),
+ Arrays.asList(
+ LocalTime.parse("12:00:00.123"),
+ LocalTime.parse("12:45:00.345"),
+ null,
+ LocalTime.parse("18:00:00.123"),
+ LocalTime.parse("18:45:00.345"),
+ LocalTime.parse("20:30:15.678"),
+ null
+ ),
+ Arrays.asList(
+ null,
+ null,
+ null
+ ),
+ Arrays.asList(
+ null,
+ LocalTime.parse("18:00:00.345")
+ ));
+ }
+
+ @Override
+ protected List<List<Long>> getInputOrderSets() {
+ return Arrays.asList(
+ Arrays.asList(
+ 6L,
+ 2L,
+ 3L
+ ),
+ Arrays.asList(
+ 1L,
+ 2L,
+ 3L
+ ),
+ Arrays.asList(
+ 10L,
+ 2L,
+ 5L,
+ 3L,
+ 11L,
+ 7L,
+ 5L
+ ),
+ Arrays.asList(
+ 6L,
+ 9L,
+ 5L
+ ),
+ Arrays.asList(
+ 4L,
+ 3L
+ )
+ );
+ }
+
+ @Override
+ protected List<LocalTime> getExpectedResults() {
+ return Arrays.asList(
+ LocalTime.parse("12:45:00.345"),
+ LocalTime.parse("18:00:00.123"),
+ LocalTime.parse("12:45:00.345"),
+ null,
+ LocalTime.parse("18:00:00.345")
+ );
+ }
+
+ @Override
+ protected AggregateFunction<LocalTime, RowData> getAggregator()
{
+ return new
FirstValueAggFunction<>(DataTypes.TIME(precision).getLogicalType());
+ }
+ }
+
+ /**
+ * Test for {@link TimestampType}.
+ */
+ public static final class TimestampFirstValueAggFunctionWithOrderTest
+ extends FirstValueAggFunctionWithOrderTestBase<TimestampData> {
+
+ private int precision = 3;
Review comment:
we can move this variable to `getAggregator` internal, it only used here
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
