diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowAggregateQueryOperation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowAggregateQueryOperation.java index aebced129b1fbb..b476de9834d0c3 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowAggregateQueryOperation.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/WindowAggregateQueryOperation.java @@ -21,11 +21,17 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.table.api.TableException; import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.expressions.CallExpression; import org.apache.flink.table.expressions.FieldReferenceExpression; import org.apache.flink.table.expressions.ResolvedExpression; import org.apache.flink.table.expressions.SqlFactory; import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.functions.BuiltInFunctionDefinitions; +import org.apache.flink.table.functions.FunctionDefinition; import org.apache.flink.table.operations.utils.OperationExpressionsUtils; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.table.utils.EncodingUtils; import org.apache.flink.util.StringUtils; import javax.annotation.Nullable; @@ -35,7 +41,6 @@ import java.util.List; import java.util.Map; import java.util.Optional; -import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -53,6 +58,14 @@ public class WindowAggregateQueryOperation implements QueryOperation { private static final String INPUT_ALIAS = "$$T_WIN_AGG"; + + // Output columns of a windowing TVF, which is what window properties become in SQL. + private static final String WINDOW_START_COLUMN = + BuiltInFunctionDefinitions.WINDOW_START.getSqlName(); + private static final String WINDOW_END_COLUMN = + BuiltInFunctionDefinitions.WINDOW_END.getSqlName(); + private static final String WINDOW_TIME_COLUMN = "window_time"; + private final List groupingExpressions; private final List aggregateExpressions; private final List windowPropertiesExpressions; @@ -94,35 +107,112 @@ public String asSummaryString() { @Override public String asSerializableString(SqlFactory sqlFactory) { + final List 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 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 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)) { + return property; + } + } + throw new TableException( + "Expected an aliased window property call. Got: " + aliasedProperty); + } + + /** + * Rejects a processing-time property on an event-time group window. + * + *

The {@code window_time} column of a windowing TVF derives its time attribute kind from the + * window, not from the requested property, so it cannot express a processing-time attribute of + * an event-time window. Since a {@code rowtime} property on a processing-time window is + * rejected during expression resolution (see {@code WindowTimeIndictorInputTypeStrategy}), we + * reject the other case here rather than serializing it. + */ + private void checkWindowIsProcessingTime() { + final LogicalType windowTimeAttribute = + groupWindow.getTimeAttribute().getOutputDataType().getLogicalType(); + if (LogicalTypeChecks.isProctimeAttribute(windowTimeAttribute)) { + return; + } + throw new TableException( + String.format( + "The processing-time property of the event-time group window '%s' " + + "cannot be expressed in windowing-TVF syntax. The window_time " + + "column of a windowing TVF always has the time attribute kind " + + "of the window itself, so it cannot represent a wall-clock " + + "processing-time attribute. Define the group window on a " + + "processing-time attribute instead.", + groupWindow.getAlias())); + } + + private String serializeSelectList(List windowColumns, SqlFactory sqlFactory) { + return Stream.concat( + Stream.concat(groupingExpressions.stream(), aggregateExpressions.stream()) + .map(expr -> scopedToInput(expr, sqlFactory)), + windowColumns.stream().map(WindowColumn::asSelectItem)) + .collect(Collectors.joining(", ")); + } + + private String scopedToInput(ResolvedExpression expr, SqlFactory sqlFactory) { + return OperationExpressionsUtils.scopeReferencesWithAlias(INPUT_ALIAS, expr) + .asSerializableString(sqlFactory); + } + + private String serializeGroupBy(List windowColumns, SqlFactory sqlFactory) { + final Stream groupedWindowColumns = + windowColumns.stream().anyMatch(WindowColumn::isWindowTime) + ? Stream.of(WINDOW_START_COLUMN, WINDOW_END_COLUMN, WINDOW_TIME_COLUMN) + : Stream.of(WINDOW_START_COLUMN, WINDOW_END_COLUMN); + return Stream.concat( + groupedWindowColumns, + groupingExpressions.stream().map(expr -> scopedToInput(expr, sqlFactory))) + .collect(Collectors.joining(", ")); } public List getGroupingExpressions() { @@ -151,6 +241,26 @@ public T accept(QueryOperationVisitor visitor) { return visitor.visit(this); } + /** A windowing TVF output column, projected under the alias of a window property. */ + private static final class WindowColumn { + + private final String alias; + private final String column; + + private WindowColumn(String alias, String column) { + this.alias = alias; + this.column = column; + } + + private boolean isWindowTime() { + return WINDOW_TIME_COLUMN.equals(column); + } + + private String asSelectItem() { + return String.format("(%s) AS %s", column, EncodingUtils.escapeIdentifier(alias)); + } + } + /** Wrapper for resolved expressions of a {@link org.apache.flink.table.api.GroupWindow}. */ @Internal public static class ResolvedGroupWindow { diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java index f0aa2f0744cab0..1b75be773d18e7 100644 --- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java +++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/QueryOperationTest.java @@ -26,14 +26,20 @@ import org.apache.flink.table.catalog.ResolvedCatalogTable; import org.apache.flink.table.catalog.ResolvedSchema; import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.DefaultSqlFactory; import org.apache.flink.table.expressions.FieldReferenceExpression; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.functions.BuiltInFunctionDefinition; import org.apache.flink.table.functions.BuiltInFunctionDefinitions; import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.Collections; import static org.apache.flink.table.expressions.ApiExpressionUtils.intervalOfMillis; +import static org.apache.flink.table.expressions.ApiExpressionUtils.localRef; +import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral; import static org.assertj.core.api.Assertions.assertThat; /** Tests for describing {@link Operation}s. */ @@ -134,4 +140,52 @@ void testIndentation() { + " secondLevel1\n" + " thirdLevel1"); } + + @Test + void testWindowPropertiesSharingAnAliasAreAllSerialized() { + final ResolvedSchema childSchema = + ResolvedSchema.physical( + Collections.singletonList("a"), Collections.singletonList(DataTypes.INT())); + final FieldReferenceExpression field = + new FieldReferenceExpression("a", DataTypes.INT(), 0, 0); + final ResolvedSchema schema = + ResolvedSchema.physical( + Arrays.asList("a", "dup", "dup"), + Arrays.asList( + DataTypes.INT(), DataTypes.TIMESTAMP(3), DataTypes.TIMESTAMP(3))); + + final WindowAggregateQueryOperation operation = + new WindowAggregateQueryOperation( + Collections.singletonList(field), + Collections.emptyList(), + Arrays.asList( + aliasedWindowProperty(BuiltInFunctionDefinitions.WINDOW_START), + aliasedWindowProperty(BuiltInFunctionDefinitions.WINDOW_END)), + WindowAggregateQueryOperation.ResolvedGroupWindow.tumblingWindow( + "w", field, intervalOfMillis(10)), + new SourceQueryOperation( + ContextResolvedTable.temporary( + ObjectIdentifier.of("cat1", "db1", "tab1"), + new ResolvedCatalogTable( + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .build(), + childSchema))), + schema); + + assertThat(operation.asSerializableString(DefaultSqlFactory.INSTANCE)) + .contains("(window_start) AS `dup`", "(window_end) AS `dup`"); + } + + private static ResolvedExpression aliasedWindowProperty(BuiltInFunctionDefinition property) { + final CallExpression propertyCall = + CallExpression.permanent( + property, + Collections.singletonList(localRef("w", DataTypes.TIMESTAMP(3))), + DataTypes.TIMESTAMP(3)); + return CallExpression.permanent( + BuiltInFunctionDefinitions.AS, + Arrays.asList(propertyCall, valueLiteral("dup")), + DataTypes.TIMESTAMP(3)); + } } diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java index aa1c11a7601508..30d8aab4809fde 100644 --- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java +++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java @@ -219,6 +219,19 @@ public List getSetupTemporalFunctionTestSteps() { .collect(Collectors.toList()); } + /** + * A helper method to avoid boilerplate code. It assumes that only a single Table API statement + * is tested. + */ + public TableApiTestStep getRunTableApiTestStep() { + final List tableApiSteps = + runSteps.stream() + .filter(s -> s.getKind() == TestKind.TABLE_API) + .collect(Collectors.toList()); + Preconditions.checkArgument(tableApiSteps.size() == 1, "Single Table API step expected."); + return (TableApiTestStep) tableApiSteps.get(0); + } + /** * A helper method to avoid boilerplate code. It assumes that only a single SQL statement is * tested. diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java index 26735396b585c3..2862284602d991 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java @@ -49,6 +49,7 @@ public List programs() { QueryOperationTestPrograms.AGGREGATE_HAVING_QUERY_OPERATION, QueryOperationTestPrograms.LIMIT_QUERY_OPERATION, QueryOperationTestPrograms.WINDOW_AGGREGATE_QUERY_OPERATION, + QueryOperationTestPrograms.WINDOW_AGGREGATE_ROWTIME_QUERY_OPERATION, QueryOperationTestPrograms.UNION_ALL_QUERY_OPERATION, QueryOperationTestPrograms.LATERAL_JOIN_QUERY_OPERATION, QueryOperationTestPrograms.GROUP_HOP_WINDOW_EVENT_TIME, diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java index c513c277cd58e7..d3175a773bb0f1 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java @@ -31,6 +31,7 @@ import org.apache.flink.table.test.program.TableTestProgramRunner; import org.apache.flink.table.test.program.TestStep.TestKind; +import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; @@ -42,7 +43,11 @@ import java.util.List; import java.util.Map; +import static org.apache.flink.table.api.Expressions.$; +import static org.apache.flink.table.api.Expressions.lit; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for serialization of {@link org.apache.flink.table.operations.QueryOperation}. */ @TestInstance(TestInstance.Lifecycle.PER_CLASS) @@ -64,6 +69,8 @@ public List programs() { QueryOperationTestPrograms.AGGREGATE_HAVING_QUERY_OPERATION, QueryOperationTestPrograms.LIMIT_QUERY_OPERATION, QueryOperationTestPrograms.WINDOW_AGGREGATE_QUERY_OPERATION, + QueryOperationTestPrograms.WINDOW_AGGREGATE_ROWTIME_QUERY_OPERATION, + QueryOperationTestPrograms.WINDOW_AGGREGATE_PROCTIME_QUERY_OPERATION, QueryOperationTestPrograms.UNION_ALL_QUERY_OPERATION, QueryOperationTestPrograms.LATERAL_JOIN_QUERY_OPERATION, QueryOperationTestPrograms.SQL_QUERY_OPERATION, @@ -87,19 +94,9 @@ public List programs() { void testSqlSerialization(TableTestProgram program) { final TableEnvironment env = setupEnv(program); - final TableApiTestStep tableApiStep = - (TableApiTestStep) - program.runSteps.stream() - .filter(s -> s instanceof TableApiTestStep) - .findFirst() - .get(); - - final SqlTestStep sqlStep = - (SqlTestStep) - program.runSteps.stream() - .filter(s -> s instanceof SqlTestStep) - .findFirst() - .get(); + final TableApiTestStep tableApiStep = program.getRunTableApiTestStep(); + final SqlTestStep sqlStep = program.getRunSqlTestStep(); + final Table table = tableApiStep.toTable(env); assertThat(table.getQueryOperation().asSerializableString(new InlineFunctionSqlFactory())) .isEqualTo(sqlStep.sql); @@ -110,19 +107,8 @@ void testSqlSerialization(TableTestProgram program) { void testSqlAsJobNameForQueryOperation(TableTestProgram program) { final TableEnvironmentImpl env = (TableEnvironmentImpl) setupEnv(program); - final TableApiTestStep tableApiStep = - (TableApiTestStep) - program.runSteps.stream() - .filter(s -> s instanceof TableApiTestStep) - .findFirst() - .get(); - - final SqlTestStep sqlStep = - (SqlTestStep) - program.runSteps.stream() - .filter(s -> s instanceof SqlTestStep) - .findFirst() - .get(); + final TableApiTestStep tableApiStep = program.getRunTableApiTestStep(); + final SqlTestStep sqlStep = program.getRunSqlTestStep(); final Table table = tableApiStep.toTable(env); @@ -138,6 +124,46 @@ void testSqlAsJobNameForQueryOperation(TableTestProgram program) { assertThat(streamGraph.getJobName()).isEqualTo(sqlStep.sql); } + @Test + void testProctimeWindowGeneratedSqlPlans() { + final TableTestProgram program = + QueryOperationTestPrograms.WINDOW_AGGREGATE_PROCTIME_QUERY_OPERATION; + final TableEnvironment env = setupEnv(program); + final Table tableApiTable = program.getRunTableApiTestStep().toTable(env); + + final String generatedSql = + tableApiTable + .getQueryOperation() + .asSerializableString(new InlineFunctionSqlFactory()); + + final Table sqlTable = env.sqlQuery(generatedSql); + + assertThat(sqlTable.getResolvedSchema().getColumnNames()) + .isEqualTo(tableApiTable.getResolvedSchema().getColumnNames()); + assertThatCode(sqlTable::explain).doesNotThrowAnyException(); + } + + @Test + void testProctimePropertyOfEventTimeWindowCannotBeExpressedInWindowingTvfSyntax() { + final TableEnvironment env = + setupEnv(QueryOperationTestPrograms.WINDOW_AGGREGATE_ROWTIME_QUERY_OPERATION); + + final Table table = + env.from("s") + .window(Tumble.over(lit(5).seconds()).on($("ts")).as("w")) + .groupBy($("w"), $("b")) + .select($("b"), $("w").proctime(), $("a").sum()); + + assertThatThrownBy( + () -> + table.getQueryOperation() + .asSerializableString(new InlineFunctionSqlFactory())) + .isInstanceOf(TableException.class) + .hasMessageContaining( + "The processing-time property of the event-time group window 'w' cannot be " + + "expressed in windowing-TVF syntax."); + } + private static TableEnvironment setupEnv(TableTestProgram program) { final TableEnvironment env = TableEnvironment.create( diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java index 16ea3502ab82e5..3724b4a623abb3 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java @@ -286,6 +286,107 @@ public class QueryOperationTestPrograms { + ") $$T_PROJECT") .build(); + static final TableTestProgram WINDOW_AGGREGATE_ROWTIME_QUERY_OPERATION = + TableTestProgram.of( + "window-aggregate-rowtime-query-operation", + "verifies sql serialization of the window rowtime property") + .setupTableSource( + SourceTestStep.newBuilder("s") + .addSchema( + "a bigint", + "b string", + "ts TIMESTAMP_LTZ(3)", + "WATERMARK FOR ts AS ts - INTERVAL '1' SECOND") + .producedValues( + Row.of(2L, "apple", dayOfSeconds(0)), + Row.of(3L, "apple", dayOfSeconds(4)), + Row.of(1L, "apple", dayOfSeconds(7))) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema( + "b string", + "w_start TIMESTAMP_LTZ(3)", + "w_end TIMESTAMP_LTZ(3)", + "w_rowtime TIMESTAMP_LTZ(3)", + "a_sum bigint") + .consumedValues( + Row.of( + "apple", + dayOfSeconds(0), + dayOfSeconds(5), + dayOfSeconds(5).minusMillis(1), + 5L), + Row.of( + "apple", + dayOfSeconds(5), + dayOfSeconds(10), + dayOfSeconds(10).minusMillis(1), + 1L)) + .build()) + .runTableApi( + t -> + t.from("s") + .window( + Tumble.over(lit(5).seconds()) + .on($("ts")) + .as("w")) + .groupBy($("w"), $("b")) + .select( + $("b"), + $("w").start(), + $("w").end(), + $("w").rowtime(), + $("a").sum()), + "sink") + .runSql( + "SELECT `$$T_PROJECT`.`b`, `$$T_PROJECT`.`EXPR$0`, `$$T_PROJECT`.`EXPR$1`, " + + "`$$T_PROJECT`.`EXPR$2`, `$$T_PROJECT`.`EXPR$3` FROM (\n" + + " SELECT `$$T_WIN_AGG`.`b`, (SUM(`$$T_WIN_AGG`.`a`)) AS `EXPR$3`, " + + "(window_start) AS `EXPR$0`, (window_end) AS `EXPR$1`, " + + "(window_time) AS `EXPR$2` FROM TABLE(\n" + + " TUMBLE((\n" + + " SELECT `$$T_SOURCE`.`a`, `$$T_SOURCE`.`b`, " + + "`$$T_SOURCE`.`ts` FROM `default_catalog`.`default_database`.`s` $$T_SOURCE\n" + + " ), DESCRIPTOR(`ts`), INTERVAL '0 00:00:05.000' DAY(2) TO SECOND(3))\n" + + " ) $$T_WIN_AGG GROUP BY window_start, window_end, window_time, " + + "`$$T_WIN_AGG`.`b`\n" + + ") $$T_PROJECT") + .build(); + + static final TableTestProgram WINDOW_AGGREGATE_PROCTIME_QUERY_OPERATION = + TableTestProgram.of( + "window-aggregate-proctime-query-operation", + "verifies sql serialization of the window proctime property") + .setupTableSource( + SourceTestStep.newBuilder("s") + .addSchema("a bigint", "b string", "proctime AS PROCTIME()") + .producedValues(Row.of(2L, "apple"), Row.of(3L, "apple")) + .build()) + .runTableApi( + t -> + t.from("s") + .window( + Tumble.over(lit(5).seconds()) + .on($("proctime")) + .as("w")) + .groupBy($("w"), $("b")) + .select($("b"), $("w").proctime(), $("a").sum()), + "sink") + .runSql( + "SELECT `$$T_PROJECT`.`b`, `$$T_PROJECT`.`EXPR$0`, " + + "`$$T_PROJECT`.`EXPR$1` FROM (\n" + + " SELECT `$$T_WIN_AGG`.`b`, (SUM(`$$T_WIN_AGG`.`a`)) AS `EXPR$1`, " + + "(window_time) AS `EXPR$0` FROM TABLE(\n" + + " TUMBLE((\n" + + " SELECT `$$T_SOURCE`.`a`, `$$T_SOURCE`.`b`, " + + "`$$T_SOURCE`.`proctime` FROM `default_catalog`.`default_database`.`s` $$T_SOURCE\n" + + " ), DESCRIPTOR(`proctime`), INTERVAL '0 00:00:05.000' DAY(2) TO SECOND(3))\n" + + " ) $$T_WIN_AGG GROUP BY window_start, window_end, window_time, " + + "`$$T_WIN_AGG`.`b`\n" + + ") $$T_PROJECT") + .build(); + private static Instant dayOfSeconds(int second) { return LocalDateTime.of(2024, 1, 1, 0, 0, second).atZone(ZoneId.of("UTC")).toInstant(); }