Skip to content

[FLINK-31597][table] Cleanup usage of deprecated TableEnvironment#registerFunction #23751

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
snuyanzin merged 2 commits into from
Nov 22, 2023
Merged

[FLINK-31597][table] Cleanup usage of deprecated TableEnvironment#registerFunction #23751

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ae7c7e5
[FLINK-33591][table] Cleanup usage of deprecated TableTestBase#addFun…
snuyanzin Nov 18, 2023
ae90c41
[FLINK-31597][table] Cleanup usage of deprecated TableEnvironment#reg…
snuyanzin Nov 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/themes/book
Open in desktop
Submodule book updated 102 files
2 changes: 1 addition & 1 deletion ...che/flink/table/runtime/operators/python/scalar/PythonScalarFunctionOperatorTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,7 +220,7 @@ public void testPythonScalarFunctionOperatorIsChainedByDefault() {
env.setParallelism(1);
StreamTableEnvironment tEnv = createTableEnvironment(env);
tEnv.getConfig().set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.parse("80mb"));
tEnv.registerFunction("pyFunc", new PythonScalarFunction("pyFunc"));
tEnv.createTemporarySystemFunction("pyFunc", new PythonScalarFunction("pyFunc"));
DataStream<Tuple2<Integer, Integer>> ds = env.fromElements(new Tuple2<>(1, 2));
Table t = tEnv.fromDataStream(ds, $("a"), $("b")).select(call("pyFunc", $("a"), $("b")));
// force generating the physical plan for the given table
Expand Down
3 changes: 2 additions & 1 deletion ...e-planner/src/test/java/org/apache/flink/table/planner/plan/common/CalcMergeTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ void setup() {
+ isBatchMode()
+ "'\n"
+ ")");
util.addFunction("random_udf", new JavaUserDefinedScalarFunctions.NonDeterministicUdf());
util.addTemporarySystemFunction(
"random_udf", new JavaUserDefinedScalarFunctions.NonDeterministicUdf());
}

@Test
Expand Down
2 changes: 1 addition & 1 deletion ...che/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ void testUnconvertedExpression() {

@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");
}

Expand Down
5 changes: 3 additions & 2 deletions .../src/test/java/org/apache/flink/table/planner/plan/utils/JavaUserDefinedAggFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
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;
Expand Down Expand Up @@ -297,7 +298,7 @@ public CountDistinctAccum createAccumulator() {
}

// 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) {
Expand All @@ -313,7 +314,7 @@ public void accumulate(CountDistinctAccum accumulator, String id) {
}

// 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) {
Expand Down
5 changes: 4 additions & 1 deletion ...c/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedAggFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,10 @@ public void accumulate(
}

// 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;
}
Expand Down
2 changes: 1 addition & 1 deletion ...est/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ public int eval() {
return random.nextInt();
}

public int eval(int v) {
public int eval(@DataTypeHint("INT") int v) {
return v + random.nextInt();
}

Expand Down
12 changes: 6 additions & 6 deletions .../src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DagOptimizationTest.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}'))])
Expand All @@ -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}'))])
Expand All @@ -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])

Expand Down Expand Up @@ -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])
Expand Down
2 changes: 1 addition & 1 deletion ...rc/test/resources/org/apache/flink/table/planner/plan/batch/sql/LegacyTableSourceTest.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
26 changes: 13 additions & 13 deletions .../src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveCollationTest.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 cnt])\n +- Sort(orderBy=[f ASC])\n +- [#2] Exchange(distribution=[hash[f]])\n])
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(count$0) AS cnt])\n +- Sort(orderBy=[f ASC])\n +- [#2] Exchange(distribution=[hash[f]])\n])
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
Expand Down Expand Up @@ -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 +- Sort(orderBy=[f ASC])\n +- [#2] Exchange(distribution=[hash[f]])\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) AS cnt])\n +- Sort(orderBy=[f ASC])\n +- [#2] Exchange(distribution=[hash[f]])\n])
:- Exchange(distribution=[hash[c]])
: +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+- Exchange(distribution=[hash[f]])
Expand Down Expand Up @@ -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">
Expand All @@ -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])
Expand Down Expand Up @@ -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','])
Expand All @@ -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)]]])
Expand All @@ -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])
Expand All @@ -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])
Expand Down
Loading