Timm0 commented on code in PR #28886:
URL: https://github.com/apache/flink/pull/28886#discussion_r3735795860
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowAggregateQueryOperation.java:
##########
@@ -94,35 +107,112 @@ public String asSummaryString() {
@Override
public String asSerializableString(SqlFactory sqlFactory) {
+ final List<WindowColumn> windowColumns = resolveWindowColumns();
return String.format(
"SELECT %s FROM TABLE(%s\n) %s GROUP BY %s",
- Stream.of(
- groupingExpressions.stream(),
- aggregateExpressions.stream(),
- windowPropertiesExpressions.stream())
- .flatMap(Function.identity())
- .map(
- expr ->
-
OperationExpressionsUtils.scopeReferencesWithAlias(
- INPUT_ALIAS, expr))
- .map(
- resolvedExpression ->
-
resolvedExpression.asSerializableString(sqlFactory))
- .collect(Collectors.joining(", ")),
+ serializeSelectList(windowColumns, sqlFactory),
OperationUtils.indent(
groupWindow.asSerializableString(
child.asSerializableString(sqlFactory),
sqlFactory)),
INPUT_ALIAS,
- Stream.concat(
- Stream.of("window_start", "window_end"),
- groupingExpressions.stream()
- .map(
- expr ->
-
OperationExpressionsUtils
-
.scopeReferencesWithAlias(
-
INPUT_ALIAS, expr))
- .map(expr ->
expr.asSerializableString(sqlFactory)))
- .collect(Collectors.joining(", ")));
+ serializeGroupBy(windowColumns, sqlFactory));
+ }
+
+ private List<WindowColumn> resolveWindowColumns() {
+ return windowPropertiesExpressions.stream()
+ .map(property -> new WindowColumn(aliasOf(property),
windowColumnOf(property)))
+ .collect(Collectors.toList());
+ }
+
+ private static String aliasOf(ResolvedExpression aliasedProperty) {
+ return OperationExpressionsUtils.extractName(aliasedProperty)
+ .orElseThrow(
+ () ->
+ new TableException(
+ "Expected a named alias over a window
property. Got: "
+ + aliasedProperty));
+ }
+
+ /** The windowing TVF output column that the given window property
denotes. */
+ private String windowColumnOf(ResolvedExpression aliasedProperty) {
+ final FunctionDefinition property = windowPropertyOf(aliasedProperty);
+ if (BuiltInFunctionDefinitions.WINDOW_START == property) {
+ return WINDOW_START_COLUMN;
+ }
+ if (BuiltInFunctionDefinitions.WINDOW_END == property) {
+ return WINDOW_END_COLUMN;
+ }
+ if (BuiltInFunctionDefinitions.ROWTIME == property) {
+ return WINDOW_TIME_COLUMN;
+ }
+ if (BuiltInFunctionDefinitions.PROCTIME == property) {
+ checkWindowIsProcessingTime();
+ return WINDOW_TIME_COLUMN;
+ }
+ throw new TableException("Unsupported window property: " + property);
+ }
+
+ private static FunctionDefinition windowPropertyOf(ResolvedExpression
aliasedProperty) {
+ final List<ResolvedExpression> children =
aliasedProperty.getResolvedChildren();
+ if (!children.isEmpty() && children.get(0) instanceof CallExpression) {
+ final FunctionDefinition property =
+ ((CallExpression) children.get(0)).getFunctionDefinition();
+ if
(BuiltInFunctionDefinitions.WINDOW_PROPERTIES.contains(property)) {
Review Comment:
Moved them to `OperationExpressionsUtils` and added a doc string for each of
those.
--
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]
