You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2016/01/29 22:29:40 UTC
[1/3] flink git commit: [FLINK-3225] Implemented optimization of
Table API queries via Calcite
Repository: flink
Updated Branches:
refs/heads/tableOnCalcite 364928e74 -> 7ecb70105
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
index da82b8a..a3d31da 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.api.java.table.test;
-import org.apache.flink.api.table.ExpressionException;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.java.DataSet;
@@ -30,6 +29,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -41,7 +41,7 @@ public class SelectITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testSimpleSelectAllWithAs() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -53,19 +53,19 @@ public class SelectITCase extends MultipleProgramsTestBase {
Table result = in
.select("a, b, c");
-// DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testSimpleSelectWithNaming() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -78,12 +78,12 @@ public class SelectITCase extends MultipleProgramsTestBase {
.select("f0 as a, f1 as b")
.select("a, b");
-// DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = "1,1\n" + "2,2\n" + "3,2\n" + "4,3\n" + "5,3\n" + "6,3\n" + "7,4\n" +
-// "8,4\n" + "9,4\n" + "10,4\n" + "11,5\n" + "12,5\n" + "13,5\n" + "14,5\n" + "15,5\n" +
-// "16,6\n" + "17,6\n" + "18,6\n" + "19,6\n" + "20,6\n" + "21,6\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "1,1\n" + "2,2\n" + "3,2\n" + "4,3\n" + "5,3\n" + "6,3\n" + "7,4\n" +
+ "8,4\n" + "9,4\n" + "10,4\n" + "11,5\n" + "12,5\n" + "13,5\n" + "14,5\n" + "15,5\n" +
+ "16,6\n" + "17,6\n" + "18,6\n" + "19,6\n" + "20,6\n" + "21,6\n";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -95,10 +95,10 @@ public class SelectITCase extends MultipleProgramsTestBase {
Table in = tableEnv.fromDataSet(ds, "a, b");
-// DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = " sorry dude ";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = " sorry dude ";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -110,10 +110,10 @@ public class SelectITCase extends MultipleProgramsTestBase {
Table in = tableEnv.fromDataSet(ds, "a, b, c, d");
-// DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = " sorry dude ";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = " sorry dude ";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -125,10 +125,10 @@ public class SelectITCase extends MultipleProgramsTestBase {
Table in = tableEnv.fromDataSet(ds, "a, b, c, b");
-// DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = " today's not your day ";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = " today's not your day ";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -140,9 +140,9 @@ public class SelectITCase extends MultipleProgramsTestBase {
Table in = tableEnv.fromDataSet(ds, "a, b as c, d");
-// DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = "sorry bro";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "sorry bro";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
index 12e7203..c4b2f01 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.java.table.test;
+import org.apache.flink.api.table.Row;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
@@ -28,6 +29,9 @@ import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
+
+import java.util.List;
@RunWith(Parameterized.class)
public class StringExpressionsITCase extends MultipleProgramsTestBase {
@@ -37,7 +41,7 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testSubstring() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -51,13 +55,13 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
Table result = in
.select("a.substring(0, b)");
-// DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = "AA\nB";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "AA\nB";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testSubstringWithMaxEnd() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -71,10 +75,10 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
Table result = in
.select("a.substring(b)");
-// DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = "CD\nBCD";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "CD\nBCD";
+ compareResultAsText(results, expected);
}
// Calcite does eagerly check expression types
@@ -93,10 +97,10 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
Table result = in
.select("a.substring(0, b)");
-// DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
// Calcite does eagerly check expression types
@@ -115,9 +119,9 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase {
Table result = in
.select("a.substring(b, 15)");
-// DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = resultSet.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = resultSet.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/UnionITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/UnionITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/UnionITCase.java
index de02ee1..d6297d9 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/UnionITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/UnionITCase.java
@@ -31,6 +31,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -42,7 +43,7 @@ public class UnionITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testUnion() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -55,13 +56,13 @@ public class UnionITCase extends MultipleProgramsTestBase {
Table selected = in1.unionAll(in2).select("c");
-// DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hi\n" + "Hello\n" + "Hello world\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hi\n" + "Hello\n" + "Hello world\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testUnionWithFilter() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -74,10 +75,10 @@ public class UnionITCase extends MultipleProgramsTestBase {
Table selected = in1.unionAll(in2).where("b < 2").select("c");
-// DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi\n" + "Hallo\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi\n" + "Hallo\n";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -93,10 +94,10 @@ public class UnionITCase extends MultipleProgramsTestBase {
Table selected = in1.unionAll(in2);
-// DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -112,13 +113,13 @@ public class UnionITCase extends MultipleProgramsTestBase {
Table selected = in1.unionAll(in2);
-// DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testUnionWithAggregation() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -131,10 +132,10 @@ public class UnionITCase extends MultipleProgramsTestBase {
Table selected = in1.unionAll(in2).select("c.count");
-// DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "18";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(selected, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "18";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
index 75a113d..76bdcba 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AggregationsITCase.scala
@@ -33,16 +33,16 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testAggregationTypes(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).toTable
.select('_1.sum, '_1.min, '_1.max, '_1.count, '_1.avg)
-// val results = t.toDataSet[Row].collect()
-// val expected = "231,1,21,21,11"
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val results = t.toDataSet[Row].collect()
+ val expected = "231,1,21,21,11"
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -52,12 +52,12 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
val t = CollectionDataSets.get3TupleDataSet(env).toTable
.select('foo.avg)
-// val expected = ""
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = ""
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testWorkingAggregationDataTypes(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -66,33 +66,33 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
(2: Byte, 2: Short, 2, 2L, 2.0f, 2.0d, "Ciao")).toTable
.select('_1.avg, '_2.avg, '_3.avg, '_4.avg, '_5.avg, '_6.avg, '_7.count)
-// val expected = "1,1,1,1,1.5,1.5,2"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1,1,1,1,1.5,1.5,2"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testAggregationWithArithmetic(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements((1f, "Hello"), (2f, "Ciao")).toTable
.select(('_1 + 2).avg + 2, '_2.count + 5)
-// val expected = "5.5,7"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "5.5,7"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testAggregationWithTwoCount(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements((1f, "Hello"), (2f, "Ciao")).toTable
.select('_1.count, '_2.count)
-// val expected = "2,2"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "2,2"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Ignore // Calcite does not eagerly check types
@@ -103,9 +103,9 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
val t = env.fromElements(("Hello", 1)).toTable
.select('_1.sum)
-// val expected = ""
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = ""
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -115,12 +115,12 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
val t = env.fromElements(("Hello", 1)).toTable
.select('_2.sum.sum)
-// val expected = ""
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = ""
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testSQLStyleAggregations(): Unit = {
// the grouping key needs to be forwarded to the intermediate DataSet, even
@@ -136,10 +136,9 @@ class AggregationsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBa
|Count(a) as e1, a.count as e2
""".stripMargin)
-// val expected = "231,231,1,1,21,21,11,11,21,21"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "231,231,1,1,21,21,11,11,21,21"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
index a32774f..5ff2b82 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/AsITCase.scala
@@ -18,7 +18,7 @@
package org.apache.flink.api.scala.table.test
-import org.apache.flink.api.table.{Row, ExpressionException}
+import org.apache.flink.api.table.Row
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.api.scala.util.CollectionDataSets
@@ -33,20 +33,20 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testAs(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
-// val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -55,9 +55,9 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -66,9 +66,9 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c, 'd)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -77,9 +77,9 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'b)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -89,9 +89,9 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
// as can only have field references
val t = CollectionDataSets.get3TupleDataSet(env).as('a + 1, 'b, 'b)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -101,8 +101,8 @@ class AsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
// as can only have field references
val t = CollectionDataSets.get3TupleDataSet(env).as('a as 'foo, 'b, 'b)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
index 8199f6b..4a37737 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/CastingITCase.scala
@@ -44,12 +44,12 @@ class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
val t = env.fromElements((1: Byte, 1: Short, 1, 1L, 1.0f, 1.0d, new Date(0))).toTable
.select('_1 + "b", '_2 + "s", '_3 + "i", '_4 + "L", '_5 + "f", '_6 + "d", '_7 + "Date")
-// val expected = "1b,1s,1i,1L,1.0f,1.0d,1970-01-01 00:00:00.000Date"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1b,1s,1i,1L,1.0f,1.0d,1970-01-01 00:00:00.000Date"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testNumericAutoCastInArithmetic(): Unit = {
// don't test everything, just some common cast directions
@@ -58,12 +58,12 @@ class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
val t = env.fromElements((1: Byte, 1: Short, 1, 1L, 1.0f, 1.0d)).toTable
.select('_1 + 1, '_2 + 1, '_3 + 1L, '_4 + 1.0f, '_5 + 1.0d, '_6 + 1)
-// val expected = "2,2,2,2.0,2.0,2.0"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "2,2,2,2.0,2.0,2.0"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testNumericAutoCastInComparison(): Unit = {
// don't test everything, just some common cast directions
@@ -74,12 +74,12 @@ class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
(2: Byte, 2: Short, 2, 2L, 2.0f, 2.0d)).as('a, 'b, 'c, 'd, 'e, 'f)
.filter('a > 1 && 'b > 1 && 'c > 1L && 'd > 1.0f && 'e > 1.0d && 'f > 1)
-// val expected = "2,2,2,2,2.0,2.0"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "2,2,2,2,2.0,2.0"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testCastFromString: Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -99,14 +99,14 @@ class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
'_6.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO),
'_7.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO))
-// val expected = "1,1,1,1,2.0,2.0,true," +
-// "2011-05-03 00:00:00.000,1970-01-01 15:51:36.000,2011-05-03 15:51:36.000," +
-// "1970-01-17 17:47:53.775\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1,1,1,1,2.0,2.0,true," +
+ "2011-05-03 00:00:00.000,1970-01-01 15:51:36.000,2011-05-03 15:51:36.000," +
+ "1970-01-17 17:47:53.775\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testCastDateToStringAndLong {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds = env.fromElements(("2011-05-03 15:51:36.000", "1304437896000"))
@@ -118,9 +118,9 @@ class CastingITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mo
'f1.cast(BasicTypeInfo.STRING_TYPE_INFO),
'f1.cast(BasicTypeInfo.LONG_TYPE_INFO))
-// val expected = "2011-05-03 15:51:36.000,1304437896000," +
-// "2011-05-03 15:51:36.000,1304437896000\n"
-// val result = t.toDataSet[Row].collect
-// TestBaseUtils.compareResultAsText(result.asJava, expected)
+ val expected = "2011-05-03 15:51:36.000,1304437896000," +
+ "2011-05-03 15:51:36.000,1304437896000\n"
+ val result = t.toDataSet[Row].collect
+ TestBaseUtils.compareResultAsText(result.asJava, expected)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
index a25f3e3..9f20043 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/ExpressionsITCase.scala
@@ -22,7 +22,7 @@ import java.util.Date
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.table.expressions.Literal
-import org.apache.flink.api.table.{Row, ExpressionException}
+import org.apache.flink.api.table.Row
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
@@ -36,43 +36,43 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class ExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testArithmetic(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements((5, 10)).as('a, 'b)
.select('a - 5, 'a + 5, 'a / 2, 'a * 2, 'a % 2, -'a)
-// val expected = "0,10,2,10,1,-5"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "0,10,2,10,1,-5"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testLogic(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements((5, true)).as('a, 'b)
.select('b && true, 'b && false, 'b || false, !'b)
-// val expected = "true,false,true,false"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "true,false,true,false"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testComparisons(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements((5, 5, 4)).as('a, 'b, 'c)
.select('a > 'c, 'a >= 'b, 'a < 'c, 'a.isNull, 'a.isNotNull)
-// val expected = "true,true,false,false,true"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "true,true,false,false,true"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testCaseInsensitiveForAs(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -80,9 +80,9 @@ class ExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBas
val t = env.fromElements((3, 5.toByte)).as('a, 'b)
.groupBy("a").select("a, a.count As cnt")
-// val expected = "3,1"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "3,1"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
// Date literals not yet supported
@@ -96,9 +96,9 @@ class ExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBas
Literal(new Date(0)).cast(BasicTypeInfo.STRING_TYPE_INFO),
'a.cast(BasicTypeInfo.DATE_TYPE_INFO).cast(BasicTypeInfo.STRING_TYPE_INFO))
-// val expected = "0,1970-01-01 00:00:00.000,1970-01-01 00:00:00.000"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "0,1970-01-01 00:00:00.000,1970-01-01 00:00:00.000"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
index bcd5cbf..14a0cf0 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/GroupedAggregationsITCase.scala
@@ -18,11 +18,10 @@
package org.apache.flink.api.scala.table.test
-import org.apache.flink.api.table.{ExpressionException, Row}
+import org.apache.flink.api.table.Row
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.api.scala.util.CollectionDataSets
-import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.test.util.{TestBaseUtils, MultipleProgramsTestBase}
import org.apache.flink.test.util.MultipleProgramsTestBase.TestExecutionMode
import org.junit._
@@ -42,12 +41,12 @@ class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgram
.groupBy('_foo)
.select('a.avg)
-// val expected = ""
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = ""
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testGroupedAggregate(): Unit = {
// the grouping key needs to be forwarded to the intermediate DataSet, even
@@ -58,12 +57,12 @@ class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgram
.groupBy('b)
.select('b, 'a.sum)
-// val expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testGroupingKeyForwardIfNotUsed(): Unit = {
// the grouping key needs to be forwarded to the intermediate DataSet, even
@@ -74,12 +73,12 @@ class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgram
.groupBy('b)
.select('a.sum)
-// val expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testGroupNoAggregation(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -90,8 +89,8 @@ class GroupedAggregationsITCase(mode: TestExecutionMode) extends MultipleProgram
.groupBy('b, 'd)
.select('b)
-// val expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
index a98b7c8..628613e 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/JoinITCase.scala
@@ -18,7 +18,7 @@
package org.apache.flink.api.scala.table.test
-import org.apache.flink.api.table.{Row, ExpressionException}
+import org.apache.flink.api.table.Row
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.table._
import org.apache.flink.api.scala.util.CollectionDataSets
@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testJoin(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
@@ -41,12 +41,12 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val joinT = ds1.join(ds2).where('b === 'e).select('c, 'g)
-// val expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
-// val results = joinT.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n"
+ val results = joinT.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testJoinWithFilter(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
@@ -54,12 +54,12 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val joinT = ds1.join(ds2).where('b === 'e && 'b < 2).select('c, 'g)
-// val expected = "Hi,Hallo\n"
-// val results = joinT.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "Hi,Hallo\n"
+ val results = joinT.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testJoinWithMultipleKeys(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c)
@@ -67,10 +67,10 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val joinT = ds1.join(ds2).filter('a === 'd && 'b === 'h).select('c, 'g)
-// val expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
-// "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n"
-// val results = joinT.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
+ "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n"
+ val results = joinT.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -81,9 +81,9 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val joinT = ds1.join(ds2).where('foo === 'e).select('c, 'g)
-// val expected = ""
-// val results = joinT.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = ""
+ val results = joinT.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
// Calcite does not eagerly check the compatibility of compared types
@@ -96,9 +96,9 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val joinT = ds1.join(ds2).where('a === 'g).select('c, 'g)
-// val expected = ""
-// val results = joinT.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = ""
+ val results = joinT.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -109,12 +109,12 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val joinT = ds1.join(ds2).where('a === 'd).select('c, 'g)
-// val expected = ""
-// val results = joinT.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = ""
+ val results = joinT.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testJoinWithAggregation(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
@@ -122,9 +122,9 @@ class JoinITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode)
val joinT = ds1.join(ds2).where('a === 'd).select('g.count)
-// val expected = "6"
-// val results = joinT.toDataSet[Row]collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "6"
+ val results = joinT.toDataSet[Row]collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
index 5791d2e..4dadfe4 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/SelectITCase.scala
@@ -33,39 +33,39 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testSimpleSelectAll(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).toTable.select('_1, '_2, '_3)
-// val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testSimpleSelectAllWithAs(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c).select('a, 'b, 'c)
-// val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testSimpleSelectWithNaming(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
@@ -73,11 +73,11 @@ class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
.select('_1 as 'a, '_2 as 'b)
.select('a, 'b)
-// val expected = "1,1\n" + "2,2\n" + "3,2\n" + "4,3\n" + "5,3\n" + "6,3\n" + "7,4\n" +
-// "8,4\n" + "9,4\n" + "10,4\n" + "11,5\n" + "12,5\n" + "13,5\n" + "14,5\n" + "15,5\n" +
-// "16,6\n" + "17,6\n" + "18,6\n" + "19,6\n" + "20,6\n" + "21,6\n"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "1,1\n" + "2,2\n" + "3,2\n" + "4,3\n" + "5,3\n" + "6,3\n" + "7,4\n" +
+ "8,4\n" + "9,4\n" + "10,4\n" + "11,5\n" + "12,5\n" + "13,5\n" + "14,5\n" + "15,5\n" +
+ "16,6\n" + "17,6\n" + "18,6\n" + "19,6\n" + "20,6\n" + "21,6\n"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -86,9 +86,9 @@ class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -97,9 +97,9 @@ class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'c, 'd)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -108,9 +108,9 @@ class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b, 'b)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@@ -120,8 +120,8 @@ class SelectITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mod
val env = ExecutionEnvironment.getExecutionEnvironment
val t = CollectionDataSets.get3TupleDataSet(env).as('a, 'b as 'c, 'd)
-// val expected = "no"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "no"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
index e3a3170..565f444 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/StringExpressionsITCase.scala
@@ -32,26 +32,26 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class StringExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testSubstring(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements(("AAAA", 2), ("BBBB", 1)).as('a, 'b)
.select('a.substring(0, 'b))
-// val expected = "AA\nB"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "AA\nB"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testSubstringWithMaxEnd(): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment
val t = env.fromElements(("ABCD", 2), ("ABCD", 1)).as('a, 'b)
.select('a.substring('b))
-// val expected = "CD\nBCD"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "CD\nBCD"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
// Calcite does eagerly check expression types
@@ -63,9 +63,9 @@ class StringExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsT
val t = env.fromElements(("AAAA", 2.0), ("BBBB", 1.0)).as('a, 'b)
.select('a.substring(0, 'b))
-// val expected = "AAA\nBB"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "AAA\nBB"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
// Calcite does eagerly check expression types
@@ -77,9 +77,9 @@ class StringExpressionsITCase(mode: TestExecutionMode) extends MultipleProgramsT
val t = env.fromElements(("AAAA", "c"), ("BBBB", "d")).as('a, 'b)
.select('a.substring('b, 15))
-// val expected = "AAA\nBB"
-// val results = t.toDataSet[Row].collect()
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val expected = "AAA\nBB"
+ val results = t.toDataSet[Row].collect()
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
index d3c6127..3d03f23 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/api/scala/table/test/UnionITCase.scala
@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
@RunWith(classOf[Parameterized])
class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode) {
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testUnion(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
@@ -41,12 +41,12 @@ class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode
val unionDs = ds1.unionAll(ds2).select('c)
-// val results = unionDs.toDataSet[Row].collect()
-// val expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hi\n" + "Hello\n" + "Hello world\n"
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val results = unionDs.toDataSet[Row].collect()
+ val expected = "Hi\n" + "Hello\n" + "Hello world\n" + "Hi\n" + "Hello\n" + "Hello world\n"
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testUnionWithFilter(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
@@ -54,9 +54,9 @@ class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode
val joinDs = ds1.unionAll(ds2.select('a, 'b, 'c)).filter('b < 2).select('c)
-// val results = joinDs.toDataSet[Row].collect()
-// val expected = "Hi\n" + "Hallo\n"
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val results = joinDs.toDataSet[Row].collect()
+ val expected = "Hi\n" + "Hallo\n"
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -67,9 +67,9 @@ class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode
val unionDs = ds1.unionAll(ds2)
-// val results = unionDs.toDataSet[Row].collect()
-// val expected = ""
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val results = unionDs.toDataSet[Row].collect()
+ val expected = ""
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
@Test(expected = classOf[IllegalArgumentException])
@@ -80,12 +80,12 @@ class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode
val unionDs = ds1.unionAll(ds2)
-// val results = unionDs.toDataSet[Row].collect()
-// val expected = ""
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val results = unionDs.toDataSet[Row].collect()
+ val expected = ""
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
- @Test
+ @Test(expected = classOf[NotImplementedError])
def testUnionWithAggregation(): Unit = {
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
val ds1 = CollectionDataSets.getSmall3TupleDataSet(env).as('a, 'b, 'c)
@@ -93,8 +93,8 @@ class UnionITCase(mode: TestExecutionMode) extends MultipleProgramsTestBase(mode
val unionDs = ds1.unionAll(ds2.select('a, 'b, 'c)).select('c.count)
-// val results = unionDs.toDataSet[Row].collect()
-// val expected = "18"
-// TestBaseUtils.compareResultAsText(results.asJava, expected)
+ val results = unionDs.toDataSet[Row].collect()
+ val expected = "18"
+ TestBaseUtils.compareResultAsText(results.asJava, expected)
}
}
[3/3] flink git commit: [FLINK-3225] Implemented optimization of
Table API queries via Calcite
Posted by fh...@apache.org.
[FLINK-3225] Implemented optimization of Table API queries via Calcite
- added logical Flink nodes and translation rules
- added stubs for DataSet translation rules
- ported DataSetNodes to Scala
- reactivated tests and added expected NotImplementedError
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7ecb7010
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7ecb7010
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7ecb7010
Branch: refs/heads/tableOnCalcite
Commit: 7ecb70105254052514494d2aae482f335f7c6835
Parents: 364928e
Author: Fabian Hueske <fh...@apache.org>
Authored: Tue Jan 26 13:22:38 2016 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Fri Jan 29 22:29:12 2016 +0100
----------------------------------------------------------------------
.../apache/flink/api/table/package-info.java | 33 -----
.../api/table/sql/calcite/DataSetRelNode.java | 29 -----
.../table/sql/calcite/node/DataSetExchange.java | 60 ----------
.../table/sql/calcite/node/DataSetFlatMap.java | 56 ---------
.../api/table/sql/calcite/node/DataSetJoin.java | 80 -------------
.../api/table/sql/calcite/node/DataSetMap.java | 58 ---------
.../table/sql/calcite/node/DataSetReduce.java | 58 ---------
.../sql/calcite/node/DataSetReduceGroup.java | 62 ----------
.../api/table/sql/calcite/node/DataSetSort.java | 59 ---------
.../table/sql/calcite/node/DataSetSource.java | 55 ---------
.../table/sql/calcite/node/DataSetUnion.java | 51 --------
.../api/java/table/JavaBatchTranslator.scala | 68 +++++++----
.../api/table/plan/TranslationContext.scala | 79 ++++++++++++
.../plan/nodes/dataset/DataSetConvention.scala | 42 +++++++
.../plan/nodes/dataset/DataSetExchange.scala | 63 ++++++++++
.../plan/nodes/dataset/DataSetFlatMap.scala | 62 ++++++++++
.../plan/nodes/dataset/DataSetGroupReduce.scala | 63 ++++++++++
.../table/plan/nodes/dataset/DataSetJoin.scala | 73 +++++++++++
.../table/plan/nodes/dataset/DataSetMap.scala | 63 ++++++++++
.../plan/nodes/dataset/DataSetReduce.scala | 63 ++++++++++
.../table/plan/nodes/dataset/DataSetRel.scala | 33 +++++
.../table/plan/nodes/dataset/DataSetSort.scala | 62 ++++++++++
.../plan/nodes/dataset/DataSetSource.scala | 55 +++++++++
.../table/plan/nodes/dataset/DataSetUnion.scala | 62 ++++++++++
.../plan/nodes/logical/FlinkAggregate.scala | 76 ++++++++++++
.../table/plan/nodes/logical/FlinkCalc.scala | 37 ++++++
.../plan/nodes/logical/FlinkConvention.scala | 42 +++++++
.../table/plan/nodes/logical/FlinkFilter.scala | 42 +++++++
.../table/plan/nodes/logical/FlinkJoin.scala | 46 +++++++
.../table/plan/nodes/logical/FlinkProject.scala | 45 +++++++
.../api/table/plan/nodes/logical/FlinkRel.scala | 25 ++++
.../table/plan/nodes/logical/FlinkScan.scala | 31 +++++
.../table/plan/nodes/logical/FlinkUnion.scala | 38 ++++++
.../api/table/plan/operators/DataSetTable.scala | 66 ----------
.../api/table/plan/rules/FlinkRuleSets.scala | 120 +++++++++++++++++++
.../rules/dataset/DataSetAggregateRule.scala | 53 ++++++++
.../plan/rules/dataset/DataSetCalcRule.scala | 52 ++++++++
.../plan/rules/dataset/DataSetFilterRule.scala | 52 ++++++++
.../plan/rules/dataset/DataSetJoinRule.scala | 59 +++++++++
.../plan/rules/dataset/DataSetProjectRule.scala | 52 ++++++++
.../plan/rules/dataset/DataSetScanRule.scala | 53 ++++++++
.../plan/rules/dataset/DataSetUnionRule.scala | 53 ++++++++
.../plan/rules/logical/FlinkAggregateRule.scala | 53 ++++++++
.../plan/rules/logical/FlinkCalcRule.scala | 50 ++++++++
.../plan/rules/logical/FlinkFilterRule.scala | 50 ++++++++
.../plan/rules/logical/FlinkJoinRule.scala | 54 +++++++++
.../plan/rules/logical/FlinkProjectRule.scala | 51 ++++++++
.../plan/rules/logical/FlinkScanRule.scala | 53 ++++++++
.../plan/rules/logical/FlinkUnionRule.scala | 54 +++++++++
.../api/table/plan/schema/DataSetTable.scala | 89 ++++++++++++++
.../org/apache/flink/api/table/table.scala | 1 -
.../api/java/table/test/AggregationsITCase.java | 72 +++++------
.../flink/api/java/table/test/AsITCase.java | 62 +++++-----
.../api/java/table/test/CastingITCase.java | 53 ++++----
.../api/java/table/test/ExpressionsITCase.java | 32 ++---
.../flink/api/java/table/test/FilterITCase.java | 69 +++++------
.../table/test/GroupedAggregationsITCase.java | 40 +++----
.../flink/api/java/table/test/JoinITCase.java | 70 ++++++-----
.../api/java/table/test/PojoGroupingITCase.java | 19 +--
.../flink/api/java/table/test/SelectITCase.java | 68 +++++------
.../table/test/StringExpressionsITCase.java | 40 ++++---
.../flink/api/java/table/test/UnionITCase.java | 47 ++++----
.../scala/table/test/AggregationsITCase.scala | 59 +++++----
.../flink/api/scala/table/test/AsITCase.scala | 50 ++++----
.../api/scala/table/test/CastingITCase.scala | 44 +++----
.../scala/table/test/ExpressionsITCase.scala | 40 +++----
.../table/test/GroupedAggregationsITCase.scala | 33 +++--
.../flink/api/scala/table/test/JoinITCase.scala | 54 ++++-----
.../api/scala/table/test/SelectITCase.scala | 72 +++++------
.../table/test/StringExpressionsITCase.scala | 28 ++---
.../api/scala/table/test/UnionITCase.scala | 36 +++---
71 files changed, 2596 insertions(+), 1178 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java
deleted file mode 100644
index d7fbc8e..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * <strong>Table API</strong><br>
- *
- * This package contains the generic part of the Table API. It can be used with Flink Streaming
- * and Flink Batch. From Scala as well as from Java.
- *
- * When using the Table API, as user creates a [[org.apache.flink.api.table.Table]] from
- * a DataSet or DataStream. On this relational operations can be performed. A table can also
- * be converted back to a DataSet or DataStream.
- *
- * Packages [[org.apache.flink.api.scala.table]] and [[org.apache.flink.api.java.table]] contain
- * the language specific part of the API. Refer to these packages for documentation on how
- * the Table API can be used in Java and Scala.
- */
-package org.apache.flink.api.table;
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java
deleted file mode 100644
index df0ebc0..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/DataSetRelNode.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite;
-
-import org.apache.calcite.rel.RelNode;
-import org.apache.flink.api.java.DataSet;
-
-public interface DataSetRelNode<T> extends RelNode {
-
- /**
- * Translate the FlinkRelNode into Flink operator.
- */
- DataSet<T> translateToPlan();
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java
deleted file mode 100644
index 1ddd884..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetExchange.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.operators.base.PartitionOperatorBase.PartitionMethod;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with PartitionOperator.
- *
- * @param <T>
- */
-public class DataSetExchange<T> extends SingleRel implements DataSetRelNode<T> {
-
- public DataSetExchange(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
- super(cluster, traits, input);
- }
-
- private TypeInformation<T> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private PartitionMethod getPartitionMethod() {
- return null;
- }
-
- private int[] getPartitionKey() {
- return null;
- }
-
- @Override
- public DataSet<T> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java
deleted file mode 100644
index cf597b2..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetFlatMap.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichFlatMapFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with FlatMapOperator.
- *
- * @param <T>
- */
-public class DataSetFlatMap<T> extends SingleRel implements DataSetRelNode<T> {
-
- protected DataSetFlatMap(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
- super(cluster, traits, input);
- }
-
- private TypeInformation<T> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private RichFlatMapFunction<T, T> getFlatMapFunction() {
- return null;
- }
-
- @Override
- public DataSet<T> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java
deleted file mode 100644
index 3d1146e..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetJoin.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.BiRel;
-import org.apache.calcite.rel.RelNode;
-import org.apache.flink.api.common.functions.JoinFunction;
-import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.operators.join.JoinType;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with JoinOperator and its related operations.
- */
-public class DataSetJoin<L, R, OUT> extends BiRel implements DataSetRelNode<OUT> {
-
- public DataSetJoin(RelOptCluster cluster, RelTraitSet traitSet, RelNode left, RelNode right) {
- super(cluster, traitSet, left, right);
- }
-
- private TypeInformation<L> getLeftInputType() {
- return null;
- }
-
- private TypeInformation<R> getRightInputType() {
- return null;
- }
-
- private TypeInformation<OUT> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private JoinType getJoinType() {
- return null;
- }
-
- private JoinHint getJoinHint() {
- return null;
- }
-
- private int[] getLeftJoinKey() {
- return null;
- }
-
- private int[] getRightJoinKey() {
- return null;
- }
-
- private JoinFunction<L, R, OUT> getJoinFunction() {
- return null;
- }
-
- @Override
- public DataSet<OUT> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java
deleted file mode 100644
index 3098cb3..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetMap.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with MapOperator.
- */
-public class DataSetMap<IN, OUT> extends SingleRel implements DataSetRelNode<OUT> {
-
- protected DataSetMap(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
- super(cluster, traits, input);
- }
-
- private TypeInformation<IN> getInputType() {
- return null;
- }
-
- private TypeInformation<OUT> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private RichMapFunction<IN, OUT> getMapFunction() {
- return null;
- }
-
- @Override
- public DataSet<OUT> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java
deleted file mode 100644
index 4aa2846..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduce.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichReduceFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with ReduceOperator.
- */
-public class DataSetReduce<T> extends SingleRel implements DataSetRelNode<T> {
-
- public DataSetReduce(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
- super(cluster, traits, input);
- }
-
- private TypeInformation<T> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private RichReduceFunction<T> getReduceFunction() {
- return null;
- }
-
- private int[] getGroupingKeys() {
- return null;
- }
-
- @Override
- public DataSet<T> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java
deleted file mode 100644
index 5467c56..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetReduceGroup.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.functions.RichGroupReduceFunction;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with ReduceGroupOperator.
- */
-public class DataSetReduceGroup<IN, OUT> extends SingleRel implements DataSetRelNode<OUT> {
-
- public DataSetReduceGroup(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
- super(cluster, traits, input);
- }
-
- private TypeInformation<IN> getInputType() {
- return null;
- }
-
- private TypeInformation<OUT> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private RichGroupReduceFunction<IN, OUT> getGroupReduceFunction() {
- return null;
- }
-
- private int[] getGroupingKeys() {
- return null;
- }
-
- @Override
- public DataSet<OUT> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java
deleted file mode 100644
index 7fa1c53..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSort.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with SortPartitionOperator.
- *
- * @param <T>
- */
-public class DataSetSort<T> extends SingleRel implements DataSetRelNode<T> {
-
- public DataSetSort(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
- super(cluster, traits, input);
- }
-
- private TypeInformation<T> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private int[] getSortKey() {
- return null;
- }
-
- private boolean[] getSortKeyOrder() {
- return null;
- }
-
- @Override
- public DataSet<T> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java
deleted file mode 100644
index f24ee79..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetSource.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptTable;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.core.TableScan;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with DataSource.
- *
- * @param <T>
- */
-public class DataSetSource<T> extends TableScan implements DataSetRelNode<T> {
-
- public DataSetSource(RelOptCluster cluster, RelTraitSet traitSet, RelOptTable table) {
- super(cluster, traitSet, table);
- }
-
- private TypeInformation<T> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- private DataSet<T> getDatSource() {
- return null;
- }
-
- @Override
- public DataSet<T> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java
deleted file mode 100644
index 8b435e9..0000000
--- a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/sql/calcite/node/DataSetUnion.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.sql.calcite.node;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.BiRel;
-import org.apache.calcite.rel.RelNode;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.table.sql.calcite.DataSetRelNode;
-
-/**
- * Flink RelNode which matches along with UnionOperator.
- *
- * @param <T>
- */
-public class DataSetUnion<T> extends BiRel implements DataSetRelNode<T> {
-
- public DataSetUnion(RelOptCluster cluster, RelTraitSet traitSet, RelNode left, RelNode right) {
- super(cluster, traitSet, left, right);
- }
-
- private TypeInformation<T> getType() {
- return null;
- }
-
- private String getName() {
- return null;
- }
-
- @Override
- public DataSet<T> translateToPlan() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
index f3f4e9d..66bfbe7 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
@@ -18,14 +18,17 @@
package org.apache.flink.api.java.table
-import org.apache.calcite.plan.RelOptUtil
+import org.apache.calcite.plan.{RelTraitSet, RelOptUtil}
import org.apache.calcite.rel.RelNode
-import org.apache.calcite.tools.{RelBuilder, Frameworks}
+import org.apache.calcite.sql2rel.RelDecorrelator
+import org.apache.calcite.tools.Programs
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.{DataSet => JavaDataSet}
import org.apache.flink.api.table.plan._
-import org.apache.flink.api.table.plan.operators.DataSetTable
import org.apache.flink.api.table.Table
+import org.apache.flink.api.table.plan.nodes.dataset.DataSetRel
+import org.apache.flink.api.table.plan.rules.FlinkRuleSets
+import org.apache.flink.api.table.plan.schema.DataSetTable
/**
* [[PlanTranslator]] for creating [[Table]]s from Java [[org.apache.flink.api.java.DataSet]]s and
@@ -41,37 +44,58 @@ class JavaBatchTranslator extends PlanTranslator {
// create table representation from DataSet
val dataSetTable = new DataSetTable[A](
- repr.asInstanceOf[JavaDataSet[A]],
- fieldNames
+ repr.asInstanceOf[JavaDataSet[A]],
+ fieldNames
)
- // register table in Cascading schema
- val schema = Frameworks.createRootSchema(true)
- val tableName = repr.hashCode().toString
- schema.add(tableName, dataSetTable)
-
- // initialize RelBuilder
- val frameworkConfig = Frameworks
- .newConfigBuilder
- .defaultSchema(schema)
- .build
- val relBuilder = RelBuilder.create(frameworkConfig)
+ val tabName = TranslationContext.addDataSet(dataSetTable)
+ val relBuilder = TranslationContext.getRelBuilder
// create table scan operator
- relBuilder.scan(tableName)
+ relBuilder.scan(tabName)
new Table(relBuilder.build(), relBuilder)
}
override def translate[A](lPlan: RelNode)(implicit tpe: TypeInformation[A]): JavaDataSet[A] = {
+ // get the planner for the plan
+ val planner = lPlan.getCluster.getPlanner
+
+ // we do not have any special requirements for the output
+ val outputProps = RelTraitSet.createEmpty()
+
+ println("-----------")
+ println("Input Plan:")
+ println("-----------")
println(RelOptUtil.toString(lPlan))
- // TODO: optimize & translate:
- // - optimize RelNode plan
- // - translate to Flink RelNode plan
- // - generate DataSet program
+ // decorrelate
+ val decorPlan = RelDecorrelator.decorrelateQuery(lPlan)
+
+ // optimize the logical Flink plan
+ val optProgram = Programs.ofRules(FlinkRuleSets.DATASET_OPT_RULES)
+ val optPlan = optProgram.run(planner, decorPlan, outputProps)
+
+ println("---------------")
+ println("Optimized Plan:")
+ println("---------------")
+ println(RelOptUtil.toString(optPlan))
+
+ // optimize the logical Flink plan
+ val dataSetProgram = Programs.ofRules(FlinkRuleSets.DATASET_TRANS_RULES)
+ val dataSetPlan = dataSetProgram.run(planner, optPlan, outputProps)
+
+ println("-------------")
+ println("DataSet Plan:")
+ println("-------------")
+ println(RelOptUtil.toString(dataSetPlan))
+
+ dataSetPlan match {
+ case node: DataSetRel =>
+ node.translateToPlan.asInstanceOf[JavaDataSet[A]]
+ case _ => ???
+ }
- null
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala
new file mode 100644
index 0000000..b2b0c2b
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TranslationContext.scala
@@ -0,0 +1,79 @@
+/*
+ * 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.api.table.plan
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.apache.calcite.plan.ConventionTraitDef
+import org.apache.calcite.schema.impl.AbstractTable
+import org.apache.calcite.schema.SchemaPlus
+import org.apache.calcite.tools.{Frameworks, RelBuilder}
+import org.apache.flink.api.table.plan.schema.DataSetTable
+
+object TranslationContext {
+
+ private var relBuilder: RelBuilder = null
+ private var tables: SchemaPlus = null
+ private var tabNames: Map[AbstractTable, String] = null
+ private val nameCntr: AtomicInteger = new AtomicInteger(0)
+
+ reset()
+
+ def reset(): Unit = {
+
+ // register table in Cascading schema
+ tables = Frameworks.createRootSchema(true)
+
+ // initialize RelBuilder
+ val frameworkConfig = Frameworks
+ .newConfigBuilder
+ .defaultSchema(tables)
+ .traitDefs(ConventionTraitDef.INSTANCE)
+ .build
+
+ tabNames = Map[AbstractTable, String]()
+
+ relBuilder = RelBuilder.create(frameworkConfig)
+
+ }
+
+ def addDataSet(newTable: DataSetTable[_]): String = {
+
+ // look up name
+ val tabName = tabNames.get(newTable)
+
+ tabName match {
+ case Some(name) =>
+ name
+ case None =>
+ val tabName = "DataSetTable_" + nameCntr.getAndIncrement()
+ tabNames += (newTable -> tabName)
+ tables.add(tabName, newTable)
+ tabName
+ }
+
+ }
+
+ def getRelBuilder: RelBuilder = {
+ relBuilder
+ }
+
+}
+
+
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala
new file mode 100644
index 0000000..cbacd16
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetConvention.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan._
+
+class DataSetConvention extends Convention {
+
+ override def toString: String = getName
+
+ def getInterface: Class[_] = classOf[DataSetRel]
+
+ def getName: String = "DATASET"
+
+ def getTraitDef: RelTraitDef[_ <: RelTrait] = ConventionTraitDef.INSTANCE
+
+ def satisfies(`trait`: RelTrait): Boolean = this eq `trait`
+
+ def register(planner: RelOptPlanner): Unit = { }
+
+}
+
+object DataSetConvention {
+
+ val INSTANCE = new DataSetConvention
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
new file mode 100644
index 0000000..ec5805a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.operators.base.PartitionOperatorBase.PartitionMethod
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with PartitionOperator.
+ */
+class DataSetExchange(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ rowType: RelDataType,
+ opName: String,
+ partitionKey: Array[Int],
+ partitionMethod: PartitionMethod)
+ extends SingleRel(cluster, traitSet, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetExchange(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ opName,
+ partitionKey,
+ partitionMethod
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
new file mode 100644
index 0000000..913cca0
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
@@ -0,0 +1,62 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelOptCost, RelOptPlanner, RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.FlatMapFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with FlatMapOperator.
+ *
+ */
+class DataSetFlatMap(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ rowType: RelDataType,
+ opName: String,
+ func: FlatMapFunction[Row, Row])
+ extends SingleRel(cluster, traitSet, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetFlatMap(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ opName,
+ func
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
new file mode 100644
index 0000000..11bb160
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.GroupReduceFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with ReduceGroupOperator.
+ */
+class DataSetGroupReduce(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ rowType: RelDataType,
+ opName: String,
+ groupingKeys: Array[Int],
+ func: GroupReduceFunction[Row, Row])
+ extends SingleRel(cluster, traitSet, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetGroupReduce(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ opName,
+ groupingKeys,
+ func
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
new file mode 100644
index 0000000..c20cdc5
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
@@ -0,0 +1,73 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
+import org.apache.flink.api.common.functions.JoinFunction
+import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with JoinOperator and its related operations.
+ */
+class DataSetJoin(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ left: RelNode,
+ right: RelNode,
+ rowType: RelDataType,
+ opName: String,
+ joinKeysLeft: Array[Int],
+ joinKeysRight: Array[Int],
+ joinType: JoinType,
+ joinHint: JoinHint,
+ func: JoinFunction[Row, Row, Row])
+ extends BiRel(cluster, traitSet, left, right)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetJoin(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ inputs.get(1),
+ rowType,
+ opName,
+ joinKeysLeft,
+ joinKeysRight,
+ joinType,
+ joinHint,
+ func
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
new file mode 100644
index 0000000..be8bd9d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.MapFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with MapOperator.
+ */
+class DataSetMap(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ rowType: RelDataType,
+ opName: String,
+ func: MapFunction[Row, Row])
+ extends SingleRel(cluster, traitSet, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetMap(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ opName,
+ func
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def toString() = opName
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
new file mode 100644
index 0000000..567a91c
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
@@ -0,0 +1,63 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.common.functions.ReduceFunction
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with ReduceOperator.
+ */
+class DataSetReduce(
+ cluster: RelOptCluster,
+ traits: RelTraitSet,
+ input: RelNode,
+ rowType: RelDataType,
+ opName: String,
+ groupingKeys: Array[Int],
+ func: ReduceFunction[Row])
+ extends SingleRel(cluster, traits, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetReduce(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ opName,
+ groupingKeys,
+ func
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
new file mode 100644
index 0000000..20677b3
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.rel.RelNode
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+trait DataSetRel extends RelNode {
+
+ /**
+ * Translate the FlinkRelNode into Flink operator.
+ */
+ def translateToPlan: DataSet[Any]
+
+}
+
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
new file mode 100644
index 0000000..df5301d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
@@ -0,0 +1,62 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with SortPartitionOperator.
+ */
+class DataSetSort(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ rowType: RelDataType,
+ opName: String,
+ sortKey: Array[Int],
+ sortOrder: Array[Boolean])
+ extends SingleRel(cluster, traitSet, input)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetSort(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ rowType,
+ opName,
+ sortKey,
+ sortOrder
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
new file mode 100644
index 0000000..effaf1a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.{RelWriter, RelNode}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.TableScan
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+ * Flink RelNode which matches along with DataSource.
+ */
+class DataSetSource(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ table: RelOptTable,
+ rowType: RelDataType,
+ inputDataSet: DataSet[_])
+ extends TableScan(cluster, traitSet, table)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetSource(
+ cluster,
+ traitSet,
+ table,
+ rowType,
+ inputDataSet
+ )
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
new file mode 100644
index 0000000..a510fc9
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
@@ -0,0 +1,62 @@
+/*
+ * 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.api.table.plan.nodes.dataset
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.Row
+
+/**
+* Flink RelNode which matches along with UnionOperator.
+*
+*/
+class DataSetUnion(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ left: RelNode,
+ right: RelNode,
+ rowType: RelDataType,
+ opName: String)
+ extends BiRel(cluster, traitSet, left, right)
+ with DataSetRel {
+
+ override def deriveRowType() = rowType
+
+ override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
+ new DataSetUnion(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ inputs.get(1),
+ rowType,
+ opName
+ )
+ }
+
+ override def explainTerms(pw: RelWriter): RelWriter = {
+ super.explainTerms(pw).item("name", opName)
+ }
+
+ override def translateToPlan: DataSet[Any] = {
+ ???
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
new file mode 100644
index 0000000..f66cb71
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkAggregate.scala
@@ -0,0 +1,76 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import java.util
+
+import org.apache.calcite.plan.{RelOptCost, RelOptPlanner, RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.{AggregateCall, Aggregate}
+import org.apache.calcite.sql.fun.SqlAvgAggFunction
+import org.apache.calcite.util.ImmutableBitSet
+
+import scala.collection.JavaConversions._
+
+class FlinkAggregate(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ indicator: Boolean,
+ groupSet: ImmutableBitSet,
+ groupSets: java.util.List[ImmutableBitSet],
+ aggCalls: java.util.List[AggregateCall])
+ extends Aggregate(cluster, traitSet, input, indicator, groupSet, groupSets, aggCalls)
+ with FlinkRel {
+
+ override def copy(
+ traitSet: RelTraitSet,
+ input: RelNode,
+ indicator: Boolean,
+ groupSet: ImmutableBitSet,
+ groupSets: util.List[ImmutableBitSet],
+ aggCalls: util.List[AggregateCall]): Aggregate = {
+
+ new FlinkAggregate(
+ cluster,
+ traitSet,
+ input,
+ indicator,
+ groupSet,
+ groupSets,
+ aggCalls
+ )
+ }
+
+ override def computeSelfCost (planner: RelOptPlanner): RelOptCost = {
+
+ val origCosts = super.computeSelfCost(planner)
+ val deltaCost = planner.getCostFactory.makeHugeCost()
+
+ // only prefer aggregations with transformed Avg
+ aggCalls.toList.foldLeft[RelOptCost](origCosts){
+ (c: RelOptCost, a: AggregateCall) =>
+ if (a.getAggregation.isInstanceOf[SqlAvgAggFunction]) {
+ c.plus(deltaCost)
+ } else {
+ c
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
new file mode 100644
index 0000000..bcfe8d7
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkCalc.scala
@@ -0,0 +1,37 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.Calc
+import org.apache.calcite.rex.RexProgram
+
+class FlinkCalc(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ program: RexProgram)
+ extends Calc(cluster, traitSet, input, program)
+ with FlinkRel {
+
+ override def copy(traitSet: RelTraitSet, child: RelNode, program: RexProgram): Calc = {
+ new FlinkCalc(cluster, traitSet, child, program)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
new file mode 100644
index 0000000..80137f2
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkConvention.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import org.apache.calcite.plan._
+
+class FlinkConvention extends Convention {
+
+ override def toString: String = getName
+
+ def getInterface: Class[_] = classOf[FlinkRel]
+
+ def getName: String = "FLINK"
+
+ def getTraitDef: RelTraitDef[_ <: RelTrait] = ConventionTraitDef.INSTANCE
+
+ def satisfies(`trait`: RelTrait): Boolean = this eq `trait`
+
+ def register(planner: RelOptPlanner): Unit = { }
+
+}
+
+object FlinkConvention {
+
+ val INSTANCE = new FlinkConvention
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala
new file mode 100644
index 0000000..9f0bf30
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkFilter.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.Filter
+import org.apache.calcite.rex.RexNode
+
+class FlinkFilter(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ condition: RexNode)
+ extends Filter(cluster, traitSet, input, condition)
+ with FlinkRel {
+
+ override def copy(traitSet: RelTraitSet, input: RelNode, condition: RexNode): Filter = {
+ new FlinkFilter(
+ cluster,
+ traitSet,
+ input,
+ condition
+ )
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
new file mode 100644
index 0000000..8b04b50
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkJoin.scala
@@ -0,0 +1,46 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.{JoinRelType, Join}
+import org.apache.calcite.rex.RexNode
+
+class FlinkJoin(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ left: RelNode,
+ right: RelNode,
+ condition: RexNode,
+ joinType: JoinRelType,
+ variablesStopped: java.util.Set[String])
+ extends Join(cluster, traitSet, left, right, condition, joinType, variablesStopped)
+ with FlinkRel {
+
+ override def copy(
+ traitSet: RelTraitSet,
+ condition: RexNode,
+ left: RelNode,
+ right: RelNode,
+ joinType: JoinRelType,
+ semiJoinDone: Boolean): Join = {
+ new FlinkJoin(cluster, traitSet, left, right, condition, joinType, getVariablesStopped)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala
new file mode 100644
index 0000000..1d93036
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkProject.scala
@@ -0,0 +1,45 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import java.util
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.Project
+import org.apache.calcite.rex.RexNode
+
+class FlinkProject(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ input: RelNode,
+ projects: java.util.List[RexNode],
+ rowType: RelDataType)
+ extends Project(cluster, traitSet, input, projects, rowType)
+ with FlinkRel {
+
+ override def copy(
+ traitSet: RelTraitSet,
+ input: RelNode,
+ projects: util.List[RexNode],
+ rowType: RelDataType): Project = {
+ new FlinkProject(cluster, traitSet, input, projects, rowType)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
new file mode 100644
index 0000000..9ebd7e4
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkRel.scala
@@ -0,0 +1,25 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import org.apache.calcite.rel.RelNode
+
+trait FlinkRel extends RelNode {
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
new file mode 100644
index 0000000..6d53a75
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkScan.scala
@@ -0,0 +1,31 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import org.apache.calcite.plan.{RelOptTable, RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.core.TableScan
+
+class FlinkScan(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ table: RelOptTable)
+ extends TableScan(cluster, traitSet, table)
+ with FlinkRel {
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
new file mode 100644
index 0000000..fd791d3
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/logical/FlinkUnion.scala
@@ -0,0 +1,38 @@
+/*
+ * 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.api.table.plan.nodes.logical
+
+import java.util
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.{SetOp, Union}
+
+class FlinkUnion(
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ inputs: java.util.List[RelNode],
+ all: Boolean)
+ extends Union(cluster, traitSet, inputs, all)
+ with FlinkRel {
+
+ override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode], all: Boolean): SetOp = {
+ new FlinkUnion(cluster, traitSet, inputs, all)
+ }
+}
[2/3] flink git commit: [FLINK-3225] Implemented optimization of
Table API queries via Calcite
Posted by fh...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala
deleted file mode 100644
index 65b97fb..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/operators/DataSetTable.scala
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.api.table.plan.operators
-
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
-import org.apache.calcite.schema.impl.AbstractTable
-import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.plan.TypeConverter
-
-class DataSetTable[T](
- val dataSet: DataSet[T],
- val fieldNames: Array[String]) extends AbstractTable {
-
- // check uniquenss of field names
- if (fieldNames.length != fieldNames.toSet.size) {
- throw new scala.IllegalArgumentException(
- "Table field names must be unique.")
- }
-
- val dataSetType: CompositeType[T] =
- dataSet.getType match {
- case cType: CompositeType[T] =>
- cType
- case _ =>
- throw new scala.IllegalArgumentException(
- "DataSet must have a composite type.")
- }
-
- val fieldTypes: Array[SqlTypeName] =
- if (fieldNames.length == dataSetType.getArity) {
- (0 until dataSetType.getArity)
- .map(i => dataSetType.getTypeAt(i))
- .map(TypeConverter.typeInfoToSqlType)
- .toArray
- }
- else {
- throw new IllegalArgumentException(
- "Arity of DataSet type not equal to number of field names.")
- }
-
- override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
- val builder = typeFactory.builder
- fieldNames.zip(fieldTypes)
- .foreach( f => builder.add(f._1, f._2).nullable(true) )
- builder.build
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
new file mode 100644
index 0000000..97e8b32
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/FlinkRuleSets.scala
@@ -0,0 +1,120 @@
+/*
+ * 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.api.table.plan.rules
+
+import org.apache.calcite.rel.rules._
+import org.apache.calcite.tools.{RuleSets, RuleSet}
+import org.apache.flink.api.table.plan.rules.logical._
+import org.apache.flink.api.table.plan.rules.dataset._
+
+object FlinkRuleSets {
+
+ /**
+ * RuleSet to optimize plans for batch / DataSet exeuction
+ */
+ val DATASET_OPT_RULES: RuleSet = RuleSets.ofList(
+
+ // filter rules
+ FilterJoinRule.FILTER_ON_JOIN,
+ FilterJoinRule.JOIN,
+ FilterMergeRule.INSTANCE,
+ FilterAggregateTransposeRule.INSTANCE,
+
+ // push and merge projection rules
+ AggregateProjectMergeRule.INSTANCE,
+ ProjectMergeRule.INSTANCE,
+ ProjectFilterTransposeRule.INSTANCE,
+ FilterProjectTransposeRule.INSTANCE,
+ AggregateProjectPullUpConstantsRule.INSTANCE,
+ JoinPushExpressionsRule.INSTANCE,
+ ProjectJoinTransposeRule.INSTANCE,
+ ProjectRemoveRule.INSTANCE,
+ SortProjectTransposeRule.INSTANCE,
+ ProjectSortTransposeRule.INSTANCE,
+
+ // merge and push unions rules
+ // TODO: Add a rule to enforce binary unions
+ UnionEliminatorRule.INSTANCE,
+ JoinUnionTransposeRule.LEFT_UNION,
+ JoinUnionTransposeRule.RIGHT_UNION,
+ // non-all Union to all-union + distinct
+ UnionToDistinctRule.INSTANCE,
+
+ // aggregation rules
+ AggregateRemoveRule.INSTANCE,
+ AggregateJoinTransposeRule.EXTENDED,
+ AggregateUnionAggregateRule.INSTANCE,
+ AggregateReduceFunctionsRule.INSTANCE,
+ AggregateExpandDistinctAggregatesRule.INSTANCE,
+
+ // remove unnecessary sort rule
+ SortRemoveRule.INSTANCE,
+
+ // join reordering rules
+ JoinPushThroughJoinRule.LEFT,
+ JoinPushThroughJoinRule.RIGHT,
+ JoinAssociateRule.INSTANCE,
+ JoinCommuteRule.INSTANCE,
+ JoinCommuteRule.SWAP_OUTER,
+
+ // simplify expressions rules
+ ReduceExpressionsRule.CALC_INSTANCE,
+ ReduceExpressionsRule.FILTER_INSTANCE,
+ ReduceExpressionsRule.JOIN_INSTANCE,
+ ReduceExpressionsRule.PROJECT_INSTANCE,
+
+ // prune empty results rules
+ PruneEmptyRules.AGGREGATE_INSTANCE,
+ PruneEmptyRules.FILTER_INSTANCE,
+ PruneEmptyRules.JOIN_LEFT_INSTANCE,
+ PruneEmptyRules.JOIN_RIGHT_INSTANCE,
+ PruneEmptyRules.PROJECT_INSTANCE,
+ PruneEmptyRules.SORT_INSTANCE,
+ PruneEmptyRules.UNION_INSTANCE,
+
+ // calc rules
+ FilterCalcMergeRule.INSTANCE,
+ ProjectCalcMergeRule.INSTANCE,
+ FilterToCalcRule.INSTANCE,
+ ProjectToCalcRule.INSTANCE,
+ CalcMergeRule.INSTANCE,
+
+ // translate to logical Flink nodes
+ FlinkAggregateRule.INSTANCE,
+ FlinkCalcRule.INSTANCE,
+ FlinkFilterRule.INSTANCE,
+ FlinkJoinRule.INSTANCE,
+ FlinkProjectRule.INSTANCE,
+ FlinkScanRule.INSTANCE,
+ FlinkUnionRule.INSTANCE
+ )
+
+ val DATASET_TRANS_RULES: RuleSet = RuleSets.ofList(
+
+ // translate to DataSet nodes
+ DataSetAggregateRule.INSTANCE,
+ DataSetCalcRule.INSTANCE,
+ DataSetFilterRule.INSTANCE,
+ DataSetJoinRule.INSTANCE,
+ DataSetProjectRule.INSTANCE,
+ DataSetScanRule.INSTANCE,
+ DataSetUnionRule.INSTANCE
+ )
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.scala
new file mode 100644
index 0000000..1d17d63
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetAggregateRule.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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetReduce}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkAggregate, FlinkConvention}
+
+class DataSetAggregateRule
+ extends ConverterRule(
+ classOf[FlinkAggregate],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetAggregateRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val agg: FlinkAggregate = rel.asInstanceOf[FlinkAggregate]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(agg.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetReduce(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ agg.toString,
+ Array[Int](),
+ null)
+ }
+}
+
+object DataSetAggregateRule {
+ val INSTANCE: RelOptRule = new DataSetAggregateRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
new file mode 100644
index 0000000..85e090c
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetCalcRule.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetFlatMap}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkCalc, FlinkConvention}
+
+class DataSetCalcRule
+ extends ConverterRule(
+ classOf[FlinkCalc],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetCalcRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val calc: FlinkCalc = rel.asInstanceOf[FlinkCalc]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(calc.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetFlatMap(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ calc.toString,
+ null)
+ }
+}
+
+object DataSetCalcRule {
+ val INSTANCE: RelOptRule = new DataSetCalcRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
new file mode 100644
index 0000000..383c965
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetFlatMap}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkFilter, FlinkConvention}
+
+class DataSetFilterRule
+ extends ConverterRule(
+ classOf[FlinkFilter],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetFilterRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val filter: FlinkFilter = rel.asInstanceOf[FlinkFilter]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(filter.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetFlatMap(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ filter.toString,
+ null)
+ }
+}
+
+object DataSetFilterRule {
+ val INSTANCE: RelOptRule = new DataSetFilterRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
new file mode 100644
index 0000000..3d2117d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetJoinRule.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.java.operators.join.JoinType
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetJoin}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkJoin, FlinkConvention}
+
+class DataSetJoinRule
+ extends ConverterRule(
+ classOf[FlinkJoin],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetJoinRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val join: FlinkJoin = rel.asInstanceOf[FlinkJoin]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(join.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(join.getInput(1), DataSetConvention.INSTANCE)
+
+ new DataSetJoin(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ rel.getRowType,
+ join.toString,
+ Array[Int](),
+ Array[Int](),
+ JoinType.INNER,
+ null,
+ null)
+ }
+}
+
+object DataSetJoinRule {
+ val INSTANCE: RelOptRule = new DataSetJoinRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
new file mode 100644
index 0000000..7796d66
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetMap}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkProject, FlinkConvention}
+
+class DataSetProjectRule
+ extends ConverterRule(
+ classOf[FlinkProject],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetProjectRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val proj: FlinkProject = rel.asInstanceOf[FlinkProject]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(proj.getInput, DataSetConvention.INSTANCE)
+
+ new DataSetMap(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ rel.getRowType,
+ proj.toString,
+ null)
+ }
+}
+
+object DataSetProjectRule {
+ val INSTANCE: RelOptRule = new DataSetProjectRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.scala
new file mode 100644
index 0000000..937f3e2
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetScanRule.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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkScan, FlinkConvention}
+import org.apache.flink.api.table.plan.schema.DataSetTable
+
+class DataSetScanRule
+ extends ConverterRule(
+ classOf[FlinkScan],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetScanRule")
+{
+ def convert(rel: RelNode): RelNode = {
+ val scan: FlinkScan = rel.asInstanceOf[FlinkScan]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val dataSet: DataSet[_] = scan.getTable().unwrap(classOf[DataSetTable[_]]).dataSet
+
+ new DataSetSource(
+ rel.getCluster,
+ traitSet,
+ scan.getTable,
+ rel.getRowType,
+ dataSet
+ )
+ }
+}
+
+object DataSetScanRule {
+ val INSTANCE: RelOptRule = new DataSetScanRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.scala
new file mode 100644
index 0000000..a390374
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetUnionRule.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.api.table.plan.rules.dataset
+
+import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetUnion}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkUnion, FlinkConvention}
+
+class DataSetUnionRule
+ extends ConverterRule(
+ classOf[FlinkUnion],
+ FlinkConvention.INSTANCE,
+ DataSetConvention.INSTANCE,
+ "DataSetUnionRule")
+{
+
+ def convert(rel: RelNode): RelNode = {
+ val union: FlinkUnion = rel.asInstanceOf[FlinkUnion]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(union.getInput(0), DataSetConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(union.getInput(1), DataSetConvention.INSTANCE)
+
+ new DataSetUnion(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ rel.getRowType,
+ union.toString)
+ }
+}
+
+object DataSetUnionRule {
+ val INSTANCE: RelOptRule = new DataSetUnionRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.scala
new file mode 100644
index 0000000..a5bfeb6
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkAggregateRule.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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalAggregate
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkAggregate, FlinkConvention}
+
+class FlinkAggregateRule
+ extends ConverterRule(
+ classOf[LogicalAggregate],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkAggregateRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val agg: LogicalAggregate = rel.asInstanceOf[LogicalAggregate]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(agg.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkAggregate(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ agg.indicator,
+ agg.getGroupSet,
+ agg.getGroupSets,
+ agg.getAggCallList)
+ }
+ }
+
+object FlinkAggregateRule {
+ val INSTANCE: RelOptRule = new FlinkAggregateRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
new file mode 100644
index 0000000..f40b04d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkCalcRule.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalCalc
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkCalc, FlinkConvention}
+
+class FlinkCalcRule
+ extends ConverterRule(
+ classOf[LogicalCalc],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkCalcRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val calc: LogicalCalc = rel.asInstanceOf[LogicalCalc]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(calc.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkCalc(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ calc.getProgram)
+ }
+ }
+
+object FlinkCalcRule {
+ val INSTANCE: RelOptRule = new FlinkCalcRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala
new file mode 100644
index 0000000..25df8e8
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkFilterRule.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalFilter
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkFilter, FlinkConvention}
+
+class FlinkFilterRule
+ extends ConverterRule(
+ classOf[LogicalFilter],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkFilterRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val filter: LogicalFilter = rel.asInstanceOf[LogicalFilter]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(filter.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkFilter(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ filter.getCondition)
+ }
+ }
+
+object FlinkFilterRule {
+ val INSTANCE: RelOptRule = new FlinkFilterRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
new file mode 100644
index 0000000..3826c9a
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkJoinRule.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalJoin
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkJoin, FlinkConvention}
+
+class FlinkJoinRule
+ extends ConverterRule(
+ classOf[LogicalJoin],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkJoinRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val join: LogicalJoin = rel.asInstanceOf[LogicalJoin]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convLeft: RelNode = RelOptRule.convert(join.getInput(0), FlinkConvention.INSTANCE)
+ val convRight: RelNode = RelOptRule.convert(join.getInput(1), FlinkConvention.INSTANCE)
+
+ new FlinkJoin(
+ rel.getCluster,
+ traitSet,
+ convLeft,
+ convRight,
+ join.getCondition,
+ join.getJoinType,
+ join.getVariablesStopped)
+ }
+ }
+
+object FlinkJoinRule {
+ val INSTANCE: RelOptRule = new FlinkJoinRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala
new file mode 100644
index 0000000..d0e1410
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkProjectRule.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalProject
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkProject, FlinkConvention}
+
+class FlinkProjectRule
+ extends ConverterRule(
+ classOf[LogicalProject],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkProjectRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val proj: LogicalProject = rel.asInstanceOf[LogicalProject]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInput: RelNode = RelOptRule.convert(proj.getInput, FlinkConvention.INSTANCE)
+
+ new FlinkProject(
+ rel.getCluster,
+ traitSet,
+ convInput,
+ proj.getProjects,
+ proj.getRowType)
+ }
+ }
+
+object FlinkProjectRule {
+ val INSTANCE: RelOptRule = new FlinkProjectRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.scala
new file mode 100644
index 0000000..d789770
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkScanRule.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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.core.TableScan
+import org.apache.calcite.rel.logical.LogicalTableScan
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetSource}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkScan, FlinkConvention}
+import org.apache.flink.api.table.plan.schema.DataSetTable
+
+class FlinkScanRule
+ extends ConverterRule(
+ classOf[LogicalTableScan],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkScanRule")
+ {
+ def convert(rel: RelNode): RelNode = {
+ val scan: TableScan = rel.asInstanceOf[TableScan]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val dataSet: DataSet[_] = scan.getTable().unwrap(classOf[DataSetTable[_]]).dataSet
+
+ new FlinkScan(
+ rel.getCluster,
+ traitSet,
+ scan.getTable
+ )
+ }
+ }
+
+object FlinkScanRule {
+ val INSTANCE: RelOptRule = new FlinkScanRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
new file mode 100644
index 0000000..d9869f8
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/logical/FlinkUnionRule.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.api.table.plan.rules.logical
+
+import org.apache.calcite.plan.{Convention, RelOptRule, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.calcite.rel.logical.LogicalUnion
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkUnion, FlinkConvention}
+
+import scala.collection.JavaConversions._
+
+class FlinkUnionRule
+ extends ConverterRule(
+ classOf[LogicalUnion],
+ Convention.NONE,
+ FlinkConvention.INSTANCE,
+ "FlinkUnionRule")
+ {
+
+ def convert(rel: RelNode): RelNode = {
+ val union: LogicalUnion = rel.asInstanceOf[LogicalUnion]
+ val traitSet: RelTraitSet = rel.getTraitSet.replace(FlinkConvention.INSTANCE)
+ val convInputs = union.getInputs.toList.map(
+ RelOptRule.convert(_, FlinkConvention.INSTANCE)
+ )
+
+ new FlinkUnion(
+ rel.getCluster,
+ traitSet,
+ convInputs,
+ union.all)
+ }
+ }
+
+object FlinkUnionRule {
+ val INSTANCE: RelOptRule = new FlinkUnionRule
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
new file mode 100644
index 0000000..e6aecab
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/schema/DataSetTable.scala
@@ -0,0 +1,89 @@
+/*
+ * 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.api.table.plan.schema
+
+import java.lang.Double
+import java.util
+import java.util.Collections
+
+import org.apache.calcite.rel.{RelCollation, RelDistribution}
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
+import org.apache.calcite.schema.Statistic
+import org.apache.calcite.schema.impl.AbstractTable
+import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.util.ImmutableBitSet
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.table.plan.TypeConverter
+
+class DataSetTable[T](
+ val dataSet: DataSet[T],
+ val fieldNames: Array[String])
+ extends AbstractTable {
+
+ // check uniquenss of field names
+ if (fieldNames.length != fieldNames.toSet.size) {
+ throw new scala.IllegalArgumentException(
+ "Table field names must be unique.")
+ }
+
+ val dataSetType: CompositeType[T] =
+ dataSet.getType match {
+ case cType: CompositeType[T] =>
+ cType
+ case _ =>
+ throw new scala.IllegalArgumentException(
+ "DataSet must have a composite type.")
+ }
+
+ val fieldTypes: Array[SqlTypeName] =
+ if (fieldNames.length == dataSetType.getArity) {
+ (0 until dataSetType.getArity)
+ .map(i => dataSetType.getTypeAt(i))
+ .map(TypeConverter.typeInfoToSqlType)
+ .toArray
+ }
+ else {
+ throw new IllegalArgumentException(
+ "Arity of DataSet type not equal to number of field names.")
+ }
+
+ override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
+ val builder = typeFactory.builder
+ fieldNames.zip(fieldTypes)
+ .foreach( f => builder.add(f._1, f._2).nullable(true) )
+ builder.build
+ }
+
+// override def getStatistic: Statistic = {
+// new DefaultDataSetStatistic
+// }
+
+}
+
+class DefaultDataSetStatistic extends Statistic {
+
+ override def getRowCount: Double = 1000d
+
+ override def getCollations: util.List[RelCollation] = Collections.emptyList()
+
+ override def isKey(columns: ImmutableBitSet): Boolean = false
+
+ override def getDistribution: RelDistribution = null
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/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 271aa99..138cd70 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
@@ -270,7 +270,6 @@ class Table(
relBuilder.join(JoinRelType.INNER, relBuilder.literal(true))
val join = relBuilder.build()
- val rowT = join.getRowType()
new Table(join, relBuilder)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
index 447acad..a3d5edc 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
@@ -35,7 +35,9 @@ package org.apache.flink.api.java.table.test;
* limitations under the License.
*/
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.table.ExpressionException;
+import org.apache.flink.api.table.Row;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.table.TableEnvironment;
@@ -48,6 +50,9 @@ import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
+
+import java.util.List;
@RunWith(Parameterized.class)
public class AggregationsITCase extends MultipleProgramsTestBase {
@@ -57,7 +62,7 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAggregationTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -66,10 +71,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result = table.select("f0.sum, f0.min, f0.max, f0.count, f0.avg");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "231,1,21,21,11";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "231,1,21,21,11";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -83,13 +88,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("foo.avg");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testWorkingAggregationDataTypes() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -105,13 +110,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f0.avg, f1.avg, f2.avg, f3.avg, f4.avg, f5.avg, f6.count");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,1,1,1.5,1.5,2";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,1,1,1.5,1.5,2";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAggregationWithArithmetic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -128,13 +133,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
table.select("(f0 + 2).avg + 2, f1.count + 5");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "5.5,7";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "5.5,7";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAggregationWithTwoCount() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -150,11 +155,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f0.count, f1.count");
-
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2";
+ compareResultAsText(results, expected);
}
// Calcite does not eagerly check type compatibility
@@ -172,11 +176,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f1.sum");
-
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = ExpressionException.class)
@@ -192,11 +195,10 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
Table result =
table.select("f0.sum.sum");
-
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
index f706b48..f257c32 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.api.java.table.test;
-import org.apache.flink.api.table.ExpressionException;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.java.DataSet;
@@ -29,6 +28,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -40,7 +40,7 @@ public class AsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAs() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -48,15 +48,15 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -67,10 +67,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -81,10 +81,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c, d");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -95,10 +95,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, b");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -109,10 +109,10 @@ public class AsITCase extends MultipleProgramsTestBase {
Table table =
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a + 1, b, c");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -124,10 +124,10 @@ public class AsITCase extends MultipleProgramsTestBase {
tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a as foo, b," +
" c");
-// DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
index 5e99fd9..cde78ce 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
@@ -32,6 +32,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -43,7 +44,7 @@ public class CastingITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testNumericAutocastInArithmetic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -57,13 +58,13 @@ public class CastingITCase extends MultipleProgramsTestBase {
Table result = table.select("f0 + 1, f1 +" +
" 1, f2 + 1L, f3 + 1.0f, f4 + 1.0d, f5 + 1");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,2,2.0,2.0,2.0";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,2,2.0,2.0,2.0";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testNumericAutocastInComparison() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -79,13 +80,13 @@ public class CastingITCase extends MultipleProgramsTestBase {
Table result = table
.filter("a > 1 && b > 1 && c > 1L && d > 1.0f && e > 1.0d && f > 1");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,2,2,2.0,2.0,Hello";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,2,2,2.0,2.0,Hello";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testCastFromString() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -99,13 +100,13 @@ public class CastingITCase extends MultipleProgramsTestBase {
Table result = table.select(
"f0.cast(BYTE), f0.cast(SHORT), f0.cast(INT), f0.cast(LONG), f2.cast(DOUBLE), f2.cast(FLOAT), f1.cast(BOOL)");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,1,1,2.0,2.0,true\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,1,1,2.0,2.0,true\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testCastDateFromString() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -120,14 +121,14 @@ public class CastingITCase extends MultipleProgramsTestBase {
.select("f0.cast(DATE) AS f0, f1.cast(DATE) AS f1, f2.cast(DATE) AS f2, f3.cast(DATE) AS f3")
.select("f0.cast(STRING), f1.cast(STRING), f2.cast(STRING), f3.cast(STRING)");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2011-05-03 00:00:00.000,1970-01-01 15:51:36.000,2011-05-03 15:51:36.000," +
-// "1970-01-17 17:47:53.775\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2011-05-03 00:00:00.000,1970-01-01 15:51:36.000,2011-05-03 15:51:36.000," +
+ "1970-01-17 17:47:53.775\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testCastDateToStringAndLong() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -142,10 +143,10 @@ public class CastingITCase extends MultipleProgramsTestBase {
.select("f0.cast(DATE) AS f0, f1.cast(DATE) AS f1")
.select("f0.cast(STRING), f0.cast(LONG), f1.cast(STRING), f1.cast(LONG)");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2011-05-03 15:51:36.000,1304437896000,2011-05-03 15:51:36.000,1304437896000\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2011-05-03 15:51:36.000,1304437896000,2011-05-03 15:51:36.000,1304437896000\n";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
index 3bbc120..2a17087 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.api.java.table.test;
-import org.apache.flink.api.table.ExpressionException;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.java.DataSet;
@@ -31,6 +30,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -42,7 +42,7 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testArithmetic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -56,13 +56,13 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
Table result = table.select(
"a - 5, a + 5, a / 2, a * 2, a % 2, -a");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "0,10,2,10,1,-5";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "0,10,2,10,1,-5";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testLogic() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -76,13 +76,13 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
Table result = table.select(
"b && true, b && false, b || false, !b");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "true,false,true,false";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "true,false,true,false";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testComparisons() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -96,10 +96,10 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
Table result = table.select(
"a > c, a >= b, a < c, a.isNull, a.isNotNull");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "true,true,false,false,true";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "true,true,false,false,true";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
index 68925fb..a3ab10f 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
@@ -29,6 +29,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -40,7 +41,7 @@ public class FilterITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAllRejectingFilter() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -53,13 +54,13 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter("false");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testAllPassingFilter() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -72,18 +73,18 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter("true");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
-// "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
-// "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
-// "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
-// "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
-// "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1,Hi\n" + "2,2,Hello\n" + "3,2,Hello world\n" + "4,3,Hello world, " +
+ "how are you?\n" + "5,3,I am fine.\n" + "6,3,Luke Skywalker\n" + "7,4," +
+ "Comment#1\n" + "8,4,Comment#2\n" + "9,4,Comment#3\n" + "10,4,Comment#4\n" + "11,5," +
+ "Comment#5\n" + "12,5,Comment#6\n" + "13,5,Comment#7\n" + "14,5,Comment#8\n" + "15,5," +
+ "Comment#9\n" + "16,6,Comment#10\n" + "17,6,Comment#11\n" + "18,6,Comment#12\n" + "19," +
+ "6,Comment#13\n" + "20,6,Comment#14\n" + "21,6,Comment#15\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testFilterOnIntegerTupleField() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -96,15 +97,15 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter(" a % 2 = 0 ");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
-// "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
-// "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
+ "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
+ "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testNotEquals() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -117,15 +118,15 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table
.filter("!( a % 2 <> 0 ) ");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
-// "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
-// "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "2,2,Hello\n" + "4,3,Hello world, how are you?\n" + "6,3,Luke Skywalker\n" + "8,4," +
+ "Comment#2\n" + "10,4,Comment#4\n" + "12,5,Comment#6\n" + "14,5,Comment#8\n" + "16,6," +
+ "Comment#10\n" + "18,6,Comment#12\n" + "20,6,Comment#14\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testIntegerBiggerThan128() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -136,10 +137,10 @@ public class FilterITCase extends MultipleProgramsTestBase {
Table result = table.filter("a = 300 ");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "300,1,Hello\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "300,1,Hello\n";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
index 2add694..d5dc56a 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
@@ -18,7 +18,6 @@
package org.apache.flink.api.java.table.test;
-import org.apache.flink.api.table.ExpressionException;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.table.Row;
import org.apache.flink.api.java.DataSet;
@@ -30,6 +29,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
import java.util.List;
@@ -54,13 +54,13 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("foo").select("a.avg");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testGroupedAggregate() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -73,13 +73,13 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("b").select("b, a.sum");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1,1\n" + "2,5\n" + "3,15\n" + "4,34\n" + "5,65\n" + "6,111\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testGroupingKeyForwardIfNotUsed() throws Exception {
// the grouping key needs to be forwarded to the intermediate DataSet, even
@@ -96,13 +96,13 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("b").select("a.sum");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "1\n" + "5\n" + "15\n" + "34\n" + "65\n" + "111\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testGroupNoAggregation() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -116,10 +116,10 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase {
Table result = table
.groupBy("b").select("a.sum as d, b").groupBy("b, d").select("b");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// String expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n";
-// List<Row> results = ds.collect();
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ String expected = "1\n" + "2\n" + "3\n" + "4\n" + "5\n" + "6\n";
+ List<Row> results = ds.collect();
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
index 2b44a87..2e1f4a7 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.java.table.test;
+import org.apache.flink.api.table.Row;
import org.apache.flink.api.table.Table;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
@@ -30,6 +31,9 @@ import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
+
+import java.util.List;
@RunWith(Parameterized.class)
@@ -40,7 +44,7 @@ public class JoinITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoin() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -53,13 +57,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("b === e").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoinWithFilter() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -72,13 +76,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("b === e && b < 2").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi,Hallo\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi,Hallo\n";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoinWithMultipleKeys() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -91,11 +95,11 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("a === d && b === h").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
-// "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "Hi,Hallo\n" + "Hello,Hallo Welt\n" + "Hello world,Hallo Welt wie gehts?\n" +
+ "Hello world,ABC\n" + "I am fine.,HIJ\n" + "I am fine.,IJK\n";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -111,10 +115,10 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1.join(in2).where("foo === e").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
// Calcite does not eagerly check the compatibility of compared types
@@ -133,10 +137,10 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1
.join(in2).where("a === g").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
@Test(expected = IllegalArgumentException.class)
@@ -153,13 +157,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1
.join(in2).where("a === d").select("c, g");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "";
+ compareResultAsText(results, expected);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testJoinWithAggregation() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = new TableEnvironment();
@@ -173,10 +177,10 @@ public class JoinITCase extends MultipleProgramsTestBase {
Table result = in1
.join(in2).where("a === d").select("g.count");
-// DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
-// List<Row> results = ds.collect();
-// String expected = "6";
-// compareResultAsText(results, expected);
+ DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
+ List<Row> results = ds.collect();
+ String expected = "6";
+ compareResultAsText(results, expected);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7ecb7010/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
index f19b8c1..5ef0235 100644
--- a/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
+++ b/flink-libraries/flink-table/src/test/java/org/apache/flink/api/java/table/test/PojoGroupingITCase.java
@@ -30,6 +30,7 @@ import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import scala.NotImplementedError;
@RunWith(Parameterized.class)
public class PojoGroupingITCase extends MultipleProgramsTestBase {
@@ -38,7 +39,7 @@ public class PojoGroupingITCase extends MultipleProgramsTestBase {
super(mode);
}
- @Test
+ @Test(expected = NotImplementedError.class)
public void testPojoGrouping() throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -54,14 +55,14 @@ public class PojoGroupingITCase extends MultipleProgramsTestBase {
.select("groupMe, value, name")
.where("groupMe != 'B'");
-// DataSet<MyPojo> myPojos = tableEnv.toDataSet(table, MyPojo.class);
-//
-// DataSet<MyPojo> result = myPojos.groupBy("groupMe")
-// .sortGroup("value", Order.DESCENDING)
-// .first(1);
-//
-// List<MyPojo> resultList = result.collect();
-// compareResultAsText(resultList, "A,24.0,Y");
+ DataSet<MyPojo> myPojos = tableEnv.toDataSet(table, MyPojo.class);
+
+ DataSet<MyPojo> result = myPojos.groupBy("groupMe")
+ .sortGroup("value", Order.DESCENDING)
+ .first(1);
+
+ List<MyPojo> resultList = result.collect();
+ compareResultAsText(resultList, "A,24.0,Y");
}
public static class MyPojo implements Serializable {