You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2025/03/20 02:43:42 UTC
(flink) 03/03: [FLINK-37222][table] Don't resolve the CatalogView to QueryOperationCatalogView
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 73c556c756897ae7825dd76b6adfaba759e16a7f
Author: Shengkai <10...@qq.com>
AuthorDate: Tue Mar 18 16:29:15 2025 +0800
[FLINK-37222][table] Don't resolve the CatalogView to QueryOperationCatalogView
---
.../src/test/resources/nexmark.sql | 40 ++++++++++++-
.../EnvironmentReusableInMemoryCatalog.java | 68 ----------------------
.../gateway/service/context/SessionContext.java | 3 +-
.../operations/converters/SqlNodeConvertUtils.java | 15 +++--
.../analyze/NonDeterministicUpdateAnalyzerTest.xml | 27 +++++----
.../table/planner/catalog/JavaCatalogTableTest.xml | 6 +-
.../plan/stream/sql/NonDeterministicDagTest.xml | 36 +++++++-----
.../plan/stream/sql/agg/WindowAggregateTest.xml | 36 +++++-------
8 files changed, 101 insertions(+), 130 deletions(-)
diff --git a/flink-table/flink-sql-client/src/test/resources/nexmark.sql b/flink-table/flink-sql-client/src/test/resources/nexmark.sql
index 5475f471dd9..adbaa918f92 100644
--- a/flink-table/flink-sql-client/src/test/resources/nexmark.sql
+++ b/flink-table/flink-sql-client/src/test/resources/nexmark.sql
@@ -56,7 +56,7 @@ CREATE TABLE datagen
'connector' = 'datagen',
'number-of-rows' = '10'
);
-CREATE VIEW person AS
+CREATE TEMPORARY VIEW person AS
SELECT person.id,
person.name,
person.emailAddress,
@@ -93,7 +93,6 @@ SELECT bid.auction,
FROM datagen
WHERE event_type = 2;
-
CREATE TABLE nexmark_q7
(
auction BIGINT,
@@ -105,6 +104,9 @@ CREATE TABLE nexmark_q7
'connector' = 'blackhole'
);
+CREATE TABLE nexmark_q8 (id BIGINT, name VARCHAR, stime TIMESTAMP(3)) WITH ('connector' = 'blackhole');
+
+BEGIN STATEMENT SET;
INSERT INTO nexmark_q7
SELECT B.auction, B.price, B.bidder, B.`dateTime`, B.extra
from bid B
@@ -114,3 +116,37 @@ from bid B
GROUP BY window_start, window_end) B1
ON B.price = B1.maxprice
WHERE B.`dateTime` BETWEEN B1.`dateTime` - INTERVAL '10' SECOND AND B1.`dateTime`;
+
+INSERT INTO nexmark_q8
+ SELECT
+ P.id, P.name, P.starttime
+ FROM
+ (
+ SELECT
+ P.id,
+ P.name,
+ TUMBLE_START(P.dateTime, INTERVAL '10' SECOND) AS starttime,
+ TUMBLE_END(P.dateTime, INTERVAL '10' SECOND) AS endtime
+ FROM
+ person P
+ GROUP BY
+ P.id,
+ P.name,
+ TUMBLE(P.dateTime, INTERVAL '10' SECOND)
+ ) P
+ JOIN (
+ SELECT
+ A.seller,
+ TUMBLE_START(A.dateTime, INTERVAL '10' SECOND) AS starttime,
+ TUMBLE_END(A.dateTime, INTERVAL '10' SECOND) AS endtime
+ FROM
+ auction A
+ GROUP BY
+ A.seller,
+ TUMBLE(A.dateTime, INTERVAL '10' SECOND)
+ ) A
+ ON P.id = A.seller
+ AND P.starttime = A.starttime
+ AND P.endtime = A.endtime;
+END;
+
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/EnvironmentReusableInMemoryCatalog.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/EnvironmentReusableInMemoryCatalog.java
deleted file mode 100644
index ae8efc1a3e7..00000000000
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/EnvironmentReusableInMemoryCatalog.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.table.gateway.service.context;
-
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.CatalogView;
-import org.apache.flink.table.catalog.GenericInMemoryCatalog;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.QueryOperationCatalogView;
-import org.apache.flink.table.catalog.ResolvedCatalogView;
-import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
-import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
-
-import java.util.Optional;
-
-/**
- * An in-memory catalog that can be reused across different {@link TableEnvironment}. The SQL client
- * works against {@link TableEnvironment} design and reuses some of the components (e.g.
- * CatalogManager), but not all (e.g. Planner) which causes e.g. views registered in an in-memory
- * catalog to fail. This class is a workaround not to keep Planner bound parts of a view reused
- * across different {@link TableEnvironment}.
- */
-public class EnvironmentReusableInMemoryCatalog extends GenericInMemoryCatalog {
- public EnvironmentReusableInMemoryCatalog(String name, String defaultDatabase) {
- super(name, defaultDatabase);
- }
-
- @Override
- public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws TableAlreadyExistException, DatabaseNotExistException {
- CatalogBaseTable tableToRegister =
- extractView(table)
- .flatMap(QueryOperationCatalogView::getOriginalView)
- .map(v -> (CatalogBaseTable) v)
- .orElse(table);
- super.createTable(tablePath, tableToRegister, ignoreIfExists);
- }
-
- private Optional<QueryOperationCatalogView> extractView(CatalogBaseTable table) {
- if (table instanceof ResolvedCatalogView) {
- final CatalogView origin = ((ResolvedCatalogView) table).getOrigin();
- if (origin instanceof QueryOperationCatalogView) {
- return Optional.of((QueryOperationCatalogView) origin);
- }
- return Optional.empty();
- } else if (table instanceof QueryOperationCatalogView) {
- return Optional.of((QueryOperationCatalogView) table);
- }
- return Optional.empty();
- }
-}
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
index 602a1f027fd..cf1597ecea9 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java
@@ -30,6 +30,7 @@ import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.CatalogStoreHolder;
import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
import org.apache.flink.table.factories.CatalogStoreFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.TableFactoryUtil;
@@ -447,7 +448,7 @@ public class SessionContext {
catalogStore.config(),
catalogStore.classLoader()))
.orElse(
- new EnvironmentReusableInMemoryCatalog(
+ new GenericInMemoryCatalog(
defaultCatalogName, settings.getBuiltInDatabaseName()));
}
defaultCatalog.open();
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
index 991bbf744b5..dbdf700109a 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/converters/SqlNodeConvertUtils.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.ContextResolvedTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogView;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.planner.operations.PlannerQueryOperation;
@@ -110,12 +111,14 @@ class SqlNodeConvertUtils {
schema = ResolvedSchema.physical(aliasFieldNames, schema.getColumnDataTypes());
}
- return CatalogView.of(
- Schema.newBuilder().fromResolvedSchema(schema).build(),
- viewComment,
- originalQuery,
- expandedQuery,
- viewOptions);
+ return new ResolvedCatalogView(
+ CatalogView.of(
+ Schema.newBuilder().fromResolvedSchema(schema).build(),
+ viewComment,
+ originalQuery,
+ expandedQuery,
+ viewOptions),
+ schema);
}
/**
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml
index a2bb0a112ab..ce4b07b9c67 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml
@@ -177,31 +177,30 @@ Calc(select=[a, b, DATE_FORMAT(CURRENT_TIMESTAMP(), _UTF-16LE'yyMMdd') AS day],
Sink(table=[default_catalog.default_database.sink1], fields=[a, day, EXPR$2, EXPR$3])
+- GroupAggregate(advice=[1], groupBy=[a, day], select=[a, day, SUM_RETRACT(b) AS EXPR$2, COUNT_RETRACT(DISTINCT c) AS EXPR$3])
+- Exchange(distribution=[hash[a, day]])
- +- Calc(select=[a, day, b0 AS b, c])
- +- Join(joinType=[InnerJoin], where=[=(a, d)], select=[a, b, day, b0, day0, c, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ +- Calc(select=[a, day, b, c])
+ +- Join(joinType=[InnerJoin], where=[=(a, d)], select=[a, day, b, c, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
:- Exchange(distribution=[hash[a]])
- : +- Calc(select=[a, b, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day])
- : +- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a, b], metadata=[]]], fields=[a, b])
+ : +- Calc(select=[a, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day])
+ : +- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a], metadata=[]]], fields=[a])
+- Exchange(distribution=[hash[d]])
- +- Calc(select=[b, CONCAT(c, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd')) AS day, c, d])
- +- TableSourceScan(table=[[default_catalog, default_database, src2, project=[b, c, d], metadata=[]]], fields=[b, c, d])
+ +- TableSourceScan(table=[[default_catalog, default_database, src2, project=[b, c, d], metadata=[]]], fields=[b, c, d])
-Sink(table=[default_catalog.default_database.sink2], fields=[a, day, b0, c])
-+- Calc(select=[a, day, b0, c], where=[>(b0, 100)])
- +- Join(joinType=[InnerJoin], where=[=(a, d)], select=[a, b, day, b0, day0, c, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+Sink(table=[default_catalog.default_database.sink2], fields=[a, day, b, c])
++- Calc(select=[a, day, b, c])
+ +- Join(joinType=[InnerJoin], where=[=(a, d)], select=[a, day, b, c, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
:- Exchange(distribution=[hash[a]])
- : +- Calc(select=[a, b, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day])
- : +- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a, b], metadata=[]]], fields=[a, b])
+ : +- Calc(select=[a, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day])
+ : +- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a], metadata=[]]], fields=[a])
+- Exchange(distribution=[hash[d]])
- +- Calc(select=[b, CONCAT(c, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd')) AS day, c, d])
+ +- Calc(select=[b, c, d], where=[>(b, 100)])
+- TableSourceScan(table=[[default_catalog, default_database, src2, project=[b, c, d], metadata=[]]], fields=[b, c, d])
advice[1]: [ADVICE] You might want to enable local-global two-phase optimization by configuring ('table.exec.mini-batch.enabled' to 'true', 'table.exec.mini-batch.allow-latency' to a positive long value, 'table.exec.mini-batch.size' to a positive long value).
advice[2]: [WARNING] The column(s): day(generated by non-deterministic function: CURRENT_TIMESTAMP ) can not satisfy the determinism requirement for correctly processing update message('UB'/'UA'/'D' in changelogMode, not 'I' only), this usually happens when input node has no upsertKey(upsertKeys=[{}]) or current node outputs non-deterministic update messages. Please consider removing these non-deterministic columns or making them deterministic by using deterministic functions.
related rel plan:
-Calc(select=[a, b, DATE_FORMAT(CURRENT_TIMESTAMP(), _UTF-16LE'yyMMdd') AS day], changelogMode=[I,UB,UA,D])
-+- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a, b], metadata=[]]], fields=[a, b], changelogMode=[I,UB,UA,D])
+Calc(select=[a, DATE_FORMAT(CURRENT_TIMESTAMP(), _UTF-16LE'yyMMdd') AS day], changelogMode=[I,UB,UA,D])
++- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a], metadata=[]]], fields=[a], changelogMode=[I,UB,UA,D])
]]>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.xml
index a8b689b40b0..3c03f118985 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/catalog/JavaCatalogTableTest.xml
@@ -187,19 +187,19 @@ Calc(select=[EXPR$0, window_start])
<![CDATA[
LogicalProject(order_id=[$0], customer_id=[$1], product_id=[$2], ts=[$3])
+- LogicalProject(order_id=[$0], customer_id=[$1], product_id=[$5], ts=[$4])
- +- LogicalCorrelate(correlation=[$cor3], joinType=[inner], requiredColumns=[{3}])
+ +- LogicalCorrelate(correlation=[$cor2], joinType=[inner], requiredColumns=[{3}])
:- LogicalWatermarkAssigner(rowtime=[ts], watermark=[$4])
: +- LogicalTableScan(table=[[cat, default, t]])
+- LogicalProject(product_id=[$0])
+- Uncollect
- +- LogicalProject(product_ids=[$cor3.product_ids])
+ +- LogicalProject(product_ids=[$cor2.product_ids])
+- LogicalValues(tuples=[[{ 0 }]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[order_id, customer_id, f0 AS product_id, ts])
-+- Correlate(invocation=[$UNNEST_ROWS$1($cor3.product_ids)], correlate=[table($UNNEST_ROWS$1($cor3.product_ids))], select=[order_id,customer_id,product_id,product_ids,ts,f0], rowType=[RecordType(INTEGER order_id, INTEGER customer_id, INTEGER product_id, INTEGER ARRAY product_ids, TIMESTAMP_LTZ(3) *ROWTIME* ts, INTEGER f0)], joinType=[INNER])
++- Correlate(invocation=[$UNNEST_ROWS$1($cor2.product_ids)], correlate=[table($UNNEST_ROWS$1($cor2.product_ids))], select=[order_id,customer_id,product_id,product_ids,ts,f0], rowType=[RecordType(INTEGER order_id, INTEGER customer_id, INTEGER product_id, INTEGER ARRAY product_ids, TIMESTAMP_LTZ(3) *ROWTIME* ts, INTEGER f0)], joinType=[INNER])
+- WatermarkAssigner(rowtime=[ts], watermark=[ts])
+- TableSourceScan(table=[[cat, default, t]], fields=[order_id, customer_id, product_id, product_ids, ts])
]]>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml
index 742207b6bd0..b15149604fe 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml
@@ -2462,23 +2462,28 @@ LogicalSink(table=[default_catalog.default_database.sink2], fields=[a, day, b, c
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Join(joinType=[InnerJoin], where=[(a = d)], select=[a, b, day, b0, day0, c, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])(reuse_id=[1])
-:- Exchange(distribution=[hash[a]])
-: +- Calc(select=[a, b, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day])
-: +- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a, b], metadata=[]]], fields=[a, b])
-+- Exchange(distribution=[hash[d]])
- +- Calc(select=[b, CONCAT(c, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd')) AS day, c, d])
- +- TableSourceScan(table=[[default_catalog, default_database, src2, project=[b, c, d], metadata=[]]], fields=[b, c, d])
+Exchange(distribution=[hash[a]])(reuse_id=[1])
++- Calc(select=[a, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day])
+ +- TableSourceScan(table=[[default_catalog, default_database, src1, project=[a], metadata=[]]], fields=[a])
+
+TableSourceScan(table=[[default_catalog, default_database, src2, project=[b, c, d], metadata=[]]], fields=[b, c, d])(reuse_id=[2])
Sink(table=[default_catalog.default_database.sink1], fields=[a, day, EXPR$2, EXPR$3])
+- GroupAggregate(groupBy=[a, day], select=[a, day, SUM_RETRACT(b) AS EXPR$2, COUNT_RETRACT(DISTINCT c) AS EXPR$3])
+- Exchange(distribution=[hash[a, day]])
- +- Calc(select=[a, day, b0 AS b, c])
- +- Reused(reference_id=[1])
+ +- Calc(select=[a, day, b, c])
+ +- Join(joinType=[InnerJoin], where=[(a = d)], select=[a, day, b, c, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Reused(reference_id=[1])
+ +- Exchange(distribution=[hash[d]])
+ +- Reused(reference_id=[2])
-Sink(table=[default_catalog.default_database.sink2], fields=[a, day, b0, c])
-+- Calc(select=[a, day, b0, c], where=[(b0 > 100)])
- +- Reused(reference_id=[1])
+Sink(table=[default_catalog.default_database.sink2], fields=[a, day, b, c])
++- Calc(select=[a, day, b, c])
+ +- Join(joinType=[InnerJoin], where=[(a = d)], select=[a, day, b, c, d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+ :- Reused(reference_id=[1])
+ +- Exchange(distribution=[hash[d]])
+ +- Calc(select=[b, c, d], where=[(b > 100)])
+ +- Reused(reference_id=[2])
]]>
</Resource>
</TestCase>
@@ -2542,18 +2547,17 @@ LogicalSink(table=[default_catalog.default_database.sink2], fields=[a, b, d])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[id, deepNested_nested2_num AS a, deepNested_nested1_name AS name, ((deepNested_nested1_value + deepNested_nested2_num) + metadata_1) AS b])(reuse_id=[1])
-+- TableSourceScan(table=[[default_catalog, default_database, nested_src, project=[id, deepNested_nested2_num, deepNested_nested1_name, deepNested_nested1_value], metadata=[metadata_1]]], fields=[id, deepNested_nested2_num, deepNested_nested1_name, deepNested_nested1_value, metadata_1])
+TableSourceScan(table=[[default_catalog, default_database, nested_src, project=[deepNested, deepNested_nested1_value, deepNested_nested2_num, metadata_1], metadata=[metadata_1]]], fields=[deepNested, deepNested_nested1_value, deepNested_nested2_num, metadata_1])(reuse_id=[1])
Sink(table=[default_catalog.default_database.sink1], fields=[a, b, d])
+- Calc(select=[a, day AS b, CAST(EXPR$2 AS BIGINT) AS d])
+- GroupAggregate(groupBy=[a, day], select=[a, day, SUM_RETRACT(b) AS EXPR$2])
+- Exchange(distribution=[hash[a, day]])
- +- Calc(select=[a, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day, b])
+ +- Calc(select=[deepNested_nested2_num AS a, DATE_FORMAT(CURRENT_TIMESTAMP(), 'yyMMdd') AS day, ((deepNested_nested1_value + deepNested_nested2_num) + metadata_1) AS b])
+- Reused(reference_id=[1])
Sink(table=[default_catalog.default_database.sink2], fields=[a, b, d])
-+- Calc(select=[a, name AS b, CAST(b AS BIGINT) AS d], where=[(b > 100)])
++- Calc(select=[deepNested.nested2.num AS a, deepNested.nested1.name AS b, CAST(((deepNested.nested1.value + deepNested.nested2.num) + metadata_1) AS BIGINT) AS d], where=[(((deepNested.nested1.value + deepNested.nested2.num) + metadata_1) > 100)])
+- Reused(reference_id=[1])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml
index 9d83070a4f1..75fb934413f 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/agg/WindowAggregateTest.xml
@@ -2209,24 +2209,22 @@ LogicalSink(table=[default_catalog.default_database.s1], fields=[window_start, w
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])(reuse_id=[1])
-+- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, e, rowtime], metadata=[]]], fields=[b, e, rowtime])(reuse_id=[1])
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, wAvg])
+- Calc(select=[window_start, window_end, wAvg])
- +- WindowAggregate(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], select=[weightedAvg(b, e) AS wAvg, start('w$) AS window_start, end('w$) AS window_end])
+ +- WindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[weightedAvg(b, e) AS wAvg, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
- +- Calc(select=[window_start, window_end, b, e])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+- Reused(reference_id=[1])
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, EXPR$2])
+- Calc(select=[window_start, window_end, EXPR$2])
- +- WindowAggregate(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], select=[COUNT(*) AS EXPR$2, start('w$) AS window_start, end('w$) AS window_end])
+ +- WindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[COUNT(*) AS EXPR$2, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
- +- Calc(select=[window_start, window_end])
- +- Reused(reference_id=[1])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+ +- Calc(select=[rowtime])
+ +- Reused(reference_id=[1])
]]>
</Resource>
</TestCase>
@@ -2256,26 +2254,24 @@ LogicalSink(table=[default_catalog.default_database.s1], fields=[window_start, w
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-WindowTableFunction(window=[TUMBLE(time_col=[rowtime], size=[15 min])])(reuse_id=[1])
-+- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
- +- Calc(select=[a, b, c, d, e, rowtime, PROCTIME() AS proctime])
- +- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, d, e, rowtime])
+TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[b, e, rowtime], metadata=[]]], fields=[b, e, rowtime])(reuse_id=[1])
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, wAvg])
+- Calc(select=[window_start, window_end, wAvg])
- +- GlobalWindowAggregate(window=[TUMBLE(win_end=[$window_end], size=[15 min])], select=[weightedAvg(weightedavg$0) AS wAvg, start('w$) AS window_start, end('w$) AS window_end])
+ +- GlobalWindowAggregate(window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[weightedAvg(weightedavg$0) AS wAvg, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
- +- LocalWindowAggregate(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], select=[weightedAvg(b, e) AS weightedavg$0, slice_end('w$) AS $window_end])
- +- Calc(select=[window_start, window_end, b, e])
+ +- LocalWindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[weightedAvg(b, e) AS weightedavg$0, slice_end('w$) AS $slice_end])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+- Reused(reference_id=[1])
Sink(table=[default_catalog.default_database.s1], fields=[window_start, window_end, EXPR$2])
+- Calc(select=[window_start, window_end, EXPR$2])
- +- GlobalWindowAggregate(window=[TUMBLE(win_end=[$window_end], size=[15 min])], select=[COUNT(count1$0) AS EXPR$2, start('w$) AS window_start, end('w$) AS window_end])
+ +- GlobalWindowAggregate(window=[TUMBLE(slice_end=[$slice_end], size=[15 min])], select=[COUNT(count1$0) AS EXPR$2, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[single])
- +- LocalWindowAggregate(window=[TUMBLE(win_start=[window_start], win_end=[window_end], size=[15 min])], select=[COUNT(*) AS count1$0, slice_end('w$) AS $window_end])
- +- Calc(select=[window_start, window_end])
- +- Reused(reference_id=[1])
+ +- LocalWindowAggregate(window=[TUMBLE(time_col=[rowtime], size=[15 min])], select=[COUNT(*) AS count1$0, slice_end('w$) AS $slice_end])
+ +- WatermarkAssigner(rowtime=[rowtime], watermark=[(rowtime - 1000:INTERVAL SECOND)])
+ +- Calc(select=[rowtime])
+ +- Reused(reference_id=[1])
]]>
</Resource>
</TestCase>