This is an automated email from the ASF dual-hosted git repository.
xuyangzhong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7f707f3ece7 [FLINK-40095][table-planner] Fix JSON aggregation
functions failing in group window aggregate (#28682)
7f707f3ece7 is described below
commit 7f707f3ece7c0fa5bf8bb44f9db499dc73d932ec
Author: Xuyang <[email protected]>
AuthorDate: Wed Jul 15 11:21:57 2026 +0800
[FLINK-40095][table-planner] Fix JSON aggregation functions failing in
group window aggregate (#28682)
---
.../planner/calcite/RelTimeIndicatorConverter.java | 2 +
.../table/planner/hint/FlinkHintStrategies.java | 4 +-
.../logical/WrapJsonAggFunctionArgumentsRule.java | 32 +-
.../nodes/calcite/LogicalWindowAggregate.scala | 28 +-
.../plan/nodes/calcite/WindowAggregate.scala | 3 +-
.../logical/FlinkLogicalWindowAggregate.scala | 25 +-
.../planner/plan/rules/FlinkBatchRuleSets.scala | 3 +-
.../planner/plan/rules/FlinkStreamRuleSets.scala | 3 +-
.../BuiltInAggregateFunctionTestBase.java | 29 +-
.../functions/JsonAggregationFunctionsITCase.java | 100 +++-
.../WrapJsonAggFunctionArgumentsRuleTest.java | 81 ++-
.../WrapJsonAggFunctionArgumentsRuleTest.xml | 547 ++++++++++++++++++++-
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 6 +
.../plan/metadata/FlinkRelMdRowCountTest.scala | 3 +
14 files changed, 800 insertions(+), 66 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
index 8f6ce596c49..86954858c16 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
@@ -584,6 +584,7 @@ public final class RelTimeIndicatorConverter extends
RelHomogeneousShuttle {
return new FlinkLogicalWindowAggregate(
agg.getCluster(),
agg.getTraitSet(),
+ agg.getHints(),
newInput,
agg.getGroupSet(),
updatedAggCalls,
@@ -596,6 +597,7 @@ public final class RelTimeIndicatorConverter extends
RelHomogeneousShuttle {
new FlinkLogicalWindowAggregate(
tableAgg.getCluster(),
tableAgg.getTraitSet(),
+ List.of(),
tableAgg.getInput(),
tableAgg.getGroupSet(),
tableAgg.getAggCallList(),
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java
index 33e6d4346dd..5978f98017d 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java
@@ -82,7 +82,9 @@ public abstract class FlinkHintStrategies {
.hintStrategy(
FlinkHints.HINT_NAME_JSON_AGGREGATE_WRAPPED,
HintStrategy.builder(HintPredicates.AGGREGATE)
-
.excludedRules(WrapJsonAggFunctionArgumentsRule.INSTANCE)
+ .excludedRules(
+
WrapJsonAggFunctionArgumentsRule.AGGREGATE_INSTANCE,
+
WrapJsonAggFunctionArgumentsRule.WINDOW_AGGREGATE_INSTANCE)
.build())
// internal query hint used for alias
.hintStrategy(
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRule.java
index 813034adf29..089e542f4c9 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRule.java
@@ -22,12 +22,14 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
import org.apache.flink.table.planner.hint.FlinkHints;
+import
org.apache.flink.table.planner.plan.nodes.calcite.LogicalWindowAggregate;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelRule;
import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.hint.RelHint;
import org.apache.calcite.rel.logical.LogicalAggregate;
@@ -63,13 +65,19 @@ import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_S
* their correct representation, and the actual aggregation function's
implementation can simply
* insert the values as raw nodes instead. This avoids having to re-implement
the logic for all
* supported types in the aggregation function again.
+ *
+ * <p>This rule supports both {@link LogicalAggregate} and {@link
LogicalWindowAggregate}.
*/
@Internal
@Value.Enclosing
public class WrapJsonAggFunctionArgumentsRule
extends RelRule<WrapJsonAggFunctionArgumentsRule.Config> {
- public static final RelOptRule INSTANCE = new
WrapJsonAggFunctionArgumentsRule(Config.DEFAULT);
+ public static final RelOptRule AGGREGATE_INSTANCE =
+ new WrapJsonAggFunctionArgumentsRule(Config.AGGREGATE_DEFAULT);
+
+ public static final RelOptRule WINDOW_AGGREGATE_INSTANCE =
+ new
WrapJsonAggFunctionArgumentsRule(Config.WINDOW_AGGREGATE_DEFAULT);
/** Marker hint that a call has already been transformed. */
private static final RelHint MARKER_HINT =
@@ -81,15 +89,15 @@ public class WrapJsonAggFunctionArgumentsRule
@Override
public void onMatch(RelOptRuleCall call) {
- final LogicalAggregate aggregate = call.rel(0);
+ final Aggregate aggregate = call.rel(0);
final RelNode aggInput = aggregate.getInput();
final RelBuilder relBuilder = call.builder().push(aggInput);
- final LogicalAggregate wrappedAggregate = wrapJsonAggregate(aggregate,
relBuilder);
+ final Aggregate wrappedAggregate = wrapJsonAggregate(aggregate,
relBuilder);
call.transformTo(wrappedAggregate.withHints(Collections.singletonList(MARKER_HINT)));
}
- private LogicalAggregate wrapJsonAggregate(LogicalAggregate aggregate,
RelBuilder relBuilder) {
+ private Aggregate wrapJsonAggregate(Aggregate aggregate, RelBuilder
relBuilder) {
final int inputCount =
aggregate.getInput().getRowType().getFieldCount();
List<AggregateCall> aggCallList = new
ArrayList<>(aggregate.getAggCallList());
// This map is a mapping relationship between jsonObjectAggCall and
the argument index
@@ -175,19 +183,25 @@ public class WrapJsonAggFunctionArgumentsRule
/** Configuration for {@link WrapJsonAggFunctionArgumentsRule}. */
@Value.Immutable(singleton = false)
public interface Config extends RelRule.Config {
- Config DEFAULT =
+ Config AGGREGATE_DEFAULT =
+ ImmutableWrapJsonAggFunctionArgumentsRule.Config.builder()
+ .build()
+ .as(Config.class)
+ .onAggregateClass(LogicalAggregate.class);
+
+ Config WINDOW_AGGREGATE_DEFAULT =
ImmutableWrapJsonAggFunctionArgumentsRule.Config.builder()
.build()
.as(Config.class)
- .onJsonAggregateFunctions();
+ .onAggregateClass(LogicalWindowAggregate.class);
@Override
default RelOptRule toRule() {
return new WrapJsonAggFunctionArgumentsRule(this);
}
- default Config onJsonAggregateFunctions() {
- final Predicate<LogicalAggregate> jsonAggPredicate =
+ default <T extends Aggregate> Config onAggregateClass(Class<T>
aggregateClass) {
+ final Predicate<T> jsonAggPredicate =
aggregate ->
aggregate.getAggCallList().stream()
.anyMatch(WrapJsonAggFunctionArgumentsRule::isJsonAggregation);
@@ -195,7 +209,7 @@ public class WrapJsonAggFunctionArgumentsRule
final RelRule.OperandTransform aggTransform =
operandBuilder ->
operandBuilder
- .operand(LogicalAggregate.class)
+ .operand(aggregateClass)
.predicate(jsonAggPredicate)
.anyInputs();
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/LogicalWindowAggregate.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/LogicalWindowAggregate.scala
index 0d9e223b268..16eb483a901 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/LogicalWindowAggregate.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/LogicalWindowAggregate.scala
@@ -24,6 +24,7 @@ import org.apache.calcite.plan.{Convention, RelOptCluster,
RelTraitSet}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.core.{Aggregate, AggregateCall}
import org.apache.calcite.rel.core.Aggregate.Group
+import org.apache.calcite.rel.hint.RelHint
import org.apache.calcite.util.ImmutableBitSet
import java.util
@@ -31,12 +32,33 @@ import java.util
final class LogicalWindowAggregate(
cluster: RelOptCluster,
traitSet: RelTraitSet,
+ hints: util.List[RelHint],
child: RelNode,
groupSet: ImmutableBitSet,
aggCalls: util.List[AggregateCall],
window: LogicalWindow,
namedProperties: util.List[NamedWindowProperty])
- extends WindowAggregate(cluster, traitSet, child, groupSet, aggCalls,
window, namedProperties) {
+ extends WindowAggregate(
+ cluster,
+ traitSet,
+ hints,
+ child,
+ groupSet,
+ aggCalls,
+ window,
+ namedProperties) {
+
+ override def withHints(hintList: util.List[RelHint]): RelNode = {
+ new LogicalWindowAggregate(
+ cluster,
+ traitSet,
+ hintList,
+ input,
+ getGroupSet,
+ aggCalls,
+ window,
+ namedProperties)
+ }
override def copy(
traitSet: RelTraitSet,
@@ -47,6 +69,7 @@ final class LogicalWindowAggregate(
new LogicalWindowAggregate(
cluster,
traitSet,
+ getHints,
input,
groupSet,
aggCalls,
@@ -58,6 +81,7 @@ final class LogicalWindowAggregate(
new LogicalWindowAggregate(
cluster,
traitSet,
+ getHints,
input,
getGroupSet,
aggCalls,
@@ -76,6 +100,7 @@ final class LogicalWindowAggregate(
new LogicalWindowAggregate(
cluster,
traitSet,
+ getHints,
input,
groupSet,
aggCalls,
@@ -97,6 +122,7 @@ object LogicalWindowAggregate {
new LogicalWindowAggregate(
cluster,
traitSet,
+ agg.getHints,
agg.getInput,
agg.getGroupSet,
agg.getAggCallList,
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/WindowAggregate.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/WindowAggregate.scala
index 333d80961d0..fd50f547871 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/WindowAggregate.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/WindowAggregate.scala
@@ -41,6 +41,7 @@ import scala.collection.JavaConverters._
abstract class WindowAggregate(
cluster: RelOptCluster,
traitSet: RelTraitSet,
+ hints: util.List[RelHint],
child: RelNode,
groupSet: ImmutableBitSet,
aggCalls: util.List[AggregateCall],
@@ -49,7 +50,7 @@ abstract class WindowAggregate(
extends Aggregate(
cluster,
traitSet,
- new util.ArrayList[RelHint],
+ hints,
child,
groupSet,
ImmutableList.of(groupSet),
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalWindowAggregate.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalWindowAggregate.scala
index c483d660f0a..1f2ddab0806 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalWindowAggregate.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalWindowAggregate.scala
@@ -28,6 +28,7 @@ import org.apache.calcite.rel.convert.ConverterRule
import org.apache.calcite.rel.convert.ConverterRule.Config
import org.apache.calcite.rel.core.{Aggregate, AggregateCall}
import org.apache.calcite.rel.core.Aggregate.Group
+import org.apache.calcite.rel.hint.RelHint
import org.apache.calcite.rel.metadata.RelMetadataQuery
import org.apache.calcite.sql.SqlKind
import org.apache.calcite.util.ImmutableBitSet
@@ -39,12 +40,21 @@ import scala.collection.JavaConverters._
class FlinkLogicalWindowAggregate(
cluster: RelOptCluster,
traitSet: RelTraitSet,
+ hints: util.List[RelHint],
child: RelNode,
groupSet: ImmutableBitSet,
aggCalls: util.List[AggregateCall],
window: LogicalWindow,
namedProperties: util.List[NamedWindowProperty])
- extends WindowAggregate(cluster, traitSet, child, groupSet, aggCalls,
window, namedProperties)
+ extends WindowAggregate(
+ cluster,
+ traitSet,
+ hints,
+ child,
+ groupSet,
+ aggCalls,
+ window,
+ namedProperties)
with FlinkLogicalRel {
override def copy(
@@ -56,6 +66,7 @@ class FlinkLogicalWindowAggregate(
new FlinkLogicalWindowAggregate(
cluster,
traitSet,
+ getHints,
input,
groupSet,
aggCalls,
@@ -73,6 +84,17 @@ class FlinkLogicalWindowAggregate(
planner.getCostFactory.makeCost(rowCnt, cpuCost, rowCnt * rowSize)
}
+ override def withHints(hintList: util.List[RelHint]): RelNode = {
+ new FlinkLogicalWindowAggregate(
+ cluster,
+ traitSet,
+ hintList,
+ input,
+ getGroupSet,
+ aggCalls,
+ window,
+ namedProperties)
+ }
}
class FlinkLogicalWindowAggregateConverter(config: Config) extends
ConverterRule(config) {
@@ -100,6 +122,7 @@ class FlinkLogicalWindowAggregateConverter(config: Config)
extends ConverterRule
new FlinkLogicalWindowAggregate(
rel.getCluster,
traitSet,
+ agg.getHints,
newInput,
agg.getGroupSet,
agg.getAggCallList,
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
index e840808b5e1..3fb057bc08d 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
@@ -139,7 +139,8 @@ object FlinkBatchRuleSets {
// vector search rule.
ConstantVectorSearchCallToCorrelateRule.INSTANCE,
// Wrap arguments for JSON aggregate functions
- WrapJsonAggFunctionArgumentsRule.INSTANCE,
+ WrapJsonAggFunctionArgumentsRule.AGGREGATE_INSTANCE,
+ WrapJsonAggFunctionArgumentsRule.WINDOW_AGGREGATE_INSTANCE,
// prune COUNT(*) input to project a constant before aggregation
PruneCountStarInputRule.INSTANCE
)).asJava)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
index 65e6adbf772..0617400e4ae 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala
@@ -139,7 +139,8 @@ object FlinkStreamRuleSets {
// rewrite constant table function scan to correlate
JoinTableFunctionScanToCorrelateRule.INSTANCE,
// Wrap arguments for JSON aggregate functions
- WrapJsonAggFunctionArgumentsRule.INSTANCE,
+ WrapJsonAggFunctionArgumentsRule.AGGREGATE_INSTANCE,
+ WrapJsonAggFunctionArgumentsRule.WINDOW_AGGREGATE_INSTANCE,
// prune COUNT(*) input to project a constant before aggregation
PruneCountStarInputRule.INSTANCE
)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
index 4c5f85a057c..f6b748948f4 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/BuiltInAggregateFunctionTestBase.java
@@ -106,10 +106,19 @@ abstract class BuiltInAggregateFunctionTestBase {
testCase.execute(new MiniClusterClient(miniCluster.getConfiguration(),
miniCluster));
}
- protected static Table asTable(TableEnvironment tEnv, DataType
sourceRowType, List<Row> rows) {
+ protected static Table asTable(
+ TableEnvironment tEnv,
+ DataType sourceRowType,
+ List<Row> rows,
+ @Nullable String rowtimeColumn,
+ @Nullable String watermarkExpression) {
+ final Schema.Builder schemaBuilder =
Schema.newBuilder().fromRowDataType(sourceRowType);
+ if (rowtimeColumn != null) {
+ schemaBuilder.watermark(rowtimeColumn, watermarkExpression);
+ }
final TableDescriptor descriptor =
TableFactoryHarness.newBuilder()
-
.schema(Schema.newBuilder().fromRowDataType(sourceRowType).build())
+ .schema(schemaBuilder.build())
.source(asSource(rows, sourceRowType))
.build();
@@ -204,6 +213,8 @@ abstract class BuiltInAggregateFunctionTestBase {
private final Configuration configuration = new Configuration();
private DataType sourceRowType;
private List<Row> sourceRows;
+ private @Nullable String rowtimeColumn;
+ private @Nullable String watermarkExpression;
private TestSpec(BuiltInFunctionDefinition definition) {
this.definition = definition;
@@ -238,6 +249,12 @@ abstract class BuiltInAggregateFunctionTestBase {
return this;
}
+ TestSpec withWatermark(String rowtimeColumn, String
watermarkExpression) {
+ this.rowtimeColumn = rowtimeColumn;
+ this.watermarkExpression = watermarkExpression;
+ return this;
+ }
+
TestSpec testSqlResult(
Function<Table, String> sqlSpec, DataType expectedRowType,
List<Row> expectedRows) {
this.testItems.add(new SqlTestItem(sqlSpec, expectedRowType,
expectedRows));
@@ -343,7 +360,13 @@ abstract class BuiltInAggregateFunctionTestBase {
.inStreamingMode()
.withConfiguration(configuration)
.build());
- final Table sourceTable = asTable(tEnv, sourceRowType,
sourceRows);
+ final Table sourceTable =
+ asTable(
+ tEnv,
+ sourceRowType,
+ sourceRows,
+ rowtimeColumn,
+ watermarkExpression);
testItem.execute(tEnv, sourceTable, clusterClient);
};
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonAggregationFunctionsITCase.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonAggregationFunctionsITCase.java
index 34b00813d75..32827b79d5a 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonAggregationFunctionsITCase.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/JsonAggregationFunctionsITCase.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.api.JsonOnNull;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.types.Row;
+import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Collections;
import java.util.stream.Stream;
@@ -29,6 +30,7 @@ import java.util.stream.Stream;
import static org.apache.flink.table.api.DataTypes.INT;
import static org.apache.flink.table.api.DataTypes.ROW;
import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
import static org.apache.flink.table.api.DataTypes.VARCHAR;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.jsonArrayAgg;
@@ -240,6 +242,102 @@ class JsonAggregationFunctionsITCase extends
BuiltInAggregateFunctionTestBase {
ROW(INT(), STRING(), STRING().notNull()),
Arrays.asList(
Row.of(1, "A", "[\"A\"]"),
- Row.of(2, "D", "[\"C\",\"D\"]"))));
+ Row.of(2, "D", "[\"C\",\"D\"]"))),
+
+ // JSON_OBJECTAGG - Window (group window) Aggregation
+
TestSpec.forFunction(BuiltInFunctionDefinitions.JSON_OBJECTAGG_NULL_ON_NULL)
+ .withDescription("Window Aggregation")
+ .withSource(
+ ROW(STRING(), INT(), TIMESTAMP(3)),
+ Arrays.asList(
+ Row.ofKind(
+ INSERT,
+ "A",
+ 1,
+
LocalDateTime.parse("2020-01-01T00:00:01")),
+ Row.ofKind(
+ INSERT,
+ "B",
+ 2,
+
LocalDateTime.parse("2020-01-01T00:00:02")),
+ Row.ofKind(
+ INSERT,
+ "C",
+ 3,
+
LocalDateTime.parse("2020-01-01T00:00:06"))))
+ .withWatermark("f2", "f2 - INTERVAL '1' SECOND")
+ .testSqlResult(
+ source ->
+ "SELECT JSON_OBJECTAGG(f0 VALUE f1)
FROM "
+ + source
+ + " GROUP BY TUMBLE(f2,
INTERVAL '5' SECOND)",
+ ROW(VARCHAR(2000).notNull()),
+ Arrays.asList(Row.of("{\"A\":1,\"B\":2}"),
Row.of("{\"C\":3}"))),
+
TestSpec.forFunction(BuiltInFunctionDefinitions.JSON_OBJECTAGG_NULL_ON_NULL)
+ .withDescription("Window Group Aggregation With Other
Aggs")
+ .withSource(
+ ROW(INT(), STRING(), INT(), TIMESTAMP(3)),
+ Arrays.asList(
+ Row.ofKind(
+ INSERT,
+ 1,
+ "A",
+ 1,
+
LocalDateTime.parse("2020-01-01T00:00:01")),
+ Row.ofKind(
+ INSERT,
+ 1,
+ "B",
+ 3,
+
LocalDateTime.parse("2020-01-01T00:00:02")),
+ Row.ofKind(
+ INSERT,
+ 2,
+ "A",
+ 2,
+
LocalDateTime.parse("2020-01-01T00:00:06")),
+ Row.ofKind(
+ INSERT,
+ 2,
+ "C",
+ 5,
+
LocalDateTime.parse("2020-01-01T00:00:07"))))
+ .withWatermark("f3", "f3 - INTERVAL '1' SECOND")
+ .testSqlResult(
+ source ->
+ "SELECT f0, JSON_OBJECTAGG(f1 VALUE
f2), max(f2) FROM "
+ + source
+ + " GROUP BY f0, TUMBLE(f3,
INTERVAL '5' SECOND)",
+ ROW(INT(), VARCHAR(2000).notNull(), INT()),
+ Arrays.asList(
+ Row.of(1, "{\"A\":1,\"B\":3}", 3),
+ Row.of(2, "{\"A\":2,\"C\":5}", 5))),
+
+ // JSON_ARRAYAGG - Window (group window) Aggregation
+
TestSpec.forFunction(BuiltInFunctionDefinitions.JSON_ARRAYAGG_ABSENT_ON_NULL)
+ .withDescription("Window Aggregation")
+ .withSource(
+ ROW(STRING(), TIMESTAMP(3)),
+ Arrays.asList(
+ Row.ofKind(
+ INSERT,
+ "A",
+
LocalDateTime.parse("2020-01-01T00:00:01")),
+ Row.ofKind(
+ INSERT,
+ "B",
+
LocalDateTime.parse("2020-01-01T00:00:02")),
+ Row.ofKind(
+ INSERT,
+ "C",
+
LocalDateTime.parse("2020-01-01T00:00:06"))))
+ .withWatermark("f1", "f1 - INTERVAL '1' SECOND")
+ .testSqlResult(
+ source ->
+ "SELECT JSON_ARRAYAGG(f0) FROM "
+ + source
+ + " GROUP BY TUMBLE(f1,
INTERVAL '5' SECOND)",
+ ROW(VARCHAR(2000).notNull()),
+ Arrays.asList(Row.of("[\"A\",\"B\"]"),
Row.of("[\"C\"]"))));
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.java
index e512d3b26d5..e424cd2aec1 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.java
@@ -24,27 +24,35 @@ import org.apache.flink.table.planner.utils.TableTestUtil;
import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
+import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
+import java.util.List;
/** Tests for {@link WrapJsonAggFunctionArgumentsRule}. */
@ExtendWith(ParameterizedTestExtension.class)
class WrapJsonAggFunctionArgumentsRuleTest extends TableTestBase {
private final boolean batchMode;
+ private final boolean isGroupWindowAgg;
private TableTestUtil util;
- @Parameters(name = "batchMode = {0}")
- private static Collection<Boolean> data() {
- return Arrays.asList(true, false);
+ @Parameters(name = "batchMode = {0}, isGroupWindowAgg = {1}")
+ private static List<Boolean[]> data() {
+ return Arrays.asList(
+ new Boolean[] {true, false},
+ new Boolean[] {true, true},
+ new Boolean[] {false, false},
+ new Boolean[] {false, true});
}
- WrapJsonAggFunctionArgumentsRuleTest(boolean batchMode) {
+ WrapJsonAggFunctionArgumentsRuleTest(boolean batchMode, boolean
isGroupWindowAgg) {
this.batchMode = batchMode;
+ this.isGroupWindowAgg = isGroupWindowAgg;
}
@BeforeEach
@@ -67,70 +75,89 @@ class WrapJsonAggFunctionArgumentsRuleTest extends
TableTestBase {
+ batchMode
+ "'\n)");
- util.tableEnv()
- .executeSql(
- "CREATE TABLE T1(\n"
- + " f0 INTEGER,\n"
- + " f1 VARCHAR,\n"
- + " f2 BIGINT\n"
- + ") WITH (\n"
- + " 'connector' = 'values'\n"
- + " ,'bounded' = '"
- + batchMode
- + "'\n)");
+ if (isGroupWindowAgg) {
+ util.tableEnv()
+ .executeSql(
+ "ALTER TABLE T add (rt TIMESTAMP(3), WATERMARK FOR
rt as rt - interval '1' second)");
+ }
}
@TestTemplate
void testJsonObjectAgg() {
- util.verifyRelPlan("SELECT JSON_OBJECTAGG(f1 VALUE f1) FROM T");
+ util.verifyRelPlan(renderAggregateSQL("JSON_OBJECTAGG(f1 VALUE f1)"));
}
@TestTemplate
void testJsonObjectAggInGroupWindow() {
- util.verifyRelPlan("SELECT f0, JSON_OBJECTAGG(f1 VALUE f0) FROM T
GROUP BY f0");
+ util.verifyRelPlan(renderAggregateSQL("f0, JSON_OBJECTAGG(f1 VALUE
f0)", "f0"));
}
@TestTemplate
void testJsonArrayAgg() {
- util.verifyRelPlan("SELECT JSON_ARRAYAGG(f0) FROM T");
+ util.verifyRelPlan(renderAggregateSQL("JSON_ARRAYAGG(f0)"));
}
@TestTemplate
void testJsonArrayAggInGroupWindow() {
- util.verifyRelPlan("SELECT f0, JSON_ARRAYAGG(f0) FROM T GROUP BY f0");
+ util.verifyRelPlan(renderAggregateSQL("f0, JSON_ARRAYAGG(f0)", "f0"));
}
@TestTemplate
void testJsonObjectAggWithOtherAggs() {
- util.verifyRelPlan("SELECT COUNT(*), JSON_OBJECTAGG(f1 VALUE f1) FROM
T");
+ util.verifyRelPlan(renderAggregateSQL("COUNT(*), JSON_OBJECTAGG(f1
VALUE f1)"));
}
@TestTemplate
void testGroupJsonObjectAggWithOtherAggs() {
util.verifyRelPlan(
- "SELECT f0, COUNT(*), JSON_OBJECTAGG(f1 VALUE f0), SUM(f2)
FROM T GROUP BY f0");
+ renderAggregateSQL("f0, COUNT(*), JSON_OBJECTAGG(f1 VALUE f0),
SUM(f2)", "f0"));
}
@TestTemplate
void testJsonArrayAggWithOtherAggs() {
- util.verifyRelPlan("SELECT COUNT(*), JSON_ARRAYAGG(f0) FROM T");
+ util.verifyRelPlan(renderAggregateSQL("COUNT(*), JSON_ARRAYAGG(f0)"));
}
@TestTemplate
void testGroupJsonArrayAggInWithOtherAggs() {
- util.verifyRelPlan("SELECT f0, COUNT(*), JSON_ARRAYAGG(f0), SUM(f2)
FROM T GROUP BY f0");
+ util.verifyRelPlan(renderAggregateSQL("f0, COUNT(*),
JSON_ARRAYAGG(f0), SUM(f2)", "f0"));
}
@TestTemplate
void testJsonArrayAggAndJsonObjectAggWithOtherAggs() {
util.verifyRelPlan(
- "SELECT MAX(f0), JSON_OBJECTAGG(f1 VALUE f0),
JSON_ARRAYAGG(f1), JSON_ARRAYAGG(f0) FROM T");
+ renderAggregateSQL(
+ "MAX(f0), JSON_OBJECTAGG(f1 VALUE f0),
JSON_ARRAYAGG(f1), JSON_ARRAYAGG(f0)"));
}
@TestTemplate
void testGroupJsonArrayAggAndJsonObjectAggWithOtherAggs() {
util.verifyRelPlan(
- "SELECT f0, JSON_OBJECTAGG(f1 VALUE f2), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f2),"
- + " SUM(f2) FROM T GROUP BY f0");
+ renderAggregateSQL(
+ "f0, JSON_OBJECTAGG(f1 VALUE f2), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f2),"
+ + " SUM(f2)",
+ "f0"));
+ }
+
+ @TestTemplate
+ void testJsonObjectAggWithWindowTVF() {
+ Assumptions.assumeTrue(isGroupWindowAgg);
+ util.verifyRelPlan(
+ "SELECT JSON_OBJECTAGG(f1 VALUE f1) "
+ + "FROM TABLE(TUMBLE(TABLE T, DESCRIPTOR(rt), INTERVAL
'5' SECOND))");
+ }
+
+ private String renderAggregateSQL(String projects, String... groupKeys) {
+ List<String> newGroupKeys = new ArrayList<>(Arrays.asList(groupKeys));
+ if (isGroupWindowAgg) {
+ newGroupKeys.add("TUMBLE(rt, INTERVAL '5' SECOND)");
+ }
+ String groupKeyStr;
+ if (newGroupKeys.isEmpty()) {
+ groupKeyStr = "";
+ } else {
+ groupKeyStr = " GROUP BY " + String.join(",", newGroupKeys);
+ }
+ return String.format("SELECT %s FROM T%s", projects, groupKeyStr);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.xml
index de4f8bb7341..e5480a1e27b 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/WrapJsonAggFunctionArgumentsRuleTest.xml
@@ -16,7 +16,7 @@ See the License for the specific language governing
permissions and
limitations under the License.
-->
<Root>
- <TestCase name="testGroupJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode
= false]">
+ <TestCase name="testGroupJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode
= false, isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f2), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f2), SUM(f2) FROM T GROUP BY f0]]>
</Resource>
@@ -35,7 +35,30 @@ GroupAggregate(groupBy=[f0], select=[f0,
JSON_OBJECTAGG_NULL_ON_NULL(f1, $f4) AS
]]>
</Resource>
</TestCase>
- <TestCase name="testGroupJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode
= true]">
+ <TestCase name="testGroupJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode
= false, isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f2), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f2), SUM(f2) FROM T GROUP BY f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4], EXPR$4=[$5])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($2,
$3)], EXPR$2=[JSON_ARRAYAGG_ABSENT_ON_NULL($2)],
EXPR$3=[JSON_ARRAYAGG_ABSENT_ON_NULL($3)], EXPR$4=[SUM($3)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)],
f1=[$1], f2=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt,
5000)], select=[f0, JSON_OBJECTAGG_NULL_ON_NULL(f1, $f5) AS EXPR$1,
JSON_ARRAYAGG_ABSENT_ON_NULL($f4) AS EXPR$2, JSON_ARRAYAGG_ABSENT_ON_NULL($f5)
AS EXPR$3, SUM(f2) AS EXPR$4])
++- Exchange(distribution=[hash[f0]])
+ +- Calc(select=[f0, rt, f1, f2, JSON_STRING(f1) AS $f4, JSON_STRING(f2) AS
$f5])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testGroupJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode
= true, isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f2), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f2), SUM(f2) FROM T GROUP BY f0]]>
</Resource>
@@ -55,7 +78,29 @@ SortAggregate(isMerge=[false], groupBy=[f0], select=[f0,
JSON_OBJECTAGG_NULL_ON_
]]>
</Resource>
</TestCase>
- <TestCase name="testGroupJsonArrayAggInWithOtherAggs[batchMode = false]">
+ <TestCase name="testGroupJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode
= true, isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f2), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f2), SUM(f2) FROM T GROUP BY f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4], EXPR$4=[$5])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($2,
$3)], EXPR$2=[JSON_ARRAYAGG_ABSENT_ON_NULL($2)],
EXPR$3=[JSON_ARRAYAGG_ABSENT_ON_NULL($3)], EXPR$4=[SUM($3)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)],
f1=[$1], f2=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt, 5000)],
select=[f0, JSON_OBJECTAGG_NULL_ON_NULL(f1, $f5) AS EXPR$1,
JSON_ARRAYAGG_ABSENT_ON_NULL($f4) AS EXPR$2, JSON_ARRAYAGG_ABSENT_ON_NULL($f5)
AS EXPR$3, SUM(f2) AS EXPR$4])
++- Calc(select=[f0, rt, f1, f2, JSON_STRING(f1) AS $f4, JSON_STRING(f2) AS
$f5])
+ +- Sort(orderBy=[f0 ASC, rt ASC])
+ +- Exchange(distribution=[hash[f0]])
+ +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testGroupJsonArrayAggInWithOtherAggs[batchMode = false,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, COUNT(*), JSON_ARRAYAGG(f0), SUM(f2) FROM T GROUP BY
f0]]>
</Resource>
@@ -75,7 +120,30 @@ GroupAggregate(groupBy=[f0], select=[f0, COUNT(*) AS
EXPR$1, JSON_ARRAYAGG_ABSEN
]]>
</Resource>
</TestCase>
- <TestCase name="testGroupJsonArrayAggInWithOtherAggs[batchMode = true]">
+ <TestCase name="testGroupJsonArrayAggInWithOtherAggs[batchMode = false,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, COUNT(*), JSON_ARRAYAGG(f0), SUM(f2) FROM T GROUP BY
f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[COUNT()],
EXPR$2=[JSON_ARRAYAGG_ABSENT_ON_NULL($0)], EXPR$3=[SUM($2)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)], f2=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt,
5000)], select=[f0, COUNT(*) AS EXPR$1, JSON_ARRAYAGG_ABSENT_ON_NULL($f3) AS
EXPR$2, SUM(f2) AS EXPR$3])
++- Exchange(distribution=[hash[f0]])
+ +- Calc(select=[f0, rt, f2, JSON_STRING(f0) AS $f3])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f0, rt, f2], metadata=[]]], fields=[f0, rt, f2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testGroupJsonArrayAggInWithOtherAggs[batchMode = true,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, COUNT(*), JSON_ARRAYAGG(f0), SUM(f2) FROM T GROUP BY
f0]]>
</Resource>
@@ -96,7 +164,29 @@ SortAggregate(isMerge=[false], groupBy=[f0], select=[f0,
COUNT(*) AS EXPR$1, JSO
]]>
</Resource>
</TestCase>
- <TestCase name="testGroupJsonObjectAggWithOtherAggs[batchMode = false]">
+ <TestCase name="testGroupJsonArrayAggInWithOtherAggs[batchMode = true,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, COUNT(*), JSON_ARRAYAGG(f0), SUM(f2) FROM T GROUP BY
f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[COUNT()],
EXPR$2=[JSON_ARRAYAGG_ABSENT_ON_NULL($0)], EXPR$3=[SUM($2)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)], f2=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt, 5000)],
select=[f0, COUNT(*) AS EXPR$1, JSON_ARRAYAGG_ABSENT_ON_NULL($f3) AS EXPR$2,
SUM(f2) AS EXPR$3])
++- Calc(select=[f0, rt, f2, JSON_STRING(f0) AS $f3])
+ +- Sort(orderBy=[f0 ASC, rt ASC])
+ +- Exchange(distribution=[hash[f0]])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f0, rt, f2], metadata=[]]], fields=[f0, rt, f2])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testGroupJsonObjectAggWithOtherAggs[batchMode = false,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, COUNT(*), JSON_OBJECTAGG(f1 VALUE f0), SUM(f2) FROM
T GROUP BY f0]]>
</Resource>
@@ -115,7 +205,30 @@ GroupAggregate(groupBy=[f0], select=[f0, COUNT(*) AS
EXPR$1, JSON_OBJECTAGG_NULL
]]>
</Resource>
</TestCase>
- <TestCase name="testGroupJsonObjectAggWithOtherAggs[batchMode = true]">
+ <TestCase name="testGroupJsonObjectAggWithOtherAggs[batchMode = false,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, COUNT(*), JSON_OBJECTAGG(f1 VALUE f0), SUM(f2) FROM
T GROUP BY f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[COUNT()],
EXPR$2=[JSON_OBJECTAGG_NULL_ON_NULL($2, $0)], EXPR$3=[SUM($3)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)],
f1=[$1], f2=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt,
5000)], select=[f0, COUNT(*) AS EXPR$1, JSON_OBJECTAGG_NULL_ON_NULL(f1, $f4) AS
EXPR$2, SUM(f2) AS EXPR$3])
++- Exchange(distribution=[hash[f0]])
+ +- Calc(select=[f0, rt, f1, f2, JSON_STRING(f0) AS $f4])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testGroupJsonObjectAggWithOtherAggs[batchMode = true,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, COUNT(*), JSON_OBJECTAGG(f1 VALUE f0), SUM(f2) FROM
T GROUP BY f0]]>
</Resource>
@@ -137,7 +250,31 @@ SortAggregate(isMerge=[true], groupBy=[f0], select=[f0,
Final_COUNT(count1$0) AS
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode =
false]">
+ <TestCase name="testGroupJsonObjectAggWithOtherAggs[batchMode = true,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, COUNT(*), JSON_OBJECTAGG(f1 VALUE f0), SUM(f2) FROM
T GROUP BY f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[COUNT()],
EXPR$2=[JSON_OBJECTAGG_NULL_ON_NULL($2, $0)], EXPR$3=[SUM($3)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)],
f1=[$1], f2=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt, 5000)],
select=[f0, Final_COUNT(count1$0) AS EXPR$1,
Final_JSON_OBJECTAGG_NULL_ON_NULL(EXPR$2) AS EXPR$2, Final_SUM(sum$1) AS
EXPR$3])
++- Sort(orderBy=[f0 ASC, assignedWindow$ ASC])
+ +- Exchange(distribution=[hash[f0]])
+ +- LocalSortWindowAggregate(groupBy=[f0],
window=[TumblingGroupWindow('w$, rt, 5000)], select=[f0, Partial_COUNT(*) AS
count1$0, Partial_JSON_OBJECTAGG_NULL_ON_NULL(f1, $f4) AS EXPR$2,
Partial_SUM(f2) AS sum$1])
+ +- Calc(select=[f0, rt, f1, f2, JSON_STRING(f0) AS $f4])
+ +- Sort(orderBy=[f0 ASC, rt ASC])
+ +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode =
false, isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT MAX(f0), JSON_OBJECTAGG(f1 VALUE f0), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f0) FROM T]]>
</Resource>
@@ -157,7 +294,30 @@ GroupAggregate(select=[MAX(f0) AS EXPR$0,
JSON_OBJECTAGG_NULL_ON_NULL(f1, $f2) A
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode =
true]">
+ <TestCase name="testJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode =
false, isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT MAX(f0), JSON_OBJECTAGG(f1 VALUE f0), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f0) FROM T GROUP BY TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4])
++- LogicalAggregate(group=[{0}], EXPR$0=[MAX($1)],
EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($2, $1)],
EXPR$2=[JSON_ARRAYAGG_ABSENT_ON_NULL($2)],
EXPR$3=[JSON_ARRAYAGG_ABSENT_ON_NULL($1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f0=[$0], f1=[$1])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[MAX(f0) AS EXPR$0, JSON_OBJECTAGG_NULL_ON_NULL(f1, $f3) AS EXPR$1,
JSON_ARRAYAGG_ABSENT_ON_NULL($f4) AS EXPR$2, JSON_ARRAYAGG_ABSENT_ON_NULL($f3)
AS EXPR$3])
++- Exchange(distribution=[single])
+ +- Calc(select=[rt, f0, f1, JSON_STRING(f0) AS $f3, JSON_STRING(f1) AS $f4])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f0, f1], metadata=[]]], fields=[rt, f0, f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode =
true, isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT MAX(f0), JSON_OBJECTAGG(f1 VALUE f0), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f0) FROM T]]>
</Resource>
@@ -177,7 +337,29 @@ SortAggregate(isMerge=[false], select=[MAX(f0) AS EXPR$0,
JSON_OBJECTAGG_NULL_ON
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAggInGroupWindow[batchMode = false]">
+ <TestCase name="testJsonArrayAggAndJsonObjectAggWithOtherAggs[batchMode =
true, isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT MAX(f0), JSON_OBJECTAGG(f1 VALUE f0), JSON_ARRAYAGG(f1),
JSON_ARRAYAGG(f0) FROM T GROUP BY TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1], EXPR$1=[$2], EXPR$2=[$3], EXPR$3=[$4])
++- LogicalAggregate(group=[{0}], EXPR$0=[MAX($1)],
EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($2, $1)],
EXPR$2=[JSON_ARRAYAGG_ABSENT_ON_NULL($2)],
EXPR$3=[JSON_ARRAYAGG_ABSENT_ON_NULL($1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f0=[$0], f1=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[MAX(f0) AS EXPR$0, JSON_OBJECTAGG_NULL_ON_NULL(f1, $f3) AS EXPR$1,
JSON_ARRAYAGG_ABSENT_ON_NULL($f4) AS EXPR$2, JSON_ARRAYAGG_ABSENT_ON_NULL($f3)
AS EXPR$3])
++- Calc(select=[rt, f0, f1, JSON_STRING(f0) AS $f3, JSON_STRING(f1) AS $f4])
+ +- Sort(orderBy=[rt ASC])
+ +- Exchange(distribution=[single])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f0, f1], metadata=[]]], fields=[rt, f0, f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAggInGroupWindow[batchMode = false,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, JSON_ARRAYAGG(f0) FROM T GROUP BY f0]]>
</Resource>
@@ -197,7 +379,30 @@ GroupAggregate(groupBy=[f0], select=[f0,
JSON_ARRAYAGG_ABSENT_ON_NULL($f1) AS EX
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAggInGroupWindow[batchMode = true]">
+ <TestCase name="testJsonArrayAggInGroupWindow[batchMode = false,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, JSON_ARRAYAGG(f0) FROM T GROUP BY f0,TUMBLE(rt,
INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[JSON_ARRAYAGG_ABSENT_ON_NULL($0)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt,
5000)], select=[f0, JSON_ARRAYAGG_ABSENT_ON_NULL($f2) AS EXPR$1])
++- Exchange(distribution=[hash[f0]])
+ +- Calc(select=[f0, rt, JSON_STRING(f0) AS $f2])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f0, rt], metadata=[]]], fields=[f0, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAggInGroupWindow[batchMode = true,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, JSON_ARRAYAGG(f0) FROM T GROUP BY f0]]>
</Resource>
@@ -218,7 +423,29 @@ SortAggregate(isMerge=[false], groupBy=[f0], select=[f0,
JSON_ARRAYAGG_ABSENT_ON
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAggWithOtherAggs[batchMode = false]">
+ <TestCase name="testJsonArrayAggInGroupWindow[batchMode = true,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, JSON_ARRAYAGG(f0) FROM T GROUP BY f0,TUMBLE(rt,
INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[JSON_ARRAYAGG_ABSENT_ON_NULL($0)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt, 5000)],
select=[f0, JSON_ARRAYAGG_ABSENT_ON_NULL($f2) AS EXPR$1])
++- Calc(select=[f0, rt, JSON_STRING(f0) AS $f2])
+ +- Sort(orderBy=[f0 ASC, rt ASC])
+ +- Exchange(distribution=[hash[f0]])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f0, rt], metadata=[]]], fields=[f0, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAggWithOtherAggs[batchMode = false,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT COUNT(*), JSON_ARRAYAGG(f0) FROM T]]>
</Resource>
@@ -238,7 +465,30 @@ GroupAggregate(select=[COUNT(*) AS EXPR$0,
JSON_ARRAYAGG_ABSENT_ON_NULL($f1) AS
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAggWithOtherAggs[batchMode = true]">
+ <TestCase name="testJsonArrayAggWithOtherAggs[batchMode = false,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT COUNT(*), JSON_ARRAYAGG(f0) FROM T GROUP BY TUMBLE(rt,
INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1], EXPR$1=[$2])
++- LogicalAggregate(group=[{0}], EXPR$0=[COUNT()],
EXPR$1=[JSON_ARRAYAGG_ABSENT_ON_NULL($1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f0=[$0])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[COUNT(*) AS EXPR$0, JSON_ARRAYAGG_ABSENT_ON_NULL($f2) AS EXPR$1])
++- Exchange(distribution=[single])
+ +- Calc(select=[rt, f0, JSON_STRING(f0) AS $f2])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f0], metadata=[]]], fields=[rt, f0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAggWithOtherAggs[batchMode = true,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT COUNT(*), JSON_ARRAYAGG(f0) FROM T]]>
</Resource>
@@ -258,7 +508,29 @@ SortAggregate(isMerge=[false], select=[COUNT(*) AS EXPR$0,
JSON_ARRAYAGG_ABSENT_
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAgg[batchMode = false]">
+ <TestCase name="testJsonArrayAggWithOtherAggs[batchMode = true,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT COUNT(*), JSON_ARRAYAGG(f0) FROM T GROUP BY TUMBLE(rt,
INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1], EXPR$1=[$2])
++- LogicalAggregate(group=[{0}], EXPR$0=[COUNT()],
EXPR$1=[JSON_ARRAYAGG_ABSENT_ON_NULL($1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f0=[$0])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[COUNT(*) AS EXPR$0, JSON_ARRAYAGG_ABSENT_ON_NULL($f2) AS EXPR$1])
++- Calc(select=[rt, f0, JSON_STRING(f0) AS $f2])
+ +- Sort(orderBy=[rt ASC])
+ +- Exchange(distribution=[single])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f0], metadata=[]]], fields=[rt, f0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAgg[batchMode = false, isGroupWindowAgg =
false]">
<Resource name="sql">
<![CDATA[SELECT JSON_ARRAYAGG(f0) FROM T]]>
</Resource>
@@ -278,7 +550,30 @@ GroupAggregate(select=[JSON_ARRAYAGG_ABSENT_ON_NULL($f1)
AS EXPR$0])
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonArrayAgg[batchMode = true]">
+ <TestCase name="testJsonArrayAgg[batchMode = false, isGroupWindowAgg =
true]">
+ <Resource name="sql">
+ <![CDATA[SELECT JSON_ARRAYAGG(f0) FROM T GROUP BY TUMBLE(rt, INTERVAL
'5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1])
++- LogicalAggregate(group=[{0}], EXPR$0=[JSON_ARRAYAGG_ABSENT_ON_NULL($1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f0=[$0])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[JSON_ARRAYAGG_ABSENT_ON_NULL($f2) AS EXPR$0])
++- Exchange(distribution=[single])
+ +- Calc(select=[rt, f0, JSON_STRING(f0) AS $f2])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f0], metadata=[]]], fields=[rt, f0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonArrayAgg[batchMode = true, isGroupWindowAgg =
false]">
<Resource name="sql">
<![CDATA[SELECT JSON_ARRAYAGG(f0) FROM T]]>
</Resource>
@@ -298,7 +593,29 @@ SortAggregate(isMerge=[false],
select=[JSON_ARRAYAGG_ABSENT_ON_NULL($f1) AS EXPR
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonObjectAggInGroupWindow[batchMode = false]">
+ <TestCase name="testJsonArrayAgg[batchMode = true, isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT JSON_ARRAYAGG(f0) FROM T GROUP BY TUMBLE(rt, INTERVAL
'5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1])
++- LogicalAggregate(group=[{0}], EXPR$0=[JSON_ARRAYAGG_ABSENT_ON_NULL($1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f0=[$0])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[JSON_ARRAYAGG_ABSENT_ON_NULL($f2) AS EXPR$0])
++- Calc(select=[rt, f0, JSON_STRING(f0) AS $f2])
+ +- Sort(orderBy=[rt ASC])
+ +- Exchange(distribution=[single])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f0], metadata=[]]], fields=[rt, f0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAggInGroupWindow[batchMode = false,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f0) FROM T GROUP BY f0]]>
</Resource>
@@ -318,7 +635,30 @@ GroupAggregate(groupBy=[f0], select=[f0,
JSON_OBJECTAGG_NULL_ON_NULL(f1, $f2) AS
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonObjectAggInGroupWindow[batchMode = true]">
+ <TestCase name="testJsonObjectAggInGroupWindow[batchMode = false,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f0) FROM T GROUP BY
f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($2,
$0)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)], f1=[$1])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt,
5000)], select=[f0, JSON_OBJECTAGG_NULL_ON_NULL(f1, $f3) AS EXPR$1])
++- Exchange(distribution=[hash[f0]])
+ +- Calc(select=[f0, rt, f1, JSON_STRING(f0) AS $f3])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f0, rt, f1], metadata=[]]], fields=[f0, rt, f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAggInGroupWindow[batchMode = true,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f0) FROM T GROUP BY f0]]>
</Resource>
@@ -341,7 +681,31 @@ SortAggregate(isMerge=[true], groupBy=[f0], select=[f0,
Final_JSON_OBJECTAGG_NUL
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonObjectAggWithOtherAggs[batchMode = false]">
+ <TestCase name="testJsonObjectAggInGroupWindow[batchMode = true,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT f0, JSON_OBJECTAGG(f1 VALUE f0) FROM T GROUP BY
f0,TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(f0=[$0], EXPR$1=[$2])
++- LogicalAggregate(group=[{0, 1}], EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($2,
$0)])
+ +- LogicalProject(f0=[$0], $f1=[$TUMBLE($3, 5000:INTERVAL SECOND)], f1=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(groupBy=[f0], window=[TumblingGroupWindow('w$, rt, 5000)],
select=[f0, Final_JSON_OBJECTAGG_NULL_ON_NULL(EXPR$1) AS EXPR$1])
++- Sort(orderBy=[f0 ASC, assignedWindow$ ASC])
+ +- Exchange(distribution=[hash[f0]])
+ +- LocalSortWindowAggregate(groupBy=[f0],
window=[TumblingGroupWindow('w$, rt, 5000)], select=[f0,
Partial_JSON_OBJECTAGG_NULL_ON_NULL(f1, $f3) AS EXPR$1])
+ +- Calc(select=[f0, rt, f1, JSON_STRING(f0) AS $f3])
+ +- Sort(orderBy=[f0 ASC, rt ASC])
+ +- TableSourceScan(table=[[default_catalog, default_database,
T, project=[f0, rt, f1], metadata=[]]], fields=[f0, rt, f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAggWithOtherAggs[batchMode = false,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT COUNT(*), JSON_OBJECTAGG(f1 VALUE f1) FROM T]]>
</Resource>
@@ -361,7 +725,30 @@ GroupAggregate(select=[COUNT(*) AS EXPR$0,
JSON_OBJECTAGG_NULL_ON_NULL($f1, $f1)
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonObjectAggWithOtherAggs[batchMode = true]">
+ <TestCase name="testJsonObjectAggWithOtherAggs[batchMode = false,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT COUNT(*), JSON_OBJECTAGG(f1 VALUE f1) FROM T GROUP BY
TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1], EXPR$1=[$2])
++- LogicalAggregate(group=[{0}], EXPR$0=[COUNT()],
EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($1, $1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f1=[$1])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[COUNT(*) AS EXPR$0, JSON_OBJECTAGG_NULL_ON_NULL($f2, $f2) AS EXPR$1])
++- Exchange(distribution=[single])
+ +- Calc(select=[rt, f1, JSON_STRING(f1) AS $f2])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f1], metadata=[]]], fields=[rt, f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAggWithOtherAggs[batchMode = true,
isGroupWindowAgg = false]">
<Resource name="sql">
<![CDATA[SELECT COUNT(*), JSON_OBJECTAGG(f1 VALUE f1) FROM T]]>
</Resource>
@@ -382,7 +769,80 @@ SortAggregate(isMerge=[true],
select=[Final_COUNT(count1$0) AS EXPR$0, Final_JSO
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonObjectAgg[batchMode = false]">
+ <TestCase name="testJsonObjectAggWithOtherAggs[batchMode = true,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT COUNT(*), JSON_OBJECTAGG(f1 VALUE f1) FROM T GROUP BY
TUMBLE(rt, INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1], EXPR$1=[$2])
++- LogicalAggregate(group=[{0}], EXPR$0=[COUNT()],
EXPR$1=[JSON_OBJECTAGG_NULL_ON_NULL($1, $1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f1=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[Final_COUNT(count1$0) AS EXPR$0,
Final_JSON_OBJECTAGG_NULL_ON_NULL(EXPR$1) AS EXPR$1])
++- Sort(orderBy=[assignedWindow$ ASC])
+ +- Exchange(distribution=[single])
+ +- LocalSortWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[Partial_COUNT(*) AS count1$0, Partial_JSON_OBJECTAGG_NULL_ON_NULL($f2,
$f2) AS EXPR$1])
+ +- Calc(select=[rt, f1, JSON_STRING(f1) AS $f2])
+ +- Sort(orderBy=[rt ASC])
+ +- TableSourceScan(table=[[default_catalog, default_database,
T, project=[rt, f1], metadata=[]]], fields=[rt, f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAggWithWindowTVF[batchMode = false,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT JSON_OBJECTAGG(f1 VALUE f1) FROM TABLE(TUMBLE(TABLE T,
DESCRIPTOR(rt), INTERVAL '5' SECOND))]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{}], EXPR$0=[JSON_OBJECTAGG_NULL_ON_NULL($0, $0)])
++- LogicalProject(f1=[$1])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rt'), 5000:INTERVAL SECOND)], rowType=[RecordType(INTEGER
f0, VARCHAR(2147483647) f1, BIGINT f2, TIMESTAMP(3) *ROWTIME* rt, TIMESTAMP(3)
window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) *ROWTIME* window_time)])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], rt=[$3])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3,
1000:INTERVAL SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupAggregate(select=[JSON_OBJECTAGG_NULL_ON_NULL($f1, $f1) AS EXPR$0])
++- Exchange(distribution=[single])
+ +- Calc(select=[f1, JSON_STRING(f1) AS $f1])
+ +- WindowTableFunction(window=[TUMBLE(time_col=[rt], size=[5 s])])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f1, rt], metadata=[]]], fields=[f1, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAggWithWindowTVF[batchMode = true,
isGroupWindowAgg = true]">
+ <Resource name="sql">
+ <![CDATA[SELECT JSON_OBJECTAGG(f1 VALUE f1) FROM TABLE(TUMBLE(TABLE T,
DESCRIPTOR(rt), INTERVAL '5' SECOND))]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalAggregate(group=[{}], EXPR$0=[JSON_OBJECTAGG_NULL_ON_NULL($0, $0)])
++- LogicalProject(f1=[$1])
+ +- LogicalTableFunctionScan(invocation=[TUMBLE(TABLE(#0),
DESCRIPTOR(_UTF-16LE'rt'), 5000:INTERVAL SECOND)], rowType=[RecordType(INTEGER
f0, VARCHAR(2147483647) f1, BIGINT f2, TIMESTAMP(3) rt, TIMESTAMP(3)
window_start, TIMESTAMP(3) window_end, TIMESTAMP(3) window_time)])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], rt=[$3])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortAggregate(isMerge=[true],
select=[Final_JSON_OBJECTAGG_NULL_ON_NULL(EXPR$0) AS EXPR$0])
++- Exchange(distribution=[single])
+ +- LocalSortAggregate(select=[Partial_JSON_OBJECTAGG_NULL_ON_NULL($f1, $f1)
AS EXPR$0])
+ +- Calc(select=[f1, JSON_STRING(f1) AS $f1])
+ +- WindowTableFunction(window=[TUMBLE(time_col=[rt], size=[5 s])])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f1, rt], metadata=[]]], fields=[f1, rt])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAgg[batchMode = false, isGroupWindowAgg =
false]">
<Resource name="sql">
<![CDATA[SELECT JSON_OBJECTAGG(f1 VALUE f1) FROM T]]>
</Resource>
@@ -402,7 +862,30 @@ GroupAggregate(select=[JSON_OBJECTAGG_NULL_ON_NULL($f1,
$f1) AS EXPR$0])
]]>
</Resource>
</TestCase>
- <TestCase name="testJsonObjectAgg[batchMode = true]">
+ <TestCase name="testJsonObjectAgg[batchMode = false, isGroupWindowAgg =
true]">
+ <Resource name="sql">
+ <![CDATA[SELECT JSON_OBJECTAGG(f1 VALUE f1) FROM T GROUP BY TUMBLE(rt,
INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1])
++- LogicalAggregate(group=[{0}], EXPR$0=[JSON_OBJECTAGG_NULL_ON_NULL($1, $1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f1=[$1])
+ +- LogicalWatermarkAssigner(rowtime=[rt], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+GroupWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[JSON_OBJECTAGG_NULL_ON_NULL($f2, $f2) AS EXPR$0])
++- Exchange(distribution=[single])
+ +- Calc(select=[rt, f1, JSON_STRING(f1) AS $f2])
+ +- WatermarkAssigner(rowtime=[rt], watermark=[-(rt, 1000:INTERVAL
SECOND)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T,
project=[rt, f1], metadata=[]]], fields=[rt, f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAgg[batchMode = true, isGroupWindowAgg =
false]">
<Resource name="sql">
<![CDATA[SELECT JSON_OBJECTAGG(f1 VALUE f1) FROM T]]>
</Resource>
@@ -420,6 +903,30 @@ SortAggregate(isMerge=[true],
select=[Final_JSON_OBJECTAGG_NULL_ON_NULL(EXPR$0)
+- LocalSortAggregate(select=[Partial_JSON_OBJECTAGG_NULL_ON_NULL($f1, $f1)
AS EXPR$0])
+- Calc(select=[f1, JSON_STRING(f1) AS $f1])
+- TableSourceScan(table=[[default_catalog, default_database, T,
project=[f1], metadata=[]]], fields=[f1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJsonObjectAgg[batchMode = true, isGroupWindowAgg =
true]">
+ <Resource name="sql">
+ <![CDATA[SELECT JSON_OBJECTAGG(f1 VALUE f1) FROM T GROUP BY TUMBLE(rt,
INTERVAL '5' SECOND)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[$1])
++- LogicalAggregate(group=[{0}], EXPR$0=[JSON_OBJECTAGG_NULL_ON_NULL($1, $1)])
+ +- LogicalProject($f0=[$TUMBLE($3, 5000:INTERVAL SECOND)], f1=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+SortWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[Final_JSON_OBJECTAGG_NULL_ON_NULL(EXPR$0) AS EXPR$0])
++- Sort(orderBy=[assignedWindow$ ASC])
+ +- Exchange(distribution=[single])
+ +- LocalSortWindowAggregate(window=[TumblingGroupWindow('w$, rt, 5000)],
select=[Partial_JSON_OBJECTAGG_NULL_ON_NULL($f2, $f2) AS EXPR$0])
+ +- Calc(select=[rt, f1, JSON_STRING(f1) AS $f2])
+ +- Sort(orderBy=[rt ASC])
+ +- TableSourceScan(table=[[default_catalog, default_database,
T, project=[rt, f1], metadata=[]]], fields=[rt, f1])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
index d210e06b43f..0837f65b3e2 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
@@ -1834,6 +1834,7 @@ class FlinkRelMdHandlerTestBase {
val logicalWindowAgg = new LogicalWindowAggregate(
ts.getCluster,
ts.getTraitSet,
+ util.List.of(),
project,
ImmutableBitSet.of(0, 1),
aggCallOfWindowAgg,
@@ -1845,6 +1846,7 @@ class FlinkRelMdHandlerTestBase {
val flinkLogicalWindowAgg = new FlinkLogicalWindowAggregate(
ts.getCluster,
logicalTraits,
+ util.List.of(),
new FlinkLogicalCalc(ts.getCluster, flinkLogicalTraits, flinkLogicalTs,
program),
ImmutableBitSet.of(0, 1),
aggCallOfWindowAgg,
@@ -2004,6 +2006,7 @@ class FlinkRelMdHandlerTestBase {
val logicalWindowAgg = new LogicalWindowAggregate(
ts.getCluster,
ts.getTraitSet,
+ util.List.of(),
project,
ImmutableBitSet.of(1),
aggCallOfWindowAgg,
@@ -2015,6 +2018,7 @@ class FlinkRelMdHandlerTestBase {
val flinkLogicalWindowAgg = new FlinkLogicalWindowAggregate(
ts.getCluster,
logicalTraits,
+ util.List.of(),
new FlinkLogicalCalc(ts.getCluster, flinkLogicalTraits, flinkLogicalTs,
program),
ImmutableBitSet.of(1),
aggCallOfWindowAgg,
@@ -2187,6 +2191,7 @@ class FlinkRelMdHandlerTestBase {
val logicalWindowAggWithAuxGroup = new LogicalWindowAggregate(
ts.getCluster,
ts.getTraitSet,
+ util.List.of(),
project,
ImmutableBitSet.of(0),
aggCallOfWindowAgg,
@@ -2198,6 +2203,7 @@ class FlinkRelMdHandlerTestBase {
val flinkLogicalWindowAggWithAuxGroup = new FlinkLogicalWindowAggregate(
ts.getCluster,
logicalTraits,
+ util.List.of(),
new FlinkLogicalCalc(ts.getCluster, flinkLogicalTraits, flinkLogicalTs,
program),
ImmutableBitSet.of(0),
aggCallOfWindowAgg,
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCountTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCountTest.scala
index ad44ac902db..8985ae6e69c 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCountTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCountTest.scala
@@ -28,6 +28,8 @@ import org.apache.calcite.util.ImmutableBitSet
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
+import java.util
+
class FlinkRelMdRowCountTest extends FlinkRelMdHandlerTestBase {
@Test
@@ -183,6 +185,7 @@ class FlinkRelMdRowCountTest extends
FlinkRelMdHandlerTestBase {
val windowAgg = new LogicalWindowAggregate(
ts.getCluster,
ts.getTraitSet,
+ util.List.of(),
ts,
ImmutableBitSet.of(0, 1),
aggCallOfWindowAgg,
