flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From twal...@apache.org
Subject flink git commit: [FLINK-4183] [table] Move checking for StreamTableEnvironment into validation layer
Date Tue, 19 Jul 2016 14:41:04 GMT
Repository: flink
Updated Branches:
  refs/heads/master dd53831aa -> e85f787b2


[FLINK-4183] [table] Move checking for StreamTableEnvironment into validation layer

This closes #2221.


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

Branch: refs/heads/master
Commit: e85f787b280b63960e7f3add5aa8613b4ee23795
Parents: dd53831
Author: twalthr <twalthr@apache.org>
Authored: Fri Jul 8 16:50:17 2016 +0200
Committer: twalthr <twalthr@apache.org>
Committed: Tue Jul 19 16:27:35 2016 +0200

----------------------------------------------------------------------
 .../api/table/plan/logical/operators.scala      | 16 ++++++--
 .../org/apache/flink/api/table/table.scala      | 16 ++------
 .../scala/stream/table/UnsupportedOpsTest.scala | 41 ++++++++++++++++----
 3 files changed, 49 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/e85f787b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
index 4df8a5e..381244e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/logical/operators.scala
@@ -128,7 +128,7 @@ case class Distinct(child: LogicalNode) extends UnaryNode {
 
   override def validate(tableEnv: TableEnvironment): LogicalNode = {
     if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Distinct on stream tables is currently not supported.")
+      failValidation(s"Distinct on stream tables is currently not supported.")
     }
     this
   }
@@ -144,7 +144,7 @@ case class Sort(order: Seq[Ordering], child: LogicalNode) extends UnaryNode
{
 
   override def validate(tableEnv: TableEnvironment): LogicalNode = {
     if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Distinct on stream tables is currently not supported.")
+      failValidation(s"Distinct on stream tables is currently not supported.")
     }
     super.validate(tableEnv)
   }
