You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/06/08 09:14:51 UTC
[flink] branch master updated: [FLINK-12685][table-planner-blink]
Supports UNNEST query in blink planner
This is an automated email from the ASF dual-hosted git repository.
kurt 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 1a5c69a [FLINK-12685][table-planner-blink] Supports UNNEST query in blink planner
1a5c69a is described below
commit 1a5c69a00cb2739ea244e6c655cc77487a3731ba
Author: godfrey he <go...@163.com>
AuthorDate: Sat Jun 8 17:14:31 2019 +0800
[FLINK-12685][table-planner-blink] Supports UNNEST query in blink planner
This closes #8578
---
.../flink/table/calcite/FlinkTypeFactory.scala | 5 +-
.../codegen/agg/batch/WindowCodeGenerator.scala | 6 +-
.../plan/optimize/program/FlinkBatchProgram.scala | 5 +-
.../optimize/program/FlinkDecorrelateProgram.scala | 15 +-
.../plan/optimize/program/FlinkStreamProgram.scala | 5 +-
.../table/plan/rules/FlinkBatchRuleSets.scala | 4 +-
.../table/plan/rules/FlinkStreamRuleSets.scala | 4 +-
.../plan/rules/logical/LogicalUnnestRule.scala | 179 ++++++++++++
.../flink/table/plan/util/AggFunctionFactory.scala | 2 +-
.../table/plan/util/ExplodeFunctionUtil.scala | 170 +++++++++++
.../flink/table/plan/batch/sql/UnnestTest.xml | 297 +++++++++++++++++++
.../plan/rules/logical/LogicalUnnestRuleTest.xml | 314 ++++++++++++++++++++
.../flink/table/plan/stream/sql/UnnestTest.xml | 288 ++++++++++++++++++
.../plan/stream/sql/agg/WindowAggregateTest.xml | 21 +-
.../flink/table/plan/batch/sql/UnnestTest.scala | 26 ++
.../flink/table/plan/common/UnnestTestBase.scala | 128 ++++++++
.../plan/rules/logical/LogicalUnnestRuleTest.scala | 53 ++++
.../flink/table/plan/stream/sql/UnnestTest.scala | 26 ++
.../plan/stream/sql/agg/WindowAggregateTest.scala | 2 +-
.../table/runtime/batch/sql/UnnestITCase.scala | 256 ++++++++++++++++
.../sql/agg/AggregateReduceGroupingITCase.scala | 4 +-
.../table/runtime/batch/sql/join/JoinITCase.scala | 45 +--
.../table/runtime/stream/sql/AggregateITCase.scala | 1 -
.../table/runtime/stream/sql/UnnestITCase.scala | 325 +++++++++++++++++++++
.../apache/flink/table/util/TableTestBase.scala | 2 +-
25 files changed, 2112 insertions(+), 71 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
index 196a4cd..67642ff 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
@@ -18,7 +18,7 @@
package org.apache.flink.table.calcite
-import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeinfo.{NothingTypeInfo, TypeInformation}
import org.apache.flink.table.`type`.TypeConverters.createInternalTypeFromTypeInfo
import org.apache.flink.table.`type`._
import org.apache.flink.table.api.{TableException, TableSchema}
@@ -467,6 +467,9 @@ object FlinkTypeFactory {
val mapRelDataType = relDataType.asInstanceOf[MapRelDataType]
mapRelDataType.mapType
+ // CURSOR for UDTF case, whose type info will never be used, just a placeholder
+ case CURSOR => new GenericType(new NothingTypeInfo)
+
case _@t =>
throw new TableException(s"Type is not supported: $t")
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala
index 1085141..fd9d83f 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/agg/batch/WindowCodeGenerator.scala
@@ -184,7 +184,7 @@ abstract class WindowCodeGenerator(
val functionName = CodeGenUtils.newName("triggerWindowProcess")
val functionCode =
s"""
- |private void $functionName() {
+ |private void $functionName() throws java.lang.Exception {
| ${ctx.reuseLocalVariableCode()}
| $statements
|}
@@ -561,7 +561,7 @@ abstract class WindowCodeGenerator(
val inputTypeTerm = boxedTypeTermForType(inputType)
ctx.addReusableMember(
s"""
- |private void $processFuncName($inputTypeTerm $inputTerm) throws java.io.IOException {
+ |private void $processFuncName($inputTypeTerm $inputTerm) throws java.lang.Exception {
| ${ctx.reuseLocalVariableCode()}
| // assign timestamp (pane/window)
| ${ctx.reuseInputUnboxingCode(inputTerm)}
@@ -579,7 +579,7 @@ abstract class WindowCodeGenerator(
""".stripMargin
ctx.addReusableMember(
s"""
- |private void $endProcessFuncName() throws java.io.IOException {
+ |private void $endProcessFuncName() throws java.lang.Exception {
| ${ctx.reuseLocalVariableCode()}
| $setLastPaneAggResultCode
|}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkBatchProgram.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkBatchProgram.scala
index 0191d79..921b36f 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkBatchProgram.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkBatchProgram.scala
@@ -29,7 +29,7 @@ import org.apache.calcite.plan.hep.HepMatchOrder
*/
object FlinkBatchProgram {
val SUBQUERY_REWRITE = "subquery_rewrite"
- val CORRELATE_REWRITE = "correlate_rewrite"
+ val TEMPORAL_JOIN_REWRITE = "temporal_join_rewrite"
val DECORRELATE = "decorrelate"
val DEFAULT_REWRITE = "default_rewrite"
val PREDICATE_PUSHDOWN = "predicate_pushdown"
@@ -72,8 +72,9 @@ object FlinkBatchProgram {
)
// rewrite special temporal join plan
+ // TODO remove this program after upgraded to CALCITE-1.20.0 (CALCITE-2004 is fixed)
chainedProgram.addLast(
- CORRELATE_REWRITE,
+ TEMPORAL_JOIN_REWRITE,
FlinkGroupProgramBuilder.newBuilder[BatchOptimizeContext]
.addProgram(
FlinkHepRuleSetProgramBuilder.newBuilder
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkDecorrelateProgram.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkDecorrelateProgram.scala
index f210ad2..f89b5da 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkDecorrelateProgram.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkDecorrelateProgram.scala
@@ -20,6 +20,7 @@ package org.apache.flink.table.plan.optimize.program
import org.apache.flink.table.api.TableException
+import org.apache.calcite.rel.core.Uncollect
import org.apache.calcite.rel.logical.{LogicalFilter, LogicalJoin, LogicalProject}
import org.apache.calcite.rel.{RelNode, RelShuttleImpl}
import org.apache.calcite.rex.{RexCorrelVariable, RexVisitorImpl}
@@ -43,7 +44,10 @@ class FlinkDecorrelateProgram[OC <: FlinkOptimizeContext] extends FlinkOptimizeP
}
/**
- * Check if there is still correlate variables after decorrelate.
+ * Check if there is still correlate variables after decorrelating.
+ *
+ * NOTES: this method only checks correlate variables in join, project and filter,
+ * and will ignore the correlate variables from UNNEST (inputs of Uncollect).
*/
private def checkCorrelVariableExists(root: RelNode): Unit = {
try {
@@ -77,6 +81,15 @@ class FlinkDecorrelateProgram[OC <: FlinkOptimizeContext] extends FlinkOptimizeP
join.getCondition.accept(visitor)
super.visit(join)
}
+
+ override def visit(other: RelNode): RelNode = {
+ other match {
+ // ignore Uncollect's inputs due to the correlate variables are from UNNEST directly,
+ // not from cases (project, filter and join) which RelDecorrelator handles
+ case r: Uncollect => r
+ case _ => super.visit(other)
+ }
+ }
}
input.accept(shuttle)
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkStreamProgram.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkStreamProgram.scala
index 64cf5c9..34834d1 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkStreamProgram.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/optimize/program/FlinkStreamProgram.scala
@@ -29,7 +29,7 @@ import org.apache.flink.table.plan.rules.FlinkStreamRuleSets
object FlinkStreamProgram {
val SUBQUERY_REWRITE = "subquery_rewrite"
- val CORRELATE_REWRITE = "correlate_rewrite"
+ val TEMPORAL_JOIN_REWRITE = "temporal_join_rewrite"
val DECORRELATE = "decorrelate"
val TIME_INDICATOR = "time_indicator"
val DEFAULT_REWRITE = "default_rewrite"
@@ -71,8 +71,9 @@ object FlinkStreamProgram {
.build())
// rewrite special temporal join plan
+ // TODO remove this program after upgraded to CALCITE-1.20.0 (CALCITE-2004 is fixed)
chainedProgram.addLast(
- CORRELATE_REWRITE,
+ TEMPORAL_JOIN_REWRITE,
FlinkGroupProgramBuilder.newBuilder[StreamOptimizeContext]
.addProgram(
FlinkHepRuleSetProgramBuilder.newBuilder
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkBatchRuleSets.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkBatchRuleSets.scala
index 9852dbd..700d2c5 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkBatchRuleSets.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkBatchRuleSets.scala
@@ -119,7 +119,9 @@ object FlinkBatchRuleSets {
new CoerceInputsRule(classOf[LogicalMinus], false),
ConvertToNotInOrInRule.INSTANCE,
// optimize limit 0
- FlinkLimit0RemoveRule.INSTANCE
+ FlinkLimit0RemoveRule.INSTANCE,
+ // unnest rule
+ LogicalUnnestRule.INSTANCE
)).asJava)
/**
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkStreamRuleSets.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkStreamRuleSets.scala
index eae74a1..e65f9b1 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkStreamRuleSets.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/FlinkStreamRuleSets.scala
@@ -121,7 +121,9 @@ object FlinkStreamRuleSets {
new CoerceInputsRule(classOf[LogicalMinus], false),
ConvertToNotInOrInRule.INSTANCE,
// optimize limit 0
- FlinkLimit0RemoveRule.INSTANCE
+ FlinkLimit0RemoveRule.INSTANCE,
+ // unnest rule
+ LogicalUnnestRule.INSTANCE
)
).asJava)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
new file mode 100644
index 0000000..d28b659
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
@@ -0,0 +1,179 @@
+/*
+ * 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.plan.rules.logical
+
+import org.apache.flink.api.java.typeutils.MapTypeInfo
+import org.apache.flink.table.`type`.{InternalTypes, TypeConverters}
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
+import org.apache.flink.table.plan.schema.{ArrayRelDataType, MapRelDataType, MultisetRelDataType}
+import org.apache.flink.table.plan.util.ExplodeFunctionUtil
+
+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
+
+import java.util.Collections
+
+/**
+ * Planner rule that rewrites UNNEST to explode function.
+ *
+ * Note: This class can only be used in HepPlanner.
+ */
+class LogicalUnnestRule(
+ operand: RelOptRuleOperand,
+ description: String)
+ extends RelOptRule(operand, description) {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+ val join: LogicalCorrelate = call.rel(0)
+ val right = getRel(join.getRight)
+
+ right match {
+ // a filter is pushed above the table function
+ case filter: LogicalFilter =>
+ getRel(filter.getInput) match {
+ case u: Uncollect => !u.withOrdinality
+ case p: LogicalProject => getRel(p.getInput) match {
+ case u: Uncollect => !u.withOrdinality
+ case _ => false
+ }
+ case _ => false
+ }
+ case project: LogicalProject =>
+ getRel(project.getInput) match {
+ case u: Uncollect => !u.withOrdinality
+ case _ => false
+ }
+ case u: Uncollect => !u.withOrdinality
+ case _ => false
+ }
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+ val correlate: LogicalCorrelate = call.rel(0)
+ val outer = getRel(correlate.getLeft)
+ val array = getRel(correlate.getRight)
+
+ def convert(relNode: RelNode): RelNode = {
+ relNode match {
+ case rs: HepRelVertex =>
+ convert(getRel(rs))
+
+ case f: LogicalProject =>
+ f.copy(f.getTraitSet, ImmutableList.of(convert(getRel(f.getInput))))
+
+ case f: LogicalFilter =>
+ f.copy(f.getTraitSet, ImmutableList.of(convert(getRel(f.getInput))))
+
+ 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: ArrayRelDataType =>
+ (arrayType.getComponentType,
+ ExplodeFunctionUtil.explodeTableFuncFromType(
+ TypeConverters.createExternalTypeInfoFromInternalType(arrayType.arrayType)))
+
+ case map: MapRelDataType =>
+ val rowInternalType = InternalTypes.createRowType(
+ map.mapType.getKeyType, map.mapType.getValueType)
+ val componentType = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
+ .createTypeFromInternalType(rowInternalType, isNullable = true)
+ val mapTypeInfo = new MapTypeInfo(
+ TypeConverters.createExternalTypeInfoFromInternalType(map.mapType.getKeyType),
+ TypeConverters.createExternalTypeInfoFromInternalType(map.mapType.getValueType)
+ )
+ val explodeFunction = ExplodeFunctionUtil.explodeTableFuncFromType(mapTypeInfo)
+ (componentType, explodeFunction)
+
+ case mt: MultisetRelDataType =>
+ (mt.getComponentType,
+ ExplodeFunctionUtil.explodeTableFuncFromType(
+ TypeConverters.createExternalTypeInfoFromInternalType(mt.multisetType)))
+ case _ => throw new TableException(s"Unsupported UNNEST on type: ${dataType.toString}")
+ }
+
+ // create sql function
+ val explodeSqlFunc = UserDefinedFunctionUtils.createTableSqlFunction(
+ "explode",
+ "explode",
+ explodeTableFunc,
+ TypeConverters.createExternalTypeInfoFromInternalType(
+ FlinkTypeFactory.toInternalType(componentType)),
+ cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory])
+
+ // create table function call
+ val rexCall = cluster.getRexBuilder.makeCall(
+ 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}")
+ }
+
+ // create table function scan
+ new LogicalTableFunctionScan(
+ cluster,
+ correlate.getTraitSet,
+ Collections.emptyList(),
+ rexCall,
+ classOf[Array[Object]],
+ rowType,
+ null)
+ }
+ }
+
+ // convert unnest into table function scan
+ val tableFunctionScan = convert(array)
+ // create correlate with table function scan as input
+ val newCorrelate =
+ correlate.copy(correlate.getTraitSet, ImmutableList.of(outer, tableFunctionScan))
+ call.transformTo(newCorrelate)
+ }
+
+ private def getRel(rel: RelNode): RelNode = {
+ rel match {
+ case vertex: HepRelVertex => vertex.getCurrentRel
+ case _ => rel
+ }
+ }
+}
+
+object LogicalUnnestRule {
+ val INSTANCE = new LogicalUnnestRule(
+ operand(classOf[LogicalCorrelate], any),
+ "LogicalUnnestRule")
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggFunctionFactory.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggFunctionFactory.scala
index ecc6cde..d047a93 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggFunctionFactory.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/AggFunctionFactory.scala
@@ -624,7 +624,7 @@ class AggFunctionFactory(
private def createCollectAggFunction(argTypes: Array[InternalType]): UserDefinedFunction = {
val elementTypeInfo = argTypes(0) match {
case gt: GenericType[_] => gt.getTypeInfo
- case t => TypeConverters.createInternalTypeInfoFromInternalType(t)
+ case t => TypeConverters.createExternalTypeInfoFromInternalType(t)
}
new CollectAggFunction(elementTypeInfo)
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala
new file mode 100644
index 0000000..d0f9475
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala
@@ -0,0 +1,170 @@
+/*
+ * 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.plan.util
+
+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._
+
+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))
+ }
+ }
+}
+
+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)
+ }
+}
+
+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)
+ }
+}
+
+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)
+ }
+}
+
+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)
+ }
+}
+
+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)
+ }
+}
+
+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)
+ }
+}
+
+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)
+ }
+}
+
+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/plan/batch/sql/UnnestTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/batch/sql/UnnestTest.xml
new file mode 100644
index 0000000..6e73bd5
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/batch/sql/UnnestTest.xml
@@ -0,0 +1,297 @@
+<?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="testCrossWithUnnest">
+ <Resource name="sql">
+ <![CDATA[SELECT a, s FROM MyTable, UNNEST(MyTable.c) as A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=StringType}) c, VARCHAR(65536) f0)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testTumbleWindowAggregateWithCollectUnnest">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT b, COLLECT(b) as `set`
+ FROM MyTable
+ GROUP BY b, TUMBLE(rowtime, INTERVAL '3' SECOND)
+)
+SELECT b, s FROM T, UNNEST(T.`set`) AS A(s) where b < 3
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(b=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(b=[$0], set=[$2])
+ : +- LogicalAggregate(group=[{0, 1}], set=[COLLECT($0)])
+ : +- LogicalProject(b=[$1], $f1=[TUMBLE($3, 3000:INTERVAL SECOND)])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MULTISET(BIGINT) set, BIGINT f0)], joinType=[INNER])
+ +- SortWindowAggregate(groupBy=[b], window=[TumblingGroupWindow], select=[b, Final_COLLECT(set) AS set])
+ +- Sort(orderBy=[b ASC, assignedWindow$ ASC])
+ +- Exchange(distribution=[hash[b]])
+ +- LocalSortWindowAggregate(groupBy=[b], window=[TumblingGroupWindow], select=[b, Partial_COLLECT(b) AS set])
+ +- Sort(orderBy=[b ASC, rowtime ASC])
+ +- Calc(select=[b, rowtime], where=[<(b, 3)])
+ +- BoundedStreamScan(table=[[MyTable]], fields=[a, b, c, rowtime])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCrossWithUnnestForMap">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, v FROM MyTable CROSS JOIN UNNEST(c) as f(k, v)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], v=[$4])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(k=[$0], v=[$1])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MAP(MapType{keyType=StringType, valueType=StringType}) c, VARCHAR(65536) f0, VARCHAR(65536) f1)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJoinWithUnnestOfTuple">
+ <Resource name="sql">
+ <![CDATA[
+SELECT a, b, x, y FROM
+ (SELECT a, b FROM MyTable WHERE a < 3) as tf,
+ UNNEST(tf.b) as A (x, y)
+WHERE x > a
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], x=[$2], y=[$3])
++- LogicalFilter(condition=[>($2, $0)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalFilter(condition=[<($0, 3)])
+ : +- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(x=[$0], y=[$1])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=RowType{types=[IntType, StringType], fieldNames=[_1, _2]}}) b, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER])
+ +- Calc(select=[a, b], where=[<(a, 3)])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLeftUnnestMultiSetFromCollectResult">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT a, COLLECT(b) as `set` FROM MyTable GROUP BY a)
+SELECT a, s FROM T LEFT JOIN UNNEST(T.`set`) AS A(s) ON TRUE WHERE a < 5
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 5)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MULTISET(VARCHAR(65536)) set, VARCHAR(65536) f0)], joinType=[LEFT])
+ +- SortAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COLLECT(set) AS set])
+ +- Sort(orderBy=[a ASC])
+ +- Exchange(distribution=[hash[a]])
+ +- LocalSortAggregate(groupBy=[a], select=[a, Partial_COLLECT(b) AS set])
+ +- Sort(orderBy=[a ASC])
+ +- Calc(select=[a, b], where=[<(a, 5)])
+ +- BoundedStreamScan(table=[[MyTable]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestArrayOfArrayFromTable">
+ <Resource name="sql">
+ <![CDATA[SELECT a, s FROM MyTable, UNNEST(MyTable.c) AS A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=IntType}) b, ARRAY(ArrayType{elementType=ArrayType{elementType=IntType}}) c, ARRAY(ArrayType{elementType=IntType}) f0)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestMultiSetFromCollectResult">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT b, COLLECT(c) as `set` FROM MyTable GROUP BY b)
+SELECT b, id, point FROM T, UNNEST(T.`set`) AS A(id, point) WHERE b < 3
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(b=[$0], id=[$2], point=[$3])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(b=[$1], c=[$2])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(id=[$0], point=[$1])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MULTISET(ROW(RowType{types=[IntType, StringType], fieldNames=[_1, _2]})) set, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER])
+ +- SortAggregate(isMerge=[true], groupBy=[b], select=[b, Final_COLLECT(set) AS set])
+ +- Sort(orderBy=[b ASC])
+ +- Exchange(distribution=[hash[b]])
+ +- LocalSortAggregate(groupBy=[b], select=[b, Partial_COLLECT(c) AS set])
+ +- Sort(orderBy=[b ASC])
+ +- Calc(select=[b, c], where=[<(b, 3)])
+ +- BoundedStreamScan(table=[[MyTable]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestPrimitiveArrayFromTable">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, s FROM MyTable, UNNEST(MyTable.b) AS A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=IntType}) b, ARRAY(ArrayType{elementType=ArrayType{elementType=IntType}}) c, INTEGER f0)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestObjectArrayFromTableWithFilter">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, s, t FROM MyTable, UNNEST(MyTable.b) AS A (s, t) WHERE s > 13]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$2], t=[$3])
++- LogicalFilter(condition=[>($2, 13)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(s=[$0], t=[$1])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, ARRAY(ArrayType{elementType=RowType{types=[IntType, StringType], fieldNames=[_1, _2]}}) b, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER], condition=[>($0, 13)])
++- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestObjectArrayWithoutAlias">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, A._1, A._2 FROM MyTable, UNNEST(MyTable.b) AS A where A._1 > 1]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3])
++- LogicalFilter(condition=[>($2, 1)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, ARRAY(ArrayType{elementType=RowType{types=[IntType, StringType], fieldNames=[_1, _2]}}) b, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER], condition=[>($0, 1)])
++- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/rules/logical/LogicalUnnestRuleTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/rules/logical/LogicalUnnestRuleTest.xml
new file mode 100644
index 0000000..daab94e
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/rules/logical/LogicalUnnestRuleTest.xml
@@ -0,0 +1,314 @@
+<?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="testCrossWithUnnest">
+ <Resource name="sql">
+ <![CDATA[SELECT a, s FROM MyTable, UNNEST(MyTable.c) as A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.c)], rowType=[RecordType(VARCHAR(65536) f0)], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testTumbleWindowAggregateWithCollectUnnest">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT b, COLLECT(b) as `set`
+ FROM MyTable
+ GROUP BY b, TUMBLE(rowtime, INTERVAL '3' SECOND)
+)
+SELECT b, s FROM T, UNNEST(T.`set`) AS A(s) where b < 3
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(b=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(b=[$0], set=[$2])
+ : +- LogicalAggregate(group=[{0, 1}], set=[COLLECT($0)])
+ : +- LogicalProject(b=[$1], $f1=[TUMBLE($3, 3000:INTERVAL SECOND)])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(b=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(b=[$0], set=[$2])
+ : +- LogicalAggregate(group=[{0, 1}], set=[COLLECT($0)])
+ : +- LogicalProject(b=[$1], $f1=[TUMBLE($3, 3000:INTERVAL SECOND)])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.set)], rowType=[RecordType(BIGINT f0)], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCrossWithUnnestForMap">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, v FROM MyTable CROSS JOIN UNNEST(c) as f(k, v)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], v=[$4])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(k=[$0], v=[$1])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], v=[$4])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(k=[$0], v=[$1])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.c)], rowType=[ROW(RowType{types=[StringType, StringType], fieldNames=[f0, f1]})], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJoinWithUnnestOfTuple">
+ <Resource name="sql">
+ <![CDATA[
+SELECT a, b, x, y FROM
+ (SELECT a, b FROM MyTable WHERE a < 3) as tf,
+ UNNEST(tf.b) as A (x, y)
+WHERE x > a
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], x=[$2], y=[$3])
++- LogicalFilter(condition=[>($2, $0)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalFilter(condition=[<($0, 3)])
+ : +- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(x=[$0], y=[$1])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], x=[$2], y=[$3])
++- LogicalFilter(condition=[>($2, $0)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalFilter(condition=[<($0, 3)])
+ : +- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(x=[$0], y=[$1])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[ROW(RowType{types=[IntType, StringType], fieldNames=[_1, _2]})], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLeftUnnestMultiSetFromCollectResult">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT a, COLLECT(b) as `set` FROM MyTable GROUP BY a)
+SELECT a, s FROM T LEFT JOIN UNNEST(T.`set`) AS A(s) ON TRUE WHERE a < 5
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 5)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 5)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.set)], rowType=[RecordType(VARCHAR(65536) f0)], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestArrayOfArrayFromTable">
+ <Resource name="sql">
+ <![CDATA[SELECT a, s FROM MyTable, UNNEST(MyTable.c) AS A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.c)], rowType=[RecordType(ARRAY(ArrayType{elementType=IntType}) f0)], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestMultiSetFromCollectResult">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT b, COLLECT(c) as `set` FROM MyTable GROUP BY b)
+SELECT b, id, point FROM T, UNNEST(T.`set`) AS A(id, point) WHERE b < 3
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(b=[$0], id=[$2], point=[$3])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(b=[$1], c=[$2])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(id=[$0], point=[$1])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(b=[$0], id=[$2], point=[$3])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(b=[$1], c=[$2])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(id=[$0], point=[$1])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.set)], rowType=[ROW(RowType{types=[IntType, StringType], fieldNames=[_1, _2]})], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestPrimitiveArrayFromTable">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, s FROM MyTable, UNNEST(MyTable.b) AS A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[RecordType(INTEGER f0)], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestObjectArrayFromTableWithFilter">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, s, t FROM MyTable, UNNEST(MyTable.b) AS A (s, t) WHERE s > 13]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$2], t=[$3])
++- LogicalFilter(condition=[>($2, 13)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(s=[$0], t=[$1])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$2], t=[$3])
++- LogicalFilter(condition=[>($2, 13)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(s=[$0], t=[$1])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[ROW(RowType{types=[IntType, StringType], fieldNames=[_1, _2]})], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestObjectArrayWithoutAlias">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, A._1, A._2 FROM MyTable, UNNEST(MyTable.b) AS A where A._1 > 1]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3])
++- LogicalFilter(condition=[>($2, 1)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3])
++- LogicalFilter(condition=[>($2, 1)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalTableFunctionScan(invocation=[explode($cor0.b)], rowType=[ROW(RowType{types=[IntType, StringType], fieldNames=[_1, _2]})], elementType=[class [Ljava.lang.Object;])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/UnnestTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/UnnestTest.xml
new file mode 100644
index 0000000..3a14e7c
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/UnnestTest.xml
@@ -0,0 +1,288 @@
+<?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="testCrossWithUnnest">
+ <Resource name="sql">
+ <![CDATA[SELECT a, s FROM MyTable, UNNEST(MyTable.c) as A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=StringType}) c, VARCHAR(65536) f0)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testTumbleWindowAggregateWithCollectUnnest">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT b, COLLECT(b) as `set`
+ FROM MyTable
+ GROUP BY b, TUMBLE(rowtime, INTERVAL '3' SECOND)
+)
+SELECT b, s FROM T, UNNEST(T.`set`) AS A(s) where b < 3
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(b=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(b=[$0], set=[$2])
+ : +- LogicalAggregate(group=[{0, 1}], set=[COLLECT($0)])
+ : +- LogicalProject(b=[$1], $f1=[TUMBLE($3, 3000:INTERVAL SECOND)])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MULTISET(BIGINT) set, BIGINT f0)], joinType=[INNER])
+ +- GroupWindowAggregate(groupBy=[b], window=[TumblingGroupWindow], select=[b, COLLECT(b) AS set])
+ +- Exchange(distribution=[hash[b]])
+ +- Calc(select=[b, rowtime], where=[<(b, 3)])
+ +- DataStreamScan(table=[[_DataStreamTable_0]], fields=[a, b, c, rowtime])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testCrossWithUnnestForMap">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, v FROM MyTable CROSS JOIN UNNEST(c) as f(k, v)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], v=[$4])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(k=[$0], v=[$1])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MAP(MapType{keyType=StringType, valueType=StringType}) c, VARCHAR(65536) f0, VARCHAR(65536) f1)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testJoinWithUnnestOfTuple">
+ <Resource name="sql">
+ <![CDATA[
+SELECT a, b, x, y FROM
+ (SELECT a, b FROM MyTable WHERE a < 3) as tf,
+ UNNEST(tf.b) as A (x, y)
+WHERE x > a
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], x=[$2], y=[$3])
++- LogicalFilter(condition=[>($2, $0)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalFilter(condition=[<($0, 3)])
+ : +- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(x=[$0], y=[$1])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=RowType{types=[IntType, StringType], fieldNames=[_1, _2]}}) b, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER])
+ +- Calc(select=[a, b], where=[<(a, 3)])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLeftUnnestMultiSetFromCollectResult">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT a, COLLECT(b) as `set` FROM MyTable GROUP BY a)
+SELECT a, s FROM T LEFT JOIN UNNEST(T.`set`) AS A(s) ON TRUE WHERE a < 5
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$2])
++- LogicalFilter(condition=[<($0, 5)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[left], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(a=[$0], b=[$1])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MULTISET(VARCHAR(65536)) set, VARCHAR(65536) f0)], joinType=[LEFT])
+ +- GroupAggregate(groupBy=[a], select=[a, COLLECT(b) AS set])
+ +- Exchange(distribution=[hash[a]])
+ +- Calc(select=[a, b], where=[<(a, 5)])
+ +- DataStreamScan(table=[[_DataStreamTable_0]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestArrayOfArrayFromTable">
+ <Resource name="sql">
+ <![CDATA[SELECT a, s FROM MyTable, UNNEST(MyTable.c) AS A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{2}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(c=[$cor0.c])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=IntType}) b, ARRAY(ArrayType{elementType=ArrayType{elementType=IntType}}) c, ARRAY(ArrayType{elementType=IntType}) f0)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestMultiSetFromCollectResult">
+ <Resource name="sql">
+ <![CDATA[
+WITH T AS (SELECT b, COLLECT(c) as `set` FROM MyTable GROUP BY b)
+SELECT b, id, point FROM T, UNNEST(T.`set`) AS A(id, point) WHERE b < 3
+ ]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(b=[$0], id=[$2], point=[$3])
++- LogicalFilter(condition=[<($0, 3)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalAggregate(group=[{0}], set=[COLLECT($1)])
+ : +- LogicalProject(b=[$1], c=[$2])
+ : +- LogicalTableScan(table=[[MyTable]])
+ +- LogicalProject(id=[$0], point=[$1])
+ +- Uncollect
+ +- LogicalProject(set=[$cor0.set])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, MULTISET(ROW(RowType{types=[IntType, StringType], fieldNames=[_1, _2]})) set, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER])
+ +- GroupAggregate(groupBy=[b], select=[b, COLLECT(c) AS set])
+ +- Exchange(distribution=[hash[b]])
+ +- Calc(select=[b, c], where=[<(b, 3)])
+ +- DataStreamScan(table=[[_DataStreamTable_0]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestPrimitiveArrayFromTable">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, s FROM MyTable, UNNEST(MyTable.b) AS A (s)]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$3])
++- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]])
+ +- LogicalProject(s=[$0])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <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, ARRAY(ArrayType{elementType=IntType}) b, ARRAY(ArrayType{elementType=ArrayType{elementType=IntType}}) c, INTEGER f0)], joinType=[INNER])
+ +- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestObjectArrayFromTableWithFilter">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, s, t FROM MyTable, UNNEST(MyTable.b) AS A (s, t) WHERE s > 13]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], s=[$2], t=[$3])
++- LogicalFilter(condition=[>($2, 13)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- LogicalProject(s=[$0], t=[$1])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, ARRAY(ArrayType{elementType=RowType{types=[IntType, StringType], fieldNames=[_1, _2]}}) b, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER], condition=[>($0, 13)])
++- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUnnestObjectArrayWithoutAlias">
+ <Resource name="sql">
+ <![CDATA[SELECT a, b, A._1, A._2 FROM MyTable, UNNEST(MyTable.b) AS A where A._1 > 1]]>
+ </Resource>
+ <Resource name="planBefore">
+ <![CDATA[
+LogicalProject(a=[$0], b=[$1], _1=[$2], _2=[$3])
++- LogicalFilter(condition=[>($2, 1)])
+ +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{1}])
+ :- LogicalTableScan(table=[[MyTable, source: [TestTableSource(a, b)]]])
+ +- Uncollect
+ +- LogicalProject(b=[$cor0.b])
+ +- LogicalValues(tuples=[[{ 0 }]])
+]]>
+ </Resource>
+ <Resource name="planAfter">
+ <![CDATA[
+Correlate(invocation=[explode($cor0.b)], correlate=[table(explode($cor0.b))], select=[a,b,_1,_2], rowType=[RecordType(INTEGER a, ARRAY(ArrayType{elementType=RowType{types=[IntType, StringType], fieldNames=[_1, _2]}}) b, INTEGER _1, VARCHAR(65536) _2)], joinType=[INNER], condition=[>($0, 1)])
++- TableSourceScan(table=[[MyTable, source: [TestTableSource(a, b)]]], fields=[a, b])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
diff --git a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.xml b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.xml
index 0c31ff0..59a46f3 100644
--- a/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.xml
+++ b/flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.xml
@@ -142,14 +142,14 @@ SELECT weightedAvg(c, a) FROM
TUMBLE_START(rowtime, INTERVAL '15' MINUTE) as ping_start
FROM MyTable
GROUP BY a, b, c, TUMBLE(rowtime, INTERVAL '15' MINUTE)) AS t1
-GROUP BY b, ping_start
+GROUP BY b
]]>
</Resource>
<Resource name="planBefore">
<![CDATA[
-LogicalProject(EXPR$0=[$2])
-+- LogicalAggregate(group=[{0, 1}], EXPR$0=[weightedAvg($2, $3)])
- +- LogicalProject(b=[$1], ping_start=[TUMBLE_START($3)], c=[$2], a=[$0])
+LogicalProject(EXPR$0=[$1])
++- LogicalAggregate(group=[{0}], EXPR$0=[weightedAvg($1, $2)])
+ +- LogicalProject(b=[$1], c=[$2], a=[$0])
+- LogicalAggregate(group=[{0, 1, 2, 3}])
+- LogicalProject(a=[$0], b=[$1], c=[$2], $f3=[TUMBLE($4, 900000:INTERVAL MINUTE)])
+- LogicalTableScan(table=[[MyTable]])
@@ -158,13 +158,12 @@ LogicalProject(EXPR$0=[$2])
<Resource name="planAfter">
<![CDATA[
Calc(select=[EXPR$0])
-+- GroupAggregate(groupBy=[b, ping_start], select=[b, ping_start, weightedAvg(c, a) AS EXPR$0])
- +- Exchange(distribution=[hash[b, ping_start]])
- +- Calc(select=[b, CAST(w$start) AS ping_start, c, a])
- +- GroupWindowAggregate(groupBy=[a, b, c], window=[TumblingGroupWindow], properties=[w$start, w$end, w$rowtime, w$proctime], select=[a, b, c, start('w$) AS w$start, end('w$) AS w$end, rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime])
- +- Exchange(distribution=[hash[a, b, c]])
- +- Calc(select=[a, b, c, rowtime])
- +- DataStreamScan(table=[[_DataStreamTable_0]], fields=[a, b, c, proctime, rowtime])
++- GroupAggregate(groupBy=[b], select=[b, weightedAvg(c, a) AS EXPR$0])
+ +- Exchange(distribution=[hash[b]])
+ +- GroupWindowAggregate(groupBy=[a, b, c], window=[TumblingGroupWindow], select=[a, b, c])
+ +- Exchange(distribution=[hash[a, b, c]])
+ +- Calc(select=[a, b, c, rowtime])
+ +- DataStreamScan(table=[[_DataStreamTable_0]], fields=[a, b, c, proctime, rowtime])
]]>
</Resource>
</TestCase>
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/batch/sql/UnnestTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/batch/sql/UnnestTest.scala
new file mode 100644
index 0000000..e39917d
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/batch/sql/UnnestTest.scala
@@ -0,0 +1,26 @@
+/*
+ * 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.plan.batch.sql
+
+import org.apache.flink.table.plan.common.UnnestTestBase
+import org.apache.flink.table.util.TableTestUtil
+
+class UnnestTest extends UnnestTestBase {
+
+ override def getTableTestUtil: TableTestUtil = batchTestUtil()
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/common/UnnestTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/common/UnnestTestBase.scala
new file mode 100644
index 0000000..6193773
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/common/UnnestTestBase.scala
@@ -0,0 +1,128 @@
+/*
+ * 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.plan.common
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.util.{TableTestBase, TableTestUtil}
+
+import org.junit.Test
+
+import java.sql.Timestamp
+
+/**
+ * Test for UNNEST queries.
+ */
+abstract class UnnestTestBase extends TableTestBase {
+
+ protected val util: TableTestUtil = getTableTestUtil
+
+ protected def getTableTestUtil: TableTestUtil
+
+ @Test
+ def testUnnestPrimitiveArrayFromTable(): Unit = {
+ util.addTableSource[(Int, Array[Int], Array[Array[Int]])]("MyTable", 'a, 'b, 'c)
+ util.verifyPlan("SELECT a, b, s FROM MyTable, UNNEST(MyTable.b) AS A (s)")
+ }
+
+ @Test
+ def testUnnestArrayOfArrayFromTable(): Unit = {
+ util.addTableSource[(Int, Array[Int], Array[Array[Int]])]("MyTable", 'a, 'b, 'c)
+ util.verifyPlan("SELECT a, s FROM MyTable, UNNEST(MyTable.c) AS A (s)")
+ }
+
+ @Test
+ def testUnnestObjectArrayFromTableWithFilter(): Unit = {
+ util.addTableSource[(Int, Array[(Int, String)])]("MyTable", 'a, 'b)
+ util.verifyPlan("SELECT a, b, s, t FROM MyTable, UNNEST(MyTable.b) AS A (s, t) WHERE s > 13")
+ }
+
+ @Test
+ def testUnnestMultiSetFromCollectResult(): Unit = {
+ util.addDataStream[(Int, Int, (Int, String))]("MyTable", 'a, 'b, 'c)
+ val sqlQuery =
+ """
+ |WITH T AS (SELECT b, COLLECT(c) as `set` FROM MyTable GROUP BY b)
+ |SELECT b, id, point FROM T, UNNEST(T.`set`) AS A(id, point) WHERE b < 3
+ """.stripMargin
+ util.verifyPlan(sqlQuery)
+ }
+
+ @Test
+ def testLeftUnnestMultiSetFromCollectResult(): Unit = {
+ util.addDataStream[(Int, String, String)]("MyTable", 'a, 'b, 'c)
+ val sqlQuery =
+ """
+ |WITH T AS (SELECT a, COLLECT(b) as `set` FROM MyTable GROUP BY a)
+ |SELECT a, s FROM T LEFT JOIN UNNEST(T.`set`) AS A(s) ON TRUE WHERE a < 5
+ """.stripMargin
+ util.verifyPlan(sqlQuery)
+ }
+
+ @Test
+ def testTumbleWindowAggregateWithCollectUnnest(): Unit = {
+ util.addDataStream[(Int, Long, String, Timestamp)]("MyTable", 'a, 'b, 'c, 'rowtime)
+ val sqlQuery =
+ """
+ |WITH T AS (SELECT b, COLLECT(b) as `set`
+ | FROM MyTable
+ | GROUP BY b, TUMBLE(rowtime, INTERVAL '3' SECOND)
+ |)
+ |SELECT b, s FROM T, UNNEST(T.`set`) AS A(s) where b < 3
+ """.stripMargin
+ util.verifyPlan(sqlQuery)
+ }
+
+ @Test
+ def testCrossWithUnnest(): Unit = {
+ util.addTableSource[(Int, Long, Array[String])]("MyTable", 'a, 'b, 'c)
+ util.verifyPlan("SELECT a, s FROM MyTable, UNNEST(MyTable.c) as A (s)")
+ }
+
+ @Test
+ def testCrossWithUnnestForMap(): Unit = {
+ util.addTableSource("MyTable",
+ Array[TypeInformation[_]](Types.INT,
+ Types.LONG,
+ Types.MAP(Types.STRING, Types.STRING)),
+ Array("a", "b", "c"))
+ util.verifyPlan("SELECT a, b, v FROM MyTable CROSS JOIN UNNEST(c) as f(k, v)")
+ }
+
+ @Test
+ def testJoinWithUnnestOfTuple(): Unit = {
+ util.addTableSource[(Int, Array[(Int, String)])]("MyTable", 'a, 'b)
+ val sqlQuery =
+ """
+ |SELECT a, b, x, y FROM
+ | (SELECT a, b FROM MyTable WHERE a < 3) as tf,
+ | UNNEST(tf.b) as A (x, y)
+ |WHERE x > a
+ """.stripMargin
+ util.verifyPlan(sqlQuery)
+ }
+
+ @Test
+ def testUnnestObjectArrayWithoutAlias(): Unit = {
+ util.addTableSource[(Int, Array[(Int, String)])]("MyTable", 'a, 'b)
+ util.verifyPlan("SELECT a, b, A._1, A._2 FROM MyTable, UNNEST(MyTable.b) AS A where A._1 > 1")
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRuleTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRuleTest.scala
new file mode 100644
index 0000000..a601853
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRuleTest.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.plan.rules.logical
+
+import org.apache.flink.table.calcite.CalciteConfig
+import org.apache.flink.table.plan.common.UnnestTestBase
+import org.apache.flink.table.plan.optimize.program.{BatchOptimizeContext, FlinkChainedProgram, FlinkHepRuleSetProgramBuilder, HEP_RULES_EXECUTION_TYPE}
+import org.apache.flink.table.util.TableTestUtil
+
+import org.apache.calcite.plan.hep.HepMatchOrder
+import org.apache.calcite.tools.RuleSets
+import org.junit.Before
+
+/**
+ * Test for [[LogicalUnnestRule]].
+ */
+class LogicalUnnestRuleTest extends UnnestTestBase {
+
+ override protected def getTableTestUtil: TableTestUtil = batchTestUtil()
+
+ @Before
+ def setup(): Unit = {
+ val programs = new FlinkChainedProgram[BatchOptimizeContext]()
+ programs.addLast(
+ "rules",
+ FlinkHepRuleSetProgramBuilder.newBuilder
+ .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE)
+ .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+ .add(RuleSets.ofList(LogicalUnnestRule.INSTANCE))
+ .build()
+ )
+ val calciteConfig = CalciteConfig.createBuilder(util.getTableEnv.getConfig.getCalciteConfig)
+ .replaceBatchProgram(programs).build()
+ util.getTableEnv.getConfig.setCalciteConfig(calciteConfig)
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/UnnestTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/UnnestTest.scala
new file mode 100644
index 0000000..fe984dc
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/UnnestTest.scala
@@ -0,0 +1,26 @@
+/*
+ * 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.plan.stream.sql
+
+import org.apache.flink.table.plan.common.UnnestTestBase
+import org.apache.flink.table.util.TableTestUtil
+
+class UnnestTest extends UnnestTestBase {
+
+ override def getTableTestUtil: TableTestUtil = streamTestUtil()
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.scala
index a2b4b4d..599f16a 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/plan/stream/sql/agg/WindowAggregateTest.scala
@@ -229,7 +229,7 @@ class WindowAggregateTest extends TableTestBase {
| TUMBLE_START(rowtime, INTERVAL '15' MINUTE) as ping_start
| FROM MyTable
| GROUP BY a, b, c, TUMBLE(rowtime, INTERVAL '15' MINUTE)) AS t1
- |GROUP BY b, ping_start
+ |GROUP BY b
""".stripMargin
util.verifyPlan(sql)
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/UnnestITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/UnnestITCase.scala
new file mode 100644
index 0000000..70da594
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/UnnestITCase.scala
@@ -0,0 +1,256 @@
+/*
+ * 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.batch.sql
+
+import org.apache.flink.api.java.typeutils.{ObjectArrayTypeInfo, RowTypeInfo}
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.runtime.utils.{BatchTestBase, TestData}
+import org.apache.flink.table.runtime.utils.BatchTestBase.row
+import org.apache.flink.types.Row
+
+import org.junit.Test
+
+import java.sql.Timestamp
+
+import scala.collection.Seq
+
+import scala.collection.JavaConverters._
+
+class UnnestITCase extends BatchTestBase {
+
+ @Test
+ def testUnnestPrimitiveArrayFromTable(): Unit = {
+ val data = List(
+ row(1, Array(12, 45), Array(Array(12, 45))),
+ row(2, Array(41, 5), Array(Array(18), Array(87))),
+ row(3, Array(18, 42), Array(Array(1), Array(45)))
+ )
+ registerCollection("T", data,
+ new RowTypeInfo(
+ Types.INT,
+ Types.PRIMITIVE_ARRAY(Types.INT),
+ Types.OBJECT_ARRAY(Types.PRIMITIVE_ARRAY(Types.INT))),
+ "a, b, c")
+
+ checkResult(
+ "SELECT a, b, s FROM T, UNNEST(T.b) AS A (s)",
+ Seq(row(1, Array(12, 45), 12),
+ row(1, Array(12, 45), 45),
+ row(2, Array(41, 5), 41),
+ row(2, Array(41, 5), 5),
+ row(3, Array(18, 42), 18),
+ row(3, Array(18, 42), 42))
+ )
+ }
+
+ @Test
+ def testUnnestArrayOfArrayFromTable(): Unit = {
+ val data = List(
+ row(1, Array(12, 45), Array(Array(12, 45))),
+ row(2, Array(41, 5), Array(Array(18), Array(87))),
+ row(3, Array(18, 42), Array(Array(1), Array(45)))
+ )
+ registerCollection("T", data,
+ new RowTypeInfo(
+ Types.INT,
+ Types.PRIMITIVE_ARRAY(Types.INT),
+ Types.OBJECT_ARRAY(Types.PRIMITIVE_ARRAY(Types.INT))),
+ "a, b, c")
+
+ checkResult(
+ "SELECT a, s FROM T, UNNEST(T.c) AS A (s)",
+ Seq(row(1, Array(12, 45)),
+ row(2, Array(18)),
+ row(2, Array(87)),
+ row(3, Array(1)),
+ row(3, Array(45)))
+ )
+ }
+
+ @Test
+ def testUnnestObjectArrayFromTableWithFilter(): Unit = {
+ val data = List(
+ row(1, Array(row(12, "45.6"), row(12, "45.612"))),
+ row(2, Array(row(13, "41.6"), row(14, "45.2136"))),
+ row(3, Array(row(18, "42.6")))
+ )
+ registerCollection("T", data,
+ new RowTypeInfo(
+ Types.INT,
+ ObjectArrayTypeInfo.getInfoFor(classOf[Array[Row]],
+ new RowTypeInfo(Types.INT, Types.STRING))),
+ "a, b")
+
+ checkResult(
+ "SELECT a, b, s, t FROM T, UNNEST(T.b) AS A (s, t) WHERE s > 13",
+ Seq(row(2, Array(row(13, 41.6), row(14, 45.2136)), 14, 45.2136),
+ row(3, Array(row(18, 42.6)), 18, 42.6))
+ )
+ }
+
+ @Test
+ def testUnnestMultiSetFromCollectResult(): Unit = {
+ val data = List(
+ row(1, 1, row(12, "45.6")),
+ row(2, 2, row(12, "45.612")),
+ row(3, 2, row(13, "41.6")),
+ row(4, 3, row(14, "45.2136")),
+ row(5, 3, row(18, "42.6")))
+ registerCollection("T", data,
+ new RowTypeInfo(Types.INT, Types.INT, new RowTypeInfo(Types.INT, Types.STRING)),
+ "a, b, c")
+
+ checkResult(
+ """
+ |WITH T1 AS (SELECT b, COLLECT(c) as `set` FROM T GROUP BY b)
+ |SELECT b, id, point FROM T1, UNNEST(T1.`set`) AS A(id, point) WHERE b < 3
+ """.stripMargin,
+ Seq(row(1, 12, "45.6"), row(2, 12, "45.612"), row(2, 13, "41.6"))
+ )
+ }
+
+ @Test
+ def testLeftUnnestMultiSetFromCollectResult(): Unit = {
+ val data = List(
+ row(1, "1", "Hello"),
+ row(1, "2", "Hello2"),
+ row(2, "2", "Hello"),
+ row(3, null.asInstanceOf[String], "Hello"),
+ row(4, "4", "Hello"),
+ row(5, "5", "Hello"),
+ row(5, null.asInstanceOf[String], "Hello"),
+ row(6, "6", "Hello"),
+ row(7, "7", "Hello World"),
+ row(7, "8", "Hello World"))
+
+ registerCollection("T", data, new RowTypeInfo(Types.INT, Types.STRING, Types.STRING), "a, b, c")
+
+ checkResult(
+ """
+ |WITH T1 AS (SELECT a, COLLECT(b) as `set` FROM T GROUP BY a)
+ |SELECT a, s FROM T1 LEFT JOIN UNNEST(T1.`set`) AS A(s) ON TRUE WHERE a < 5
+ """.stripMargin,
+ Seq(row(1, "1"), row(1, "2"), row(2, "2"), row(3, null), row(4, "4"))
+ )
+ }
+
+ @Test
+ def testTumbleWindowAggregateWithCollectUnnest(): Unit = {
+ val data = TestData.tupleData3.map {
+ case (i, l, s) => row(i, l, s, new Timestamp(i * 1000))
+ }
+ registerCollection("T", data,
+ new RowTypeInfo(Types.INT, Types.LONG, Types.STRING, Types.SQL_TIMESTAMP),
+ "a, b, c, ts")
+
+ checkResult(
+ """
+ |WITH T1 AS (SELECT b, COLLECT(b) as `set`
+ | FROM T
+ | GROUP BY b, TUMBLE(ts, INTERVAL '3' SECOND)
+ |)
+ |SELECT b, s FROM T1, UNNEST(T1.`set`) AS A(s) where b < 3
+ """.stripMargin,
+ Seq(row(1, 1), row(2, 2), row(2, 2))
+ )
+ }
+
+ @Test
+ def testCrossWithUnnest(): Unit = {
+ val data = List(
+ row(1, 1L, Array("Hi", "w")),
+ row(2, 2L, Array("Hello", "k")),
+ row(3, 2L, Array("Hello world", "x"))
+ )
+ registerCollection("T", data,
+ new RowTypeInfo(Types.INT, Types.LONG, Types.OBJECT_ARRAY(Types.STRING)),
+ "a, b, c")
+
+ checkResult(
+ "SELECT a, s FROM T, UNNEST(T.c) as A (s)",
+ Seq(row(1, "Hi"), row(1, "w"), row(2, "Hello"),
+ row(2, "k"), row(3, "Hello world"), row(3, "x"))
+ )
+ }
+
+ @Test
+ def testCrossWithUnnestForMap(): Unit = {
+ val data = List(
+ row(1, 11L, Map("a" -> "10", "b" -> "11").asJava),
+ row(2, 22L, Map("c" -> "20").asJava),
+ row(3, 33L, Map("d" -> "30", "e" -> "31").asJava)
+ )
+
+ registerCollection("T", data,
+ new RowTypeInfo(Types.INT, Types.LONG, Types.MAP(Types.STRING, Types.STRING)),
+ "a, b, c")
+
+ checkResult(
+ "SELECT a, b, v FROM T CROSS JOIN UNNEST(c) as f (k, v)",
+ Seq(row(1, "11", "10"), row(1, "11", "11"), row(2, "22", "20"),
+ row(3, "33", "30"), row(3, "33", "31"))
+ )
+ }
+
+ @Test
+ def testJoinWithUnnestOfTuple(): Unit = {
+ val data = List(
+ row(1, Array(row(12, "45.6"), row(2, "45.612"))),
+ row(2, Array(row(13, "41.6"), row(1, "45.2136"))),
+ row(3, Array(row(18, "42.6"))))
+ registerCollection("T", data,
+ new RowTypeInfo(Types.INT,
+ ObjectArrayTypeInfo.getInfoFor(classOf[Array[Row]],
+ new RowTypeInfo(Types.INT, Types.STRING))),
+ "a, b")
+
+ checkResult(
+ "SELECT a, b, x, y " +
+ "FROM " +
+ " (SELECT a, b FROM T WHERE a < 3) as tf, " +
+ " UNNEST(tf.b) as A (x, y) " +
+ "WHERE x > a",
+ Seq(row(1, Array(row(12, 45.6), row(2, 45.612)), 12, 45.6),
+ row(1, Array(row(12, 45.6), row(2, 45.612)), 2, 45.612),
+ row(2, Array(row(13, 41.6), row(1, 45.2136)), 13, 41.6))
+ )
+ }
+
+ @Test
+ def testUnnestObjectArrayWithoutAlias(): Unit = {
+ val data = List(
+ row(1, Array(row(12, "45.6"), row(12, "45.612"))),
+ row(2, Array(row(13, "41.6"), row(14, "45.2136"))),
+ row(3, Array(row(18, "42.6")))
+ )
+ registerCollection("T", data,
+ new RowTypeInfo(
+ Types.INT,
+ ObjectArrayTypeInfo.getInfoFor(classOf[Array[Row]],
+ new RowTypeInfo(Types.INT, Types.STRING))),
+ "a, b")
+
+ checkResult(
+ "SELECT a, b, A.f0, A.f1 FROM T, UNNEST(T.b) AS A where A.f0 > 13",
+ Seq(row(2, Array(row(13, 41.6), row(14, 45.2136)), 14, 45.2136),
+ row(3, Array(row(18, 42.6)), 18, 42.6))
+ )
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala
index 5bbb23a..9d922c8 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/agg/AggregateReduceGroupingITCase.scala
@@ -126,7 +126,7 @@ class AggregateReduceGroupingITCase extends BatchTestBase {
PlannerConfigOptions.SQL_OPTIMIZER_AGG_PHASE_ENFORCER, "TWO_PHASE")
// set smaller parallelism to avoid MemoryAllocationException
tEnv.getConfig.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_DEFAULT_PARALLELISM, 2)
- tEnv.getConfig.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_HASH_AGG_TABLE_MEM, 2) // 1M
+ tEnv.getConfig.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_HASH_AGG_TABLE_MEM, 2) // 2M
testSingleAggOnTable()
}
@@ -135,7 +135,7 @@ class AggregateReduceGroupingITCase extends BatchTestBase {
tEnv.getConfig.getConf.setString(TableConfigOptions.SQL_EXEC_DISABLED_OPERATORS, "SortAgg")
tEnv.getConfig.getConf.setString(
PlannerConfigOptions.SQL_OPTIMIZER_AGG_PHASE_ENFORCER, "ONE_PHASE")
- tEnv.getConfig.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_HASH_AGG_TABLE_MEM, 2) // 1M
+ tEnv.getConfig.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_HASH_AGG_TABLE_MEM, 2) // 2M
testSingleAggOnTable()
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
index d9ac9d0..61d55cb 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
@@ -19,10 +19,9 @@
package org.apache.flink.table.runtime.batch.sql.join
import org.apache.flink.api.common.ExecutionConfig
-import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo.STRING_ARRAY_TYPE_INFO
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, LONG_TYPE_INFO, STRING_TYPE_INFO}
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, LONG_TYPE_INFO}
import org.apache.flink.api.common.typeutils.TypeComparator
-import org.apache.flink.api.java.typeutils.{GenericTypeInfo, ObjectArrayTypeInfo, RowTypeInfo}
+import org.apache.flink.api.java.typeutils.{GenericTypeInfo, RowTypeInfo}
import org.apache.flink.table.api.{TableConfigOptions, Types}
import org.apache.flink.table.runtime.CodeGenOperatorFactory
import org.apache.flink.table.runtime.batch.sql.join.JoinType.{BroadcastHashJoin, HashJoin, JoinType, NestedLoopJoin, SortMergeJoin}
@@ -30,7 +29,6 @@ import org.apache.flink.table.runtime.utils.BatchTestBase
import org.apache.flink.table.runtime.utils.BatchTestBase.row
import org.apache.flink.table.runtime.utils.TestData._
import org.apache.flink.table.sinks.CollectRowTableSink
-import org.apache.flink.types.Row
import org.junit.{Assert, Before, Ignore, Test}
@@ -495,45 +493,6 @@ class JoinITCase() extends BatchTestBase {
@Ignore
@Test
- def testCrossWithUnnest(): Unit = {
- val data = List(
- row(1, 1L, Array("Hi", "w")),
- row(2, 2L, Array("Hello", "k")),
- row(3, 2L, Array("Hello world", "x"))
- )
- registerCollection("T", data,
- new RowTypeInfo(INT_TYPE_INFO, LONG_TYPE_INFO, STRING_ARRAY_TYPE_INFO),
- "a, b, c")
-
- checkResult(
- "SELECT a, s FROM T, UNNEST(T.c) as A (s)",
- Seq())
- }
-
- @Ignore
- @Test
- def testJoinWithUnnestOfTuple(): Unit = {
- val data = List(
- row(1, Array(row(12, "45.6"), row(2, "45.612"))),
- row(2, Array(row(13, "41.6"), row(1, "45.2136"))),
- row(3, Array(row(18, "42.6"))))
- registerCollection("T", data,
- new RowTypeInfo(INT_TYPE_INFO,
- ObjectArrayTypeInfo.getInfoFor(classOf[Array[Row]],
- new RowTypeInfo(INT_TYPE_INFO, STRING_TYPE_INFO))),
- "a, b")
-
- checkResult(
- "SELECT a, b, x, y " +
- "FROM " +
- " (SELECT a, b FROM T WHERE a < 3) as tf, " +
- " UNNEST(tf.b) as A (x, y) " +
- "WHERE x > a",
- Seq())
- }
-
- @Ignore
- @Test
def testJoinConditionNeedSimplify(): Unit = {
checkResult(
"SELECT A.d FROM Table5 A JOIN SmallTable3 B ON (A.d=B.a and B.a>2) or (A.d=B.a and B.b=1)",
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/AggregateITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/AggregateITCase.scala
index 0e20799..b3d7f3f 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/AggregateITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/AggregateITCase.scala
@@ -982,7 +982,6 @@ class AggregateITCase(
assertEquals(expected.sorted, sink.getRetractResults.sorted)
}
- @Ignore("Fix correlate variable")
@Test
def testCollectOnClusteredFields(): Unit = {
val data = List(
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/UnnestITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/UnnestITCase.scala
new file mode 100644
index 0000000..03432d7
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/UnnestITCase.scala
@@ -0,0 +1,325 @@
+/*
+ * 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.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.runtime.utils.StreamingWithStateTestBase.StateBackendMode
+import org.apache.flink.table.runtime.utils.TimeTestUtil.TimestampAndWatermarkWithOffset
+import org.apache.flink.table.runtime.utils.{StreamingWithStateTestBase, TestData, TestingAppendSink, TestingRetractSink, TestingRetractTableSink}
+import org.apache.flink.types.Row
+
+import org.junit.Assert.assertEquals
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+
+@RunWith(classOf[Parameterized])
+class UnnestITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode) {
+
+ @Test
+ def testUnnestPrimitiveArrayFromTable(): Unit = {
+ val data = List(
+ (1, Array(12, 45), Array(Array(12, 45))),
+ (2, Array(41, 5), Array(Array(18), Array(87))),
+ (3, Array(18, 42), Array(Array(1), Array(45)))
+ )
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery = "SELECT a, b, s FROM T, UNNEST(T.b) AS A (s)"
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ val sink = new TestingAppendSink
+ result.addSink(sink)
+ env.execute()
+
+ val expected = List(
+ "1,[12, 45],12",
+ "1,[12, 45],45",
+ "2,[41, 5],41",
+ "2,[41, 5],5",
+ "3,[18, 42],18",
+ "3,[18, 42],42")
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testUnnestArrayOfArrayFromTable(): Unit = {
+ val data = List(
+ (1, Array(12, 45), Array(Array(12, 45))),
+ (2, Array(41, 5), Array(Array(18), Array(87))),
+ (3, Array(18, 42), Array(Array(1), Array(45)))
+ )
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery = "SELECT a, s FROM T, UNNEST(T.c) AS A (s)"
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ val sink = new TestingAppendSink
+ result.addSink(sink)
+ env.execute()
+
+ val expected = List(
+ "1,[12, 45]",
+ "2,[18]",
+ "2,[87]",
+ "3,[1]",
+ "3,[45]")
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testUnnestObjectArrayFromTableWithFilter(): Unit = {
+ val data = List(
+ (1, Array((12, "45.6"), (12, "45.612"))),
+ (2, Array((13, "41.6"), (14, "45.2136"))),
+ (3, Array((18, "42.6")))
+ )
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery = "SELECT a, b, s, t FROM T, UNNEST(T.b) AS A (s, t) WHERE s > 13"
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ val sink = new TestingAppendSink
+ result.addSink(sink)
+ env.execute()
+
+ val expected = List(
+ "2,[13,41.6, 14,45.2136],14,45.2136",
+ "3,[18,42.6],18,42.6")
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testUnnestMultiSetFromCollectResult(): Unit = {
+ val data = List(
+ (1, 1, (12, "45.6")),
+ (2, 2, (12, "45.612")),
+ (3, 2, (13, "41.6")),
+ (4, 3, (14, "45.2136")),
+ (5, 3, (18, "42.6")))
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery =
+ """
+ |WITH T1 AS (SELECT b, COLLECT(c) as `set` FROM T GROUP BY b)
+ |SELECT b, id, point FROM T1, UNNEST(T1.`set`) AS A(id, point) WHERE b < 3
+ """.stripMargin
+ val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
+ val sink = new TestingRetractSink
+ result.addSink(sink)
+ env.execute()
+
+ val expected = List(
+ "1,12,45.6",
+ "2,12,45.612",
+ "2,13,41.6")
+ assertEquals(expected.sorted, sink.getRetractResults.sorted)
+ }
+
+ @Test
+ def testLeftUnnestMultiSetFromCollectResult(): Unit = {
+ val data = List(
+ (1, "1", "Hello"),
+ (1, "2", "Hello2"),
+ (2, "2", "Hello"),
+ (3, null.asInstanceOf[String], "Hello"),
+ (4, "4", "Hello"),
+ (5, "5", "Hello"),
+ (5, null.asInstanceOf[String], "Hello"),
+ (6, "6", "Hello"),
+ (7, "7", "Hello World"),
+ (7, "8", "Hello World")
+ )
+
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery =
+ """
+ |WITH T1 AS (SELECT a, COLLECT(b) as `set` FROM T GROUP BY a)
+ |SELECT a, s FROM T1 LEFT JOIN UNNEST(T1.`set`) AS A(s) ON TRUE WHERE a < 5
+ """.stripMargin
+ val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
+ val sink = new TestingRetractSink
+ result.addSink(sink).setParallelism(1)
+ env.execute()
+
+ val expected = List(
+ "1,1",
+ "1,2",
+ "2,2",
+ "3,null",
+ "4,4"
+ )
+ assertEquals(expected.sorted, sink.getRetractResults.sorted)
+ }
+
+ @Test
+ def testTumbleWindowAggregateWithCollectUnnest(): Unit = {
+ val data = TestData.tupleData3.map {
+ case (i, l, s) => (l, i, s)
+ }
+ val stream = failingDataSource(data)
+ .assignTimestampsAndWatermarks(new TimestampAndWatermarkWithOffset[(Long, Int, String)](0L))
+ val t = stream.toTable(tEnv, 'b, 'a, 'c, 'rowtime)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery =
+ """
+ |WITH T1 AS (SELECT b, COLLECT(b) as `set`
+ | FROM T
+ | GROUP BY b, TUMBLE(rowtime, INTERVAL '3' SECOND)
+ |)
+ |SELECT b, s FROM T1, UNNEST(T1.`set`) AS A(s) where b < 3
+ """.stripMargin
+ val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
+ val sink = new TestingRetractSink
+ result.addSink(sink).setParallelism(1)
+ env.execute()
+
+ val expected = List(
+ "1,1",
+ "2,2",
+ "2,2"
+ )
+ assertEquals(expected.sorted, sink.getRetractResults.sorted)
+ }
+
+ @Test
+ def testCrossWithUnnest(): Unit = {
+ val data = List(
+ (1, 1L, Array("Hi", "w")),
+ (2, 2L, Array("Hello", "k")),
+ (3, 2L, Array("Hello world", "x"))
+ )
+
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery = "SELECT a, s FROM T, UNNEST(T.c) as A (s)"
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ val sink = new TestingAppendSink
+ result.addSink(sink)
+ env.execute()
+
+ val expected = List("1,Hi", "1,w", "2,Hello", "2,k", "3,Hello world", "3,x")
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testCrossWithUnnestForMap(): Unit = {
+ val data = List(
+ Row.of(Int.box(1),
+ Long.box(11L), {
+ val map = new java.util.HashMap[String, String]()
+ map.put("a", "10")
+ map.put("b", "11")
+ map
+ }),
+ Row.of(Int.box(2),
+ Long.box(22L), {
+ val map = new java.util.HashMap[String, String]()
+ map.put("c", "20")
+ map
+ }),
+ Row.of(Int.box(3),
+ Long.box(33L), {
+ val map = new java.util.HashMap[String, String]()
+ map.put("d", "30")
+ map.put("e", "31")
+ map
+ })
+ )
+
+ implicit val typeInfo = Types.ROW(
+ fieldNames = Array("a", "b", "c"),
+ types = Array(Types.INT,
+ Types.LONG,
+ Types.MAP(Types.STRING, Types.STRING))
+ )
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery = "SELECT a, b, v FROM T CROSS JOIN UNNEST(c) as f (k, v)"
+ val result = tEnv.sqlQuery(sqlQuery)
+
+ val sink = new TestingRetractTableSink().configure(
+ Array("a", "b", "v"),
+ Array(Types.INT, Types.LONG, Types.STRING))
+ tEnv.writeToSink(result, sink)
+ tEnv.execute()
+
+ val expected = List("1,11,10", "1,11,11", "2,22,20", "3,33,30", "3,33,31")
+ assertEquals(expected.sorted, sink.getRetractResults.sorted)
+ }
+
+ @Test
+ def testJoinWithUnnestOfTuple(): Unit = {
+ val data = List(
+ (1, Array((12, "45.6"), (2, "45.612"))),
+ (2, Array((13, "41.6"), (1, "45.2136"))),
+ (3, Array((18, "42.6")))
+ )
+
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery = "SELECT a, b, x, y " +
+ "FROM " +
+ " (SELECT a, b FROM T WHERE a < 3) as tf, " +
+ " UNNEST(tf.b) as A (x, y) " +
+ "WHERE x > a"
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ val sink = new TestingAppendSink
+ result.addSink(sink)
+ env.execute()
+
+ val expected = List(
+ "1,[12,45.6, 2,45.612],12,45.6",
+ "1,[12,45.6, 2,45.612],2,45.612",
+ "2,[13,41.6, 1,45.2136],13,41.6")
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testUnnestObjectArrayWithoutAlias(): Unit = {
+ val data = List(
+ (1, Array((12, "45.6"), (12, "45.612"))),
+ (2, Array((13, "41.6"), (14, "45.2136"))),
+ (3, Array((18, "42.6")))
+ )
+ val t = env.fromCollection(data).toTable(tEnv, 'a, 'b)
+ tEnv.registerTable("T", t)
+
+ val sqlQuery = "SELECT a, b, A._1, A._2 FROM T, UNNEST(T.b) AS A where A._1 > 13"
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ val sink = new TestingAppendSink
+ result.addSink(sink)
+ env.execute()
+
+ val expected = List(
+ "2,[13,41.6, 14,45.2136],14,45.2136",
+ "3,[18,42.6],18,42.6")
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/util/TableTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/util/TableTestBase.scala
index be85b3c..2081749 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/util/TableTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/util/TableTestBase.scala
@@ -92,7 +92,7 @@ abstract class TableTestUtil(test: TableTestBase) {
// scala env
val env = new StreamExecutionEnvironment(javaEnv)
- protected def getTableEnv: TableEnvironment
+ def getTableEnv: TableEnvironment
// a counter for unique table names
private var counter = 0