WeiZhong94 commented on a change in pull request #14791:
URL: https://github.com/apache/flink/pull/14791#discussion_r584514927
##########
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecPythonGroupWindowAggregate.java
##########
@@ -161,16 +181,64 @@ public StreamExecPythonGroupWindowAggregate(
WindowAssigner<?> windowAssigner = windowAssignerAndTrigger.f0;
Trigger<?> trigger = windowAssignerAndTrigger.f1;
Configuration config =
CommonPythonUtil.getMergedConfig(planner.getExecEnv(), tableConfig);
- OneInputTransformation<RowData, RowData> transform =
- createPythonStreamWindowGroupOneInputTransformation(
- inputTransform,
- inputRowType,
- outputRowType,
- inputTimeFieldIndex,
- windowAssigner,
- trigger,
- emitStrategy.getAllowLateness(),
- config);
+ boolean isGeneralPythonUDAF =
+ Arrays.stream(aggCalls)
+ .anyMatch(x -> PythonUtil.isPythonAggregate(x,
PythonFunctionKind.GENERAL));
+ int[] namePropertyTypeArray =
+ Arrays.stream(namedWindowProperties)
+ .mapToInt(
+ p -> {
+ PlannerWindowProperty property =
p.property();
+ if (property instanceof
PlannerWindowStart) {
+ return 0;
Review comment:
Using enumerations or directly using case classes to represent the
namedProperties looks more readable than using magic numbers.
##########
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecPythonGroupWindowAggregate.java
##########
@@ -343,4 +444,60 @@ public StreamExecPythonGroupWindowAggregate(
e);
}
}
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private OneInputStreamOperator<RowData, RowData>
+ getGeneralPythonStreamGroupWindowAggregateFunctionOperator(
Review comment:
It seems we still need to support `allowLateness` in the
GeneralPythonStreamGroupWindowAggregateFunctionOperator in order to align with
java window.
----------------------------------------------------------------
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]
