You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2023/11/22 14:23:21 UTC
(flink) 01/02: [FLINK-33591][table] Cleanup usage of deprecated TableTestBase#addFunction
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d92423069b0538f58fd6c72fae0952b873c3b352
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Sat Nov 18 03:11:24 2023 +0100
[FLINK-33591][table] Cleanup usage of deprecated TableTestBase#addFunction
This closes #23748
---
.../planner/plan/common/CalcMergeTestBase.java | 3 +-
.../PushFilterIntoTableSourceScanRuleTestBase.java | 2 +-
.../plan/utils/JavaUserDefinedAggFunctions.java | 5 +-
.../runtime/utils/JavaUserDefinedAggFunctions.java | 5 +-
.../utils/JavaUserDefinedScalarFunctions.java | 2 +-
.../planner/plan/batch/sql/DagOptimizationTest.xml | 12 ++--
.../plan/batch/sql/LegacyTableSourceTest.xml | 2 +-
.../planner/plan/batch/sql/RemoveCollationTest.xml | 26 ++++----
.../planner/plan/batch/sql/RemoveShuffleTest.xml | 14 ++--
.../planner/plan/batch/sql/SubplanReuseTest.xml | 24 +++----
.../sql/join/BroadcastHashSemiAntiJoinTest.xml | 20 +++---
.../batch/sql/join/NestedLoopSemiAntiJoinTest.xml | 20 +++---
.../plan/batch/sql/join/SemiAntiJoinTest.xml | 20 +++---
.../sql/join/ShuffledHashSemiAntiJoinTest.xml | 20 +++---
.../batch/sql/join/SortMergeSemiAntiJoinTest.xml | 20 +++---
.../CalcPythonCorrelateTransposeRuleTest.xml | 8 +--
.../ProjectSemiAntiJoinTransposeRuleTest.xml | 4 +-
.../rules/logical/PythonCorrelateSplitRuleTest.xml | 24 +++----
.../SplitPythonConditionFromCorrelateRuleTest.xml | 4 +-
.../logical/subquery/SubQueryAntiJoinTest.xml | 16 ++---
.../logical/subquery/SubQuerySemiJoinTest.xml | 40 ++++++------
.../plan/stream/sql/DagOptimizationTest.xml | 16 ++---
.../plan/stream/sql/LegacyTableSourceTest.xml | 2 +-
.../stream/sql/RelTimeIndicatorConverterTest.xml | 6 +-
.../planner/plan/stream/sql/SubplanReuseTest.xml | 26 ++++----
.../plan/stream/sql/join/SemiAntiJoinTest.xml | 20 +++---
.../plan/stream/table/ColumnFunctionsTest.xml | 6 +-
.../table/validation/AggregateValidationTest.xml | 36 +++++++++++
.../sql/validation/OverWindowValidationTest.scala | 2 +-
.../table/planner/plan/batch/sql/CalcTest.scala | 2 +-
.../plan/batch/sql/DagOptimizationTest.scala | 4 +-
.../plan/batch/sql/LegacyTableSourceTest.scala | 4 +-
.../plan/batch/sql/PartitionableSourceTest.scala | 2 +-
.../plan/batch/sql/RemoveCollationTest.scala | 8 +--
.../planner/plan/batch/sql/RemoveShuffleTest.scala | 6 +-
.../planner/plan/batch/sql/SubplanReuseTest.scala | 4 +-
.../plan/batch/sql/agg/AggregateTestBase.scala | 4 +-
.../plan/batch/sql/agg/GroupWindowTest.scala | 10 +--
.../plan/batch/sql/agg/OverAggregateTest.scala | 4 +-
.../plan/batch/sql/join/LookupJoinTest.scala | 2 +-
.../plan/batch/sql/join/SemiAntiJoinTestBase.scala | 2 +-
.../planner/plan/batch/table/PythonCalcTest.scala | 2 +-
.../table/validation/AggregateValidationTest.scala | 4 +-
.../CalcPythonCorrelateTransposeRuleTest.scala | 4 +-
.../logical/ExpressionReductionRulesTest.scala | 10 +--
.../rules/logical/FlinkCalcMergeRuleTest.scala | 2 +-
.../ProjectSemiAntiJoinTransposeRuleTest.scala | 2 +-
...artitionIntoLegacyTableSourceScanRuleTest.scala | 4 +-
.../logical/PythonCorrelateSplitRuleTest.scala | 6 +-
...SplitPythonConditionFromCorrelateRuleTest.scala | 4 +-
.../SplitPythonConditionFromJoinRuleTest.scala | 2 +-
.../logical/subquery/SubQueryAntiJoinTest.scala | 2 +-
.../logical/subquery/SubQuerySemiJoinTest.scala | 2 +-
.../batch/EnforceLocalSortAggRuleTest.scala | 2 +-
.../table/planner/plan/stream/sql/CalcTest.scala | 2 +-
.../plan/stream/sql/DagOptimizationTest.scala | 4 +-
.../plan/stream/sql/LegacyTableSourceTest.scala | 4 +-
.../plan/stream/sql/ModifiedMonotonicityTest.scala | 2 +-
.../plan/stream/sql/PartitionableSourceTest.scala | 2 +-
.../stream/sql/RelTimeIndicatorConverterTest.scala | 11 +++-
.../planner/plan/stream/sql/SubplanReuseTest.scala | 4 +-
.../plan/stream/sql/agg/OverAggregateTest.scala | 4 +-
.../plan/stream/sql/join/IntervalJoinTest.scala | 2 +-
.../plan/stream/sql/join/SemiAntiJoinTest.scala | 2 +-
.../plan/stream/table/ColumnFunctionsTest.scala | 13 ++--
.../planner/plan/stream/table/PythonCalcTest.scala | 2 +-
.../table/TemporalTableFunctionJoinTest.scala | 2 +-
.../table/validation/AggregateValidationTest.scala | 8 +--
.../table/validation/CalcValidationTest.scala | 4 +-
.../table/validation/CorrelateValidationTest.scala | 24 ++++---
.../table/planner/utils/CountAggFunction.scala | 10 +--
.../flink/table/planner/utils/TableTestBase.scala | 74 ----------------------
72 files changed, 329 insertions(+), 348 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/common/CalcMergeTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/common/CalcMergeTestBase.java
index 728ef1edfdb..8cbf395bde4 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/common/CalcMergeTestBase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/common/CalcMergeTestBase.java
@@ -53,7 +53,8 @@ public abstract class CalcMergeTestBase extends TableTestBase {
+ isBatchMode()
+ "'\n"
+ ")");
- util.addFunction("random_udf", new JavaUserDefinedScalarFunctions.NonDeterministicUdf());
+ util.addTemporarySystemFunction(
+ "random_udf", new JavaUserDefinedScalarFunctions.NonDeterministicUdf());
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTestBase.java
index 435474926b6..3debdf2d481 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTestBase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTestBase.java
@@ -111,7 +111,7 @@ abstract class PushFilterIntoTableSourceScanRuleTestBase extends TableTestBase {
@Test
void testWithUdf() {
- util.addFunction("myUdf", Func1$.MODULE$);
+ util.addTemporarySystemFunction("myUdf", Func1$.MODULE$);
util.verifyRelPlan("SELECT * FROM MyTable WHERE amount > 2 AND myUdf(amount) < 32");
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/JavaUserDefinedAggFunctions.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/JavaUserDefinedAggFunctions.java
index 4e722be8ee3..d1b2fff068a 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/JavaUserDefinedAggFunctions.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/JavaUserDefinedAggFunctions.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.planner.plan.utils;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.api.dataview.ListView;
import org.apache.flink.table.api.dataview.MapView;
import org.apache.flink.table.functions.AggregateFunction;
@@ -297,7 +298,7 @@ public class JavaUserDefinedAggFunctions {
}
// Overloaded accumulate method
- public void accumulate(CountDistinctAccum accumulator, String id) {
+ public void accumulate(CountDistinctAccum accumulator, @DataTypeHint("STRING") String id) {
try {
Integer cnt = accumulator.map.get(id);
if (cnt != null) {
@@ -313,7 +314,7 @@ public class JavaUserDefinedAggFunctions {
}
// Overloaded accumulate method
- public void accumulate(CountDistinctAccum accumulator, long id) {
+ public void accumulate(CountDistinctAccum accumulator, @DataTypeHint("BIGINT") long id) {
try {
Integer cnt = accumulator.map.get(String.valueOf(id));
if (cnt != null) {
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedAggFunctions.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedAggFunctions.java
index 3a9993fb550..8118db3918b 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedAggFunctions.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedAggFunctions.java
@@ -96,7 +96,10 @@ public class JavaUserDefinedAggFunctions {
}
// overloaded accumulate method
- public void accumulate(WeightedAvgAccum accumulator, long iValue, int iWeight) {
+ public void accumulate(
+ WeightedAvgAccum accumulator,
+ @DataTypeHint("BIGINT") long iValue,
+ @DataTypeHint("INT") int iWeight) {
accumulator.sum += iValue * iWeight;
accumulator.count += iWeight;
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
index 5db0abf9cc8..bbd0aa2b478 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
@@ -127,7 +127,7 @@ public class JavaUserDefinedScalarFunctions {
return random.nextInt();
}
- public int eval(int v) {
+ public int eval(@DataTypeHint("INT") int v) {
return v + random.nextInt();
}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.xml
index 7c91bb1b865..2944844181a 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.xml
@@ -627,7 +627,7 @@ LogicalLegacySink(name=[`default_catalog`.`default_database`.`sink1`], fields=[a
: :- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
: : +- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
: : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], total_c=[0])
+- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
+- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
@@ -645,7 +645,7 @@ LogicalLegacySink(name=[`default_catalog`.`default_database`.`sink2`], fields=[a
: :- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
: : +- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
: : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], total_c=[0])
+- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
+- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
@@ -663,8 +663,8 @@ Calc(select=[a, total_c])(reuse_id=[2])
+- HashAggregate(isMerge=[true], groupBy=[a, b, c], select=[a, b, c])
+- Exchange(distribution=[hash[a, b, c]])
+- LocalHashAggregate(groupBy=[a, b, c], select=[a, b, c])
- +- Calc(select=[a, b, f0 AS c])
- +- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[a, b, EXPR$0 AS c])
+ +- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- Calc(select=[a, (b - MOD(b, 300)) AS b, c], where=[((b >= UNIX_TIMESTAMP('${startTime}')) AND (c <> ''))])
+- Reused(reference_id=[1])
@@ -848,13 +848,13 @@ LogicalLegacySink(name=[`default_catalog`.`default_database`.`sink`], fields=[a,
: : :- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
: : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(d, e, f)]]])
: +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(i, j, k, l, m)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
LegacySink(name=[`default_catalog`.`default_database`.`sink`], fields=[a, b, c, d, e, f, i, j, k, l, m, s])
-+- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,d,e,f,i,j,k,l,m,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, INTEGER d, BIGINT e, VARCHAR(2147483647) f, INTEGER i, BIGINT j, INTEGER k, VARCHAR(2147483647) l, BIGINT m, VARCHAR(2147483647) f0)], joinType=[INNER])
++- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,d,e,f,i,j,k,l,m,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, INTEGER d, BIGINT e, VARCHAR(2147483647) f, INTEGER i, BIGINT j, INTEGER k, VARCHAR(2147483647) l, BIGINT m, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- HashJoin(joinType=[InnerJoin], where=[(a = i)], select=[a, b, c, d, e, f, i, j, k, l, m], build=[left])
:- Exchange(distribution=[hash[a]])
: +- HashJoin(joinType=[InnerJoin], where=[(b = e)], select=[a, b, c, d, e, f], build=[right])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.xml
index cbac8411d39..b447d6b1f56 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.xml
@@ -172,7 +172,7 @@ LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[name, id, amount, price], where=[(Func1$(amount) < 32)])
+Calc(select=[name, id, amount, price], where=[(myUdf(amount) < 32)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, FilterableTable, source: [filterPushedDown=[true], filter=[greaterThan(amount, 2)]]]], fields=[name, id, amount, price])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml
index b7cb75e2f9a..50be25b3c72 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml
@@ -134,12 +134,12 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f1=[$3], f=[$4], cnt=[$5])
:- LogicalAggregate(group=[{0}], cnt=[COUNT($0)])
: +- LogicalProject(f=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f1, f, cnt], rightSorted=[true])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f0 AS f1, f, cnt])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,f0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) f0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) AS [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f1, f, cnt], rightSorted=[true])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[EXPR$0 AS f1, f, cnt])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT [...]
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
@@ -170,12 +170,12 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f=[$3], f1=[$4])
:- LogicalAggregate(group=[{0}], cnt=[COUNT($0)])
: +- LogicalProject(f=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f, f1], rightSorted=[true])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f, f0 AS f1])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,f0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) f0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) AS cnt])\n [...]
+MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f, f1], rightSorted=[true])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f, EXPR$0 AS f1])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) [...]
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
@@ -206,7 +206,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f1=[$3])
:- LogicalAggregate(group=[{0}], cnt=[COUNT($0)])
: +- LogicalProject(f=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
@@ -215,8 +215,8 @@ SortMergeJoin(joinType=[InnerJoin], where=[(c = f1)], select=[a, b, c, f1])
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,f0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- SortAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) AS cnt])
+- Exchange(distribution=[forward])
+- Sort(orderBy=[f ASC])
@@ -356,7 +356,7 @@ LogicalProject(id=[$0], tb2_ids=[$2], tb3_ids=[$3], name=[$4], tb2_names=[$6], t
: : : :- LogicalProject(id=[$0], tb2_id=[$5])
: : : : +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: : : : :- LogicalTableScan(table=[[default_catalog, default_database, tb1, source: [TestTableSource(id, key, tb2_ids, tb3_ids, name)]]])
- : : : : +- LogicalTableFunctionScan(invocation=[split($cor0.tb2_ids)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : : : : +- LogicalTableFunctionScan(invocation=[split($cor0.tb2_ids)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
: : : +- LogicalTableScan(table=[[default_catalog, default_database, tb2, source: [TestTableSource(id, name)]]])
: : +- LogicalAggregate(group=[{0}], tb3_names=[LISTAGG($1, $2)])
: : +- LogicalProject(tb1_id=[$0], tb3_name=[$1], $f2=[_UTF-16LE','])
@@ -365,7 +365,7 @@ LogicalProject(id=[$0], tb2_ids=[$2], tb3_ids=[$3], name=[$4], tb2_names=[$6], t
: : :- LogicalProject(id=[$0], tb3_id=[$5])
: : : +- LogicalCorrelate(correlation=[$cor1], joinType=[inner], requiredColumns=[{3}])
: : : :- LogicalTableScan(table=[[default_catalog, default_database, tb1, source: [TestTableSource(id, key, tb2_ids, tb3_ids, name)]]])
- : : : +- LogicalTableFunctionScan(invocation=[split($cor1.tb3_ids)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : : : +- LogicalTableFunctionScan(invocation=[split($cor1.tb3_ids)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
: : +- LogicalTableScan(table=[[default_catalog, default_database, tb3, source: [TestTableSource(id, name)]]])
: +- LogicalTableScan(table=[[default_catalog, default_database, tb4, source: [TestTableSource(id, name)]]])
+- LogicalTableScan(table=[[default_catalog, default_database, tb5, source: [TestTableSource(id, name)]]])
@@ -389,8 +389,8 @@ Calc(select=[id, tb2_ids, tb3_ids, name, tb2_names, tb3_names, name0, name1])
: : +- Calc(select=[id AS tb1_id, name AS tb3_name, ',' AS $f2])
: : +- SortMergeJoin(joinType=[LeftOuterJoin], where=[(tb3_id = id0)], select=[id, tb3_id, id0, name])
: : :- Exchange(distribution=[hash[tb3_id]])
- : : : +- Calc(select=[id, f0 AS tb3_id])
- : : : +- Correlate(invocation=[split($cor1.tb3_ids)], correlate=[table(split($cor1.tb3_ids))], select=[id,key,tb2_ids,tb3_ids,name,f0], rowType=[RecordType(VARCHAR(2147483647) id, VARCHAR(2147483647) key, VARCHAR(2147483647) tb2_ids, VARCHAR(2147483647) tb3_ids, VARCHAR(2147483647) name, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : : : +- Calc(select=[id, EXPR$0 AS tb3_id])
+ : : : +- Correlate(invocation=[split($cor1.tb3_ids)], correlate=[table(split($cor1.tb3_ids))], select=[id,key,tb2_ids,tb3_ids,name,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) id, VARCHAR(2147483647) key, VARCHAR(2147483647) tb2_ids, VARCHAR(2147483647) tb3_ids, VARCHAR(2147483647) name, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: : : +- Reused(reference_id=[1])
: : +- Exchange(distribution=[hash[id]])
: : +- LegacyTableSourceScan(table=[[default_catalog, default_database, tb3, source: [TestTableSource(id, name)]]], fields=[id, name])
@@ -401,8 +401,8 @@ Calc(select=[id, tb2_ids, tb3_ids, name, tb2_names, tb3_names, name0, name1])
: +- Calc(select=[id AS tb1_id, name AS tb2_name, ',' AS $f2])
: +- SortMergeJoin(joinType=[LeftOuterJoin], where=[(tb2_id = id0)], select=[id, tb2_id, id0, name])
: :- Exchange(distribution=[hash[tb2_id]])
- : : +- Calc(select=[id, f0 AS tb2_id])
- : : +- Correlate(invocation=[split($cor0.tb2_ids)], correlate=[table(split($cor0.tb2_ids))], select=[id,key,tb2_ids,tb3_ids,name,f0], rowType=[RecordType(VARCHAR(2147483647) id, VARCHAR(2147483647) key, VARCHAR(2147483647) tb2_ids, VARCHAR(2147483647) tb3_ids, VARCHAR(2147483647) name, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : : +- Calc(select=[id, EXPR$0 AS tb2_id])
+ : : +- Correlate(invocation=[split($cor0.tb2_ids)], correlate=[table(split($cor0.tb2_ids))], select=[id,key,tb2_ids,tb3_ids,name,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) id, VARCHAR(2147483647) key, VARCHAR(2147483647) tb2_ids, VARCHAR(2147483647) tb3_ids, VARCHAR(2147483647) name, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: : +- Reused(reference_id=[1])
: +- Exchange(distribution=[hash[id]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, tb2, source: [TestTableSource(id, name)]]], fields=[id, name])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
index 9b5170ca2de..975a8a430ad 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
@@ -106,12 +106,12 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f1=[$3], f=[$4], cnt=[$5])
:- LogicalAggregate(group=[{0}], cnt=[COUNT($0)])
: +- LogicalProject(f=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(readOrder=[1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f1, f, cnt], build=[right])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f0 AS f1, f, cnt])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,f0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) f0)], joinType=[INNER])\n +- HashAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(coun [...]
+MultipleInput(readOrder=[1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f1, f, cnt], build=[right])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[EXPR$0 AS f1, f, cnt])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- HashAggregate(isMerge=[true], groupBy=[f], select=[f, Fina [...]
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
@@ -140,12 +140,12 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f=[$3], f1=[$4])
:- LogicalAggregate(group=[{0}], cnt=[COUNT($0)])
: +- LogicalProject(f=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(readOrder=[1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f, f1], build=[right])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f, f0 AS f1])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,f0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) f0)], joinType=[INNER])\n +- HashAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) AS cn [...]
+MultipleInput(readOrder=[1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[(c = f)], select=[a, b, c, f, f1], build=[right])\n:- [#1] Exchange(distribution=[hash[c]])\n+- Calc(select=[f, EXPR$0 AS f1])\n +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n +- HashAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(co [...]
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
@@ -174,7 +174,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], f1=[$3])
:- LogicalAggregate(group=[{0}], cnt=[COUNT($0)])
: +- LogicalProject(f=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
@@ -183,8 +183,8 @@ HashJoin(joinType=[InnerJoin], where=[(c = f1)], select=[a, b, c, f1], build=[ri
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,f0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[split($cor0.f)], correlate=[table(split($cor0.f))], select=[f,cnt,EXPR$0], rowType=[RecordType(VARCHAR(2147483647) f, BIGINT cnt, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- HashAggregate(isMerge=[true], groupBy=[f], select=[f, Final_COUNT(count$0) AS cnt])
+- Exchange(distribution=[hash[f]])
+- LocalHashAggregate(groupBy=[f], select=[f, Partial_COUNT(f) AS count$0])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
index 9806eb1cb0c..bbc137a2477 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
@@ -650,21 +650,21 @@ LogicalProject(a=[$0], b=[$1], c=[$2], v=[$3], a0=[$4], b0=[$5], c0=[$6], v0=[$7
:- LogicalProject(a=[$0], b=[$1], c=[$2], v=[$3])
: +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[str_split($cor0.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[str_split($cor0.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], b=[$1], c=[$2], v=[$3])
+- LogicalCorrelate(correlation=[$cor1], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[str_split($cor1.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[str_split($cor1.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-HashJoin(joinType=[InnerJoin], where=[(f0 = f00)], select=[a, b, c, f0, a0, b0, c0, f00], build=[right])
-:- Exchange(distribution=[hash[f0]])
-: +- Correlate(invocation=[str_split($cor0.c, _UTF-16LE'-')], correlate=[table(str_split($cor0.c,'-'))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
+HashJoin(joinType=[InnerJoin], where=[(EXPR$0 = EXPR$00)], select=[a, b, c, EXPR$0, a0, b0, c0, EXPR$00], build=[right])
+:- Exchange(distribution=[hash[EXPR$0]])
+: +- Correlate(invocation=[str_split($cor0.c, _UTF-16LE'-')], correlate=[table(str_split($cor0.c,'-'))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-+- Exchange(distribution=[hash[f0]])
- +- Correlate(invocation=[str_split($cor1.c, _UTF-16LE'-')], correlate=[table(str_split($cor1.c,'-'))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
++- Exchange(distribution=[hash[EXPR$0]])
+ +- Correlate(invocation=[str_split($cor1.c, _UTF-16LE'-')], correlate=[table(str_split($cor1.c,'-'))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -684,18 +684,18 @@ LogicalProject(a=[$0], b=[$1], c=[$2], s=[$3], a0=[$4], b0=[$5], c0=[$6], s0=[$7
:- LogicalProject(a=[$0], b=[$1], c=[$2], s=[$3])
: +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[TableFun($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[TableFun($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], b=[$1], c=[$2], s=[$3])
+- LogicalCorrelate(correlation=[$cor1], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[TableFun($cor1.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[TableFun($cor1.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(c = f00)], select=[a, b, c, f0, a0, b0, c0, f00], build=[right])\n:- Correlate(invocation=[TableFun($cor0.c)], correlate=[table(TableFun($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])\n: +- [#2] Exchange(distribution=[hash[c]])\n+- [#1] Exchange(distribution=[hash[f0]])\n])
-:- Exchange(distribution=[hash[f0]])
-: +- Correlate(invocation=[TableFun($cor1.c)], correlate=[table(TableFun($cor1.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
+MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(c = EXPR$00)], select=[a, b, c, EXPR$0, a0, b0, c0, EXPR$00], build=[right])\n:- Correlate(invocation=[TableFun($cor0.c)], correlate=[table(TableFun($cor0.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])\n: +- [#2] Exchange(distribution=[hash[c]])\n+- [#1] Exchange(distribution=[hash[EXPR$0]])\n])
+:- Exchange(distribution=[hash[EXPR$0]])
+: +- Correlate(invocation=[TableFun($cor1.c)], correlate=[table(TableFun($cor1.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[c]])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashSemiAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashSemiAntiJoinTest.xml
index 6194abc2d41..441f86e8fa9 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashSemiAntiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashSemiAntiJoinTest.xml
@@ -90,7 +90,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -101,7 +101,7 @@ HashJoin(joinType=[LeftSemiJoin], where=[(a = d)], select=[a, b, c], isBroadcast
:- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[broadcast])
+- Calc(select=[d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -202,7 +202,7 @@ LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTabl
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -219,7 +219,7 @@ NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[a, b, c], build=[r
+- Calc(select=[true AS i])
+- NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[d], build=[right], singleRowJoin=[true])
:- Calc(select=[d])
- : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+- Exchange(distribution=[broadcast])
+- Calc(select=[], where=[m IS NOT NULL])
@@ -619,7 +619,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -629,8 +629,8 @@ LogicalProject(f1=[$3])
HashJoin(joinType=[LeftSemiJoin], where=[((c = f1) AND (a = d))], select=[a, b, c], isBroadcast=[true], build=[right])
:- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[broadcast])
- +- Calc(select=[f0 AS f1, d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1, d])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -1099,7 +1099,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1109,8 +1109,8 @@ LogicalProject(f1=[$3])
HashJoin(joinType=[LeftSemiJoin], where=[(c = f1)], select=[a, b, c], isBroadcast=[true], build=[right])
:- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[broadcast])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopSemiAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopSemiAntiJoinTest.xml
index 54de0c603a5..c448697a17c 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopSemiAntiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopSemiAntiJoinTest.xml
@@ -90,7 +90,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -101,7 +101,7 @@ NestedLoopJoin(joinType=[LeftSemiJoin], where=[(a = d)], select=[a, b, c], build
:- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[broadcast])
+- Calc(select=[d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -290,7 +290,7 @@ LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTabl
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -307,7 +307,7 @@ NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[a, b, c], build=[r
+- Calc(select=[true AS i])
+- NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[d], build=[right], singleRowJoin=[true])
:- Calc(select=[d])
- : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+- Exchange(distribution=[broadcast])
+- Calc(select=[], where=[m IS NOT NULL])
@@ -801,7 +801,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -811,8 +811,8 @@ LogicalProject(f1=[$3])
NestedLoopJoin(joinType=[LeftSemiJoin], where=[((c = f1) AND (a = d))], select=[a, b, c], build=[right])
:- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[broadcast])
- +- Calc(select=[f0 AS f1, d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1, d])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -1366,7 +1366,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1376,8 +1376,8 @@ LogicalProject(f1=[$3])
NestedLoopJoin(joinType=[LeftSemiJoin], where=[(c = f1)], select=[a, b, c], build=[right])
:- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[broadcast])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTest.xml
index 0c93102527e..8fcc6e99bbc 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTest.xml
@@ -93,7 +93,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -105,7 +105,7 @@ HashJoin(joinType=[LeftSemiJoin], where=[(a = d)], select=[a, b, c], build=[righ
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[d]])
+- Calc(select=[d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -296,7 +296,7 @@ LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTabl
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -313,7 +313,7 @@ NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[a, b, c], build=[r
+- Calc(select=[true AS i])
+- NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[d], build=[right], singleRowJoin=[true])
:- Calc(select=[d])
- : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+- Exchange(distribution=[broadcast])
+- Calc(select=[], where=[m IS NOT NULL])
@@ -829,7 +829,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -840,8 +840,8 @@ HashJoin(joinType=[LeftSemiJoin], where=[((c = f1) AND (a = d))], select=[a, b,
:- Exchange(distribution=[hash[c, a]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1, d]])
- +- Calc(select=[f0 AS f1, d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1, d])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -1417,7 +1417,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1428,8 +1428,8 @@ HashJoin(joinType=[LeftSemiJoin], where=[(c = f1)], select=[a, b, c], build=[rig
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashSemiAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashSemiAntiJoinTest.xml
index d98446ee0f5..0f79f0406f5 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashSemiAntiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashSemiAntiJoinTest.xml
@@ -93,7 +93,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -105,7 +105,7 @@ HashJoin(joinType=[LeftSemiJoin], where=[(a = d)], select=[a, b, c], build=[righ
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[d]])
+- Calc(select=[d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -246,7 +246,7 @@ LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTabl
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -263,7 +263,7 @@ NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[a, b, c], build=[r
+- Calc(select=[true AS i])
+- NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[d], build=[right], singleRowJoin=[true])
:- Calc(select=[d])
- : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+- Exchange(distribution=[broadcast])
+- Calc(select=[], where=[m IS NOT NULL])
@@ -679,7 +679,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -690,8 +690,8 @@ HashJoin(joinType=[LeftSemiJoin], where=[((c = f1) AND (a = d))], select=[a, b,
:- Exchange(distribution=[hash[c, a]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1, d]])
- +- Calc(select=[f0 AS f1, d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1, d])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -1220,7 +1220,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1231,8 +1231,8 @@ HashJoin(joinType=[LeftSemiJoin], where=[(c = f1)], select=[a, b, c], build=[rig
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeSemiAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeSemiAntiJoinTest.xml
index d5d9fce29ad..da4819342a7 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeSemiAntiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeSemiAntiJoinTest.xml
@@ -93,7 +93,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -105,7 +105,7 @@ SortMergeJoin(joinType=[LeftSemiJoin], where=[(a = d)], select=[a, b, c])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[d]])
+- Calc(select=[d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -246,7 +246,7 @@ LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTabl
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -263,7 +263,7 @@ NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[a, b, c], build=[r
+- Calc(select=[true AS i])
+- NestedLoopJoin(joinType=[LeftSemiJoin], where=[true], select=[d], build=[right], singleRowJoin=[true])
:- Calc(select=[d])
- : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+- Exchange(distribution=[broadcast])
+- Calc(select=[], where=[m IS NOT NULL])
@@ -679,7 +679,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -690,8 +690,8 @@ SortMergeJoin(joinType=[LeftSemiJoin], where=[((c = f1) AND (a = d))], select=[a
:- Exchange(distribution=[hash[c, a]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1, d]])
- +- Calc(select=[f0 AS f1, d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1, d])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -1220,7 +1220,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1231,8 +1231,8 @@ SortMergeJoin(joinType=[LeftSemiJoin], where=[(c = f1)], select=[a, b, c])
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.xml
index 72ac8d84668..7dd52761273 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.xml
@@ -26,17 +26,17 @@ LogicalProject(a=[$0], b=[$1], c=[$2], x=[$3], y=[$4])
+- LogicalFilter(condition=[AND(=($3, $0), =(pyFunc($3, $3), 2), =(+($4, 1), *($4, $4)))])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 1}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[func(*($cor0.a, $cor0.a), $cor0.b)], rowType=[RecordType(INTEGER f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[func(*($cor0.a, $cor0.a), $cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER x, INTEGER y)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-FlinkLogicalCalc(select=[a, b, c, f00 AS f0, f1], where=[AND(=(f0, 2), =(+(f1, 1), *(f1, f1)), =(f00, a))])
-+- FlinkLogicalCalc(select=[a, b, c, f00, f1, pyFunc(f00, f00) AS f0])
+FlinkLogicalCalc(select=[a, b, c, x, y], where=[AND(=(f0, 2), =(+(y, 1), *(y, y)), =(x, a))])
++- FlinkLogicalCalc(select=[a, b, c, x, y, pyFunc(x, x) AS f0])
+- FlinkLogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 1}])
:- FlinkLogicalCalc(select=[a, b, c, *(a, a) AS f0])
: +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
- +- FlinkLogicalTableFunctionScan(invocation=[func($3, $1)], rowType=[RecordType(INTEGER f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- FlinkLogicalTableFunctionScan(invocation=[func($3, $1)], rowType=[RecordType:peek_no_expand(INTEGER x, INTEGER y)])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.xml
index 029fa09e575..a4f8e0b18a0 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.xml
@@ -306,7 +306,7 @@ LogicalTableScan(table=[[default_catalog, default_database, MyTable3, source: [T
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
]]>
@@ -323,7 +323,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
:- LogicalProject(d=[$0])
: +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: :- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f)]]])
- : +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject($f0=[IS NOT NULL($0)])
+- LogicalAggregate(group=[{}], m=[MIN($0)])
+- LogicalProject(i=[true])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.xml
index c77a62519f8..90db9c5005c 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.xml
@@ -25,16 +25,16 @@ limitations under the License.
LogicalProject(a=[$0], b=[$1], c=[$2], x=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[javaFunc(pyFunc($cor0.c))], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[javaFunc(pyFunc($cor0.c))], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-FlinkLogicalCalc(select=[a, b, c, f00 AS f0])
+FlinkLogicalCalc(select=[a, b, c, EXPR$0])
+- FlinkLogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- FlinkLogicalCalc(select=[a, b, c, pyFunc(c) AS f0])
: +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
- +- FlinkLogicalTableFunctionScan(invocation=[javaFunc($3)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- FlinkLogicalTableFunctionScan(invocation=[javaFunc($3)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -47,16 +47,16 @@ FlinkLogicalCalc(select=[a, b, c, f00 AS f0])
LogicalProject(a=[$0], b=[$1], c=[$2], x=[$3], y=[$4])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 1, 2}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[func(*($cor0.a, $cor0.a), pyFunc($cor0.b, $cor0.c))], rowType=[RecordType(INTEGER f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[func(*($cor0.a, $cor0.a), pyFunc($cor0.b, $cor0.c))], rowType=[RecordType:peek_no_expand(INTEGER x, INTEGER y)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-FlinkLogicalCalc(select=[a, b, c, f00 AS f0, f1])
+FlinkLogicalCalc(select=[a, b, c, x, y])
+- FlinkLogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 1, 2}])
:- FlinkLogicalCalc(select=[a, b, c, *(a, a) AS f0])
: +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
- +- FlinkLogicalTableFunctionScan(invocation=[func($3, pyFunc($1, $2))], rowType=[RecordType(INTEGER f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- FlinkLogicalTableFunctionScan(invocation=[func($3, pyFunc($1, $2))], rowType=[RecordType:peek_no_expand(INTEGER x, INTEGER y)])
]]>
</Resource>
</TestCase>
@@ -69,17 +69,17 @@ FlinkLogicalCalc(select=[a, b, c, f00 AS f0, f1])
LogicalProject(a=[$0], b=[$1], c=[$2], x=[$4])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{3}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]])
- +- LogicalTableFunctionScan(invocation=[javaFunc(pyFunc($cor0.d._1))], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[javaFunc(pyFunc($cor0.d._1))], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-FlinkLogicalCalc(select=[a, b, c, f00 AS x])
+FlinkLogicalCalc(select=[a, b, c, EXPR$0 AS x])
+- FlinkLogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{3}])
:- FlinkLogicalCalc(select=[a, b, c, d, pyFunc(f0) AS f0])
: +- FlinkLogicalCalc(select=[a, b, c, d, d._1 AS f0])
: +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])
- +- FlinkLogicalTableFunctionScan(invocation=[javaFunc($4)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- FlinkLogicalTableFunctionScan(invocation=[javaFunc($4)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -92,16 +92,16 @@ FlinkLogicalCalc(select=[a, b, c, f00 AS x])
LogicalProject(a=[$0], b=[$1], c=[$2], x=[$4], y=[$5])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 2, 3}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]])
- +- LogicalTableFunctionScan(invocation=[func(*($cor0.d._1, $cor0.a), pyFunc($cor0.d._2, $cor0.c))], rowType=[RecordType(INTEGER f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[func(*($cor0.d._1, $cor0.a), pyFunc($cor0.d._2, $cor0.c))], rowType=[RecordType:peek_no_expand(INTEGER x, INTEGER y)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-FlinkLogicalCalc(select=[a, b, c, f00 AS x, f10 AS y])
+FlinkLogicalCalc(select=[a, b, c, x, y])
+- FlinkLogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 2, 3}])
:- FlinkLogicalCalc(select=[a, b, c, d, *(d._1, a) AS f0, d._2 AS f1])
: +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c, d)]]], fields=[a, b, c, d])
- +- FlinkLogicalTableFunctionScan(invocation=[func($4, pyFunc($5, $2))], rowType=[RecordType(INTEGER f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- FlinkLogicalTableFunctionScan(invocation=[func($4, pyFunc($5, $2))], rowType=[RecordType:peek_no_expand(INTEGER x, INTEGER y)])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.xml
index 8b08e63ce91..5fe73e79c78 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.xml
@@ -26,7 +26,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], s=[$3], l=[$4])
+- LogicalFilter(condition=[AND(=($4, $0), =($2, $3), =(pyFunc($4, $4), 2), =(+($4, 1), *($4, $4)))])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[func($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[func($cor0.c)], rowType=[RecordType:peek_no_expand(VARCHAR(2147483647) f0, INTEGER f1)])
]]>
</Resource>
<Resource name="optimized rel plan">
@@ -36,7 +36,7 @@ FlinkLogicalCalc(select=[a, b, c, f0, f1], where=[AND(=(f00, 2), =(f1, a), =(c,
+- FlinkLogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- FlinkLogicalCalc(select=[f0, f1], where=[=(+(f1, 1), *(f1, f1))])
- +- FlinkLogicalTableFunctionScan(invocation=[func($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0, INTEGER f1)], elementType=[class [Ljava.lang.Object;])
+ +- FlinkLogicalTableFunctionScan(invocation=[func($cor0.c)], rowType=[RecordType:peek_no_expand(VARCHAR(2147483647) f0, INTEGER f1)])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml
index 11bff246b04..8a66523e28c 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.xml
@@ -324,7 +324,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
}))], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -338,7 +338,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[true])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -583,7 +583,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[NOT(EXISTS({
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
}))])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -598,7 +598,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalProject(i=[true])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -832,7 +832,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
}))], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -846,7 +846,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[true])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -1581,7 +1581,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
}))])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1594,7 +1594,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalProject(f1=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml
index 70de33eff74..55d1aa8decb 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.xml
@@ -27,7 +27,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -41,7 +41,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[true])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -838,7 +838,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[EXISTS({
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -853,7 +853,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalProject(i=[true])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -870,7 +870,7 @@ LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTabl
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -886,7 +886,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalJoin(condition=[$4], joinType=[semi])
:- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: :- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- : +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject($f0=[IS NOT NULL($0)])
+- LogicalAggregate(group=[{}], m=[MIN($0)])
+- LogicalProject(i=[true])
@@ -908,7 +908,7 @@ LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTableSource(i, j, k)]]])
})])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -929,7 +929,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
: +- LogicalAggregate(group=[{}], m=[MIN($0)])
: +- LogicalProject(i=[true])
: +- LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTableSource(i, j, k)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -943,7 +943,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[EXISTS({
LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -958,7 +958,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalProject(i=[true])
+- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -1353,7 +1353,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1367,7 +1367,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalFilter(condition=[true])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -2815,7 +2815,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -2828,7 +2828,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalProject(f1=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -2847,7 +2847,7 @@ LogicalProject(i=[$0])
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -2861,7 +2861,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalJoin(condition=[=($0, $4)], joinType=[semi])
:- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: :- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- : +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(i=[$0])
+- LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTableSource(i, j, k)]]])
]]>
@@ -2883,7 +2883,7 @@ LogicalProject(i=[$0])
LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTableSource(i, j, k)]]])
})])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -2900,7 +2900,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
: :- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
: +- LogicalProject(i=[$0])
: +- LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTableSource(i, j, k)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
@@ -2915,7 +2915,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -2928,7 +2928,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
+- LogicalProject(f1=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
index 0dc07f2d2ed..616eed09d56 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.xml
@@ -365,7 +365,7 @@ LogicalLegacySink(name=[`default_catalog`.`default_database`.`retractSink1`], fi
: :- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
: : +- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
: : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], total_c=[0])
+- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
+- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
@@ -383,7 +383,7 @@ LogicalLegacySink(name=[`default_catalog`.`default_database`.`retractSink2`], fi
: :- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
: : +- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
: : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], total_c=[0])
+- LogicalProject(a=[$0], b=[-($1, MOD($1, 300))], c=[$2])
+- LogicalFilter(condition=[>=($1, UNIX_TIMESTAMP(_UTF-16LE'${startTime}'))])
@@ -398,8 +398,8 @@ LegacySink(name=[`default_catalog`.`default_database`.`retractSink1`], fields=[a
: +- Calc(select=[a, total_c], changelogMode=[I,UB,UA])
: +- GroupAggregate(groupBy=[a, b], select=[a, b, COUNT(DISTINCT c) AS total_c], changelogMode=[I,UB,UA])
: +- Exchange(distribution=[hash[a, b]], changelogMode=[I])
- : +- Calc(select=[a, b, f0 AS c], changelogMode=[I])
- : +- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER], changelogMode=[I])
+ : +- Calc(select=[a, b, EXPR$0 AS c], changelogMode=[I])
+ : +- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER], changelogMode=[I])
: +- Calc(select=[a, -(b, MOD(b, 300)) AS b, c], where=[AND(>=(b, UNIX_TIMESTAMP('${startTime}')), <>(c, ''))], changelogMode=[I])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], changelogMode=[I])
+- Calc(select=[a, CAST(total_c AS BIGINT) AS total_c], where=[>(a, 50)], changelogMode=[I])
@@ -412,8 +412,8 @@ LegacySink(name=[`default_catalog`.`default_database`.`retractSink2`], fields=[a
: +- Calc(select=[a, total_c], changelogMode=[I,UB,UA])
: +- GroupAggregate(groupBy=[a, b], select=[a, b, COUNT(DISTINCT c) AS total_c], changelogMode=[I,UB,UA])
: +- Exchange(distribution=[hash[a, b]], changelogMode=[I])
- : +- Calc(select=[a, b, f0 AS c], changelogMode=[I])
- : +- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER], changelogMode=[I])
+ : +- Calc(select=[a, b, EXPR$0 AS c], changelogMode=[I])
+ : +- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER], changelogMode=[I])
: +- Calc(select=[a, -(b, MOD(b, 300)) AS b, c], where=[AND(>=(b, UNIX_TIMESTAMP('${startTime}')), <>(c, ''))], changelogMode=[I])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c], changelogMode=[I])
+- Calc(select=[a, CAST(total_c AS BIGINT) AS total_c], where=[<(a, 50)], changelogMode=[I])
@@ -946,13 +946,13 @@ LogicalLegacySink(name=[`default_catalog`.`default_database`.`appendSink`], fiel
: : :- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
: : +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(d, e, f)]]])
: +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(i, j, k, l, m)]]])
- +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[split($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
LegacySink(name=[`default_catalog`.`default_database`.`appendSink`], fields=[a, b, c, d, e, f, i, j, k, l, m, s], changelogMode=[NONE])
-+- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,d,e,f,i,j,k,l,m,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, INTEGER d, BIGINT e, VARCHAR(2147483647) f, INTEGER i, BIGINT j, INTEGER k, VARCHAR(2147483647) l, BIGINT m, VARCHAR(2147483647) f0)], joinType=[INNER], changelogMode=[I])
++- Correlate(invocation=[split($cor0.c)], correlate=[table(split($cor0.c))], select=[a,b,c,d,e,f,i,j,k,l,m,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, INTEGER d, BIGINT e, VARCHAR(2147483647) f, INTEGER i, BIGINT j, INTEGER k, VARCHAR(2147483647) l, BIGINT m, VARCHAR(2147483647) EXPR$0)], joinType=[INNER], changelogMode=[I])
+- Join(joinType=[InnerJoin], where=[=(a, i)], select=[a, b, c, d, e, f, i, j, k, l, m], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey], changelogMode=[I])
:- Exchange(distribution=[hash[a]], changelogMode=[I])
: +- Join(joinType=[InnerJoin], where=[=(b, e)], select=[a, b, c, d, e, f], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey], changelogMode=[I])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml
index e17c1823556..647421d63b0 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.xml
@@ -201,7 +201,7 @@ LogicalProject(name=[$0], id=[$1], amount=[$2], price=[$3])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[name, id, amount, price], where=[(Func1$(amount) < 32)])
+Calc(select=[name, id, amount, price], where=[(myUdf(amount) < 32)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, FilterableTable, source: [filterPushedDown=[true], filter=[greaterThan(amount, 2)]]]], fields=[name, id, amount, price])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml
index 9fc53d2df1d..21ac6d3c757 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.xml
@@ -240,13 +240,13 @@ FROM MyTable, LATERAL TABLE(tableFunc(rowtime, proctime, '')) AS T(s)
LogicalProject(rowtime=[$0], proctime=[$3], s=[$4])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0, 3}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
- +- LogicalTableFunctionScan(invocation=[tableFunc($cor0.rowtime, $cor0.proctime, _UTF-16LE'')], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[tableFunc($cor0.rowtime, $cor0.proctime, _UTF-16LE'')], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[rowtime, PROCTIME_MATERIALIZE(proctime) AS proctime, f0 AS s])
-+- Correlate(invocation=[tableFunc(CAST($cor0.rowtime):TIMESTAMP(3), PROCTIME_MATERIALIZE($cor0.proctime), _UTF-16LE'')], correlate=[table(tableFunc(CAST($cor0.rowtime AS TIMESTAMP(3)),PROCTIME_MATERIALIZE($cor0.proctime),''))], select=[rowtime,long,int,proctime,f0], rowType=[RecordType(TIMESTAMP(3) *ROWTIME* rowtime, BIGINT long, INTEGER int, TIMESTAMP_LTZ(3) *PROCTIME* proctime, VARCHAR(2147483647) f0)], joinType=[INNER])
+Calc(select=[rowtime, PROCTIME_MATERIALIZE(proctime) AS proctime, EXPR$0 AS s])
++- Correlate(invocation=[tableFunc(CAST($cor0.rowtime):TIMESTAMP(3), PROCTIME_MATERIALIZE($cor0.proctime), _UTF-16LE'')], correlate=[table(tableFunc(CAST($cor0.rowtime AS TIMESTAMP(3)),PROCTIME_MATERIALIZE($cor0.proctime),''))], select=[rowtime,long,int,proctime,EXPR$0], rowType=[RecordType(TIMESTAMP(3) *ROWTIME* rowtime, BIGINT long, INTEGER int, TIMESTAMP_LTZ(3) *PROCTIME* proctime, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[rowtime, long, int, proctime])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.xml
index b0aeef1da59..c6979ac3d0c 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.xml
@@ -507,21 +507,21 @@ LogicalProject(a=[$0], b=[$1], c=[$2], v=[$3], a0=[$4], b0=[$5], c0=[$6], v0=[$7
:- LogicalProject(a=[$0], b=[$1], c=[$2], v=[$3])
: +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[str_split($cor0.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[str_split($cor0.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], b=[$1], c=[$2], v=[$3])
+- LogicalCorrelate(correlation=[$cor1], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[str_split($cor1.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[str_split($cor1.c, _UTF-16LE'-')], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Join(joinType=[InnerJoin], where=[(f0 = f00)], select=[a, b, c, f0, a0, b0, c0, f00], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
-:- Exchange(distribution=[hash[f0]])
-: +- Correlate(invocation=[str_split($cor0.c, _UTF-16LE'-')], correlate=[table(str_split($cor0.c,'-'))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
+Join(joinType=[InnerJoin], where=[(EXPR$0 = EXPR$00)], select=[a, b, c, EXPR$0, a0, b0, c0, EXPR$00], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+:- Exchange(distribution=[hash[EXPR$0]])
+: +- Correlate(invocation=[str_split($cor0.c, _UTF-16LE'-')], correlate=[table(str_split($cor0.c,'-'))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-+- Exchange(distribution=[hash[f0]])
- +- Correlate(invocation=[str_split($cor1.c, _UTF-16LE'-')], correlate=[table(str_split($cor1.c,'-'))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
++- Exchange(distribution=[hash[EXPR$0]])
+ +- Correlate(invocation=[str_split($cor1.c, _UTF-16LE'-')], correlate=[table(str_split($cor1.c,'-'))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -541,21 +541,21 @@ LogicalProject(a=[$0], b=[$1], c=[$2], s=[$3], a0=[$4], b0=[$5], c0=[$6], s0=[$7
:- LogicalProject(a=[$0], b=[$1], c=[$2], s=[$3])
: +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
: :- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- : +- LogicalTableFunctionScan(invocation=[TableFun($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ : +- LogicalTableFunctionScan(invocation=[TableFun($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
+- LogicalProject(a=[$0], b=[$1], c=[$2], s=[$3])
+- LogicalCorrelate(correlation=[$cor1], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]])
- +- LogicalTableFunctionScan(invocation=[TableFun($cor1.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[TableFun($cor1.c)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Join(joinType=[InnerJoin], where=[(c = f00)], select=[a, b, c, f0, a0, b0, c0, f00], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+Join(joinType=[InnerJoin], where=[(c = EXPR$00)], select=[a, b, c, EXPR$0, a0, b0, c0, EXPR$00], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
:- Exchange(distribution=[hash[c]])
-: +- Correlate(invocation=[TableFun($cor0.c)], correlate=[table(TableFun($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
+: +- Correlate(invocation=[TableFun($cor0.c)], correlate=[table(TableFun($cor0.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-+- Exchange(distribution=[hash[f0]])
- +- Correlate(invocation=[TableFun($cor1.c)], correlate=[table(TableFun($cor1.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)], joinType=[INNER])
++- Exchange(distribution=[hash[EXPR$0]])
+ +- Correlate(invocation=[TableFun($cor1.c)], correlate=[table(TableFun($cor1.c))], select=[a,b,c,EXPR$0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.xml
index eac4344357f..37119efe848 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.xml
@@ -92,7 +92,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -104,7 +104,7 @@ Join(joinType=[LeftSemiJoin], where=[(a = d)], select=[a, b, c], leftInputSpec=[
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[d]])
+- Calc(select=[d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -295,7 +295,7 @@ LogicalTableScan(table=[[default_catalog, default_database, t, source: [TestTabl
})])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -313,7 +313,7 @@ Join(joinType=[LeftSemiJoin], where=[true], select=[a, b, c], leftInputSpec=[NoU
+- Join(joinType=[LeftSemiJoin], where=[true], select=[d], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
:- Exchange(distribution=[single])
: +- Calc(select=[d])
- : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ : +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+- Exchange(distribution=[single])
+- Calc(select=[], where=[m IS NOT NULL])
@@ -828,7 +828,7 @@ LogicalProject(f1=[$3])
LogicalFilter(condition=[=($cor1.a, $0)])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})], variablesSet=[[$cor1]])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -839,8 +839,8 @@ Join(joinType=[LeftSemiJoin], where=[((c = f1) AND (a = d))], select=[a, b, c],
:- Exchange(distribution=[hash[c, a]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1, d]])
- +- Calc(select=[f0 AS f1, d])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1, d])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
@@ -1416,7 +1416,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2])
LogicalProject(f1=[$3])
LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
LogicalTableScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]])
- LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ LogicalTableFunctionScan(invocation=[table_func($cor0.f)], rowType=[RecordType(VARCHAR(2147483647) EXPR$0)])
})])
+- LogicalTableScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]])
]]>
@@ -1427,8 +1427,8 @@ Join(joinType=[LeftSemiJoin], where=[(c = f1)], select=[a, b, c], leftInputSpec=
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0 AS f1])
- +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,f0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) f0)], joinType=[INNER])
+ +- Calc(select=[EXPR$0 AS f1])
+ +- Correlate(invocation=[table_func($cor0.f)], correlate=[table(table_func($cor0.f))], select=[d,e,f,EXPR$0], rowType=[RecordType(INTEGER d, BIGINT e, VARCHAR(2147483647) f, VARCHAR(2147483647) EXPR$0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, r, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.xml
index 063fcca6732..7c9102940a6 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.xml
@@ -25,7 +25,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[TestFunc($0, $1)])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[a, b, c, TestFunc$(a, b) AS d])
+Calc(select=[a, b, c, TestFunc(a, b) AS d])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, Table1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -173,7 +173,7 @@ LogicalProject(c=[$2], mycount=[AS(countFun($1) OVER (PARTITION BY $2 ORDER BY $
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[c, w0$o0 AS mycount, w0$o1 AS wAvg, w0$o2 AS countDist])
-+- OverAggregate(partitionBy=[c], orderBy=[proctime ASC], window=[ ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], select=[a, b, c, proctime, $4, CountAggFunction(b) AS w0$o0, weightAvgFun(a, $4) AS w0$o1, CountDistinct(a) AS w0$o2])
++- OverAggregate(partitionBy=[c], orderBy=[proctime ASC], window=[ ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], select=[a, b, c, proctime, $4, countFun(b) AS w0$o0, weightAvgFun(a, $4) AS w0$o1, countDist(a) AS w0$o2])
+- Exchange(distribution=[hash[c]])
+- Calc(select=[a, b, c, proctime, CAST(b AS BIGINT) AS $4])
+- DataStreamScan(table=[[default_catalog, default_database, T1]], fields=[a, b, c, proctime])
@@ -203,7 +203,7 @@ LogicalProject(_c0=[TestFunc($0, $1)])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-Calc(select=[TestFunc$(double, long) AS _c0])
+Calc(select=[TestFunc(double, long) AS _c0])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, Table1, source: [TestTableSource(double, long)]]], fields=[double, long])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/validation/AggregateValidationTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/validation/AggregateValidationTest.xml
new file mode 100644
index 00000000000..5db02638d2a
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/table/validation/AggregateValidationTest.xml
@@ -0,0 +1,36 @@
+<?xml version="1.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.
+-->
+<Root>
+ <TestCase name="testTableFunctionInSelection">
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(_c0=[func(_UTF-16LE'abc')])
++- LogicalAggregate(group=[{0}], TMP_0=[SUM($1)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, Table1, source: [TestTableSource(a, b, c)]]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[func('abc') AS _c0])
++- GroupAggregate(groupBy=[a], select=[a])
+ +- Exchange(distribution=[hash[a]])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, Table1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala
index e171ab8c56b..ae165b1dc1a 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala
@@ -49,7 +49,7 @@ class OverWindowValidationTest extends TableTestBase {
/** OVER clause is necessary for [[OverAgg0]] window function. */
@Test
def testInvalidOverAggregation(): Unit = {
- streamUtil.addFunction("overAgg", new OverAgg0)
+ streamUtil.addTemporarySystemFunction("overAgg", new OverAgg0)
val sqlQuery = "SELECT overAgg(c, a) FROM MyTable"
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/CalcTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/CalcTest.scala
index 6c0767f09ca..abd196d34af 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/CalcTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/CalcTest.scala
@@ -37,7 +37,7 @@ class CalcTest extends TableTestBase {
@BeforeEach
def setup(): Unit = {
util.addTableSource[(Long, Int, String)]("MyTable", 'a, 'b, 'c)
- util.addFunction("random_udf", new NonDeterministicUdf)
+ util.addTemporarySystemFunction("random_udf", new NonDeterministicUdf)
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala
index 9a18da17f6f..1e066548c4f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.scala
@@ -101,7 +101,7 @@ class DagOptimizationTest extends TableTestBase {
@Test
def testSingleSinkWithUDTF(): Unit = {
util.addTableSource[(Int, Long, Int, String, Long)]("MyTable2", 'i, 'j, 'k, 'l, 'm)
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
@@ -299,7 +299,7 @@ class DagOptimizationTest extends TableTestBase {
util.tableEnv.getConfig.set(
RelNodeBlockPlanBuilder.TABLE_OPTIMIZER_UNIONALL_AS_BREAKPOINT_ENABLED,
Boolean.box(false))
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery1 =
"""
|SELECT a, b - MOD(b, 300) AS b, c FROM MyTable
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.scala
index a4da620ec0e..23290fddf35 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.scala
@@ -171,7 +171,7 @@ class LegacyTableSourceTest extends TableTestBase {
@Test
def testFilterPushDownWithUdf(): Unit = {
- util.addFunction("myUdf", Func1)
+ util.addTemporarySystemFunction("myUdf", Func1)
util.verifyExecPlan("SELECT * FROM FilterableTable WHERE amount > 2 AND myUdf(amount) < 32")
}
@@ -183,7 +183,7 @@ class LegacyTableSourceTest extends TableTestBase {
@Test
def testPartitionTableSourceWithUdf(): Unit = {
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyExecPlan("SELECT * FROM PartitionableTable WHERE id > 2 AND MyUdf(part2) < 3")
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/PartitionableSourceTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/PartitionableSourceTest.scala
index d2d4935e63b..3e73cd735d2 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/PartitionableSourceTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/PartitionableSourceTest.scala
@@ -125,7 +125,7 @@ class PartitionableSourceTest(val sourceFetchPartitions: Boolean, val useCatalog
@TestTemplate
def testWithUdfAndVirtualColumn(): Unit = {
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyExecPlan("SELECT * FROM PartitionableTable WHERE id > 2 AND MyUdf(part2) < 3")
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala
index 1abdd4e8007..4f1b14e33e5 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.scala
@@ -303,7 +303,7 @@ class RemoveCollationTest extends TableTestBase {
Array("id", "name"),
FlinkStatistic.builder().uniqueKeys(ImmutableSet.of(ImmutableSet.of("id"))).build()
)
- util.addFunction("split", new StringSplit())
+ util.addTemporarySystemFunction("split", new StringSplit())
val sql =
"""
@@ -353,7 +353,7 @@ class RemoveCollationTest extends TableTestBase {
def testRemoveCollation_Correlate1(): Unit = {
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin,NestedLoopJoin,HashAgg")
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
|WITH r AS (SELECT f, count(f) as cnt FROM y GROUP BY f),
@@ -367,7 +367,7 @@ class RemoveCollationTest extends TableTestBase {
def testRemoveCollation_Correlate2(): Unit = {
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin,NestedLoopJoin,HashAgg")
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
|WITH r AS (SELECT f, count(f) as cnt FROM y GROUP BY f),
@@ -382,7 +382,7 @@ class RemoveCollationTest extends TableTestBase {
// do not remove shuffle
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin,NestedLoopJoin,HashAgg")
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
|WITH r AS (SELECT f, count(f) as cnt FROM y GROUP BY f),
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala
index c65f341a29a..c56eb166f36 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.scala
@@ -518,7 +518,7 @@ class RemoveShuffleTest extends TableTestBase {
// disable BroadcastHashJoin
util.tableEnv.getConfig
.set(OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD, Long.box(-1))
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
|WITH r AS (SELECT f, count(f) as cnt FROM y GROUP BY f),
@@ -536,7 +536,7 @@ class RemoveShuffleTest extends TableTestBase {
// disable BroadcastHashJoin
util.tableEnv.getConfig
.set(OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD, Long.box(-1))
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
|WITH r AS (SELECT f, count(f) as cnt FROM y GROUP BY f),
@@ -555,7 +555,7 @@ class RemoveShuffleTest extends TableTestBase {
// disable BroadcastHashJoin
util.tableEnv.getConfig
.set(OptimizerConfigOptions.TABLE_OPTIMIZER_BROADCAST_JOIN_THRESHOLD, Long.box(-1))
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
|WITH r AS (SELECT f, count(f) as cnt FROM y GROUP BY f),
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala
index 5718002822f..791283a54a2 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.scala
@@ -397,7 +397,7 @@ class SubplanReuseTest extends TableTestBase {
@Test
def testSubplanReuseOnCorrelate(): Unit = {
- util.addFunction("str_split", new StringSplit())
+ util.addTemporarySystemFunction("str_split", new StringSplit())
util.tableEnv.getConfig
.set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "NestedLoopJoin,SortMergeJoin")
val sqlQuery =
@@ -412,7 +412,7 @@ class SubplanReuseTest extends TableTestBase {
@Test
def testSubplanReuseOnCorrelateWithNonDeterministicUDTF(): Unit = {
- util.addFunction("TableFun", new NonDeterministicTableFunc)
+ util.addTemporarySystemFunction("TableFun", new NonDeterministicTableFunc)
val sqlQuery =
"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/AggregateTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/AggregateTestBase.scala
index 3b3d092d944..1c182e55236 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/AggregateTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/AggregateTestBase.scala
@@ -224,13 +224,13 @@ abstract class AggregateTestBase extends TableTestBase {
@TestTemplate
def testAggNotSupportMerge(): Unit = {
- util.addFunction("var_sum", new VarSum2AggFunction)
+ util.addTemporarySystemFunction("var_sum", new VarSum2AggFunction)
util.verifyExecPlan("SELECT b, var_sum(a) FROM MyTable1 GROUP BY b")
}
@TestTemplate
def testPojoAccumulator(): Unit = {
- util.addFunction("var_sum", new VarSum1AggFunction)
+ util.addTemporarySystemFunction("var_sum", new VarSum1AggFunction)
util.verifyExecPlan("SELECT b, var_sum(a) FROM MyTable1 GROUP BY b")
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala
index c6dcac1a54b..661c71801f9 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/GroupWindowTest.scala
@@ -42,7 +42,7 @@ class GroupWindowTest(aggStrategy: AggregatePhaseStrategy) extends TableTestBase
def before(): Unit = {
util.tableEnv.getConfig
.set(OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY, aggStrategy.toString)
- util.addFunction("countFun", new CountAggFunction)
+ util.addTemporarySystemFunction("countFun", new CountAggFunction)
util.addTableSource[(Int, Timestamp, Int, Long)]("MyTable", 'a, 'b, 'c, 'd)
util.addTableSource[(Timestamp, Long, Int, String)]("MyTable1", 'ts, 'a, 'b, 'c)
util.addTableSource[(Int, Long, String, Int, Timestamp)]("MyTable2", 'a, 'b, 'c, 'd, 'ts)
@@ -90,14 +90,14 @@ class GroupWindowTest(aggStrategy: AggregatePhaseStrategy) extends TableTestBase
@TestTemplate
def testTumbleWindowWithInvalidUdAggArgs(): Unit = {
val weightedAvg = new WeightedAvgWithMerge
- util.addFunction("weightedAvg", weightedAvg)
+ util.addTemporarySystemFunction("weightedAvg", weightedAvg)
val sql = "SELECT weightedAvg(c, a) AS wAvg FROM MyTable2 " +
"GROUP BY TUMBLE(ts, INTERVAL '4' MINUTE)"
assertThatThrownBy(() => util.verifyExecPlan(sql))
- .hasMessageContaining("SQL validation failed. "
- + "Given parameters of function 'weightedAvg' do not match any signature.")
+ .hasMessageContaining(
+ "SQL validation failed. Invalid function call:\nweightedAvg(STRING, INT)")
.isInstanceOf[ValidationException]
}
@@ -183,7 +183,7 @@ class GroupWindowTest(aggStrategy: AggregatePhaseStrategy) extends TableTestBase
@TestTemplate
def testTumblingWindowWithUdAgg(): Unit = {
- util.addFunction("weightedAvg", new WeightedAvgWithMerge)
+ util.addTemporarySystemFunction("weightedAvg", new WeightedAvgWithMerge)
val sql = "SELECT weightedAvg(b, a) AS wAvg FROM MyTable2 " +
"GROUP BY TUMBLE(ts, INTERVAL '4' MINUTE)"
util.verifyExecPlan(sql)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala
index e95d40c8c84..1fb6ad9028a 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala
@@ -338,7 +338,7 @@ class OverAggregateTest extends TableTestBase {
/** OVER clause is necessary for [[OverAgg0]] window function. */
@Test
def testInvalidOverAggregation(): Unit = {
- util.addFunction("overAgg", new OverAgg0)
+ util.addTemporarySystemFunction("overAgg", new OverAgg0)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(() => util.verifyExecPlan("SELECT overAgg(b, a) FROM MyTable"))
}
@@ -347,7 +347,7 @@ class OverAggregateTest extends TableTestBase {
@Test
def testInvalidOverAggregation2(): Unit = {
util.addTableSource[(Int, Long, String, Timestamp)]("T", 'a, 'b, 'c, 'ts)
- util.addFunction("overAgg", new OverAgg0)
+ util.addTemporarySystemFunction("overAgg", new OverAgg0)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.scala
index 7121e02d723..424fa999517 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.scala
@@ -139,7 +139,7 @@ class LookupJoinTest(legacyTableSource: Boolean) extends TableTestBase {
@TestTemplate
def testPythonUDFInJoinCondition(): Unit = {
- testUtil.addFunction("pyFunc", new PythonScalarFunction("pyFunc"))
+ testUtil.addTemporarySystemFunction("pyFunc", new PythonScalarFunction("pyFunc"))
val sql =
"""
|SELECT * FROM MyTable AS T
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTestBase.scala
index d9afccaddcf..c3e91ea235f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SemiAntiJoinTestBase.scala
@@ -32,7 +32,7 @@ abstract class SemiAntiJoinTestBase extends TableTestBase {
util.addTableSource[(Int, Long, String)]("t", 'i, 'j, 'k)
util.addTableSource[(Int, Long)]("leftT", 'a, 'b)
util.addTableSource[(Int, Long)]("rightT", 'c, 'd)
- util.addFunction("table_func", new StringSplit)
+ util.addTemporarySystemFunction("table_func", new StringSplit)
@Test
def testInWithUncorrelated_SimpleCondition1(): Unit = {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/PythonCalcTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/PythonCalcTest.scala
index 9cf2d1d2151..46c3bfcdc79 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/PythonCalcTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/PythonCalcTest.scala
@@ -30,7 +30,7 @@ class PythonCalcTest extends TableTestBase {
@BeforeEach
def setup(): Unit = {
util.addTableSource[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
- util.addFunction("pyFunc1", new PythonScalarFunction("pyFunc1"))
+ util.addTemporarySystemFunction("pyFunc1", new PythonScalarFunction("pyFunc1"))
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/AggregateValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/AggregateValidationTest.scala
index 6fb62a24756..69e693e216d 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/AggregateValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/validation/AggregateValidationTest.scala
@@ -212,7 +212,7 @@ class AggregateValidationTest extends TableTestBase {
val t = util.addTableSource[(Int, Long, String)]("Table3", 'a, 'b, 'c)
val myWeightedAvg = new WeightedAvgWithMergeAndReset
- util.addFunction("myWeightedAvg", myWeightedAvg)
+ util.addTemporarySystemFunction("myWeightedAvg", myWeightedAvg)
// must fail. UDAGG does not accept String type
assertThatExceptionOfType(classOf[ValidationException])
@@ -225,7 +225,7 @@ class AggregateValidationTest extends TableTestBase {
val t = util.addTableSource[(Int, Long, String)]("Table3", 'a, 'b, 'c)
val myWeightedAvg = new WeightedAvgWithMergeAndReset
- util.addFunction("myWeightedAvg", myWeightedAvg)
+ util.addTemporarySystemFunction("myWeightedAvg", myWeightedAvg)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.scala
index ba5c2e075ea..a73507d9c5e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/CalcPythonCorrelateTransposeRuleTest.scala
@@ -55,8 +55,8 @@ class CalcPythonCorrelateTransposeRuleTest extends TableTestBase {
util.replaceStreamProgram(programs)
util.addTableSource[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
- util.addFunction("func", new MockPythonTableFunction)
- util.addFunction("pyFunc", new PythonScalarFunction("pyFunc"))
+ util.addTemporarySystemFunction("func", new MockPythonTableFunction)
+ util.addTemporarySystemFunction("pyFunc", new PythonScalarFunction("pyFunc"))
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ExpressionReductionRulesTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ExpressionReductionRulesTest.scala
index 943b6e4c7fa..e6e25b71082 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ExpressionReductionRulesTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ExpressionReductionRulesTest.scala
@@ -34,20 +34,20 @@ class ExpressionReductionRulesTest extends TableTestBase {
@Test
def testExpressionReductionWithUDF(): Unit = {
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyRelPlan("SELECT MyUdf(1) FROM MyTable")
}
@Test
def testExpressionReductionWithRichUDF(): Unit = {
- util.addFunction("MyUdf", new RichFunc1)
+ util.addTemporarySystemFunction("MyUdf", new RichFunc1)
util.getTableEnv.getConfig.addJobParameter("int.value", "10")
util.verifyRelPlan("SELECT myUdf(1) FROM MyTable")
}
@Test
def testExpressionReductionWithRichUDFAndInvalidOpen(): Unit = {
- util.addFunction("MyUdf", new RichFunc1)
+ util.addTemporarySystemFunction("MyUdf", new RichFunc1)
// FunctionContext.getCachedFile will fail during expression reduction
// it will be executed during runtime though
util.getTableEnv.getConfig.addJobParameter("fail-for-cached-file", "true")
@@ -56,8 +56,8 @@ class ExpressionReductionRulesTest extends TableTestBase {
@Test
def testExpressionReductionWithPythonUDF(): Unit = {
- util.addFunction("PyUdf", DeterministicPythonFunc)
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("PyUdf", DeterministicPythonFunc)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyExecPlan("SELECT PyUdf(), MyUdf(1) FROM MyTable")
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkCalcMergeRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkCalcMergeRuleTest.scala
index 4b5165e4ab1..9011ce399f6 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkCalcMergeRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkCalcMergeRuleTest.scala
@@ -63,7 +63,7 @@ class FlinkCalcMergeRuleTest extends TableTestBase {
util.replaceBatchProgram(programs)
util.addTableSource[(Int, Int, String)]("MyTable", 'a, 'b, 'c)
- util.addFunction("random_udf", new NonDeterministicUdf)
+ util.addTemporarySystemFunction("random_udf", new NonDeterministicUdf)
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.scala
index 2ec64e94bc6..35d311c35ea 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/ProjectSemiAntiJoinTransposeRuleTest.scala
@@ -138,7 +138,7 @@ class ProjectSemiAntiJoinTransposeRuleTest extends TableTestBase {
@Test
def testTransposeProject_EmptyProject(): Unit = {
util.addTableSource[(Int, Long, String)]("MyTable3", 'i, 'j, 'k)
- util.addFunction("table_func", new StringSplit)
+ util.addTemporarySystemFunction("table_func", new StringSplit)
val sqlQuery = "SELECT * FROM MyTable1 WHERE EXISTS (" +
"SELECT * FROM MyTable2, " +
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRuleTest.scala
index 03b3fa9c746..f9ed439cfd0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PushPartitionIntoLegacyTableSourceScanRuleTest.scala
@@ -175,13 +175,13 @@ class PushPartitionIntoLegacyTableSourceScanRuleTest(
@TestTemplate
def testWithUdf(): Unit = {
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyRelPlan("SELECT * FROM MyTable WHERE id > 2 AND MyUdf(part2) < 3")
}
@TestTemplate
def testWithUdfAndVirtualColumn(): Unit = {
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyRelPlan("SELECT * FROM VirtualTable WHERE id > 2 AND MyUdf(part2) < 3")
}
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.scala
index 82af9cf2ead..d6c3f5ca876 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/PythonCorrelateSplitRuleTest.scala
@@ -53,9 +53,9 @@ class PythonCorrelateSplitRuleTest extends TableTestBase {
)
util.replaceStreamProgram(programs)
- util.addFunction("func", new MockPythonTableFunction)
- util.addFunction("javaFunc", new TableFunc1)
- util.addFunction("pyFunc", new PythonScalarFunction("pyFunc"))
+ util.addTemporarySystemFunction("func", new MockPythonTableFunction)
+ util.addTemporarySystemFunction("javaFunc", new TableFunc1)
+ util.addTemporarySystemFunction("pyFunc", new PythonScalarFunction("pyFunc"))
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.scala
index fd67aeac497..59f6545562b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromCorrelateRuleTest.scala
@@ -56,8 +56,8 @@ class SplitPythonConditionFromCorrelateRuleTest extends TableTestBase {
util.replaceBatchProgram(programs)
util.addTableSource[(Int, Int, String)]("MyTable", 'a, 'b, 'c)
- util.addFunction("func", new TableFunc2)
- util.addFunction("pyFunc", new PythonScalarFunction("pyFunc"))
+ util.addTemporarySystemFunction("func", new TableFunc2)
+ util.addTemporarySystemFunction("pyFunc", new PythonScalarFunction("pyFunc"))
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromJoinRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromJoinRuleTest.scala
index a2aa865016d..757409616ea 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromJoinRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/SplitPythonConditionFromJoinRuleTest.scala
@@ -55,7 +55,7 @@ class SplitPythonConditionFromJoinRuleTest extends TableTestBase {
util.addTableSource[(Int, Int, Int)]("leftTable", 'a, 'b, 'c)
util.addTableSource[(Int, Int, Int)]("rightTable", 'd, 'e, 'f)
- util.addFunction("pyFunc", new PythonScalarFunction("pyFunc"))
+ util.addTemporarySystemFunction("pyFunc", new PythonScalarFunction("pyFunc"))
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.scala
index 03509beb0b3..b12dd24818a 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQueryAntiJoinTest.scala
@@ -35,7 +35,7 @@ class SubQueryAntiJoinTest extends SubQueryTestBase {
util.addTableSource[(Int, Long)]("x", 'a, 'b)
util.addTableSource[(Int, Long)]("y", 'c, 'd)
util.addTableSource[(Int, Long)]("z", 'e, 'f)
- util.addFunction("table_func", new StringSplit)
+ util.addTemporarySystemFunction("table_func", new StringSplit)
@Test
def testNotInOnWhere_NotSubQuery(): Unit = {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala
index 6170dcb2998..4952f65ebda 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/subquery/SubQuerySemiJoinTest.scala
@@ -35,7 +35,7 @@ class SubQuerySemiJoinTest extends SubQueryTestBase {
util.addTableSource[(Int, Long)]("x", 'a, 'b)
util.addTableSource[(Int, Long)]("y", 'c, 'd)
util.addTableSource[(Int, Long)]("z", 'e, 'f)
- util.addFunction("table_func", new StringSplit)
+ util.addTemporarySystemFunction("table_func", new StringSplit)
@Test
def testInOnWhere_NotSubQuery(): Unit = {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala
index 72aceb8ab75..3e6a5fada0c 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/physical/batch/EnforceLocalSortAggRuleTest.scala
@@ -36,7 +36,7 @@ class EnforceLocalSortAggRuleTest extends EnforceLocalAggRuleTestBase {
@BeforeEach
override def setup(): Unit = {
super.setup()
- util.addFunction("weightedAvg", new WeightedAvg)
+ util.addTemporarySystemFunction("weightedAvg", new WeightedAvg)
val program = FlinkBatchProgram.buildProgram(util.tableEnv.getConfig)
// remove the original BatchExecSortAggRule and add BatchExecSortAggRuleForOnePhase
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/CalcTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/CalcTest.scala
index ce0b864d311..a0643c21e62 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/CalcTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/CalcTest.scala
@@ -36,7 +36,7 @@ class CalcTest extends TableTestBase {
@BeforeEach
def setup(): Unit = {
util.addTableSource[(Long, Int, String)]("MyTable", 'a, 'b, 'c)
- util.addFunction("random_udf", new NonDeterministicUdf)
+ util.addTemporarySystemFunction("random_udf", new NonDeterministicUdf)
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
index d2f10011ce1..f6f081fc172 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DagOptimizationTest.scala
@@ -97,7 +97,7 @@ class DagOptimizationTest extends TableTestBase {
@Test
def testSingleSinkWithUDTF(): Unit = {
util.addTableSource[(Int, Long, Int, String, Long)]("MyTable2", 'i, 'j, 'k, 'l, 'm)
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery =
"""
@@ -274,7 +274,7 @@ class DagOptimizationTest extends TableTestBase {
util.tableEnv.getConfig.set(
RelNodeBlockPlanBuilder.TABLE_OPTIMIZER_UNIONALL_AS_BREAKPOINT_ENABLED,
Boolean.box(false))
- util.addFunction("split", new TableFunc1)
+ util.addTemporarySystemFunction("split", new TableFunc1)
val sqlQuery1 =
"""
|SELECT a, b - MOD(b, 300) AS b, c FROM MyTable
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.scala
index 73ddddb52c7..2c3a090b1b9 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/LegacyTableSourceTest.scala
@@ -464,7 +464,7 @@ class LegacyTableSourceTest extends TableTestBase {
@Test
def testFilterPushDownWithUdf(): Unit = {
- util.addFunction("myUdf", Func1)
+ util.addTemporarySystemFunction("myUdf", Func1)
util.verifyExecPlan("SELECT * FROM FilterableTable WHERE amount > 2 AND myUdf(amount) < 32")
}
@@ -476,7 +476,7 @@ class LegacyTableSourceTest extends TableTestBase {
@Test
def testPartitionTableSourceWithUdf(): Unit = {
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyExecPlan("SELECT * FROM PartitionableTable WHERE id > 2 AND MyUdf(part2) < 3")
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/ModifiedMonotonicityTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/ModifiedMonotonicityTest.scala
index 6a22ccbb889..c78735fc111 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/ModifiedMonotonicityTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/ModifiedMonotonicityTest.scala
@@ -168,7 +168,7 @@ class ModifiedMonotonicityTest extends TableTestBase {
@Test
def testMultiOperandsForCalc(): Unit = {
- util.addFunction("func1", new Func1)
+ util.addTemporarySystemFunction("func1", new Func1)
val sql = "SELECT func1(func1(a1, a3)) from " +
"(SELECT last_value(a1) as a1, last_value(a3) as a3 FROM AA group by a2) "
verifyMonotonicity(sql, new RelModifiedMonotonicity(Array(NOT_MONOTONIC)))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/PartitionableSourceTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/PartitionableSourceTest.scala
index 441bb7d1c96..cddaafe1193 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/PartitionableSourceTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/PartitionableSourceTest.scala
@@ -126,7 +126,7 @@ class PartitionableSourceTest(val sourceFetchPartitions: Boolean, val useCatalog
@TestTemplate
def testWithUdfAndVirtualColumn(): Unit = {
- util.addFunction("MyUdf", Func1)
+ util.addTemporarySystemFunction("MyUdf", Func1)
util.verifyExecPlan("SELECT * FROM PartitionableTable WHERE id > 2 AND MyUdf(part2) < 3")
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala
index 2b195426e44..b3b76930e7a 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/RelTimeIndicatorConverterTest.scala
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.stream.sql
import org.apache.flink.api.scala._
+import org.apache.flink.table.annotation.{DataTypeHint, FunctionHint}
import org.apache.flink.table.api._
import org.apache.flink.table.api.internal.TableEnvironmentInternal
import org.apache.flink.table.data.TimestampData
@@ -83,7 +84,7 @@ class RelTimeIndicatorConverterTest extends TableTestBase {
@Test
def testTableFunction(): Unit = {
- util.addFunction("tableFunc", new TableFunc)
+ util.addTemporarySystemFunction("tableFunc", new TableFunc)
val sqlQuery =
"""
|SELECT rowtime, proctime, s
@@ -200,6 +201,14 @@ object RelTimeIndicatorConverterTest {
class TableFunc extends TableFunction[String] {
val t = new Timestamp(0L)
+ @FunctionHint(
+ input = Array(
+ new DataTypeHint(
+ value = "TIMESTAMP(3)",
+ bridgedTo = classOf[org.apache.flink.table.data.TimestampData]),
+ new DataTypeHint(value = "TIMESTAMP_LTZ(3)", bridgedTo = classOf[java.sql.Timestamp]),
+ new DataTypeHint("STRING")
+ ))
def eval(time1: TimestampData, time2: Timestamp, string: String): Unit = {
collect(time1.toString + time2.after(t) + string)
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.scala
index 5cd67937c6a..a34e6f5d724 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/SubplanReuseTest.scala
@@ -264,7 +264,7 @@ class SubplanReuseTest extends TableTestBase {
@Test
def testSubplanReuseOnCorrelate(): Unit = {
- util.addFunction("str_split", new StringSplit())
+ util.addTemporarySystemFunction("str_split", new StringSplit())
val sqlQuery =
"""
|WITH r AS (SELECT a, b, c, v FROM x, LATERAL TABLE(str_split(c, '-')) AS T(v))
@@ -277,7 +277,7 @@ class SubplanReuseTest extends TableTestBase {
@Test
def testSubplanReuseOnCorrelateWithNonDeterministicUDTF(): Unit = {
- util.addFunction("TableFun", new NonDeterministicTableFunc)
+ util.addTemporarySystemFunction("TableFun", new NonDeterministicTableFunc)
val sqlQuery =
"""
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala
index 3ef4d6b4694..65e6fb40eb9 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/agg/OverAggregateTest.scala
@@ -58,7 +58,7 @@ class OverAggregateTest extends TableTestBase {
/** OVER clause is necessary for [[OverAgg0]] window function. */
@Test
def testInvalidOverAggregation(): Unit = {
- util.addFunction("overAgg", new OverAgg0)
+ util.addTemporarySystemFunction("overAgg", new OverAgg0)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(() => util.verifyExecPlan("SELECT overAgg(c, a) FROM MyTable"))
}
@@ -66,7 +66,7 @@ class OverAggregateTest extends TableTestBase {
/** OVER clause is necessary for [[OverAgg0]] window function. */
@Test
def testInvalidOverAggregation2(): Unit = {
- util.addFunction("overAgg", new OverAgg0)
+ util.addTemporarySystemFunction("overAgg", new OverAgg0)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(() => util.verifyExecPlan("SELECT overAgg(c, a) FROM MyTable"))
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/IntervalJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/IntervalJoinTest.scala
index 4333151b1be..9d6f67d4fa1 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/IntervalJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/IntervalJoinTest.scala
@@ -139,7 +139,7 @@ class IntervalJoinTest extends TableTestBase {
*/
@Test
def testWindowOuterJoinWithPythonFunctionInCondition(): Unit = {
- util.addFunction("pyFunc", new PythonScalarFunction("pyFunc"))
+ util.addTemporarySystemFunction("pyFunc", new PythonScalarFunction("pyFunc"))
val sql =
"""
|SELECT t1.a, t2.b FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.scala
index f1b8c9d584a..4a3e9b3a660 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.scala
@@ -30,7 +30,7 @@ class SemiAntiJoinTest extends TableTestBase {
util.addTableSource[(Int, Long, String)]("l", 'a, 'b, 'c)
util.addTableSource[(Int, Long, String)]("r", 'd, 'e, 'f)
util.addTableSource[(Int, Long, String)]("t", 'i, 'j, 'k)
- util.addFunction("table_func", new StringSplit)
+ util.addTemporarySystemFunction("table_func", new StringSplit)
@Test
def testInWithUncorrelated_SimpleCondition1(): Unit = {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.scala
index 2d5ea2c8634..00c980d7a9f 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/ColumnFunctionsTest.scala
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.stream.table
import org.apache.flink.api.scala._
+import org.apache.flink.table.annotation.DataTypeHint
import org.apache.flink.table.api._
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.table.planner.plan.utils.JavaUserDefinedAggFunctions.{CountDistinct, WeightedAvg}
@@ -35,7 +36,7 @@ class ColumnFunctionsTest extends TableTestBase {
val t = util.addTableSource[(Double, Long)]('double, 'long)
- util.addFunction("TestFunc", TestFunc)
+ util.addTemporarySystemFunction("TestFunc", TestFunc)
val tab1 = t.select(call("TestFunc", withColumns('*)))
util.verifyExecPlan(tab1)
}
@@ -87,7 +88,7 @@ class ColumnFunctionsTest extends TableTestBase {
def testJoinLateral(): Unit = {
val t = util.addTableSource[(Double, Long, String)]('int, 'long, 'string)
val func0 = new TableFunc0
- util.addFunction("func0", func0)
+ util.addTemporarySystemFunction("func0", func0)
val tab1 = t.joinLateral(func0(withColumns('string)))
util.verifyExecPlan(tab1)
@@ -132,9 +133,9 @@ class ColumnFunctionsTest extends TableTestBase {
val weightAvgFun = new WeightedAvg
val countDist = new CountDistinct
- util.addFunction("countFun", countFun)
+ util.addTemporarySystemFunction("countFun", countFun)
util.addTemporarySystemFunction("weightAvgFun", weightAvgFun)
- util.addFunction("countDist", countDist)
+ util.addTemporarySystemFunction("countDist", countDist)
val tab1 = table
.window(Over.partitionBy(withColumns('c)).orderBy('proctime).preceding(UNBOUNDED_ROW).as('w))
@@ -152,7 +153,7 @@ class ColumnFunctionsTest extends TableTestBase {
def testAddColumns(): Unit = {
val t = util.addTableSource[(Double, Long, String)]('a, 'b, 'c)
- util.addFunction("TestFunc", TestFunc)
+ util.addTemporarySystemFunction("TestFunc", TestFunc)
val tab1 = t.addColumns(call("TestFunc", withColumns('a, 'b)).as('d))
util.verifyExecPlan(tab1)
}
@@ -176,7 +177,7 @@ class ColumnFunctionsTest extends TableTestBase {
@SerialVersionUID(1L)
object TestFunc extends ScalarFunction {
- def eval(a: Double, b: Long): Double = {
+ def eval(@DataTypeHint("DOUBLE") a: Double, @DataTypeHint("BIGINT") b: Long): Double = {
a
}
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/PythonCalcTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/PythonCalcTest.scala
index e5e076a534f..5c323bbe977 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/PythonCalcTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/PythonCalcTest.scala
@@ -30,7 +30,7 @@ class PythonCalcTest extends TableTestBase {
@BeforeEach
def setup(): Unit = {
util.addTableSource[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
- util.addFunction("pyFunc1", new PythonScalarFunction("pyFunc1"))
+ util.addTemporarySystemFunction("pyFunc1", new PythonScalarFunction("pyFunc1"))
}
@Test
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/TemporalTableFunctionJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/TemporalTableFunctionJoinTest.scala
index af2be420d32..47a915290df 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/TemporalTableFunctionJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/TemporalTableFunctionJoinTest.scala
@@ -110,7 +110,7 @@ class TemporalTableFunctionJoinTest extends TableTestBase {
val rates = ratesHistory
.filter('rate > 110L)
.createTemporalTableFunction('rowtime, 'currency)
- util.addFunction("Rates", rates)
+ util.addTemporarySystemFunction("Rates", rates)
val result = orders
.joinLateral(rates('o_rowtime))
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/AggregateValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/AggregateValidationTest.scala
index cb8951f1a3d..cc47e1c2aeb 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/AggregateValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/AggregateValidationTest.scala
@@ -87,15 +87,13 @@ class AggregateValidationTest extends TableTestBase {
def testTableFunctionInSelection(): Unit = {
val table = util.addTableSource[(Long, Int, String)]('a, 'b, 'c)
- util.addFunction("func", new TableFunc0)
+ util.addTemporarySystemFunction("func", new TableFunc0)
val resultTable = table
.groupBy('a)
.aggregate('b.sum.as('d))
- // must fail. Cannot use TableFunction in select after aggregate
.select(call("func", "abc"))
- assertThatExceptionOfType(classOf[RuntimeException])
- .isThrownBy(() => util.verifyExecPlan(resultTable))
+ util.verifyExecPlan(resultTable)
}
@Test
@@ -116,7 +114,7 @@ class AggregateValidationTest extends TableTestBase {
def testInvalidTableFunctionInAggregate(): Unit = {
val table = util.addTableSource[(Long, Int, String)]('a, 'b, 'c)
- util.addFunction("func", new TableFunc0)
+ util.addTemporarySystemFunction("func", new TableFunc0)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(
() =>
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CalcValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CalcValidationTest.scala
index 661781dd1b8..3b56fed4159 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CalcValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CalcValidationTest.scala
@@ -181,7 +181,7 @@ class CalcValidationTest extends TableTestBase {
def testInvalidMapFunctionTypeUDAGG2(): Unit = {
val util = streamTestUtil()
- util.addFunction("weightedAvg", new WeightedAvg)
+ util.addTemporarySystemFunction("weightedAvg", new WeightedAvg)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(
@@ -195,7 +195,7 @@ class CalcValidationTest extends TableTestBase {
def testInvalidMapFunctionTypeTableFunction(): Unit = {
val util = streamTestUtil()
- util.addFunction("func", new TableFunc0)
+ util.addTemporarySystemFunction("func", new TableFunc0)
assertThatExceptionOfType(classOf[ValidationException])
.isThrownBy(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CorrelateValidationTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CorrelateValidationTest.scala
index 8c755f9b87b..ae97fbd7c9e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CorrelateValidationTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/table/validation/CorrelateValidationTest.scala
@@ -35,7 +35,9 @@ class CorrelateValidationTest extends TableTestBase {
val t = util.addTableSource[(Int, Long, String)]('a, 'b, 'c)
// check scala object is forbidden
- expectExceptionThrown(util.addFunction("func3", ObjectTableFunction), "Scala object")
+ expectExceptionThrown(
+ util.addTemporarySystemFunction("func3", ObjectTableFunction),
+ "Could not register temporary system function 'func3' due to implementation errors.")
expectExceptionThrown(t.joinLateral(ObjectTableFunction('a, 1)), "Scala object")
}
@@ -46,9 +48,13 @@ class CorrelateValidationTest extends TableTestBase {
// =================== check scala object is forbidden =====================
// Scala table environment register
- expectExceptionThrown(util.addFunction("udtf", ObjectTableFunction), "Scala object")
+ expectExceptionThrown(
+ util.addTemporarySystemFunction("udtf", ObjectTableFunction),
+ "Could not register temporary system function 'udtf' due to implementation errors.")
// Java table environment register
- expectExceptionThrown(util.addFunction("udtf", ObjectTableFunction), "Scala object")
+ expectExceptionThrown(
+ util.addTemporarySystemFunction("udtf", ObjectTableFunction),
+ "Could not register temporary system function 'udtf' due to implementation errors.")
// Scala Table API directly call
expectExceptionThrown(t.joinLateral(ObjectTableFunction('a, 1)), "Scala object")
@@ -61,7 +67,7 @@ class CorrelateValidationTest extends TableTestBase {
"No match found for function signature nonexist(<NUMERIC>)")
// ========= throw exception when the called function is a scalar function ====
- util.addFunction("func0", Func0)
+ util.addTemporarySystemFunction("func0", Func0)
// SQL API call
expectExceptionThrown(
@@ -71,14 +77,14 @@ class CorrelateValidationTest extends TableTestBase {
// ========== throw exception when the parameters is not correct ===============
// Java Table API call
- util.addFunction("func2", new TableFunc2)
+ util.addTemporarySystemFunction("func2", new TableFunc2)
expectExceptionThrown(
t.joinLateral(call("func2", $"c", $"c")),
- "Given parameters of function 'func2' do not match any signature")
+ "Invalid function call:\nfunc2(STRING, STRING)")
// SQL API call
expectExceptionThrown(
util.tableEnv.sqlQuery("SELECT * FROM MyTable, LATERAL TABLE(func2(c, c))"),
- "Given parameters of function 'func2' do not match any signature.")
+ "No match found for function signature func2(<CHARACTER>, <CHARACTER>).")
}
/**
@@ -90,7 +96,7 @@ class CorrelateValidationTest extends TableTestBase {
val util = streamTestUtil()
val table = util.addTableSource[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
val function = new TableFunc1
- util.addFunction("func1", function)
+ util.addTemporarySystemFunction("func1", function)
expectExceptionThrown(
{
@@ -131,7 +137,7 @@ class CorrelateValidationTest extends TableTestBase {
def testInvalidMapFunctionTypeUDAGG2(): Unit = {
val util = streamTestUtil()
- util.addFunction("weightedAvg", new WeightedAvg)
+ util.addTemporarySystemFunction("weightedAvg", new WeightedAvg)
expectExceptionThrown(
util
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/CountAggFunction.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/CountAggFunction.scala
index eed8499a599..394565a5f4e 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/CountAggFunction.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/CountAggFunction.scala
@@ -20,8 +20,10 @@ package org.apache.flink.table.planner.utils
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.api.java.tuple.{Tuple1 => JTuple1}
import org.apache.flink.api.java.typeutils.TupleTypeInfo
+import org.apache.flink.table.annotation.{DataTypeHint, FunctionHint}
import org.apache.flink.table.api.Types
import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.table.planner.JInt
import java.lang.{Iterable => JIterable, Long => JLong}
@@ -33,17 +35,15 @@ class CountAccumulator extends JTuple1[JLong] {
/** built-in count aggregate function */
class CountAggFunction extends AggregateFunction[JLong, CountAccumulator] {
- def accumulate(acc: CountAccumulator, value: Any): Unit = {
- if (value != null) {
- acc.f0 += 1L
- }
+ def accumulate(acc: CountAccumulator, @DataTypeHint("INT") value: Any): Unit = {
+ if (value != null) acc.f0 += 1L
}
def accumulate(acc: CountAccumulator): Unit = {
acc.f0 += 1L
}
- def retract(acc: CountAccumulator, value: Any): Unit = {
+ def retract(acc: CountAccumulator, @DataTypeHint("INT") value: Any): Unit = {
if (value != null) {
acc.f0 -= 1L
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index d8920bc62c0..26193a650f4 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -292,18 +292,6 @@ abstract class TableTestUtilBase(test: TableTestBase, isStreamingMode: Boolean)
getTableEnv.from(name)
}
- /**
- * Registers a [[ScalarFunction]] under given name into the TableEnvironment's catalog.
- *
- * @deprecated
- * Use [[addTemporarySystemFunction]].
- */
- @deprecated
- @Deprecated
- def addFunction(name: String, function: ScalarFunction): Unit = {
- getTableEnv.registerFunction(name, function)
- }
-
/** Registers a [[UserDefinedFunction]] according to FLIP-65. */
def addTemporarySystemFunction(name: String, function: UserDefinedFunction): Unit = {
getTableEnv.createTemporarySystemFunction(name, function)
@@ -1226,28 +1214,6 @@ abstract class TableTestUtil(
testingTableEnv.createTemporaryView(name, table)
testingTableEnv.from(name)
}
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation](name: String, function: TableFunction[T]): Unit =
- testingTableEnv.registerFunction(name, function)
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation, ACC: TypeInformation](
- name: String,
- function: AggregateFunction[T, ACC]): Unit = testingTableEnv.registerFunction(name, function)
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation, ACC: TypeInformation](
- name: String,
- function: TableAggregateFunction[T, ACC]): Unit = {
- testingTableEnv.registerFunction(name, function)
- }
}
abstract class ScalaTableTestUtil(test: TableTestBase, isStreamingMode: Boolean)
@@ -1258,26 +1224,6 @@ abstract class ScalaTableTestUtil(test: TableTestBase, isStreamingMode: Boolean)
val tableEnv: ScalaStreamTableEnv = ScalaStreamTableEnv.create(env, setting)
override def getTableEnv: TableEnvironment = tableEnv
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation](name: String, function: TableFunction[T]): Unit =
- tableEnv.registerFunction(name, function)
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation, ACC: TypeInformation](
- name: String,
- function: AggregateFunction[T, ACC]): Unit = tableEnv.registerFunction(name, function)
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation, ACC: TypeInformation](
- name: String,
- function: TableAggregateFunction[T, ACC]): Unit = tableEnv.registerFunction(name, function)
}
abstract class JavaTableTestUtil(test: TableTestBase, isStreamingMode: Boolean)
@@ -1288,26 +1234,6 @@ abstract class JavaTableTestUtil(test: TableTestBase, isStreamingMode: Boolean)
val tableEnv: JavaStreamTableEnv = JavaStreamTableEnv.create(env, setting)
override def getTableEnv: TableEnvironment = tableEnv
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation](name: String, function: TableFunction[T]): Unit =
- tableEnv.registerFunction(name, function)
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation, ACC: TypeInformation](
- name: String,
- function: AggregateFunction[T, ACC]): Unit = tableEnv.registerFunction(name, function)
-
- /** @deprecated Use [[addTemporarySystemFunction()]] for the new type inference. */
- @deprecated
- @Deprecated
- def addFunction[T: TypeInformation, ACC: TypeInformation](
- name: String,
- function: TableAggregateFunction[T, ACC]): Unit = tableEnv.registerFunction(name, function)
}
/** Utility for stream table test. */