flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [1/3] flink git commit: [FLINK-6033] [table] Add support for SQL UNNEST.
Date Sun, 07 May 2017 12:18:28 GMT
Repository: flink
Updated Branches:
  refs/heads/master e5b65a7fc -> fe4e96a72


[FLINK-6033] [table] Add support for SQL UNNEST.

This closes #3793.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fe4e96a7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fe4e96a7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fe4e96a7

Branch: refs/heads/master
Commit: fe4e96a726dd32fb948db050b975312e120e2461
Parents: 9f2293c
Author: Shuyi Chen <shuyi@uber.com>
Authored: Fri Apr 21 23:48:28 2017 -0700
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Sun May 7 13:32:26 2017 +0200

----------------------------------------------------------------------
 docs/dev/table_api.md                           |   2 +
 .../flink/table/calcite/FlinkTypeFactory.scala  |   5 +-
 .../utils/UserDefinedFunctionUtils.scala        |   6 +-
 .../flink/table/plan/nodes/FlinkRelNode.scala   |   3 +-
 .../flink/table/plan/rules/FlinkRuleSets.scala  |   3 +
 .../plan/rules/logical/LogicalUnnestRule.scala  | 134 +++++++++++++++++++
 .../table/plan/util/ExplodeFunctionUtil.scala   |  91 +++++++++++++
 .../flink/table/typeutils/TypeCheckUtils.scala  |   4 +-
 .../table/api/scala/batch/sql/JoinITCase.scala  |  54 ++++++++
 .../table/api/scala/stream/sql/SqlITCase.scala  |  90 +++++++++++++
 10 files changed, 388 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/docs/dev/table_api.md
