flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject flink git commit: [FLINK-2165] [TableAPI] Renamed table conversion functions in TableEnvironment
Date Fri, 05 Jun 2015 13:38:44 GMT
Repository: flink
Updated Branches:
  refs/heads/master 235b02cb5 -> 9b8818451


[FLINK-2165] [TableAPI] Renamed table conversion functions in TableEnvironment

This closes #793


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

Branch: refs/heads/master
Commit: 9b88184518c6f23f1a0f18d68623e4909b3c4476
Parents: 235b02c
Author: Fabian Hueske <fhueske@apache.org>
Authored: Fri Jun 5 11:37:39 2015 +0200
Committer: Fabian Hueske <fhueske@apache.org>
Committed: Fri Jun 5 15:37:23 2015 +0200

----------------------------------------------------------------------
 .../flink/examples/java/JavaTableExample.java   |  4 +-
 .../flink/api/java/table/TableEnvironment.scala | 22 +++++-----
 .../api/java/table/test/AggregationsITCase.java | 25 ++++++------
 .../flink/api/java/table/test/AsITCase.java     | 25 ++++++------
 .../api/java/table/test/CastingITCase.java      | 13 +++---
 .../api/java/table/test/ExpressionsITCase.java  | 25 ++++++------
 .../flink/api/java/table/test/FilterITCase.java | 21 +++++-----
 .../table/test/GroupedAggregationsITCase.java   | 13 +++---
 .../flink/api/java/table/test/JoinITCase.java   | 43 ++++++++++----------
 .../flink/api/java/table/test/SelectITCase.java | 25 ++++++------
 .../table/test/StringExpressionsITCase.java     | 17 ++++----
 11 files changed, 112 insertions(+), 121 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
b/flink-staging/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
index e898b5d..c043508 100644
--- a/flink-staging/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
+++ b/flink-staging/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
@@ -57,14 +57,14 @@ public class JavaTableExample {
 				new WC("Ciao", 1),
 				new WC("Hello", 1));
 
-		Table table = tableEnv.toTable(input);
+		Table table = tableEnv.fromDataSet(input);
 
 		Table filtered = table
 				.groupBy("word")
 				.select("word.count as count, word")
 				.filter("count = 2");
 
