fhueske commented on code in PR #26424:
URL: https://github.com/apache/flink/pull/26424#discussion_r2182926372
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunction.java:
##########
@@ -102,8 +100,7 @@ public NonTimeRangeUnboundedPrecedingFunction(
accTypes,
inputFieldTypes,
sortKeyTypes,
- sortKeySelector,
- accKeyRowTypeInfo);
+ sortKeySelector);
Review Comment:
pass `InternalTypeInfo.ofFields(sortKeyTypes)` as `accKeyRowTypeInfo` here.
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunctionTest.java:
##########
@@ -99,6 +96,55 @@ public void testInsertOnlyRecordsWithCustomSortKey() throws
Exception {
validateRows(actualRows, expectedRows);
}
+ @Test
+ public void testInsertOnlyRecordsWithCustomSortKeyAndLongSumAgg() throws
Exception {
+ KeyedProcessOperator<RowData, RowData, RowData> operator =
+ new KeyedProcessOperator<>(
+ new NonTimeRowsUnboundedPrecedingFunction<RowData>(
+ 0L,
+ aggsSumLongHandleFunction,
+ GENERATED_ROW_VALUE_EQUALISER,
+ GENERATED_SORT_KEY_EQUALISER,
+ GENERATED_SORT_KEY_COMPARATOR_ASC,
+ accTypes,
+ inputFieldTypes,
+ SORT_KEY_TYPES,
+ SORT_KEY_SELECTOR) {});
Review Comment:
We should add another test with another sort key type. `SORT_KEY_SELECTOR`
is of type `BigInt` which is the same type as the internal rowId used by the
functions. If the new test would have a sort key of type `String`, I would
expect it to fail.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/AbstractNonTimeUnboundedPrecedingOver.java:
##########
@@ -170,8 +167,7 @@ public AbstractNonTimeUnboundedPrecedingOver(
LogicalType[] accTypes,
LogicalType[] inputFieldTypes,
LogicalType[] sortKeyTypes,
- RowDataKeySelector sortKeySelector,
- InternalTypeInfo<RowData> accKeyRowTypeInfo) {
Review Comment:
Don't we need this in the abstract class to have different key types for
RANGE (sortKeys) and ROWS (rowId) for `accMapState`?
We don't need this in the RANGE and ROWS functions, but they have to pass
the correct one to the abstract class.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRowsUnboundedPrecedingFunction.java:
##########
@@ -103,8 +101,7 @@ public NonTimeRowsUnboundedPrecedingFunction(
accTypes,
inputFieldTypes,
sortKeyTypes,
- sortKeySelector,
- accKeyRowTypeInfo);
+ sortKeySelector);
Review Comment:
pass `InternalTypeInfo.ofFields(new BigIntType())` as `accKeyRowTypeInfo`
here.
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunctionTest.java:
##########
@@ -99,6 +97,55 @@ public void testInsertOnlyRecordsWithCustomSortKey() throws
Exception {
validateRows(actualRows, expectedRows);
}
+ @Test
+ public void testInsertOnlyRecordsWithCustomSortKeyAndLongSumAgg() throws
Exception {
+ KeyedProcessOperator<RowData, RowData, RowData> operator =
+ new KeyedProcessOperator<>(
+ new NonTimeRangeUnboundedPrecedingFunction<>(
+ 0L,
+ aggsSumLongHandleFunction,
+ GENERATED_ROW_VALUE_EQUALISER,
+ GENERATED_SORT_KEY_EQUALISER,
+ GENERATED_SORT_KEY_COMPARATOR_ASC,
+ accTypes,
+ inputFieldTypes,
+ SORT_KEY_TYPES,
+ SORT_KEY_SELECTOR) {});
Review Comment:
same as in `NonTimeRangeUnboundedPrecedingFunctionTest`
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