----------------------------------------------------------------------
diff --git a/docs/dev/table_api.md b/docs/dev/table_api.md
index a77d994..d105188 100644
--- a/docs/dev/table_api.md
+++ b/docs/dev/table_api.md
@@ -1482,6 +1482,7 @@ val result2 = tableEnv.sql(
 #### Limitations
 
 Joins, set operations, and non-windowed aggregations are not supported yet.
+`UNNEST` supports only arrays and does not support `WITH ORDINALITY` yet.
 
 {% top %}
 
@@ -1690,6 +1691,7 @@ tableReference:
 tablePrimary:
   [ TABLE ] [ [ catalogName . ] schemaName . ] tableName
   | LATERAL TABLE '(' functionName '(' expression [, expression ]* ')' ')'
+  | UNNEST '(' expression ')'
 
 values:
   VALUES expression [, expression ]*

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
index 9281ad8..eba1623 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
@@ -26,7 +26,7 @@ import org.apache.calcite.sql.`type`.SqlTypeName._
 import org.apache.calcite.sql.`type`.{BasicSqlType, SqlTypeName}
 import org.apache.calcite.sql.parser.SqlParserPos
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
-import org.apache.flink.api.common.typeinfo.{NothingTypeInfo, PrimitiveArrayTypeInfo, SqlTimeTypeInfo,
TypeInformation}
+import org.apache.flink.api.common.typeinfo._
 import org.apache.flink.api.common.typeutils.CompositeType
 import org.apache.flink.api.java.typeutils.ValueTypeInfo._
 import org.apache.flink.api.java.typeutils.{MapTypeInfo, ObjectArrayTypeInfo, RowTypeInfo}
@@ -180,6 +180,9 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
     case pa: PrimitiveArrayTypeInfo[_] =>
       new ArrayRelDataType(pa, createTypeFromTypeInfo(pa.getComponentType), false)
 
+    case ba: BasicArrayTypeInfo[_, _] =>
+      new ArrayRelDataType(ba, createTypeFromTypeInfo(ba.getComponentInfo), true)
+
     case oa: ObjectArrayTypeInfo[_, _] =>
       new ArrayRelDataType(oa, createTypeFromTypeInfo(oa.getComponentInfo), true)
 

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
index 689bf0e..11174de 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
@@ -436,7 +436,11 @@ object UserDefinedFunctionUtils {
     expected.isPrimitive && Primitives.wrap(expected) == candidate ||
     candidate == classOf[Date] && (expected == classOf[Int] || expected == classOf[JInt])
 ||
     candidate == classOf[Time] && (expected == classOf[Int] || expected == classOf[JInt])
||
-    candidate == classOf[Timestamp] && (expected == classOf[Long] || expected ==
classOf[JLong])
+    candidate == classOf[Timestamp] && (expected == classOf[Long] || expected ==
classOf[JLong]) ||
+    (candidate.isArray &&
+      expected.isArray &&
+      candidate.getComponentType.isInstanceOf[Object] &&
+      expected.getComponentType == classOf[Object])
 
   @throws[Exception]
   def serialize(function: UserDefinedFunction): String = {

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
index 0b244e9..8509a8e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/FlinkRelNode.scala
@@ -61,7 +61,8 @@ trait FlinkRelNode extends RelNode {
         val referenceExpr = getExpressionString(fa.getReferenceExpr, inFields, localExprsTable)
         val field = fa.getField.getName
         s"$referenceExpr.$field"
-
+      case cv: RexCorrelVariable =>
+        cv.toString
       case _ =>
         throw new IllegalArgumentException(s"Unknown expression type '${expr.getClass}':
$expr")
     }

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
index fad60fd..980dfd3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
@@ -100,6 +100,9 @@ object FlinkRuleSets {
     PushProjectIntoTableSourceScanRule.INSTANCE,
     PushFilterIntoTableSourceScanRule.INSTANCE,
 
+    // Unnest rule
+    LogicalUnnestRule.INSTANCE,
+
     // translate to flink logical rel nodes
     FlinkLogicalAggregate.CONVERTER,
     FlinkLogicalWindowAggregate.CONVERTER,

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
new file mode 100644
index 0000000..f2d9f2a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/logical/LogicalUnnestRule.scala
@@ -0,0 +1,134 @@
+/*
+ * 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 java.util.Collections
+
+import com.google.common.collect.ImmutableList
+import org.apache.calcite.plan.RelOptRule._
+import org.apache.calcite.plan.volcano.RelSubset
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelOptRuleOperand}
+import org.apache.calcite.rel.`type`.{RelDataTypeFieldImpl, RelRecordType, StructKind}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.Uncollect
+import org.apache.calcite.rel.logical._
+import org.apache.calcite.sql.`type`.AbstractSqlType
+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
+import org.apache.flink.table.plan.util.ExplodeFunctionUtil
+
+class LogicalUnnestRule(
+    operand: RelOptRuleOperand,
+    description: String)
+  extends RelOptRule(operand, description) {
+
+  override def matches(call: RelOptRuleCall): Boolean = {
+
+    val join: LogicalCorrelate = call.rel(0).asInstanceOf[LogicalCorrelate]
+    val right = join.getRight.asInstanceOf[RelSubset].getOriginal
+
+    right match {
+      // a filter is pushed above the table function
+      case filter: LogicalFilter =>
+        filter.getInput.asInstanceOf[RelSubset].getOriginal match {
+          case u: Uncollect => !u.withOrdinality
+          case _ => false
+        }
+      case u: Uncollect => !u.withOrdinality
+      case _ => false
+    }
+  }
+
+  override def onMatch(call: RelOptRuleCall): Unit = {
+    val correlate = call.rel(0).asInstanceOf[LogicalCorrelate]
+
+    val outer = correlate.getLeft.asInstanceOf[RelSubset].getOriginal
+    val array = correlate.getRight.asInstanceOf[RelSubset].getOriginal
+
+    def convert(relNode: RelNode): RelNode = {
+      relNode match {
+        case rs: RelSubset =>
+          convert(rs.getRelList.get(0))
+
+        case f: LogicalFilter =>
+          f.copy(
+            f.getTraitSet,
+            ImmutableList.of(convert(f.getInput.asInstanceOf[RelSubset].getOriginal)))
+
+        case uc: Uncollect =>
+          // convert Uncollect into TableFunctionScan
+          val cluster = correlate.getCluster
+
+          val arrayType =
+            uc.getInput.getRowType.getFieldList.get(0).getValue.asInstanceOf[ArrayRelDataType]
+          val componentType = arrayType.getComponentType
+
+          // create table function
+          val explodeTableFunc = UserDefinedFunctionUtils.createTableSqlFunctions(
+            "explode",
+            ExplodeFunctionUtil.explodeTableFuncFromType(arrayType.typeInfo),
+            FlinkTypeFactory.toTypeInfo(arrayType.getComponentType),
+            cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]).head
+
+          // create table function call
+          val rexCall = cluster.getRexBuilder.makeCall(
+            explodeTableFunc,
+            uc.getInput.asInstanceOf[RelSubset]
+              .getOriginal.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 TableException(
+              s"Unsupported array 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 newCorrleate =
+      correlate.copy(correlate.getTraitSet, ImmutableList.of(outer, tableFunctionScan))
+    call.transformTo(newCorrleate)
+  }
+}
+
+object LogicalUnnestRule {
+  val INSTANCE = new LogicalUnnestRule(
+    operand(classOf[LogicalCorrelate], any),
+    "LogicalUnnestRule")
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala
new file mode 100644
index 0000000..1bcc6d9
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/util/ExplodeFunctionUtil.scala
@@ -0,0 +1,91 @@
+/*
+ * 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.ObjectArrayTypeInfo
+import org.apache.flink.table.functions.TableFunction
+
+class ObjectExplodeTableFunc extends TableFunction[Object] {
+  def eval(arr: Array[Object]): Unit = {
+    arr.foreach(collect)
+  }
+}
+
+class FloatExplodeTableFunc extends TableFunction[Float] {
+  def eval(arr: Array[Float]): Unit = {
+    arr.foreach(collect)
+  }
+}
+
+class ShortExplodeTableFunc extends TableFunction[Short] {
+  def eval(arr: Array[Short]): Unit = {
+    arr.foreach(collect)
+  }
+}
+class IntExplodeTableFunc extends TableFunction[Int] {
+  def eval(arr: Array[Int]): Unit = {
+    arr.foreach(collect)
+  }
+}
+
+class LongExplodeTableFunc extends TableFunction[Long] {
+  def eval(arr: Array[Long]): Unit = {
+    arr.foreach(collect)
+  }
+}
+
+class DoubleExplodeTableFunc extends TableFunction[Double] {
+  def eval(arr: Array[Double]): Unit = {
+    arr.foreach(collect)
+  }
+}
+
+class ByteExplodeTableFunc extends TableFunction[Byte] {
+  def eval(arr: Array[Byte]): Unit = {
+    arr.foreach(collect)
+  }
+}
+
+class BooleanExplodeTableFunc extends TableFunction[Boolean] {
+  def eval(arr: Array[Boolean]): Unit = {
+    arr.foreach(collect)
+  }
+}
+
+object ExplodeFunctionUtil {
+  def explodeTableFuncFromType(ti: TypeInformation[_]):TableFunction[_] = {
+    ti match {
+      case pat: PrimitiveArrayTypeInfo[_] => {
+        pat.getComponentType 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 _: ObjectArrayTypeInfo[_, _] => new ObjectExplodeTableFunc
+      case _: BasicArrayTypeInfo[_, _] => new ObjectExplodeTableFunc
+      case _ => throw new UnsupportedOperationException(ti.toString + "IS NOT supported")
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
index 9896a8c..fea8c2a 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
@@ -68,7 +68,9 @@ object TypeCheckUtils {
   def isLong(dataType: TypeInformation[_]): Boolean = dataType == LONG_TYPE_INFO
 
   def isArray(dataType: TypeInformation[_]): Boolean = dataType match {
-    case _: ObjectArrayTypeInfo[_, _] | _: PrimitiveArrayTypeInfo[_] => true
+    case _: ObjectArrayTypeInfo[_, _] |
+         _: PrimitiveArrayTypeInfo[_] |
+         _: BasicArrayTypeInfo[_, _] => true
     case _ => false
   }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala
index 9df17ad..8a8c0ce 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala
@@ -26,6 +26,7 @@ import org.apache.flink.api.scala.util.CollectionDataSets
 import org.apache.flink.table.api.{TableEnvironment, TableException, ValidationException}
 import org.apache.flink.test.util.TestBaseUtils
 import org.apache.flink.types.Row
+import org.junit.Assert.assertEquals
 import org.junit._
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
@@ -375,6 +376,59 @@ class JoinITCase(
     Assert.assertEquals(0, result)
   }
 
+  @Test
+  def testCrossWithUnnest(): Unit = {
+
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env, config)
+
+    val data = List(
+      (1, 1L, Array("Hi", "w")),
+      (2, 2L, Array("Hello", "k")),
+      (3, 2L, Array("Hello world", "x"))
+    )
+    val stream = env.fromCollection(data)
+    tEnv.registerDataSet("T", stream, 'a, 'b, 'c)
+
+    val sqlQuery = "SELECT a, s FROM T, UNNEST(T.c) as A (s)"
+
+    val result = tEnv.sql(sqlQuery)
+
+    val expected = List("1,Hi", "1,w", "2,Hello", "2,k", "3,Hello world", "3,x")
+    val results = result.toDataSet[Row].collect().toList
+    assertEquals(expected.toString(), results.sortWith(_.toString < _.toString).toString())
+  }
+
+  @Test
+  def testJoinWithUnnestOfTuple(): Unit = {
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env, config)
+
+    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 stream = env.fromCollection(data)
+    tEnv.registerDataSet("T", stream, 'a, 'b)
+
+    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.sql(sqlQuery)
+
+    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").mkString(", ")
+    val results = result.toDataSet[Row].collect().map(_.toString)
+    assertEquals(expected, results.sorted.mkString(", "))
+  }
+
   @Test(expected = classOf[TableException])
   def testRightOuterJoinWithNonEquiJoinPredicate(): Unit = {
 

http://git-wip-us.apache.org/repos/asf/flink/blob/fe4e96a7/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
index 95366e1..4147358 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/SqlITCase.scala
@@ -190,4 +190,94 @@ class SqlITCase extends StreamingWithStateTestBase {
     assertEquals(expected.sorted, StreamITCase.testResults.sorted)
   }
 
+  @Test
+  def testUnnestPrimitiveArrayFromTable(): Unit = {
+
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env)
+    StreamITCase.clear
+
+    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 stream = env.fromCollection(data)
+    tEnv.registerDataStream("T", stream, 'a, 'b, 'c)
+
+    val sqlQuery = "SELECT a, b, s FROM T, UNNEST(T.b) AS A (s)"
+
+    val result = tEnv.sql(sqlQuery).toDataStream[Row]
+    result.addSink(new StreamITCase.StringSink)
+    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, StreamITCase.testResults.sorted)
+  }
+
+  @Test
+  def testUnnestArrayOfArrayFromTable(): Unit = {
+
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env)
+    StreamITCase.clear
+
+    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 stream = env.fromCollection(data)
+    tEnv.registerDataStream("T", stream, 'a, 'b, 'c)
+
+    val sqlQuery = "SELECT a, s FROM T, UNNEST(T.c) AS A (s)"
+
+    val result = tEnv.sql(sqlQuery).toDataStream[Row]
+    result.addSink(new StreamITCase.StringSink)
+    env.execute()
+
+    val expected = List(
+      "1,[12, 45]",
+      "2,[18]",
+      "2,[87]",
+      "3,[1]",
+      "3,[45]")
+    assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+  }
+
+  @Test
+  def testUnnestObjectArrayFromTableWithFilter(): Unit = {
+
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env)
+    StreamITCase.clear
+
+    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 stream = env.fromCollection(data)
+    tEnv.registerDataStream("T", stream, 'a, 'b)
+
+    val sqlQuery = "SELECT a, b, s, t FROM T, UNNEST(T.b) AS A (s, t) WHERE s > 13"
+
+    val result = tEnv.sql(sqlQuery).toDataStream[Row]
+    result.addSink(new StreamITCase.StringSink)
+    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, StreamITCase.testResults.sorted)
+  }
+
 }
+


Mime
View raw message