twalthr commented on code in PR #28886:
URL: https://github.com/apache/flink/pull/28886#discussion_r3734332008
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java:
##########
@@ -138,6 +124,46 @@ void testSqlAsJobNameForQueryOperation(TableTestProgram
program) {
assertThat(streamGraph.getJobName()).isEqualTo(sqlStep.sql);
}
+ @Test
+ void testProctimeWindowGeneratedSqlPlans() {
Review Comment:
Could also be replaced with a QueryOperationSqlSemanticTest?
##########
flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java:
##########
@@ -134,4 +140,52 @@ void testIndentation() {
+ " secondLevel1\n"
+ " thirdLevel1");
}
+
+ @Test
+ void testWindowPropertiesSharingAnAliasAreAllSerialized() {
Review Comment:
I'm wondering whether we need this test at all. The code is hard to read and
to maintain. In the end QueryOperationSqlSemanticTest is what counts.
##########
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:
I was confused to see `WINDOW_PROPERTIES` in `BuiltInFunctionDefinitions`.
Could we move these Table API specific lists (ORDERING, TIME_ATTRIBUTES etc.)
into OperationExpressionsUtils or some other Table API specific class?
--
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]