@@ -196,7 +196,7 @@ case class Aggregate(
 
   override def validate(tableEnv: TableEnvironment): LogicalNode = {
     if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Aggregate on stream tables is currently not supported.")
+      failValidation(s"Aggregate on stream tables is currently not supported.")
     }
 
     val resolvedAggregate = super.validate(tableEnv).asInstanceOf[Aggregate]
@@ -277,6 +277,10 @@ case class Union(left: LogicalNode, right: LogicalNode, all: Boolean)
extends Bi
   }
 
   override def validate(tableEnv: TableEnvironment): LogicalNode = {
+    if (tableEnv.isInstanceOf[StreamTableEnvironment] && !all) {
+      failValidation(s"Union on stream tables is currently not supported.")
+    }
+
     val resolvedUnion = super.validate(tableEnv).asInstanceOf[Union]
     if (left.output.length != right.output.length) {
       failValidation(s"Union two tables of different column sizes:" +
@@ -304,6 +308,10 @@ case class Intersect(left: LogicalNode, right: LogicalNode, all: Boolean)
extend
   }
 
   override def validate(tableEnv: TableEnvironment): LogicalNode = {
+    if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
+      failValidation(s"Intersect on stream tables is currently not supported.")
+    }
+
     val resolvedIntersect = super.validate(tableEnv).asInstanceOf[Intersect]
     if (left.output.length != right.output.length) {
       failValidation(s"Intersect two tables of different column sizes:" +
@@ -392,7 +400,7 @@ case class Join(
 
   override def validate(tableEnv: TableEnvironment): LogicalNode = {
     if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Join on stream tables is currently not supported.")
+      failValidation(s"Join on stream tables is currently not supported.")
     }
 
     val resolvedJoin = super.validate(tableEnv).asInstanceOf[Join]

http://git-wip-us.apache.org/repos/asf/flink/blob/e85f787b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
index e719782..cbb9a07 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/table.scala
@@ -202,7 +202,7 @@ class Table(
     */
   def groupBy(fields: Expression*): GroupedTable = {
     if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Group by on stream tables is currently not supported.")
+      throw new ValidationException(s"Group by on stream tables is currently not supported.")
     }
     new GroupedTable(this, fields)
   }
@@ -392,7 +392,6 @@ class Table(
   }
 
   private def join(right: Table, joinPredicate: Option[Expression], joinType: JoinType):
Table = {
-
     // check that right table belongs to the same TableEnvironment
     if (right.tableEnv != this.tableEnv) {
       throw new ValidationException("Only tables from the same TableEnvironment can be joined.")
@@ -464,14 +463,11 @@ class Table(
     * }}}
     */
   def union(right: Table): Table = {
-    if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Union on stream tables is currently not supported.")
-    }
     // check that right table belongs to the same TableEnvironment
     if (right.tableEnv != this.tableEnv) {
       throw new ValidationException("Only tables from the same TableEnvironment can be unioned.")
     }
-    new Table(tableEnv, Union(logicalPlan, right.logicalPlan, false).validate(tableEnv))
+    new Table(tableEnv, Union(logicalPlan, right.logicalPlan, all = false).validate(tableEnv))
   }
 
   /**
@@ -491,7 +487,7 @@ class Table(
     if (right.tableEnv != this.tableEnv) {
       throw new ValidationException("Only tables from the same TableEnvironment can be unioned.")
     }
-    new Table(tableEnv, Union(logicalPlan, right.logicalPlan, true).validate(tableEnv))
+    new Table(tableEnv, Union(logicalPlan, right.logicalPlan, all = true).validate(tableEnv))
   }
 
   /**
@@ -509,9 +505,6 @@ class Table(
     * }}}
     */
   def intersect(right: Table): Table = {
-    if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Intersect on stream tables is currently not supported.")
-    }
     // check that right table belongs to the same TableEnvironment
     if (right.tableEnv != this.tableEnv) {
       throw new ValidationException(
@@ -535,9 +528,6 @@ class Table(
     * }}}
     */
   def intersectAll(right: Table): Table = {
-    if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
-      throw new TableException(s"Intersect on stream tables is currently not supported.")
-    }
     // check that right table belongs to the same TableEnvironment
     if (right.tableEnv != this.tableEnv) {
       throw new ValidationException(

http://git-wip-us.apache.org/repos/asf/flink/blob/e85f787b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnsupportedOpsTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnsupportedOpsTest.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnsupportedOpsTest.scala
index df22f2f..89b0fdc 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnsupportedOpsTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/stream/table/UnsupportedOpsTest.scala
@@ -20,21 +20,21 @@ package org.apache.flink.api.scala.stream.table
 
 import org.apache.flink.api.scala.stream.utils.StreamTestData
 import org.apache.flink.api.scala.table._
-import org.apache.flink.api.table.{ValidationException, TableEnvironment, TableException}
+import org.apache.flink.api.table.{TableEnvironment, ValidationException}
 import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
 import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase
 import org.junit.Test
 
 class UnsupportedOpsTest extends StreamingMultipleProgramsTestBase {
 
-  @Test(expected = classOf[TableException])
+  @Test(expected = classOf[ValidationException])
   def testSelectWithAggregation(): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env)
     StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv).select('_1.min)
   }
 
-  @Test(expected = classOf[TableException])
+  @Test(expected = classOf[ValidationException])
   def testGroupBy(): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env)
@@ -42,21 +42,21 @@ class UnsupportedOpsTest extends StreamingMultipleProgramsTestBase {
       .groupBy('_1)
   }
 
-  @Test(expected = classOf[TableException])
+  @Test(expected = classOf[ValidationException])
   def testDistinct(): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env)
     StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv).distinct()
   }
 
-  @Test(expected = classOf[TableException])
+  @Test(expected = classOf[ValidationException])
   def testSort(): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env)
     StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv).orderBy('_1.desc)
   }
 
-  @Test(expected = classOf[TableException])
+  @Test(expected = classOf[ValidationException])
   def testJoin(): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env)
@@ -65,7 +65,7 @@ class UnsupportedOpsTest extends StreamingMultipleProgramsTestBase {
     t1.join(t2)
   }
 
-  @Test(expected = classOf[TableException])
+  @Test(expected = classOf[ValidationException])
   def testUnion(): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env)
@@ -75,6 +75,24 @@ class UnsupportedOpsTest extends StreamingMultipleProgramsTestBase {
   }
 
   @Test(expected = classOf[ValidationException])
+  def testIntersect(): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env)
+    val t1 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv)
+    val t2 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv)
+    t1.intersect(t2)
+  }
+
+  @Test(expected = classOf[ValidationException])
+  def testIntersectAll(): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env)
+    val t1 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv)
+    val t2 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv)
+    t1.intersectAll(t2)
+  }
+
+  @Test(expected = classOf[ValidationException])
   def testMinus(): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env)
@@ -82,4 +100,13 @@ class UnsupportedOpsTest extends StreamingMultipleProgramsTestBase {
     val t2 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv)
     t1.minus(t2)
   }
+
+  @Test(expected = classOf[ValidationException])
+  def testMinusAll(): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env)
+    val t1 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv)
+    val t2 = StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv)
+    t1.minusAll(t2)
+  }
 }


Mime
View raw message