-		DataSet<WC> result = tableEnv.toSet(filtered, WC.class);
+		DataSet<WC> result = tableEnv.toDataSet(filtered, WC.class);
 
 		result.print();
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
b/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
index ab88874..5614031 100644
--- a/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
+++ b/flink-staging/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
@@ -38,13 +38,13 @@ class TableEnvironment {
    * Example:
    *
    * {{{
-   *   tableEnv.toTable(set, "a, b")
+   *   tableEnv.fromDataSet(set, "a, b")
    * }}}
    *
    * This will transform the set containing elements of two fields to a table where the fields
    * are named a and b.
    */
-  def toTable[T](set: DataSet[T], fields: String): Table = {
+  def fromDataSet[T](set: DataSet[T], fields: String): Table = {
     new JavaBatchTranslator().createTable(set, fields)
   }
 
@@ -53,33 +53,33 @@ class TableEnvironment {
    * The fields of the DataSet type are used to name the
    * [[org.apache.flink.api.table.Table]] fields.
    */
-  def toTable[T](set: DataSet[T]): Table = {
+  def fromDataSet[T](set: DataSet[T]): Table = {
     new JavaBatchTranslator().createTable(set)
   }
 
   /**
    * Transforms the given DataStream to a [[org.apache.flink.api.table.Table]].
-   * The fields of the DataSet type are renamed to the given set of fields:
+   * The fields of the DataStream type are renamed to the given set of fields:
    *
    * Example:
    *
    * {{{
-   *   tableEnv.toTable(set, "a, b")
+   *   tableEnv.fromDataStream(set, "a, b")
    * }}}
    *
    * This will transform the set containing elements of two fields to a table where the fields
    * are named a and b.
    */
-  def toTable[T](set: DataStream[T], fields: String): Table = {
+  def fromDataStream[T](set: DataStream[T], fields: String): Table = {
     new JavaStreamingTranslator().createTable(set, fields)
   }
 
   /**
-   * Transforms the given DataSet to a [[org.apache.flink.api.table.Table]].
-   * The fields of the DataSet type are used to name the
+   * Transforms the given DataStream to a [[org.apache.flink.api.table.Table]].
+   * The fields of the DataStream type are used to name the
    * [[org.apache.flink.api.table.Table]] fields.
    */
-  def toTable[T](set: DataStream[T]): Table = {
+  def fromDataStream[T](set: DataStream[T]): Table = {
     new JavaStreamingTranslator().createTable(set)
   }
 
@@ -90,7 +90,7 @@ class TableEnvironment {
    * fields and the types must match.
    */
   @SuppressWarnings(Array("unchecked"))
-  def toSet[T](table: Table, clazz: Class[T]): DataSet[T] = {
+  def toDataSet[T](table: Table, clazz: Class[T]): DataSet[T] = {
     new JavaBatchTranslator().translate[T](table.operation)(
       TypeExtractor.createTypeInfo(clazz).asInstanceOf[TypeInformation[T]])
   }
@@ -102,7 +102,7 @@ class TableEnvironment {
    * fields and the types must match.
    */
   @SuppressWarnings(Array("unchecked"))
-  def toStream[T](table: Table, clazz: Class[T]): DataStream[T] = {
+  def toDataStream[T](table: Table, clazz: Class[T]): DataStream[T] = {
     new JavaStreamingTranslator().translate[T](table.operation)(
       TypeExtractor.createTypeInfo(clazz).asInstanceOf[TypeInformation[T]])
 

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
index 60fb984..0eb6a40 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AggregationsITCase.java
@@ -44,7 +44,6 @@ import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.operators.DataSource;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple7;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
@@ -85,11 +84,11 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		TableEnvironment tableEnv = new TableEnvironment();
 
-		Table table = tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env));
+		Table table = tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env));
 
 		Table result = table.select("f0.sum, f0.min, f0.max, f0.count, f0.avg");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -103,12 +102,12 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
 		TableEnvironment tableEnv = new TableEnvironment();
 
 		Table table =
-				tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env));
+				tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env));
 
 		Table result =
 				table.select("foo.avg");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -127,12 +126,12 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
 						new Tuple7<Byte, Short, Integer, Long, Float, Double, String>((byte) 2, (short)
2, 2, 2L, 2.0f, 2.0d, "Ciao"));
 
 		Table table =
-				tableEnv.toTable(input);
+				tableEnv.fromDataSet(input);
 
 		Table result =
 				table.select("f0.avg, f1.avg, f2.avg, f3.avg, f4.avg, f5.avg, f6.count");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -151,13 +150,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
 						new Tuple2<Float, String>(2f, "Ciao"));
 
 		Table table =
-				tableEnv.toTable(input);
+				tableEnv.fromDataSet(input);
 
 		Table result =
 				table.select("(f0 + 2).avg + 2, f1.count + \" THE COUNT\"");
 
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -174,13 +173,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
 				"Hello"));
 
 		Table table =
-				tableEnv.toTable(input);
+				tableEnv.fromDataSet(input);
 
 		Table result =
 				table.select("f1.sum");
 
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -196,13 +195,13 @@ public class AggregationsITCase extends MultipleProgramsTestBase {
 		DataSource<Tuple2<Float, String>> input = env.fromElements(new Tuple2<Float,
String>(1f, "Hello"));
 
 		Table table =
-				tableEnv.toTable(input);
+				tableEnv.fromDataSet(input);
 
 		Table result =
 				table.select("f0.sum.sum");
 
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
index 6ec3187..4bb5dec 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/AsITCase.java
@@ -24,7 +24,6 @@ import org.apache.flink.api.table.Row;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.table.TableEnvironment;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
@@ -66,9 +65,9 @@ public class AsITCase extends MultipleProgramsTestBase {
 		TableEnvironment tableEnv = new TableEnvironment();
 
 		Table table =
-				tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env), "a, b, c");
+				tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c");
 
-		DataSet<Row> ds = tableEnv.toSet(table, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -87,9 +86,9 @@ public class AsITCase extends MultipleProgramsTestBase {
 		TableEnvironment tableEnv = new TableEnvironment();
 
 		Table table =
-				tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env), "a, b");
+				tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b");
 
