You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2020/07/13 11:13:57 UTC
[flink] branch master updated: [FLINK-18528][table] Update UNNEST
to new type system
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new a028ba2 [FLINK-18528][table] Update UNNEST to new type system
a028ba2 is described below
commit a028ba216a35a70c4f2d337bdede195bf7701192
Author: Timo Walther <tw...@apache.org>
AuthorDate: Wed Jul 8 13:48:35 2020 +0200
[FLINK-18528][table] Update UNNEST to new type system
This closes #12862.
---
.../apache/flink/table/types/logical/RowType.java | 8 +-
.../flink/table/types/utils/DataTypeUtils.java | 9 +
.../functions/bridging/BridgingSqlFunction.java | 35 ++++
.../planner/plan/QueryOperationConverter.java | 14 +-
.../plan/rules/logical/LogicalUnnestRule.scala | 89 +++-------
.../planner/plan/utils/ExplodeFunctionUtil.scala | 178 -------------------
.../table/planner/plan/batch/sql/UnnestTest.xml | 20 +--
.../plan/rules/logical/LogicalUnnestRuleTest.xml | 20 +--
.../table/planner/plan/stream/sql/UnnestTest.xml | 20 +--
.../table/runtime/functions/SqlUnnestUtils.java | 189 +++++++++++++++++++++
10 files changed, 292 insertions(+), 290 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/RowType.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/RowType.java
index 8fe64ab..1a740e8 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/RowType.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/RowType.java
@@ -280,11 +280,15 @@ public final class RowType extends LogicalType {
}
public static RowType of(LogicalType... types) {
- List<RowField> fields = new ArrayList<>();
+ return of(true, types);
+ }
+
+ public static RowType of(boolean isNullable, LogicalType... types) {
+ final List<RowField> fields = new ArrayList<>();
for (int i = 0; i < types.length; i++) {
fields.add(new RowField("f" + i, types[i]));
}
- return new RowType(fields);
+ return new RowType(isNullable, fields);
}
public static RowType of(LogicalType[] types, String[] names) {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/DataTypeUtils.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/DataTypeUtils.java
index 1157c37..29bb6a7 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/DataTypeUtils.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/DataTypeUtils.java
@@ -30,6 +30,7 @@ import org.apache.flink.table.types.DataTypeVisitor;
import org.apache.flink.table.types.FieldsDataType;
import org.apache.flink.table.types.KeyValueDataType;
import org.apache.flink.table.types.inference.TypeTransformation;
+import org.apache.flink.table.types.inference.TypeTransformations;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.DistinctType;
import org.apache.flink.table.types.logical.LegacyTypeInformationType;
@@ -59,6 +60,14 @@ import static org.apache.flink.table.types.logical.utils.LogicalTypeUtils.toInte
public final class DataTypeUtils {
/**
+ * Creates a {@link DataType} from the given {@link LogicalType} with internal data structures.
+ */
+ public static DataType toInternalDataType(LogicalType logicalType) {
+ final DataType defaultDataType = TypeConversions.fromLogicalToDataType(logicalType);
+ return transform(defaultDataType, TypeTransformations.TO_INTERNAL_CLASS);
+ }
+
+ /**
* Checks whether a given data type is an internal data structure.
*/
public static boolean isInternal(DataType dataType) {
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
index 45cc033..a0bc987 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
@@ -23,10 +23,13 @@ import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.FunctionIdentifier;
import org.apache.flink.table.functions.FunctionKind;
+import org.apache.flink.table.planner.calcite.FlinkContext;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.sql.SqlFunction;
import org.apache.calcite.sql.SqlKind;
@@ -118,6 +121,38 @@ public final class BridgingSqlFunction extends SqlFunction {
typeInference);
}
+ /**
+ * Creates an instance of a scalar or table function during translation.
+ */
+ public static BridgingSqlFunction of(
+ FlinkContext context,
+ FlinkTypeFactory typeFactory,
+ @Nullable FunctionIdentifier identifier,
+ FunctionDefinition definition) {
+ final DataTypeFactory dataTypeFactory = context.getCatalogManager().getDataTypeFactory();
+ final TypeInference typeInference = definition.getTypeInference(dataTypeFactory);
+ return of(
+ dataTypeFactory,
+ typeFactory,
+ SqlKind.OTHER_FUNCTION,
+ identifier,
+ definition,
+ typeInference
+ );
+ }
+
+ /**
+ * Creates an instance of a scalar or table function during translation.
+ */
+ public static BridgingSqlFunction of(
+ RelOptCluster cluster,
+ @Nullable FunctionIdentifier identifier,
+ FunctionDefinition definition) {
+ final FlinkContext context = ShortcutUtils.unwrapContext(cluster);
+ final FlinkTypeFactory typeFactory = ShortcutUtils.unwrapTypeFactory(cluster);
+ return of(context, typeFactory, identifier, definition);
+ }
+
public DataTypeFactory getDataTypeFactory() {
return dataTypeFactory;
}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
index 04dc531..78326e5 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
@@ -24,7 +24,6 @@ import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.catalog.CatalogManager;
import org.apache.flink.table.catalog.ConnectorCatalogTable;
-import org.apache.flink.table.catalog.DataTypeFactory;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.expressions.CallExpression;
@@ -82,7 +81,6 @@ import org.apache.flink.table.planner.plan.schema.LegacyTableSourceTable;
import org.apache.flink.table.planner.plan.schema.TypedFlinkTableFunction;
import org.apache.flink.table.planner.plan.stats.FlinkStatistic;
import org.apache.flink.table.planner.sources.TableSourceUtil;
-import org.apache.flink.table.planner.utils.ShortcutUtils;
import org.apache.flink.table.sources.LookupableTableSource;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.table.sources.TableSource;
@@ -99,7 +97,6 @@ import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlAggFunction;
-import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.tools.RelBuilder.AggCall;
import org.apache.calcite.tools.RelBuilder.GroupKey;
@@ -290,17 +287,10 @@ public class QueryOperationConverter extends QueryOperationDefaultVisitor<RelNod
typeFactory);
}
- DataTypeFactory dataTypeFactory = ShortcutUtils.unwrapContext(relBuilder.getCluster())
- .getCatalogManager()
- .getDataTypeFactory();
-
final BridgingSqlFunction sqlFunction = BridgingSqlFunction.of(
- dataTypeFactory,
- typeFactory,
- SqlKind.OTHER_FUNCTION,
+ relBuilder.getCluster(),
calculatedTable.getFunctionIdentifier().orElse(null),
- calculatedTable.getFunctionDefinition(),
- calculatedTable.getFunctionDefinition().getTypeInference(dataTypeFactory));
+ calculatedTable.getFunctionDefinition());
return relBuilder.functionScan(
sqlFunction,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRule.scala
index 5a4f792..b1cc3ec 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRule.scala
@@ -18,28 +18,20 @@
package org.apache.flink.table.planner.plan.rules.logical
-import org.apache.flink.api.java.typeutils.MapTypeInfo
-import org.apache.flink.table.api.TableException
-import org.apache.flink.table.functions.FunctionIdentifier
-import org.apache.flink.table.planner.calcite.FlinkTypeFactory
-import org.apache.flink.table.planner.calcite.FlinkTypeFactory.toLogicalType
-import org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils
-import org.apache.flink.table.planner.plan.utils.ExplodeFunctionUtil
-import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter.fromLogicalTypeToDataType
-import org.apache.flink.table.runtime.types.TypeInfoLogicalTypeConverter.fromLogicalTypeToTypeInfo
-import org.apache.flink.table.types.logical.RowType
+import java.util.Collections
+
import com.google.common.collect.ImmutableList
import org.apache.calcite.plan.RelOptRule._
import org.apache.calcite.plan.hep.HepRelVertex
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelOptRuleOperand}
import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.`type`.{RelDataTypeFieldImpl, RelRecordType, StructKind}
import org.apache.calcite.rel.core.Uncollect
import org.apache.calcite.rel.logical._
-import org.apache.calcite.sql.`type`.{AbstractSqlType, ArraySqlType, MapSqlType, MultisetSqlType}
-import java.util.Collections
-
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTypeFactory
+import org.apache.flink.table.functions.FunctionIdentifier
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction
+import org.apache.flink.table.planner.utils.ShortcutUtils
+import org.apache.flink.table.runtime.functions.SqlUnnestUtils
/**
* Planner rule that rewrites UNNEST to explode function.
@@ -95,68 +87,29 @@ class LogicalUnnestRule(
case uc: Uncollect =>
// convert Uncollect into TableFunctionScan
val cluster = correlate.getCluster
- val dataType = uc.getInput.getRowType.getFieldList.get(0).getValue
- val (componentType, explodeTableFunc) = dataType match {
- case arrayType: ArraySqlType =>
- (arrayType.getComponentType,
- ExplodeFunctionUtil.explodeTableFuncFromType(
- fromLogicalTypeToTypeInfo(toLogicalType(arrayType))))
-
- case map: MapSqlType =>
- val keyType = toLogicalType(map.getKeyType)
- val valueType = toLogicalType(map.getValueType)
- val rowInternalType = RowType.of(keyType, valueType)
- val componentType = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
- .createFieldTypeFromLogicalType(rowInternalType)
- val mapTypeInfo = new MapTypeInfo(
- fromLogicalTypeToTypeInfo(keyType),
- fromLogicalTypeToTypeInfo(valueType)
- )
- val explodeFunction = ExplodeFunctionUtil.explodeTableFuncFromType(mapTypeInfo)
- (componentType, explodeFunction)
-
- case mt: MultisetSqlType =>
- (mt.getComponentType,
- ExplodeFunctionUtil.explodeTableFuncFromType(
- fromLogicalTypeToTypeInfo(toLogicalType(mt))))
- case _ => throw new TableException(s"Unsupported UNNEST on type: ${dataType.toString}")
- }
+ val typeFactory = ShortcutUtils.unwrapTypeFactory(cluster)
+ val relDataType = uc.getInput.getRowType.getFieldList.get(0).getValue
+ val logicalType = FlinkTypeFactory.toLogicalType(relDataType)
- // create sql function
- val explodeSqlFunc = UserDefinedFunctionUtils.createTableSqlFunction(
- FunctionIdentifier.of("explode"),
- "explode",
- explodeTableFunc,
- fromLogicalTypeToDataType(toLogicalType(componentType)),
- cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory])
+ val unnestFunction = SqlUnnestUtils.createUnnestFunction(logicalType)
+
+ val sqlFunction = BridgingSqlFunction.of(
+ cluster,
+ FunctionIdentifier.of("UNNEST"),
+ unnestFunction)
- // create table function call
- // TODO use BridgingSqlFunction once we remove TableSqlFunction
val rexCall = cluster.getRexBuilder.makeCall(
- explodeSqlFunc.getRowType(unwrapTypeFactory(cluster), Collections.emptyList()),
- explodeSqlFunc,
- getRel(uc.getInput).asInstanceOf[LogicalProject].getChildExps
- )
-
- // determine rel data type of unnest
- val rowType = componentType match {
- case _: AbstractSqlType =>
- new RelRecordType(
- StructKind.FULLY_QUALIFIED,
- ImmutableList.of(new RelDataTypeFieldImpl("f0", 0, componentType)))
- case _: RelRecordType => componentType
- case _ => throw new TableException(
- s"Unsupported multiset component type in UNNEST: ${componentType.toString}")
- }
+ typeFactory.createFieldTypeFromLogicalType(unnestFunction.getWrappedOutputType),
+ sqlFunction,
+ getRel(uc.getInput).asInstanceOf[LogicalProject].getChildExps)
- // create table function scan
new LogicalTableFunctionScan(
cluster,
correlate.getTraitSet,
Collections.emptyList(),
rexCall,
- classOf[Array[Object]],
- rowType,
+ null,
+ rexCall.getType,
null)
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/ExplodeFunctionUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/ExplodeFunctionUtil.scala
deleted file mode 100644
index a817b4e..0000000
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/ExplodeFunctionUtil.scala
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.table.planner.plan.utils
-
-import org.apache.flink.api.common.typeinfo.{BasicArrayTypeInfo, BasicTypeInfo, PrimitiveArrayTypeInfo, TypeInformation}
-import org.apache.flink.api.java.typeutils.{MapTypeInfo, MultisetTypeInfo, ObjectArrayTypeInfo}
-import org.apache.flink.table.api.Types
-import org.apache.flink.table.functions.TableFunction
-import org.apache.flink.types.Row
-
-import java.util
-
-import scala.collection.JavaConverters._
-
-@SerialVersionUID(1L)
-class ObjectExplodeTableFunc(componentType: TypeInformation[_]) extends TableFunction[Object] {
- def eval(arr: Array[Object]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Object, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-
- override def getParameterTypes(signature: Array[Class[_]]): Array[TypeInformation[_]] = {
- if (signature.head == classOf[Array[Object]]) {
- Array(Types.OBJECT_ARRAY(componentType))
- } else {
- Array(Types.MAP(componentType, Types.INT))
- }
- }
-}
-
-@SerialVersionUID(1L)
-class FloatExplodeTableFunc extends TableFunction[Float] {
- def eval(arr: Array[Float]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Float, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-}
-
-@SerialVersionUID(1L)
-class ShortExplodeTableFunc extends TableFunction[Short] {
- def eval(arr: Array[Short]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Short, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-}
-
-@SerialVersionUID(1L)
-class IntExplodeTableFunc extends TableFunction[Int] {
- def eval(arr: Array[Int]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Int, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-}
-
-@SerialVersionUID(1L)
-class LongExplodeTableFunc extends TableFunction[Long] {
- def eval(arr: Array[Long]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Long, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-}
-
-@SerialVersionUID(1L)
-class DoubleExplodeTableFunc extends TableFunction[Double] {
- def eval(arr: Array[Double]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Double, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-}
-
-@SerialVersionUID(1L)
-class ByteExplodeTableFunc extends TableFunction[Byte] {
- def eval(arr: Array[Byte]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Byte, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-}
-
-@SerialVersionUID(1L)
-class BooleanExplodeTableFunc extends TableFunction[Boolean] {
- def eval(arr: Array[Boolean]): Unit = {
- arr.foreach(collect)
- }
-
- def eval(map: util.Map[Boolean, Integer]): Unit = {
- CommonCollect.collect(map, collect)
- }
-}
-
-@SerialVersionUID(1L)
-class MapExplodeTableFunc extends TableFunction[Row] {
- def eval(map: util.Map[Object, Object]): Unit = {
- map.asScala.foreach { case (key, value) =>
- collect(Row.of(key, value))
- }
- }
-}
-
-object CommonCollect {
- def collect[T](map: util.Map[T, Integer], collectFunc: (T) => Unit): Unit = {
- map.asScala.foreach { e =>
- for (i <- 0 until e._2) {
- collectFunc(e._1)
- }
- }
- }
-}
-
-object ExplodeFunctionUtil {
- def explodeTableFuncFromType(ti: TypeInformation[_]): TableFunction[_] = {
- ti match {
- case pat: PrimitiveArrayTypeInfo[_] => createTableFuncByType(pat.getComponentType)
-
- case t: ObjectArrayTypeInfo[_, _] => new ObjectExplodeTableFunc(t.getComponentInfo)
-
- case t: BasicArrayTypeInfo[_, _] => new ObjectExplodeTableFunc(t.getComponentInfo)
-
- case mt: MultisetTypeInfo[_] => createTableFuncByType(mt.getElementTypeInfo)
-
- // should match MultisetTypeInfo before MapTypeInfo
- // due to MultisetTypeInfo is a subclass of MapTypeInfo
- case _: MapTypeInfo[_, _] => new MapExplodeTableFunc
-
- case _ => throw new UnsupportedOperationException(ti.toString + " IS NOT supported")
- }
- }
-
- def createTableFuncByType(typeInfo: TypeInformation[_]): TableFunction[_] = {
- typeInfo match {
- case BasicTypeInfo.INT_TYPE_INFO => new IntExplodeTableFunc
- case BasicTypeInfo.LONG_TYPE_INFO => new LongExplodeTableFunc
- case BasicTypeInfo.SHORT_TYPE_INFO => new ShortExplodeTableFunc
- case BasicTypeInfo.FLOAT_TYPE_INFO => new FloatExplodeTableFunc
- case BasicTypeInfo.DOUBLE_TYPE_INFO => new DoubleExplodeTableFunc
- case BasicTypeInfo.BYTE_TYPE_INFO => new ByteExplodeTableFunc
- case BasicTypeInfo.BOOLEAN_TYPE_INFO => new BooleanExplodeTableFunc
- case BasicTypeInfo.BOOLEAN_TYPE_INFO => new BooleanExplodeTableFunc
- case _ => new ObjectExplodeTableFunc(typeInfo)
- }
- }
-}
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml
index 192da32..f0f9acd 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/UnnestTest.xml
@@ -34,7 +34,7 @@ LogicalProject(a=[$0], s=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, f0 AS s])
-+- Correlate(invocation=[explode($cor0.c)], correlate=[table(explode($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) ARRAY c, VARCHAR(2147483647) f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.c)], correlate=[table(UNNEST($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) ARRAY c, VARCHAR(2147483647) f0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -67,7 +67,7 @@ LogicalProject(b=[$0], s=[$2])
<Resource name="planAfter">
<![CDATA[
Calc(select=[b, f0 AS s])
-+- Correlate(invocation=[explode($cor0.set)], correlate=[table(explode($cor0.set))], select=[b,set,f0], rowType=[RecordType(BIGINT b, BIGINT MULTISET set, BIGINT f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.set)], correlate=[table(UNNEST($cor0.set))], select=[b,set,f0], rowType=[RecordType(BIGINT b, BIGINT MULTISET set, BIGINT f0)], joinType=[INNER])
+- SortWindowAggregate(groupBy=[b], window=[TumblingGroupWindow('w$, rowtime, 3000)], select=[b, Final_COLLECT(set) AS set])
+- Sort(orderBy=[b ASC, assignedWindow$ ASC])
+- Exchange(distribution=[hash[b]])
@@ -96,7 +96,7 @@ LogicalProject(a=[$0], b=[$1], v=[$4])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, b, f1 AS v])
-+- Correlate(invocation=[explode($cor0.c)], correlate=[table(explode($cor0.c))], select=[a,b,c,f0,f1], rowType=[RecordType(INTEGER a, BIGINT b, (VARCHAR(2147483647), VARCHAR(2147483647)) MAP c, VARCHAR(2147483647) f0, VARCHAR(2147483647) f1)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.c)], correlate=[table(UNNEST($cor0.c))], select=[a,b,c,f0,f1], rowType=[RecordType(INTEGER a, BIGINT b, (VARCHAR(2147483647), VARCHAR(2147483647)) MAP c, VARCHAR(2147483647) f0, VARCHAR(2147483647) f1)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -127,7 +127,7 @@ LogicalProject(a=[$0], b=[$1], x=[$2], y=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, b, _1, _2], where=[>(_1, a)])
-+- Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
+- Calc(select=[a, b], where=[<(a, 3)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
]]>
@@ -157,7 +157,7 @@ LogicalProject(a=[$0], s=[$2])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, f0 AS s])
-+- Correlate(invocation=[explode($cor0.set)], correlate=[table(explode($cor0.set))], select=[a,set,f0], rowType=[RecordType(INTEGER a, VARCHAR(2147483647) MULTISET set, VARCHAR(2147483647) f0)], joinType=[LEFT])
++- Correlate(invocation=[UNNEST($cor0.set)], correlate=[table(UNNEST($cor0.set))], select=[a,set,f0], rowType=[RecordType(INTEGER a, VARCHAR(2147483647) MULTISET set, VARCHAR(2147483647) f0)], joinType=[LEFT])
+- SortAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COLLECT(set) AS set])
+- Sort(orderBy=[a ASC])
+- Exchange(distribution=[hash[a]])
@@ -186,7 +186,7 @@ LogicalProject(a=[$0], s=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, f0 AS s])
-+- Correlate(invocation=[explode($cor0.c)], correlate=[table(explode($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER ARRAY f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.c)], correlate=[table(UNNEST($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER ARRAY f0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -215,7 +215,7 @@ LogicalProject(b=[$0], id=[$2], point=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[b, _1 AS id, _2 AS point])
-+- Correlate(invocation=[explode($cor0.set)], correlate=[table(explode($cor0.set))], select=[b,set,_1,_2], rowType=[RecordType(INTEGER b, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) MULTISET set, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.set)], correlate=[table(UNNEST($cor0.set))], select=[b,set,_1,_2], rowType=[RecordType(INTEGER b, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) MULTISET set, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
+- SortAggregate(isMerge=[true], groupBy=[b], select=[b, Final_COLLECT(set) AS set])
+- Sort(orderBy=[b ASC])
+- Exchange(distribution=[hash[b]])
@@ -244,7 +244,7 @@ LogicalProject(a=[$0], b=[$1], s=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, b, f0 AS s])
-+- Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER f0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -267,7 +267,7 @@ LogicalProject(a=[$0], b=[$1], s=[$2], t=[$3])
</Resource>
<Resource name="planAfter">
<![CDATA[
-Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 13)])
+Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 13)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
]]>
</Resource>
@@ -289,7 +289,7 @@ LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3])
</Resource>
<Resource name="planAfter">
<![CDATA[
-Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 1)])
+Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 1)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
]]>
</Resource>
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml
index 0a35338..8f5a988 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/LogicalUnnestRuleTest.xml
@@ -37,7 +37,7 @@ LogicalProject(a=[$0], s=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
+- LogicalProject(s=[$0])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.c)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.c)], rowType=[RecordType:peek_no_expand(VARCHAR(2147483647) f0)])
]]>
</Resource>
</TestCase>
@@ -76,7 +76,7 @@ LogicalProject(b=[$0], s=[$2])
: +- LogicalProject(b=[$1], $f1=[$TUMBLE($3, 3000:INTERVAL SECOND)])
: +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+- LogicalProject(s=[$0])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.set)], rowType=[RecordType(BIGINT f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.set)], rowType=[RecordType:peek_no_expand(BIGINT f0)])
]]>
</Resource>
</TestCase>
@@ -101,7 +101,7 @@ LogicalProject(a=[$0], b=[$1], v=[$4])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
+- LogicalProject(k=[$0], v=[$1])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.c)], rowType=[RecordType:peek_no_expand(VARCHAR(2147483647) f0, VARCHAR(2147483647) f1)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.c)], rowType=[RecordType:peek_no_expand(VARCHAR(2147483647) f0, VARCHAR(2147483647) f1)])
]]>
</Resource>
</TestCase>
@@ -137,7 +137,7 @@ LogicalProject(a=[$0], b=[$1], x=[$2], y=[$3])
: +- LogicalFilter(condition=[<($0, 3)])
: +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]])
+- LogicalProject(x=[$0], y=[$1])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)])
]]>
</Resource>
</TestCase>
@@ -171,7 +171,7 @@ LogicalProject(a=[$0], s=[$2])
: +- LogicalProject(a=[$0], b=[$1])
: +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+- LogicalProject(s=[$0])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.set)], rowType=[RecordType(VARCHAR(2147483647) f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.set)], rowType=[RecordType:peek_no_expand(VARCHAR(2147483647) f0)])
]]>
</Resource>
</TestCase>
@@ -196,7 +196,7 @@ LogicalProject(a=[$0], s=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
+- LogicalProject(s=[$0])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.c)], rowType=[RecordType(INTEGER ARRAY f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.c)], rowType=[RecordType:peek_no_expand(INTEGER ARRAY f0)])
]]>
</Resource>
</TestCase>
@@ -230,7 +230,7 @@ LogicalProject(b=[$0], id=[$2], point=[$3])
: +- LogicalProject(b=[$1], c=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+- LogicalProject(id=[$0], point=[$1])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.set)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.set)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)])
]]>
</Resource>
</TestCase>
@@ -255,7 +255,7 @@ LogicalProject(a=[$0], b=[$1], s=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
+- LogicalProject(s=[$0])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[RecordType(INTEGER f0)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER f0)])
]]>
</Resource>
</TestCase>
@@ -282,7 +282,7 @@ LogicalProject(a=[$0], b=[$1], s=[$2], t=[$3])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]])
+- LogicalProject(s=[$0], t=[$1])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)])
]]>
</Resource>
</TestCase>
@@ -307,7 +307,7 @@ LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3])
+- LogicalFilter(condition=[>($2, 1)])
+- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
:- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]])
- +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)], elementType=[class [Ljava.lang.Object;])
+ +- LogicalTableFunctionScan(invocation=[UNNEST($cor0.b)], rowType=[RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2)])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml
index 69f8a77..61a1e29 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/UnnestTest.xml
@@ -34,7 +34,7 @@ LogicalProject(a=[$0], s=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, s])
-+- Correlate(invocation=[explode($cor0.c)], correlate=[table(explode($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) ARRAY c, VARCHAR(2147483647) f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.c)], correlate=[table(UNNEST($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) ARRAY c, VARCHAR(2147483647) f0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -67,7 +67,7 @@ LogicalProject(b=[$0], s=[$2])
<Resource name="planAfter">
<![CDATA[
Calc(select=[b, s])
-+- Correlate(invocation=[explode($cor0.set)], correlate=[table(explode($cor0.set))], select=[b,set,f0], rowType=[RecordType(BIGINT b, BIGINT MULTISET set, BIGINT f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.set)], correlate=[table(UNNEST($cor0.set))], select=[b,set,f0], rowType=[RecordType(BIGINT b, BIGINT MULTISET set, BIGINT f0)], joinType=[INNER])
+- GroupWindowAggregate(groupBy=[b], window=[TumblingGroupWindow('w$, rowtime, 3000)], select=[b, COLLECT(b) AS set])
+- Exchange(distribution=[hash[b]])
+- Calc(select=[b, rowtime], where=[<(b, 3)])
@@ -93,7 +93,7 @@ LogicalProject(a=[$0], b=[$1], v=[$4])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, b, v])
-+- Correlate(invocation=[explode($cor0.c)], correlate=[table(explode($cor0.c))], select=[a,b,c,f0,f1], rowType=[RecordType(INTEGER a, BIGINT b, (VARCHAR(2147483647), VARCHAR(2147483647)) MAP c, VARCHAR(2147483647) f0, VARCHAR(2147483647) f1)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.c)], correlate=[table(UNNEST($cor0.c))], select=[a,b,c,f0,f1], rowType=[RecordType(INTEGER a, BIGINT b, (VARCHAR(2147483647), VARCHAR(2147483647)) MAP c, VARCHAR(2147483647) f0, VARCHAR(2147483647) f1)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -124,7 +124,7 @@ LogicalProject(a=[$0], b=[$1], x=[$2], y=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, b, x, y], where=[>(x, a)])
-+- Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
+- Calc(select=[a, b], where=[<(a, 3)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
]]>
@@ -154,7 +154,7 @@ LogicalProject(a=[$0], s=[$2])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, s])
-+- Correlate(invocation=[explode($cor0.set)], correlate=[table(explode($cor0.set))], select=[a,set,f0], rowType=[RecordType(INTEGER a, VARCHAR(2147483647) MULTISET set, VARCHAR(2147483647) f0)], joinType=[LEFT])
++- Correlate(invocation=[UNNEST($cor0.set)], correlate=[table(UNNEST($cor0.set))], select=[a,set,f0], rowType=[RecordType(INTEGER a, VARCHAR(2147483647) MULTISET set, VARCHAR(2147483647) f0)], joinType=[LEFT])
+- GroupAggregate(groupBy=[a], select=[a, COLLECT(b) AS set])
+- Exchange(distribution=[hash[a]])
+- Calc(select=[a, b], where=[<(a, 5)])
@@ -180,7 +180,7 @@ LogicalProject(a=[$0], s=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, s])
-+- Correlate(invocation=[explode($cor0.c)], correlate=[table(explode($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER ARRAY f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.c)], correlate=[table(UNNEST($cor0.c))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER ARRAY f0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -209,7 +209,7 @@ LogicalProject(b=[$0], id=[$2], point=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[b, id, point])
-+- Correlate(invocation=[explode($cor0.set)], correlate=[table(explode($cor0.set))], select=[b,set,_1,_2], rowType=[RecordType(INTEGER b, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) MULTISET set, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.set)], correlate=[table(UNNEST($cor0.set))], select=[b,set,_1,_2], rowType=[RecordType(INTEGER b, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) MULTISET set, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER])
+- GroupAggregate(groupBy=[b], select=[b, COLLECT(c) AS set])
+- Exchange(distribution=[hash[b]])
+- Calc(select=[b, c], where=[<(b, 3)])
@@ -235,7 +235,7 @@ LogicalProject(a=[$0], b=[$1], s=[$3])
<Resource name="planAfter">
<![CDATA[
Calc(select=[a, b, s])
-+- Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER f0)], joinType=[INNER])
++- Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,c,f0], rowType=[RecordType(INTEGER a, INTEGER ARRAY b, INTEGER ARRAY ARRAY c, INTEGER f0)], joinType=[INNER])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
]]>
</Resource>
@@ -258,7 +258,7 @@ LogicalProject(a=[$0], b=[$1], s=[$2], t=[$3])
</Resource>
<Resource name="planAfter">
<![CDATA[
-Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 13)])
+Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 13)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
]]>
</Resource>
@@ -280,7 +280,7 @@ LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3])
</Resource>
<Resource name="planAfter">
<![CDATA[
-Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 1)])
+Correlate(invocation=[UNNEST($cor0.b)], correlate=[table(UNNEST($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, RecordType:peek_no_expand(INTEGER _1, VARCHAR(2147483647) _2) ARRAY b, INTEGER _1, VARCHAR(2147483647) _2)], joinType=[INNER], condition=[>($0, 1)])
+- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
]]>
</Resource>
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlUnnestUtils.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlUnnestUtils.java
new file mode 100644
index 0000000..56cdd89
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlUnnestUtils.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.runtime.functions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.flink.table.types.inference.TypeStrategies;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.utils.DataTypeUtils;
+
+import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot;
+
+/**
+ * Utility for handling SQL's {@code UNNEST}.
+ */
+@Internal
+public final class SqlUnnestUtils {
+
+ public static UnnestTableFunction createUnnestFunction(LogicalType t) {
+ switch (t.getTypeRoot()) {
+ case ARRAY:
+ final ArrayType arrayType = (ArrayType) t;
+ return new CollectionUnnestTableFunction(
+ arrayType,
+ arrayType.getElementType(),
+ ArrayData.createElementGetter(arrayType.getElementType()));
+ case MULTISET:
+ final MultisetType multisetType = (MultisetType) t;
+ return new CollectionUnnestTableFunction(
+ multisetType,
+ multisetType.getElementType(),
+ ArrayData.createElementGetter(multisetType.getElementType()));
+ case MAP:
+ final MapType mapType = (MapType) t;
+ return new MapUnnestTableFunction(
+ mapType,
+ RowType.of(false, mapType.getKeyType(), mapType.getValueType()),
+ ArrayData.createElementGetter(mapType.getKeyType()),
+ ArrayData.createElementGetter(mapType.getKeyType()));
+ default:
+ throw new UnsupportedOperationException("Unsupported type for UNNEST: " + t);
+ }
+ }
+
+ /**
+ * Base table function for all unnest functions.
+ */
+ public abstract static class UnnestTableFunction extends TableFunction<Object> {
+
+ private final transient LogicalType inputType;
+
+ private final transient LogicalType outputType;
+
+ UnnestTableFunction(LogicalType inputType, LogicalType outputType) {
+ this.inputType = inputType;
+ this.outputType = outputType;
+ }
+
+ public LogicalType getWrappedOutputType() {
+ if (hasRoot(outputType, LogicalTypeRoot.ROW) ||
+ hasRoot(outputType, LogicalTypeRoot.STRUCTURED_TYPE)) {
+ return outputType;
+ }
+ return RowType.of(outputType);
+ }
+
+ @Override
+ public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+ final DataType inputDataType = DataTypeUtils.toInternalDataType(inputType);
+ final DataType outputDataType = DataTypeUtils.toInternalDataType(outputType);
+ return TypeInference.newBuilder()
+ .typedArguments(inputDataType)
+ .outputTypeStrategy(TypeStrategies.explicit(outputDataType))
+ .build();
+ }
+ }
+
+ /**
+ * Table function that unwraps the elements of a collection (array or multiset).
+ */
+ public static final class CollectionUnnestTableFunction extends UnnestTableFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ArrayData.ElementGetter elementGetter;
+
+ public CollectionUnnestTableFunction(
+ LogicalType inputType,
+ LogicalType outputType,
+ ArrayData.ElementGetter elementGetter) {
+ super(inputType, outputType);
+ this.elementGetter = elementGetter;
+ }
+
+ public void eval(ArrayData arrayData) {
+ if (arrayData == null) {
+ return;
+ }
+ final int size = arrayData.size();
+ for (int pos = 0; pos < size; pos++) {
+ collect(elementGetter.getElementOrNull(arrayData, pos));
+ }
+ }
+
+ public void eval(MapData mapData) {
+ if (mapData == null) {
+ return;
+ }
+ final int size = mapData.size();
+ final ArrayData keys = mapData.keyArray();
+ final ArrayData values = mapData.valueArray();
+ for (int pos = 0; pos < size; pos++) {
+ final int multiplier = values.getInt(pos);
+ final Object key = elementGetter.getElementOrNull(keys, pos);
+ for (int i = 0; i < multiplier; i++) {
+ collect(key);
+ }
+ }
+ }
+ }
+
+ /**
+ * Table function that unwraps the elements of a map.
+ */
+ public static final class MapUnnestTableFunction extends UnnestTableFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ArrayData.ElementGetter keyGetter;
+
+ private final ArrayData.ElementGetter valueGetter;
+
+ public MapUnnestTableFunction(
+ LogicalType inputType,
+ LogicalType outputType,
+ ArrayData.ElementGetter keyGetter,
+ ArrayData.ElementGetter valueGetter) {
+ super(inputType, outputType);
+ this.keyGetter = keyGetter;
+ this.valueGetter = valueGetter;
+ }
+
+ public void eval(MapData mapData) {
+ if (mapData == null) {
+ return;
+ }
+ final int size = mapData.size();
+ final ArrayData keyArray = mapData.keyArray();
+ final ArrayData valueArray = mapData.valueArray();
+ for (int i = 0; i < size; i++) {
+ collect(
+ GenericRowData.of(
+ keyGetter.getElementOrNull(keyArray, i),
+ valueGetter.getElementOrNull(valueArray, i)));
+ }
+ }
+ }
+
+ private SqlUnnestUtils() {
+ // no instantiation
+ }
+}