wuchong commented on a change in pull request #15081:
URL: https://github.com/apache/flink/pull/15081#discussion_r587207786
##########
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -65,21 +70,31 @@ public CombineRecordsFunction(
StateKeyContext keyContext,
WindowValueState<Long> accState,
NamespaceAggsHandleFunction<Long> aggregator,
- boolean requiresCopyKey,
+ boolean requiresCopy,
+ RowType recordType,
boolean isEventTime) {
this.timerService = timerService;
this.keyContext = keyContext;
this.accState = accState;
this.aggregator = aggregator;
- this.requiresCopyKey = requiresCopyKey;
+ this.requiresCopy = requiresCopy;
+ if (requiresCopy) {
+ LogicalType[] recordFieldTypes =
+ recordType.getFields().stream()
+ .map(RowType.RowField::getType)
+ .toArray(LogicalType[]::new);
+ this.recordSerializer = new RowDataSerializer(recordFieldTypes);
+ } else {
+ this.recordSerializer = null;
Review comment:
I think it's fine to always initialize `recordSerializer`.
##########
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -152,13 +174,14 @@ public WindowCombineFunction create(
aggregator.open(
new PerWindowStateDataViewStore(
stateBackend, LongSerializer.INSTANCE,
runtimeContext));
- boolean requiresCopyKey =
!isStateImmutableInStateBackend(stateBackend);
+ boolean requiresCopyRecord =
!isStateImmutableInStateBackend(stateBackend);
Review comment:
`requiresCopyRecord` ==> `requiresCopy`.
##########
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -65,21 +70,31 @@ public CombineRecordsFunction(
StateKeyContext keyContext,
WindowValueState<Long> accState,
NamespaceAggsHandleFunction<Long> aggregator,
- boolean requiresCopyKey,
+ boolean requiresCopy,
+ RowType recordType,
boolean isEventTime) {
this.timerService = timerService;
this.keyContext = keyContext;
this.accState = accState;
this.aggregator = aggregator;
- this.requiresCopyKey = requiresCopyKey;
+ this.requiresCopy = requiresCopy;
+ if (requiresCopy) {
+ LogicalType[] recordFieldTypes =
+ recordType.getFields().stream()
+ .map(RowType.RowField::getType)
+ .toArray(LogicalType[]::new);
+ this.recordSerializer = new RowDataSerializer(recordFieldTypes);
Review comment:
Why not `new RowDataSerializer(recordType)`?
##########
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -54,8 +57,10 @@
/** Function used to handle all aggregates. */
private final NamespaceAggsHandleFunction<Long> aggregator;
- /** Whether to copy input key, because key is reused. */
- private final boolean requiresCopyKey;
+ /** Whether to copy input record, because record is reused. */
Review comment:
```suggestion
/** Whether to copy key and input record, because key and record are
reused. */
```
We should also mention key, because key and record are separate.
##########
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/combines/CombineRecordsFunction.java
##########
@@ -54,8 +57,10 @@
/** Function used to handle all aggregates. */
private final NamespaceAggsHandleFunction<Long> aggregator;
- /** Whether to copy input key, because key is reused. */
- private final boolean requiresCopyKey;
+ /** Whether to copy input record, because record is reused. */
+ private final boolean requiresCopy;
+
+ private final RowDataSerializer recordSerializer;
Review comment:
Add a simple Javadoc to describe it.
----------------------------------------------------------------
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]