-		DataSet<Row> ds = tableEnv.toSet(table, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -103,9 +102,9 @@ public class AsITCase extends MultipleProgramsTestBase {
 		TableEnvironment tableEnv = new TableEnvironment();
 
 		Table table =
-				tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env), "a, b, c, d");
+				tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, c, d");
 
-		DataSet<Row> ds = tableEnv.toSet(table, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -119,9 +118,9 @@ public class AsITCase extends MultipleProgramsTestBase {
 		TableEnvironment tableEnv = new TableEnvironment();
 
 		Table table =
-				tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env), "a, b, b");
+				tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a, b, b");
 
-		DataSet<Row> ds = tableEnv.toSet(table, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -135,9 +134,9 @@ public class AsITCase extends MultipleProgramsTestBase {
 		TableEnvironment tableEnv = new TableEnvironment();
 
 		Table table =
-				tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env), "a + 1, b, c");
+				tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a + 1, b, c");
 
-		DataSet<Row> ds = tableEnv.toSet(table, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -151,10 +150,10 @@ public class AsITCase extends MultipleProgramsTestBase {
 		TableEnvironment tableEnv = new TableEnvironment();
 
 		Table table =
-				tableEnv.toTable(CollectionDataSets.get3TupleDataSet(env), "a as foo, b," +
+				tableEnv.fromDataSet(CollectionDataSets.get3TupleDataSet(env), "a as foo, b," +
 						" c");
 
-		DataSet<Row> ds = tableEnv.toSet(table, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(table, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
index 898c650..8b60ed1 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/CastingITCase.java
@@ -25,7 +25,6 @@ import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.operators.DataSource;
 import org.apache.flink.api.java.tuple.Tuple7;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
 import org.junit.After;
@@ -70,12 +69,12 @@ public class CastingITCase extends MultipleProgramsTestBase {
 						(byte) 1, (short) 1, 1, 1L, 1.0f, 1.0d, "Hello"));
 
 		Table table =
-				tableEnv.toTable(input);
+				tableEnv.fromDataSet(input);
 
 		Table result = table.select(
 				"f0 + 'b', f1 + 's', f2 + 'i', f3 + 'L', f4 + 'f', f5 + \"d\"");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -93,12 +92,12 @@ public class CastingITCase extends MultipleProgramsTestBase {
 						(byte) 1, (short) 1, 1, 1L, 1.0f, 1.0d, "Hello"));
 
 		Table table =
-				tableEnv.toTable(input);
+				tableEnv.fromDataSet(input);
 
 		Table result = table.select("f0 + 1, f1 +" +
 				" 1, f2 + 1L, f3 + 1.0f, f4 + 1.0d, f5 + 1");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -117,12 +116,12 @@ public class CastingITCase extends MultipleProgramsTestBase {
 						new Tuple7<Byte, Short, Integer, Long, Float, Double, String>((byte) 2, (short)
2, 2, 2L, 2.0f, 2.0d, "Hello"));
 
 		Table table =
-				tableEnv.toTable(input, "a,b,c,d,e,f,g");
+				tableEnv.fromDataSet(input, "a,b,c,d,e,f,g");
 
 		Table result = table
 				.filter("a > 1 && b > 1 && c > 1L && d > 1.0f &&
e > 1.0d && f > 1");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
index 29e7db1..853da87 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/ExpressionsITCase.java
@@ -27,7 +27,6 @@ import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.operators.DataSource;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
 import org.junit.After;
@@ -71,12 +70,12 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
 				env.fromElements(new Tuple2<Integer, Integer>(5, 10));
 
 		Table table =
-				tableEnv.toTable(input, "a, b");
+				tableEnv.fromDataSet(input, "a, b");
 
 		Table result = table.select(
 				"a - 5, a + 5, a / 2, a * 2, a % 2, -a");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -93,12 +92,12 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
 				env.fromElements(new Tuple2<Integer, Boolean>(5, true));
 
 		Table table =
-				tableEnv.toTable(input, "a, b");
+				tableEnv.fromDataSet(input, "a, b");
 
 		Table result = table.select(
 				"b && true, b && false, b || false, !b");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -115,12 +114,12 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
 				env.fromElements(new Tuple3<Integer, Integer, Integer>(5, 5, 4));
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table.select(
 				"a > c, a >= b, a < c, a.isNull, a.isNotNull");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -137,12 +136,12 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
 				env.fromElements(new Tuple2<Byte, Byte>((byte) 3, (byte) 5));
 
 		Table table =
-				tableEnv.toTable(input, "a, b");
+				tableEnv.fromDataSet(input, "a, b");
 
 		Table result = table.select(
 				"a & b, a | b, a ^ b, ~a");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -159,12 +158,12 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
 				env.fromElements(new Tuple2<Integer, Byte>(3, (byte) 5));
 
 		Table table =
-				tableEnv.toTable(input, "a, b");
+				tableEnv.fromDataSet(input, "a, b");
 
 		Table result = table.select(
 				"a & b, a | b, a ^ b, ~a");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -181,12 +180,12 @@ public class ExpressionsITCase extends MultipleProgramsTestBase {
 				env.fromElements(new Tuple2<Float, Byte>(3.0f, (byte) 5));
 
 		Table table =
-				tableEnv.toTable(input, "a, b");
+				tableEnv.fromDataSet(input, "a, b");
 
 		Table result =
 				table.select("a & b, a | b, a ^ b, ~a");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
index 8b5fc43..54d8f42 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/FilterITCase.java
@@ -24,7 +24,6 @@ import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
@@ -68,12 +67,12 @@ public class FilterITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table
 				.filter("false");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -89,12 +88,12 @@ public class FilterITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table
 				.filter("true");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -115,12 +114,12 @@ public class FilterITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table
 				.filter(" a % 2 = 0 ");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -138,12 +137,12 @@ public class FilterITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table
 				.filter("!( a % 2 <> 0 ) ");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -160,11 +159,11 @@ public class FilterITCase extends MultipleProgramsTestBase {
 
 		DataSet<Tuple3<Integer, Long, String>> input = env.fromElements(new Tuple3<Integer,
Long, String>(300, 1L, "Hello"));
 
-		Table table = tableEnv.toTable(input, "a, b, c");
+		Table table = tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table.filter("a = 300 ");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
index c9217a1..aa0b481 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/GroupedAggregationsITCase.java
@@ -25,7 +25,6 @@ import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
@@ -69,12 +68,12 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase
{
 		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table
 				.groupBy("foo").select("a.avg");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -90,12 +89,12 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase
{
 		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table
 				.groupBy("b").select("b, a.sum");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -115,12 +114,12 @@ public class GroupedAggregationsITCase extends MultipleProgramsTestBase
{
 		DataSet<Tuple3<Integer, Long, String>> input = CollectionDataSets.get3TupleDataSet(env);
 
 		Table table =
-				tableEnv.toTable(input, "a, b, c");
+				tableEnv.fromDataSet(input, "a, b, c");
 
 		Table result = table
 				.groupBy("b").select("a.sum");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
index f87970e..58f5d23 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/JoinITCase.java
@@ -26,7 +26,6 @@ import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.api.java.tuple.Tuple5;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
@@ -70,12 +69,12 @@ public class JoinITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.getSmall3TupleDataSet(env);
 		DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds2 = CollectionDataSets.get5TupleDataSet(env);
 
-		Table in1 = tableEnv.toTable(ds1, "a, b, c");
-		Table in2 = tableEnv.toTable(ds2, "d, e, f, g, h");
+		Table in1 = tableEnv.fromDataSet(ds1, "a, b, c");
+		Table in2 = tableEnv.fromDataSet(ds2, "d, e, f, g, h");
 
 		Table result = in1.join(in2).where("b === e").select("c, g");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -91,12 +90,12 @@ public class JoinITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.getSmall3TupleDataSet(env);
 		DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds2 = CollectionDataSets.get5TupleDataSet(env);
 
-		Table in1 = tableEnv.toTable(ds1, "a, b, c");
-		Table in2 = tableEnv.toTable(ds2, "d, e, f, g, h");
+		Table in1 = tableEnv.fromDataSet(ds1, "a, b, c");
+		Table in2 = tableEnv.fromDataSet(ds2, "d, e, f, g, h");
 
 		Table result = in1.join(in2).where("b === e && b < 2").select("c, g");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -112,12 +111,12 @@ public class JoinITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.get3TupleDataSet(env);
 		DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds2 = CollectionDataSets.get5TupleDataSet(env);
 
-		Table in1 = tableEnv.toTable(ds1, "a, b, c");
-		Table in2 = tableEnv.toTable(ds2, "d, e, f, g, h");
+		Table in1 = tableEnv.fromDataSet(ds1, "a, b, c");
+		Table in2 = tableEnv.fromDataSet(ds2, "d, e, f, g, h");
 
 		Table result = in1.join(in2).where("a === d && b === h").select("c, g");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -134,12 +133,12 @@ public class JoinITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.getSmall3TupleDataSet(env);
 		DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds2 = CollectionDataSets.get5TupleDataSet(env);
 
-		Table in1 = tableEnv.toTable(ds1, "a, b, c");
-		Table in2 = tableEnv.toTable(ds2, "d, e, f, g, h");
+		Table in1 = tableEnv.fromDataSet(ds1, "a, b, c");
+		Table in2 = tableEnv.fromDataSet(ds2, "d, e, f, g, h");
 
 		Table result = in1.join(in2).where("foo === e").select("c, g");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -155,13 +154,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.getSmall3TupleDataSet(env);
 		DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds2 = CollectionDataSets.get5TupleDataSet(env);
 
-		Table in1 = tableEnv.toTable(ds1, "a, b, c");
-		Table in2 = tableEnv.toTable(ds2, "d, e, f, g, h");
+		Table in1 = tableEnv.fromDataSet(ds1, "a, b, c");
+		Table in2 = tableEnv.fromDataSet(ds2, "d, e, f, g, h");
 
 		Table result = in1
 				.join(in2).where("a === g").select("c, g");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -177,13 +176,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.getSmall3TupleDataSet(env);
 		DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds2 = CollectionDataSets.get5TupleDataSet(env);
 
-		Table in1 = tableEnv.toTable(ds1, "a, b, c");
-		Table in2 = tableEnv.toTable(ds2, "d, e, f, g, c");
+		Table in1 = tableEnv.fromDataSet(ds1, "a, b, c");
+		Table in2 = tableEnv.fromDataSet(ds2, "d, e, f, g, c");
 
 		Table result = in1
 				.join(in2).where("a === d").select("c, g");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -199,13 +198,13 @@ public class JoinITCase extends MultipleProgramsTestBase {
 		DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.getSmall3TupleDataSet(env);
 		DataSet<Tuple5<Integer, Long, Integer, String, Long>> ds2 = CollectionDataSets.get5TupleDataSet(env);
 
-		Table in1 = tableEnv.toTable(ds1, "a, b, c");
-		Table in2 = tableEnv.toTable(ds2, "d, e, f, g, h");
+		Table in1 = tableEnv.fromDataSet(ds1, "a, b, c");
+		Table in2 = tableEnv.fromDataSet(ds2, "d, e, f, g, h");
 
 		Table result = in1
 				.join(in2).where("a === d").select("g.count");
 
-		DataSet<Row> ds = tableEnv.toSet(result, Row.class);
+		DataSet<Row> ds = tableEnv.toDataSet(result, Row.class);
 		ds.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
index 446f13e..5385abd 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/SelectITCase.java
@@ -25,7 +25,6 @@ import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
@@ -68,12 +67,12 @@ public class SelectITCase extends MultipleProgramsTestBase {
 
 		DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
 
-		Table in = tableEnv.toTable(ds, "a,b,c");
+		Table in = tableEnv.fromDataSet(ds, "a,b,c");
 
 		Table result = in
 				.select("a, b, c");
 
-		DataSet<Row> resultSet = tableEnv.toSet(result, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -94,13 +93,13 @@ public class SelectITCase extends MultipleProgramsTestBase {
 
 		DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
 
-		Table in = tableEnv.toTable(ds);
+		Table in = tableEnv.fromDataSet(ds);
 
 		Table result = in
 				.select("f0 as a, f1 as b")
 				.select("a, b");
 
-		DataSet<Row> resultSet = tableEnv.toSet(result, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -117,9 +116,9 @@ public class SelectITCase extends MultipleProgramsTestBase {
 
 		DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
 
-		Table in = tableEnv.toTable(ds, "a, b");
+		Table in = tableEnv.fromDataSet(ds, "a, b");
 
-		DataSet<Row> resultSet = tableEnv.toSet(in, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -134,9 +133,9 @@ public class SelectITCase extends MultipleProgramsTestBase {
 
 		DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
 
-		Table in = tableEnv.toTable(ds, "a, b, c, d");
+		Table in = tableEnv.fromDataSet(ds, "a, b, c, d");
 
-		DataSet<Row> resultSet = tableEnv.toSet(in, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -151,9 +150,9 @@ public class SelectITCase extends MultipleProgramsTestBase {
 
 		DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
 
-		Table in = tableEnv.toTable(ds, "a, b, c, b");
+		Table in = tableEnv.fromDataSet(ds, "a, b, c, b");
 
-		DataSet<Row> resultSet = tableEnv.toSet(in, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -168,9 +167,9 @@ public class SelectITCase extends MultipleProgramsTestBase {
 
 		DataSet<Tuple3<Integer, Long, String>> ds = CollectionDataSets.get3TupleDataSet(env);
 
-		Table in = tableEnv.toTable(ds, "a, b as c, d");
+		Table in = tableEnv.fromDataSet(ds, "a, b as c, d");
 
-		DataSet<Row> resultSet = tableEnv.toSet(in, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(in, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();

http://git-wip-us.apache.org/repos/asf/flink/blob/9b881845/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
index c6a56c3..6fe7e8c 100644
--- a/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
+++ b/flink-staging/flink-table/src/test/java/org/apache/flink/api/java/table/test/StringExpressionsITCase.java
@@ -25,7 +25,6 @@ import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.table.TableEnvironment;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.table.JavaBatchTranslator;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.test.util.MultipleProgramsTestBase;
 import org.junit.After;
@@ -69,12 +68,12 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase
{
 				new Tuple2<String, Integer>("AAAA", 2),
 				new Tuple2<String, Integer>("BBBB", 1));
 
-		Table in = tableEnv.toTable(ds, "a, b");
+		Table in = tableEnv.fromDataSet(ds, "a, b");
 
 		Table result = in
 				.select("a.substring(0, b)");
 
-		DataSet<Row> resultSet = tableEnv.toSet(result, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -91,12 +90,12 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase
{
 				new Tuple2<String, Integer>("ABCD", 2),
 				new Tuple2<String, Integer>("ABCD", 1));
 
-		Table in = tableEnv.toTable(ds, "a, b");
+		Table in = tableEnv.fromDataSet(ds, "a, b");
 
 		Table result = in
 				.select("a.substring(b)");
 
-		DataSet<Row> resultSet = tableEnv.toSet(result, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -113,12 +112,12 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase
{
 				new Tuple2<String, Float>("ABCD", 2.0f),
 				new Tuple2<String, Float>("ABCD", 1.0f));
 
-		Table in = tableEnv.toTable(ds, "a, b");
+		Table in = tableEnv.fromDataSet(ds, "a, b");
 
 		Table result = in
 				.select("a.substring(0, b)");
 
-		DataSet<Row> resultSet = tableEnv.toSet(result, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();
@@ -135,12 +134,12 @@ public class StringExpressionsITCase extends MultipleProgramsTestBase
{
 				new Tuple2<String, String>("ABCD", "a"),
 				new Tuple2<String, String>("ABCD", "b"));
 
-		Table in = tableEnv.toTable(ds, "a, b");
+		Table in = tableEnv.fromDataSet(ds, "a, b");
 
 		Table result = in
 				.select("a.substring(b, 15)");
 
-		DataSet<Row> resultSet = tableEnv.toSet(result, Row.class);
+		DataSet<Row> resultSet = tableEnv.toDataSet(result, Row.class);
 		resultSet.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
 
 		env.execute();


Mime
View raw message