flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmetz...@apache.org
Subject [38/50] flink git commit: [FLINK-1712] Remove "flink-staging" module
Date Thu, 14 Jan 2016 16:16:35 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/regression/RegressionData.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/regression/RegressionData.scala b/flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/regression/RegressionData.scala
new file mode 100644
index 0000000..062f510
--- /dev/null
+++ b/flink-libraries/flink-ml/src/test/scala/org/apache/flink/ml/regression/RegressionData.scala
@@ -0,0 +1,191 @@
+/*
+ * 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.ml.regression
+
+import org.apache.flink.ml.common.LabeledVector
+import org.apache.flink.ml.math.DenseVector
+
+object RegressionData {
+
+  val expectedWeights = Array[Double](3.0094)
+  val expectedWeight0: Double = 9.8158
+  val expectedSquaredResidualSum: Double = 49.7596/2
+
+  val data: Seq[LabeledVector] = Seq(
+    LabeledVector(10.7949, DenseVector(0.2714)),
+    LabeledVector(10.6426, DenseVector(0.1008)),
+    LabeledVector(10.5603, DenseVector(0.5078)),
+    LabeledVector(12.8707, DenseVector(0.5856)),
+    LabeledVector(10.7026, DenseVector(0.7629)),
+    LabeledVector(9.8571, DenseVector(0.0830)),
+    LabeledVector(10.5001, DenseVector(0.6616)),
+    LabeledVector(11.2063, DenseVector(0.5170)),
+    LabeledVector(9.1892, DenseVector(0.1710)),
+    LabeledVector(12.2408, DenseVector(0.9386)),
+    LabeledVector(11.0307, DenseVector(0.5905)),
+    LabeledVector(10.1369, DenseVector(0.4406)),
+    LabeledVector(10.7609, DenseVector(0.9419)),
+    LabeledVector(12.5328, DenseVector(0.6559)),
+    LabeledVector(13.3560, DenseVector(0.4519)),
+    LabeledVector(14.7424, DenseVector(0.8397)),
+    LabeledVector(11.1057, DenseVector(0.5326)),
+    LabeledVector(11.6157, DenseVector(0.5539)),
+    LabeledVector(11.5744, DenseVector(0.6801)),
+    LabeledVector(11.1775, DenseVector(0.3672)),
+    LabeledVector(9.7991, DenseVector(0.2393)),
+    LabeledVector(9.8173, DenseVector(0.5789)),
+    LabeledVector(12.5642, DenseVector(0.8669)),
+    LabeledVector(9.9952, DenseVector(0.4068)),
+    LabeledVector(8.4354, DenseVector(0.1126)),
+    LabeledVector(13.7058, DenseVector(0.4438)),
+    LabeledVector(10.6672, DenseVector(0.3002)),
+    LabeledVector(11.6080, DenseVector(0.4014)),
+    LabeledVector(13.6926, DenseVector(0.8334)),
+    LabeledVector(9.5261, DenseVector(0.4036)),
+    LabeledVector(11.5837, DenseVector(0.3902)),
+    LabeledVector(11.5831, DenseVector(0.3604)),
+    LabeledVector(10.5038, DenseVector(0.1403)),
+    LabeledVector(10.9382, DenseVector(0.2601)),
+    LabeledVector(9.7325, DenseVector(0.0868)),
+    LabeledVector(12.0113, DenseVector(0.4294)),
+    LabeledVector(9.9219, DenseVector(0.2573)),
+    LabeledVector(10.0963, DenseVector(0.2976)),
+    LabeledVector(11.9999, DenseVector(0.4249)),
+    LabeledVector(12.0442, DenseVector(0.1192))
+  )
+
+  val expectedNoInterceptWeights = Array[Double](5.0)
+  val expectedNoInterceptWeight0: Double = 0.0
+
+  val noInterceptData: Seq[LabeledVector] = Seq(
+    LabeledVector(217.228709, DenseVector(43.4457419)),
+    LabeledVector(450.037048, DenseVector(90.0074095)),
+    LabeledVector( 67.553478, DenseVector(13.5106955)),
+    LabeledVector( 26.976958, DenseVector( 5.3953916)),
+    LabeledVector(403.808709, DenseVector(80.7617418)),
+    LabeledVector(203.932158, DenseVector(40.7864316)),
+    LabeledVector(146.974958, DenseVector(29.3949916)),
+    LabeledVector( 46.869291, DenseVector( 9.3738582)),
+    LabeledVector(450.780834, DenseVector(90.1561667)),
+    LabeledVector(386.535619, DenseVector(77.3071239)),
+    LabeledVector(202.644342, DenseVector(40.5288684)),
+    LabeledVector(227.586507, DenseVector(45.5173013)),
+    LabeledVector(408.801080, DenseVector(81.7602161)),
+    LabeledVector(146.118550, DenseVector(29.2237100)),
+    LabeledVector(156.475382, DenseVector(31.2950763)),
+    LabeledVector(291.822515, DenseVector(58.3645030)),
+    LabeledVector( 61.506887, DenseVector(12.3013775)),
+    LabeledVector(363.949913, DenseVector(72.7899827)),
+    LabeledVector(398.050744, DenseVector(79.6101487)),
+    LabeledVector(246.053111, DenseVector(49.2106221)),
+    LabeledVector(225.494661, DenseVector(45.0989323)),
+    LabeledVector(265.986844, DenseVector(53.1973689)),
+    LabeledVector(110.459912, DenseVector(22.0919823)),
+    LabeledVector(122.716974, DenseVector(24.5433947)),
+    LabeledVector(128.014314, DenseVector(25.6028628)),
+    LabeledVector(252.538913, DenseVector(50.5077825)),
+    LabeledVector(393.632082, DenseVector(78.7264163)),
+    LabeledVector( 77.698941, DenseVector(15.5397881)),
+    LabeledVector(206.187568, DenseVector(41.2375135)),
+    LabeledVector(244.073426, DenseVector(48.8146851)),
+    LabeledVector(364.946890, DenseVector(72.9893780)),
+    LabeledVector(  4.627494, DenseVector( 0.9254987)),
+    LabeledVector(485.359565, DenseVector(97.0719130)),
+    LabeledVector(347.359190, DenseVector(69.4718380)),
+    LabeledVector(419.663211, DenseVector(83.9326422)),
+    LabeledVector(488.518318, DenseVector(97.7036635)),
+    LabeledVector( 28.082962, DenseVector( 5.6165925)),
+    LabeledVector(211.002441, DenseVector(42.2004881)),
+    LabeledVector(250.624124, DenseVector(50.1248248)),
+    LabeledVector(489.776669, DenseVector(97.9553337))
+  )
+
+
+  val expectedPolynomialWeights = Seq(0.2375, -0.3493, -0.1674)
+  val expectedPolynomialWeight0 = 0.0233
+  val expectedPolynomialSquaredResidualSum = 1.5389e+03/2
+
+  val polynomialData: Seq[LabeledVector] = Seq(
+    LabeledVector(2.1415, DenseVector(3.6663)),
+    LabeledVector(10.9835, DenseVector(4.0761)),
+    LabeledVector(7.2507, DenseVector(0.5714)),
+    LabeledVector(11.9274, DenseVector(4.1102)),
+    LabeledVector(-4.2798, DenseVector(2.8456)),
+    LabeledVector(7.1929, DenseVector(0.4389)),
+    LabeledVector(4.5097, DenseVector(1.2532)),
+    LabeledVector(-3.6059, DenseVector(2.4610)),
+    LabeledVector(18.1132, DenseVector(4.3088)),
+    LabeledVector(19.2674, DenseVector(4.3420)),
+    LabeledVector(7.0664, DenseVector(0.7093)),
+    LabeledVector(20.1836, DenseVector(4.3677)),
+    LabeledVector(18.0609, DenseVector(4.3073)),
+    LabeledVector(-2.2090, DenseVector(2.1842)),
+    LabeledVector(1.1306, DenseVector(3.6013)),
+    LabeledVector(7.1903, DenseVector(0.6385)),
+    LabeledVector(-0.2668, DenseVector(1.8979)),
+    LabeledVector(12.2281, DenseVector(4.1208)),
+    LabeledVector(0.6086, DenseVector(3.5649)),
+    LabeledVector(18.4202, DenseVector(4.3177)),
+    LabeledVector(-4.1284, DenseVector(2.9508)),
+    LabeledVector(6.1964, DenseVector(0.1607)),
+    LabeledVector(4.9638, DenseVector(3.8211)),
+    LabeledVector(14.6677, DenseVector(4.2030)),
+    LabeledVector(-3.8132, DenseVector(3.0543)),
+    LabeledVector(-1.2891, DenseVector(3.4098)),
+    LabeledVector(-1.9390, DenseVector(3.3441)),
+    LabeledVector(0.7293, DenseVector(1.7650)),
+    LabeledVector(-4.1310, DenseVector(2.9497)),
+    LabeledVector(6.9131, DenseVector(0.7703)),
+    LabeledVector(-3.2060, DenseVector(3.1772)),
+    LabeledVector(6.0899, DenseVector(0.1432)),
+    LabeledVector(4.5567, DenseVector(1.2462)),
+    LabeledVector(6.4562, DenseVector(0.2078)),
+    LabeledVector(7.1903, DenseVector(0.4371)),
+    LabeledVector(2.8017, DenseVector(3.7056)),
+    LabeledVector(-3.4873, DenseVector(3.1267)),
+    LabeledVector(3.2918, DenseVector(1.4269)),
+    LabeledVector(17.0085, DenseVector(4.2760)),
+    LabeledVector(6.1622, DenseVector(0.1550)),
+    LabeledVector(-0.8192, DenseVector(1.9743)),
+    LabeledVector(1.0957, DenseVector(1.7170)),
+    LabeledVector(-0.9065, DenseVector(3.4448)),
+    LabeledVector(0.7986, DenseVector(3.5784)),
+    LabeledVector(6.6861, DenseVector(0.8409)),
+    LabeledVector(-2.3274, DenseVector(2.2039)),
+    LabeledVector(-1.0359, DenseVector(2.0051)),
+    LabeledVector(-4.2092, DenseVector(2.9084)),
+    LabeledVector(-3.1140, DenseVector(3.1921)),
+    LabeledVector(-1.4323, DenseVector(3.3961))
+  )
+
+  val expectedRegWeights = Array[Double](0.0, 0.0, 0.0, 0.18, 0.2, 0.24)
+  val expectedRegWeight0 = 0.74
+
+  // Example values from scikit-learn L1 test: http://git.io/vf4V2
+  val regularizationData: Seq[LabeledVector] = Seq(
+    LabeledVector(1.0, DenseVector(1.0,0.9 ,0.8 ,0.0 ,0.0 ,0.0)),
+    LabeledVector(1.0, DenseVector(1.0,0.84,0.98,0.0 ,0.0 ,0.0)),
+    LabeledVector(1.0, DenseVector(1.0,0.96,0.88,0.0 ,0.0 ,0.0)),
+    LabeledVector(1.0, DenseVector(1.0,0.91,0.99,0.0 ,0.0 ,0.0)),
+    LabeledVector(2.0, DenseVector(0.0,0.0 ,0.0 ,0.89,0.91,1.0)),
+    LabeledVector(2.0, DenseVector(0.0,0.0 ,0.0 ,0.79,0.84,1.0)),
+    LabeledVector(2.0, DenseVector(0.0,0.0 ,0.0 ,0.91,0.95,1.0)),
+    LabeledVector(2.0, DenseVector(0.0,0.0 ,0.0 ,0.93,1.0 ,1.0))
+  )
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/pom.xml
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/pom.xml b/flink-libraries/flink-table/pom.xml
new file mode 100644
index 0000000..19951cb
--- /dev/null
+++ b/flink-libraries/flink-table/pom.xml
@@ -0,0 +1,258 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-libraries</artifactId>
+		<version>1.0-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<artifactId>flink-table</artifactId>
+	<name>flink-table</name>
+
+	<packaging>jar</packaging>
+
+	<dependencies>
+
+		<dependency>
+			<groupId>com.google.guava</groupId>
+			<artifactId>guava</artifactId>
+			<version>${guava.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-scala</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-scala</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-examples-batch</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.scala-lang</groupId>
+			<artifactId>scala-reflect</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.scala-lang</groupId>
+			<artifactId>scala-library</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.scala-lang</groupId>
+			<artifactId>scala-compiler</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-test-utils</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.codehaus.janino</groupId>
+			<artifactId>janino</artifactId>
+			<version>2.7.5</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-tests</artifactId>
+			<version>${project.version}</version>
+			<type>test-jar</type>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+			<version>${jackson.version}</version>
+		</dependency>
+
+	</dependencies>
+
+	<build>
+		<plugins>
+			<!-- Scala Compiler -->
+			<plugin>
+				<groupId>net.alchim31.maven</groupId>
+				<artifactId>scala-maven-plugin</artifactId>
+				<version>3.1.4</version>
+				<executions>
+					<!-- Run scala compiler in the process-resources phase, so that dependencies on
+						scala classes can be resolved later in the (Java) compile phase -->
+					<execution>
+						<id>scala-compile-first</id>
+						<phase>process-resources</phase>
+						<goals>
+							<goal>compile</goal>
+						</goals>
+					</execution>
+
+					<!-- Run scala compiler in the process-test-resources phase, so that dependencies on
+						 scala classes can be resolved later in the (Java) test-compile phase -->
+					<execution>
+						<id>scala-test-compile</id>
+						<phase>process-test-resources</phase>
+						<goals>
+							<goal>testCompile</goal>
+						</goals>
+					</execution>
+				</executions>
+				<configuration>
+					<jvmArgs>
+						<jvmArg>-Xms128m</jvmArg>
+						<jvmArg>-Xmx512m</jvmArg>
+					</jvmArgs>
+					<compilerPlugins combine.children="append">
+						<compilerPlugin>
+							<groupId>org.scalamacros</groupId>
+							<artifactId>paradise_${scala.version}</artifactId>
+							<version>${scala.macros.version}</version>
+						</compilerPlugin>
+					</compilerPlugins>
+				</configuration>
+			</plugin>
+
+			<!-- Eclipse Integration -->
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-eclipse-plugin</artifactId>
+				<version>2.8</version>
+				<configuration>
+					<downloadSources>true</downloadSources>
+					<projectnatures>
+						<projectnature>org.scala-ide.sdt.core.scalanature</projectnature>
+						<projectnature>org.eclipse.jdt.core.javanature</projectnature>
+					</projectnatures>
+					<buildcommands>
+						<buildcommand>org.scala-ide.sdt.core.scalabuilder</buildcommand>
+					</buildcommands>
+					<classpathContainers>
+						<classpathContainer>org.scala-ide.sdt.launching.SCALA_CONTAINER</classpathContainer>
+						<classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
+					</classpathContainers>
+					<excludes>
+						<exclude>org.scala-lang:scala-library</exclude>
+						<exclude>org.scala-lang:scala-compiler</exclude>
+					</excludes>
+					<sourceIncludes>
+						<sourceInclude>**/*.scala</sourceInclude>
+						<sourceInclude>**/*.java</sourceInclude>
+					</sourceIncludes>
+				</configuration>
+			</plugin>
+
+			<!-- Adding scala source directories to build path -->
+			<plugin>
+				<groupId>org.codehaus.mojo</groupId>
+				<artifactId>build-helper-maven-plugin</artifactId>
+				<version>1.7</version>
+				<executions>
+					<!-- Add src/main/scala to eclipse build path -->
+					<execution>
+						<id>add-source</id>
+						<phase>generate-sources</phase>
+						<goals>
+							<goal>add-source</goal>
+						</goals>
+						<configuration>
+							<sources>
+								<source>src/main/scala</source>
+							</sources>
+						</configuration>
+					</execution>
+					<!-- Add src/test/scala to eclipse build path -->
+					<execution>
+						<id>add-test-source</id>
+						<phase>generate-test-sources</phase>
+						<goals>
+							<goal>add-test-source</goal>
+						</goals>
+						<configuration>
+							<sources>
+								<source>src/test/scala</source>
+							</sources>
+						</configuration>
+					</execution>
+				</executions>
+			</plugin>
+
+			<plugin>
+				<groupId>org.scalastyle</groupId>
+				<artifactId>scalastyle-maven-plugin</artifactId>
+				<version>0.5.0</version>
+				<executions>
+					<execution>
+						<goals>
+							<goal>check</goal>
+						</goals>
+					</execution>
+				</executions>
+				<configuration>
+					<verbose>false</verbose>
+					<failOnViolation>true</failOnViolation>
+					<includeTestSourceDirectory>true</includeTestSourceDirectory>
+					<failOnWarning>false</failOnWarning>
+					<sourceDirectory>${basedir}/src/main/scala</sourceDirectory>
+					<testSourceDirectory>${basedir}/src/test/scala</testSourceDirectory>
+					<configLocation>${project.basedir}/../../tools/maven/scalastyle-config.xml</configLocation>
+					<outputFile>${project.basedir}/scalastyle-output.xml</outputFile>
+					<outputEncoding>UTF-8</outputEncoding>
+				</configuration>
+			</plugin>
+
+		</plugins>
+	</build>
+
+	<profiles>
+		<profile>
+			<id>scala-2.10</id>
+			<activation>
+				<property>
+					<!-- this is the default scala profile -->
+					<name>!scala-2.11</name>
+				</property>
+			</activation>
+			<dependencies>
+				<dependency>
+					<groupId>org.scalamacros</groupId>
+					<artifactId>quasiquotes_${scala.binary.version}</artifactId>
+					<version>${scala.macros.version}</version>
+				</dependency>
+			</dependencies>
+		</profile>
+	</profiles>
+
+</project>

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/java/org/apache/flink/api/java/table/package-info.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/api/java/table/package-info.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/java/table/package-info.java
new file mode 100644
index 0000000..97113bb
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/java/table/package-info.java
@@ -0,0 +1,60 @@
+/*
+ * 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 (Java)</strong><br>
+ *
+ * {@link org.apache.flink.api.java.table.TableEnvironment} can be used to create a
+ * {@link org.apache.flink.api.table.Table} from a {@link org.apache.flink.api.java.DataSet}
+ * or {@link org.apache.flink.streaming.api.datastream.DataStream}.
+ *
+ * <p>
+ * This can be used to perform SQL-like queries on data. Please have
+ * a look at {@link org.apache.flink.api.table.Table} to see which operations are supported and
+ * how query strings are written.
+ *
+ * <p>
+ * Example:
+ *
+ * <pre>{@code
+ * ExecutionEnvironment env = ExecutionEnvironment.createCollectionsEnvironment();
+ *
+ * DataSet<WC> input = env.fromElements(
+ *   new WC("Hello", 1),
+ *   new WC("Ciao", 1),
+ *   new WC("Hello", 1));
+ *
+ * Table table = TableUtil.from(input);
+ *
+ * Table filtered = table
+ *     .groupBy("word")
+ *     .select("word.count as count, word")
+ *     .filter("count = 2");
+ *
+ * DataSet<WC> result = TableUtil.toSet(filtered, WC.class);
+ *
+ * result.print();
+ * env.execute();
+ * }</pre>
+ *
+ * <p>
+ * As seen above, a {@link org.apache.flink.api.table.Table} can be converted back to the
+ * underlying API representation using {@link org.apache.flink.api.java.table.TableEnvironment#toDataSet(org.apache.flink.api.table.Table, java.lang.Class)}
+ * or {@link org.apache.flink.api.java.table.TableEnvironment#toDataStream(org.apache.flink.api.table.Table, java.lang.Class)}}.
+ */
+package org.apache.flink.api.java.table;

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/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
new file mode 100644
index 0000000..d7fbc8e
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/java/org/apache/flink/api/table/package-info.java
@@ -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.
+ */
+
+/**
+ * <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/e9bf13d8/flink-libraries/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java b/flink-libraries/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
new file mode 100644
index 0000000..c043508
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/java/org/apache/flink/examples/java/JavaTableExample.java
@@ -0,0 +1,71 @@
+/*
+ * 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.examples.java;
+
+
+import org.apache.flink.api.table.Table;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.table.TableEnvironment;
+
+/**
+ * Very simple example that shows how the Java Table API can be used.
+ */
+public class JavaTableExample {
+
+	public static class WC {
+		public String word;
+		public int count;
+
+		// Public constructor to make it a Flink POJO
+		public WC() {
+
+		}
+
+		public WC(String word, int count) {
+			this.word = word;
+			this.count = count;
+		}
+
+		@Override
+		public String toString() {
+			return "WC " + word + " " + count;
+		}
+	}
+
+	public static void main(String[] args) throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.createCollectionsEnvironment();
+		TableEnvironment tableEnv = new TableEnvironment();
+
+		DataSet<WC> input = env.fromElements(
+				new WC("Hello", 1),
+				new WC("Ciao", 1),
+				new WC("Hello", 1));
+
+		Table table = tableEnv.fromDataSet(input);
+
+		Table filtered = table
+				.groupBy("word")
+				.select("word.count as count, word")
+				.filter("count = 2");
+
+		DataSet<WC> result = tableEnv.toDataSet(filtered, WC.class);
+
+		result.print();
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/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
new file mode 100644
index 0000000..9dc9297
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaBatchTranslator.scala
@@ -0,0 +1,346 @@
+/*
+ * 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.java.table
+
+import java.lang.reflect.Modifier
+
+import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.aggregation.AggregationFunction
+import org.apache.flink.api.java.operators.JoinOperator.EquiJoin
+import org.apache.flink.api.java.operators.Keys.ExpressionKeys
+import org.apache.flink.api.java.operators.{GroupReduceOperator, Keys, MapOperator, UnsortedGrouping}
+import org.apache.flink.api.java.{DataSet => JavaDataSet}
+import org.apache.flink.api.table.expressions.analysis.ExtractEquiJoinFields
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.table.runtime._
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.typeinfo.{RenameOperator, RenamingProxyTypeInfo, RowTypeInfo}
+import org.apache.flink.api.table.{ExpressionException, Row, Table}
+
+/**
+ * [[PlanTranslator]] for creating [[Table]]s from Java [[org.apache.flink.api.java.DataSet]]s and
+ * translating them back to Java [[org.apache.flink.api.java.DataSet]]s.
+ */
+class JavaBatchTranslator extends PlanTranslator {
+
+  type Representation[A] = JavaDataSet[A]
+
+  override def createTable[A](
+      repr: Representation[A],
+      inputType: CompositeType[A],
+      expressions: Array[Expression],
+      resultFields: Seq[(String, TypeInformation[_])]): Table = {
+
+    val rowDataSet = createSelect(expressions, repr, inputType)
+
+    Table(Root(rowDataSet, resultFields))
+  }
+
+  override def translate[A](op: PlanNode)(implicit tpe: TypeInformation[A]): JavaDataSet[A] = {
+
+    if (tpe.getTypeClass == classOf[Row]) {
+      // shortcut for DataSet[Row]
+      return translateInternal(op).asInstanceOf[JavaDataSet[A]]
+    }
+
+    val clazz = tpe.getTypeClass
+    if (clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) {
+      throw new ExpressionException("Cannot create DataSet of type " +
+        clazz.getName + ". Only top-level classes or static member classes are supported.")
+    }
+
+
+    if (!implicitly[TypeInformation[A]].isInstanceOf[CompositeType[A]]) {
+      throw new ExpressionException(
+        "A Table can only be converted to composite types, type is: " +
+          implicitly[TypeInformation[A]] +
+          ". Composite types would be tuples, case classes and POJOs.")
+    }
+
+    val resultSet = translateInternal(op)
+
+    val resultType = resultSet.getType.asInstanceOf[RowTypeInfo]
+
+    val outputType = implicitly[TypeInformation[A]].asInstanceOf[CompositeType[A]]
+
+    val resultNames = resultType.getFieldNames
+    val outputNames = outputType.getFieldNames.toSeq
+
+    if (resultNames.toSet != outputNames.toSet) {
+      throw new ExpressionException(s"Expression result type $resultType does not have the same " +
+        s"fields as output type $outputType")
+    }
+
+    for (f <- outputNames) {
+      val in = resultType.getTypeAt(resultType.getFieldIndex(f))
+      val out = outputType.getTypeAt(outputType.getFieldIndex(f))
+      if (!in.equals(out)) {
+        throw new ExpressionException(s"Types for field $f differ on input $resultType and " +
+          s"output $outputType.")
+      }
+    }
+
+    val outputFields = outputNames map {
+      f => ResolvedFieldReference(f, resultType.getTypeAt(f))
+    }
+
+    val function = new ExpressionSelectFunction(
+      resultSet.getType.asInstanceOf[RowTypeInfo],
+      outputType,
+      outputFields)
+
+    val opName = s"select(${outputFields.mkString(",")})"
+    val operator = new MapOperator(resultSet, outputType, function, opName)
+
+    operator
+  }
+
+  private def translateInternal(op: PlanNode): JavaDataSet[Row] = {
+    op match {
+      case Root(dataSet: JavaDataSet[Row], resultFields) =>
+        dataSet
+
+      case Root(_, _) =>
+        throw new ExpressionException("Invalid Root for JavaBatchTranslator: " + op + ". " +
+          "Did you try converting a Table based on a DataSet to a DataStream or vice-versa?")
+
+      case GroupBy(_, fields) =>
+        throw new ExpressionException("Dangling GroupBy operation. Did you forget a " +
+          "SELECT statement?")
+
+      case As(input, newNames) =>
+        val translatedInput = translateInternal(input)
+        val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
+        val proxyType = new RenamingProxyTypeInfo[Row](inType, newNames.toArray)
+        new RenameOperator(translatedInput, proxyType)
+
+      case sel@Select(Filter(Join(leftInput, rightInput), predicate), selection) =>
+
+        val expandedInput = ExpandAggregations(sel)
+
+        if (expandedInput.eq(sel)) {
+          val translatedLeftInput = translateInternal(leftInput)
+          val translatedRightInput = translateInternal(rightInput)
+          val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
+          val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
+
+          createJoin(
+            predicate,
+            selection,
+            translatedLeftInput,
+            translatedRightInput,
+            leftInType,
+            rightInType,
+            JoinHint.OPTIMIZER_CHOOSES)
+        } else {
+          translateInternal(expandedInput)
+        }
+
+      case Filter(Join(leftInput, rightInput), predicate) =>
+        val translatedLeftInput = translateInternal(leftInput)
+        val translatedRightInput = translateInternal(rightInput)
+        val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
+        val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
+
+        createJoin(
+          predicate,
+          leftInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)) ++
+            rightInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)),
+          translatedLeftInput,
+          translatedRightInput,
+          leftInType,
+          rightInType,
+          JoinHint.OPTIMIZER_CHOOSES)
+
+      case Join(leftInput, rightInput) =>
+        throw new ExpressionException("Join without filter condition encountered. " +
+          "Did you forget to add .where(...) ?")
+
+      case sel@Select(input, selection) =>
+
+        val expandedInput = ExpandAggregations(sel)
+
+        if (expandedInput.eq(sel)) {
+          val translatedInput = input match {
+            case GroupBy(groupByInput, groupExpressions) =>
+              val translatedGroupByInput = translateInternal(groupByInput)
+              val inType = translatedGroupByInput.getType.asInstanceOf[CompositeType[Row]]
+
+              val keyIndices = groupExpressions map {
+                case fe: ResolvedFieldReference => inType.getFieldIndex(fe.name)
+                case e =>
+                  throw new ExpressionException(s"Expression $e is not a valid key expression.")
+              }
+
+              val keys = new Keys.ExpressionKeys(keyIndices.toArray, inType, false)
+              val grouping = new UnsortedGrouping(translatedGroupByInput, keys)
+
+              new GroupReduceOperator(
+                grouping,
+                inType,
+                new NoExpressionAggregateFunction(),
+                "Nop Expression Aggregation")
+
+            case _ => translateInternal(input)
+          }
+
+          val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
+          val inputFields = inType.getFieldNames
+          createSelect(
+            selection,
+            translatedInput,
+            inType)
+        } else {
+          translateInternal(expandedInput)
+        }
+
+      case agg@Aggregate(GroupBy(input, groupExpressions), aggregations) =>
+        val translatedInput = translateInternal(input)
+        val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
+
+        val keyIndices = groupExpressions map {
+          case fe: ResolvedFieldReference => inType.getFieldIndex(fe.name)
+          case e => throw new ExpressionException(s"Expression $e is not a valid key expression.")
+        }
+
+        val keys = new Keys.ExpressionKeys(keyIndices.toArray, inType, false)
+
+        val grouping = new UnsortedGrouping(translatedInput, keys)
+
+        val aggFunctions: Seq[AggregationFunction[Any]] = aggregations map {
+          case (fieldName, fun) =>
+            fun.getFactory.createAggregationFunction[Any](
+              inType.getTypeAt[Any](inType.getFieldIndex(fieldName)).getTypeClass)
+        }
+
+        val aggIndices = aggregations map {
+          case (fieldName, _) =>
+            inType.getFieldIndex(fieldName)
+        }
+
+        val result = new GroupReduceOperator(
+          grouping,
+          inType,
+          new ExpressionAggregateFunction(aggIndices, aggFunctions),
+          "Expression Aggregation: " + agg)
+
+        result
+
+      case agg@Aggregate(input, aggregations) =>
+        val translatedInput = translateInternal(input)
+        val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
+
+        val aggFunctions: Seq[AggregationFunction[Any]] = aggregations map {
+          case (fieldName, fun) =>
+            fun.getFactory.createAggregationFunction[Any](
+              inType.getTypeAt[Any](inType.getFieldIndex(fieldName)).getTypeClass)
+        }
+
+        val aggIndices = aggregations map {
+          case (fieldName, _) =>
+            inType.getFieldIndex(fieldName)
+        }
+
+        val result = new GroupReduceOperator(
+          translatedInput,
+          inType,
+          new ExpressionAggregateFunction(aggIndices, aggFunctions),
+          "Expression Aggregation: " + agg)
+
+        result
+
+
+      case Filter(input, predicate) =>
+        val translatedInput = translateInternal(input)
+        val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
+        val filter = new ExpressionFilterFunction[Row](predicate, inType)
+        translatedInput.filter(filter).name(predicate.toString)
+
+      case uni@UnionAll(left, right) =>
+        val translatedLeft = translateInternal(left)
+        val translatedRight = translateInternal(right)
+        translatedLeft.union(translatedRight).name("Union: " + uni)
+    }
+  }
+
+  private def createSelect[I](
+      fields: Seq[Expression],
+      input: JavaDataSet[I],
+      inputType: CompositeType[I]): JavaDataSet[Row] = {
+
+    fields foreach {
+      f =>
+        if (f.exists(_.isInstanceOf[Aggregation])) {
+          throw new ExpressionException("Found aggregate in " + fields.mkString(", ") + ".")
+        }
+
+    }
+
+    val resultType = new RowTypeInfo(fields)
+
+    val function = new ExpressionSelectFunction(inputType, resultType, fields)
+
+    val opName = s"select(${fields.mkString(",")})"
+    val operator = new MapOperator(input, resultType, function, opName)
+
+    operator
+  }
+
+  private def createJoin[L, R](
+      predicate: Expression,
+      fields: Seq[Expression],
+      leftInput: JavaDataSet[L],
+      rightInput: JavaDataSet[R],
+      leftType: CompositeType[L],
+      rightType: CompositeType[R],
+      joinHint: JoinHint): JavaDataSet[Row] = {
+
+    val resultType = new RowTypeInfo(fields)
+
+    val (reducedPredicate, leftFields, rightFields) =
+      ExtractEquiJoinFields(leftType, rightType, predicate)
+
+    if (leftFields.isEmpty || rightFields.isEmpty) {
+      throw new ExpressionException("Could not derive equi-join predicates " +
+        "for predicate " + predicate + ".")
+    }
+
+    val leftKey = new ExpressionKeys[L](leftFields, leftType)
+    val rightKey = new ExpressionKeys[R](rightFields, rightType)
+
+    val joiner = new ExpressionJoinFunction[L, R, Row](
+      reducedPredicate,
+      leftType,
+      rightType,
+      resultType,
+      fields)
+
+    new EquiJoin[L, R, Row](
+      leftInput,
+      rightInput,
+      leftKey,
+      rightKey,
+      joiner,
+      resultType,
+      joinHint,
+      predicate.toString)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala
new file mode 100644
index 0000000..a37c892
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/JavaStreamingTranslator.scala
@@ -0,0 +1,241 @@
+/*
+ * 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.java.table
+
+import java.lang.reflect.Modifier
+import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.table.runtime.{ExpressionFilterFunction, ExpressionSelectFunction}
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.table.typeinfo.RowTypeInfo
+import org.apache.flink.api.table.{ExpressionException, Row, Table}
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.operators.StreamMap
+
+/**
+ * [[PlanTranslator]] for creating [[Table]]s from Java [[DataStream]]s and
+ * translating them back to Java [[DataStream]]s.
+ *
+ * This is very limited right now. Only select and filter are implemented. Also, the expression
+ * operations must be extended to allow windowing operations.
+ */
+
+class JavaStreamingTranslator extends PlanTranslator {
+
+  type Representation[A] = DataStream[A]
+
+  override def createTable[A](
+      repr: Representation[A],
+      inputType: CompositeType[A],
+      expressions: Array[Expression],
+      resultFields: Seq[(String, TypeInformation[_])]): Table = {
+
+    val rowDataStream = createSelect(expressions, repr, inputType)
+
+    new Table(Root(rowDataStream, resultFields))
+  }
+
+  override def translate[A](op: PlanNode)(implicit tpe: TypeInformation[A]): DataStream[A] = {
+
+    if (tpe.getTypeClass == classOf[Row]) {
+      // shortcut for DataSet[Row]
+      return translateInternal(op).asInstanceOf[DataStream[A]]
+    }
+
+    val clazz = tpe.getTypeClass
+    if (clazz.isMemberClass && !Modifier.isStatic(clazz.getModifiers)) {
+      throw new ExpressionException("Cannot create DataStream of type " +
+        clazz.getName + ". Only top-level classes or static member classes are supported.")
+    }
+
+    if (!implicitly[TypeInformation[A]].isInstanceOf[CompositeType[A]]) {
+      throw new ExpressionException(
+        "A Table can only be converted to composite types, type is: " +
+          implicitly[TypeInformation[A]] +
+          ". Composite types would be tuples, case classes and POJOs.")
+
+    }
+
+    val resultSet = translateInternal(op)
+
+    val resultType = resultSet.getType.asInstanceOf[RowTypeInfo]
+
+    val outputType = implicitly[TypeInformation[A]].asInstanceOf[CompositeType[A]]
+
+    val resultNames = resultType.getFieldNames
+    val outputNames = outputType.getFieldNames.toSeq
+
+    if (resultNames.toSet != outputNames.toSet) {
+      throw new ExpressionException(s"Expression result type $resultType does not have the same" +
+        s"fields as output type $outputType")
+    }
+
+    for (f <- outputNames) {
+      val in = resultType.getTypeAt(resultType.getFieldIndex(f))
+      val out = outputType.getTypeAt(outputType.getFieldIndex(f))
+      if (!in.equals(out)) {
+        throw new ExpressionException(s"Types for field $f differ on input $resultType and " +
+          s"output $outputType.")
+      }
+    }
+
+    val outputFields = outputNames map {
+      f => ResolvedFieldReference(f, resultType.getTypeAt(f))
+    }
+
+    val function = new ExpressionSelectFunction(
+      resultSet.getType.asInstanceOf[RowTypeInfo],
+      outputType,
+      outputFields)
+
+    val opName = s"select(${outputFields.mkString(",")})"
+
+    resultSet.transform(opName, outputType, new StreamMap[Row, A](function))
+  }
+
+  private def translateInternal(op: PlanNode): DataStream[Row] = {
+    op match {
+      case Root(dataSet: DataStream[Row], resultFields) =>
+        dataSet
+
+      case Root(_, _) =>
+        throw new ExpressionException("Invalid Root for JavaStreamingTranslator: " + op + ". " +
+          "Did you try converting a Table based on a DataSet to a DataStream or vice-versa?")
+
+      case GroupBy(_, fields) =>
+        throw new ExpressionException("Dangling GroupBy operation. Did you forget a " +
+          "SELECT statement?")
+
+      case As(input, newNames) =>
+        throw new ExpressionException("As operation for Streams not yet implemented.")
+
+      case sel@Select(Filter(Join(leftInput, rightInput), predicate), selection) =>
+
+        val expandedInput = ExpandAggregations(sel)
+
+        if (expandedInput.eq(sel)) {
+          val translatedLeftInput = translateInternal(leftInput)
+          val translatedRightInput = translateInternal(rightInput)
+          val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
+          val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
+
+          createJoin(
+            predicate,
+            selection,
+            translatedLeftInput,
+            translatedRightInput,
+            leftInType,
+            rightInType,
+            JoinHint.OPTIMIZER_CHOOSES)
+        } else {
+          translateInternal(expandedInput)
+        }
+
+      case Filter(Join(leftInput, rightInput), predicate) =>
+        val translatedLeftInput = translateInternal(leftInput)
+        val translatedRightInput = translateInternal(rightInput)
+        val leftInType = translatedLeftInput.getType.asInstanceOf[CompositeType[Row]]
+        val rightInType = translatedRightInput.getType.asInstanceOf[CompositeType[Row]]
+
+        createJoin(
+          predicate,
+          leftInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)) ++
+            rightInput.outputFields.map( f => ResolvedFieldReference(f._1, f._2)),
+          translatedLeftInput,
+          translatedRightInput,
+          leftInType,
+          rightInType,
+          JoinHint.OPTIMIZER_CHOOSES)
+
+      case Join(leftInput, rightInput) =>
+        throw new ExpressionException("Join without filter condition encountered. " +
+          "Did you forget to add .where(...) ?")
+
+      case sel@Select(input, selection) =>
+
+        val expandedInput = ExpandAggregations(sel)
+
+        if (expandedInput.eq(sel)) {
+          // no expansions took place
+          val translatedInput = translateInternal(input)
+          val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
+          val inputFields = inType.getFieldNames
+          createSelect(
+            selection,
+            translatedInput,
+            inType)
+        } else {
+          translateInternal(expandedInput)
+        }
+
+      case agg@Aggregate(GroupBy(input, groupExpressions), aggregations) =>
+        throw new ExpressionException("Aggregate operation for Streams not yet implemented.")
+
+      case agg@Aggregate(input, aggregations) =>
+        throw new ExpressionException("Aggregate operation for Streams not yet implemented.")
+
+      case Filter(input, predicate) =>
+        val translatedInput = translateInternal(input)
+        val inType = translatedInput.getType.asInstanceOf[CompositeType[Row]]
+        val filter = new ExpressionFilterFunction[Row](predicate, inType)
+        translatedInput.filter(filter)
+
+      case UnionAll(left, right) =>
+        val translatedLeft = translateInternal(left)
+        val translatedRight = translateInternal(right)
+        translatedLeft.union(translatedRight)
+    }
+  }
+
+  private def createSelect[I](
+      fields: Seq[Expression],
+      input: DataStream[I],
+      inputType: CompositeType[I]): DataStream[Row] = {
+
+    fields foreach {
+      f =>
+        if (f.exists(_.isInstanceOf[Aggregation])) {
+          throw new ExpressionException("Found aggregate in " + fields.mkString(", ") + ".")
+        }
+
+    }
+
+    val resultType = new RowTypeInfo(fields)
+
+    val function = new ExpressionSelectFunction(inputType, resultType, fields)
+
+    val opName = s"select(${fields.mkString(",")})"
+
+    input.transform(opName, resultType, new StreamMap[I, Row](function))
+  }
+
+  private def createJoin[L, R](
+      predicate: Expression,
+      fields: Seq[Expression],
+      leftInput: DataStream[L],
+      rightInput: DataStream[R],
+      leftType: CompositeType[L],
+      rightType: CompositeType[R],
+      joinHint: JoinHint): DataStream[Row] = {
+
+    throw new ExpressionException("Join operation for Streams not yet implemented.")
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
new file mode 100644
index 0000000..5614031
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/java/table/TableEnvironment.scala
@@ -0,0 +1,111 @@
+/*
+ * 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.java.table
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.DataSet
+import org.apache.flink.api.java.typeutils.TypeExtractor
+import org.apache.flink.api.table.Table
+import org.apache.flink.streaming.api.datastream.DataStream
+
+/**
+ * Environment for working with the Table API.
+ *
+ * This can be used to convert [[DataSet]] or [[DataStream]] to a [[Table]] and back again. You
+ * can also use the provided methods to create a [[Table]] directly from a data source.
+ */
+class TableEnvironment {
+
+  /**
+   * Transforms the given DataSet to a [[org.apache.flink.api.table.Table]].
+   * The fields of the DataSet type are renamed to the given set of fields:
+   *
+   * Example:
+   *
+   * {{{
+   *   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 fromDataSet[T](set: DataSet[T], fields: String): Table = {
+    new JavaBatchTranslator().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
+   * [[org.apache.flink.api.table.Table]] fields.
+   */
+  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 DataStream type are renamed to the given set of fields:
+   *
+   * Example:
+   *
+   * {{{
+   *   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 fromDataStream[T](set: DataStream[T], fields: String): Table = {
+    new JavaStreamingTranslator().createTable(set, fields)
+  }
+
+  /**
+   * 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 fromDataStream[T](set: DataStream[T]): Table = {
+    new JavaStreamingTranslator().createTable(set)
+  }
+
+  /**
+   * Converts the given [[org.apache.flink.api.table.Table]] to
+   * a DataSet. The given type must have exactly the same fields as the
+   * [[org.apache.flink.api.table.Table]]. That is, the names of the
+   * fields and the types must match.
+   */
+  @SuppressWarnings(Array("unchecked"))
+  def toDataSet[T](table: Table, clazz: Class[T]): DataSet[T] = {
+    new JavaBatchTranslator().translate[T](table.operation)(
+      TypeExtractor.createTypeInfo(clazz).asInstanceOf[TypeInformation[T]])
+  }
+
+  /**
+   * Converts the given [[org.apache.flink.api.table.Table]] to
+   * a DataStream. The given type must have exactly the same fields as the
+   * [[org.apache.flink.api.table.Table]]. That is, the names of the
+   * fields and the types must match.
+   */
+  @SuppressWarnings(Array("unchecked"))
+  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/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala
new file mode 100644
index 0000000..2508a3d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataSetConversions.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.scala.table
+
+import org.apache.flink.api.table._
+import org.apache.flink.api.table.expressions.{UnresolvedFieldReference, Expression}
+import org.apache.flink.api.common.typeutils.CompositeType
+
+import org.apache.flink.api.scala._
+
+/**
+ * Methods for converting a [[DataSet]] to a [[Table]]. A [[DataSet]] is
+ * wrapped in this by the implicit conversions in [[org.apache.flink.api.scala.table]].
+ */
+class DataSetConversions[T](set: DataSet[T], inputType: CompositeType[T]) {
+
+  /**
+   * Converts the [[DataSet]] to a [[Table]]. The field names can be specified like this:
+   *
+   * {{{
+   *   val in: DataSet[(String, Int)] = ...
+   *   val table = in.as('a, 'b)
+   * }}}
+   *
+   * This results in a [[Table]] that has field `a` of type `String` and field `b`
+   * of type `Int`.
+   */
+  def as(fields: Expression*): Table = {
+     new ScalaBatchTranslator().createTable(set, fields.toArray)
+  }
+
+  /**
+   * Converts the [[DataSet]] to a [[Table]]. The field names will be taken from the field names
+   * of the input type.
+   *
+   * Example:
+   *
+   * {{{
+   *   val in: DataSet[(String, Int)] = ...
+   *   val table = in.toTable
+   * }}}
+   *
+   * Here, the result is a [[Table]] that has field `_1` of type `String` and field `_2`
+   * of type `Int`.
+   */
+  def toTable: Table = {
+    val resultFields = inputType.getFieldNames.map(UnresolvedFieldReference)
+    as(resultFields: _*)
+  }
+
+}
+

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala
new file mode 100644
index 0000000..47bd100
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/DataStreamConversions.scala
@@ -0,0 +1,68 @@
+/*
+ * 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.scala.table
+
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.table._
+import org.apache.flink.api.table.expressions.{Expression, UnresolvedFieldReference}
+import org.apache.flink.streaming.api.scala.DataStream
+
+class DataStreamConversions[T](stream: DataStream[T], inputType: CompositeType[T]) {
+
+  /**
+   * Converts the [[DataStream]] to a [[Table]]. The field names can be specified like this:
+   *
+   * {{{
+   *   val in: DataSet[(String, Int)] = ...
+   *   val table = in.as('a, 'b)
+   * }}}
+   *
+   * This results in a [[Table]] that has field `a` of type `String` and field `b`
+   * of type `Int`.
+   */
+
+  def as(fields: Expression*): Table = {
+     new ScalaStreamingTranslator().createTable(
+       stream,
+       fields.toArray,
+       checkDeterministicFields = true)
+  }
+
+  /**
+   * Converts the [[DataStream]] to a [[Table]]. The field names will be taken from the field
+   * names of the input type.
+   *
+   * Example:
+   *
+   * {{{
+   *   val in: DataSet[(String, Int)] = ...
+   *   val table = in.toTable
+   * }}}
+   *
+   * This results in a [[Table]] that has field `_1` of type `String` and field `_2`
+   * of type `Int`.
+   */
+
+  def toTable: Table = {
+    val resultFields = inputType.getFieldNames.map(UnresolvedFieldReference)
+    as(resultFields: _*)
+  }
+
+}
+

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala
new file mode 100644
index 0000000..cdcf53e
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaBatchTranslator.scala
@@ -0,0 +1,68 @@
+/*
+ * 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.scala.table
+
+
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.table.JavaBatchTranslator
+import org.apache.flink.api.table.expressions.Expression
+import org.apache.flink.api.scala.wrap
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.table.Table
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.scala.DataSet
+
+import scala.reflect.ClassTag
+
+
+/**
+ * [[PlanTranslator]] for creating [[Table]]s from Scala [[DataSet]]s and
+ * translating them back to Scala [[DataSet]]s.
+ */
+class ScalaBatchTranslator extends PlanTranslator {
+
+  private val javaTranslator = new JavaBatchTranslator
+
+  type Representation[A] = DataSet[A]
+
+  def createTable[A](
+      repr: DataSet[A],
+      fields: Array[Expression]): Table = {
+
+    val result = javaTranslator.createTable(repr.javaSet, fields)
+
+    new Table(result.operation)
+  }
+
+  override def translate[O](op: PlanNode)(implicit tpe: TypeInformation[O]): DataSet[O] = {
+    // fake it till you make it ...
+    wrap(javaTranslator.translate(op))(ClassTag.AnyRef.asInstanceOf[ClassTag[O]])
+  }
+
+  override def createTable[A](
+      repr: Representation[A],
+      inputType: CompositeType[A],
+      expressions: Array[Expression],
+      resultFields: Seq[(String, TypeInformation[_])]): Table = {
+
+    val result = javaTranslator.createTable(repr.javaSet, inputType, expressions, resultFields)
+
+    Table(result.operation)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala
new file mode 100644
index 0000000..88f1b83
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/ScalaStreamingTranslator.scala
@@ -0,0 +1,58 @@
+/*
+ * 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.scala.table
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.table.JavaStreamingTranslator
+import org.apache.flink.api.table.Table
+import org.apache.flink.api.table.plan._
+import org.apache.flink.api.table.expressions.Expression
+import org.apache.flink.streaming.api.scala.{DataStream, javaToScalaStream}
+
+/**
+ * [[PlanTranslator]] for creating [[Table]]s from Scala [[DataStream]]s and
+ * translating them back to Scala [[DataStream]]s.
+ *
+ * This is very limited right now. Only select and filter are implemented. Also, the expression
+ * operations must be extended to allow windowing operations.
+ */
+class ScalaStreamingTranslator extends PlanTranslator {
+
+  private val javaTranslator = new JavaStreamingTranslator
+
+  override type Representation[A] = DataStream[A]
+
+  override def translate[O](op: PlanNode)(implicit tpe: TypeInformation[O]): DataStream[O] = {
+    // fake it till you make it ...
+    javaToScalaStream(javaTranslator.translate(op))
+  }
+
+  override def createTable[A](
+      repr: Representation[A],
+      inputType: CompositeType[A],
+      expressions: Array[Expression],
+      resultFields: Seq[(String, TypeInformation[_])]): Table = {
+
+    val result =
+      javaTranslator.createTable(repr.getJavaStream, inputType, expressions, resultFields)
+
+    new Table(result.operation)
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala
new file mode 100644
index 0000000..4f2172e
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/TableConversions.scala
@@ -0,0 +1,47 @@
+/*
+ * 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.scala.table
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.scala._
+import org.apache.flink.api.table._
+
+import org.apache.flink.streaming.api.scala.DataStream
+
+/**
+ * Methods for converting a [[Table]] to a [[DataSet]] or [[DataStream]]. A [[Table]] is
+ * wrapped in this by the implicit conversions in [[org.apache.flink.api.scala.table]].
+ */
+class TableConversions(table: Table) {
+
+  /**
+   * Converts the [[Table]] to a [[DataSet]].
+   */
+  def toDataSet[T: TypeInformation]: DataSet[T] = {
+     new ScalaBatchTranslator().translate[T](table.operation)
+  }
+
+  /**
+   * Converts the [[Table]] to a [[DataStream]].
+   */
+  def toDataStream[T: TypeInformation]: DataStream[T] = {
+    new ScalaStreamingTranslator().translate[T](table.operation)
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
new file mode 100644
index 0000000..0be6be2
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/expressionDsl.scala
@@ -0,0 +1,124 @@
+/*
+ * 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.scala.table
+
+import org.apache.flink.api.table.expressions._
+import org.apache.flink.api.common.typeinfo.TypeInformation
+
+import scala.language.implicitConversions
+
+/**
+ * These are all the operations that can be used to construct an [[Expression]] AST for expression
+ * operations.
+ *
+ * These operations must be kept in sync with the parser in
+ * [[org.apache.flink.api.table.parser.ExpressionParser]].
+ */
+trait ImplicitExpressionOperations {
+  def expr: Expression
+
+  def && (other: Expression) = And(expr, other)
+  def || (other: Expression) = Or(expr, other)
+
+  def > (other: Expression) = GreaterThan(expr, other)
+  def >= (other: Expression) = GreaterThanOrEqual(expr, other)
+  def < (other: Expression) = LessThan(expr, other)
+  def <= (other: Expression) = LessThanOrEqual(expr, other)
+
+  def === (other: Expression) = EqualTo(expr, other)
+  def !== (other: Expression) = NotEqualTo(expr, other)
+
+  def unary_! = Not(expr)
+  def unary_- = UnaryMinus(expr)
+
+  def isNull = IsNull(expr)
+  def isNotNull = IsNotNull(expr)
+
+  def + (other: Expression) = Plus(expr, other)
+  def - (other: Expression) = Minus(expr, other)
+  def / (other: Expression) = Div(expr, other)
+  def * (other: Expression) = Mul(expr, other)
+  def % (other: Expression) = Mod(expr, other)
+
+  def & (other: Expression) = BitwiseAnd(expr, other)
+  def | (other: Expression) = BitwiseOr(expr, other)
+  def ^ (other: Expression) = BitwiseXor(expr, other)
+  def unary_~ = BitwiseNot(expr)
+
+  def abs = Abs(expr)
+
+  def sum = Sum(expr)
+  def min = Min(expr)
+  def max = Max(expr)
+  def count = Count(expr)
+  def avg = Avg(expr)
+
+  def substring(beginIndex: Expression, endIndex: Expression = Literal(Int.MaxValue)) = {
+    Substring(expr, beginIndex, endIndex)
+  }
+
+  def cast(toType: TypeInformation[_]) = Cast(expr, toType)
+
+  def as(name: Symbol) = Naming(expr, name.name)
+}
+
+/**
+ * Implicit conversions from Scala Literals to Expression [[Literal]] and from [[Expression]]
+ * to [[ImplicitExpressionOperations]].
+ */
+trait ImplicitExpressionConversions {
+  implicit class WithOperations(e: Expression) extends ImplicitExpressionOperations {
+    def expr = e
+  }
+
+  implicit class SymbolExpression(s: Symbol) extends ImplicitExpressionOperations {
+    def expr = UnresolvedFieldReference(s.name)
+  }
+
+  implicit class LiteralLongExpression(l: Long) extends ImplicitExpressionOperations {
+    def expr = Literal(l)
+  }
+
+  implicit class LiteralIntExpression(i: Int) extends ImplicitExpressionOperations {
+    def expr = Literal(i)
+  }
+
+  implicit class LiteralFloatExpression(f: Float) extends ImplicitExpressionOperations {
+    def expr = Literal(f)
+  }
+
+  implicit class LiteralDoubleExpression(d: Double) extends ImplicitExpressionOperations {
+    def expr = Literal(d)
+  }
+
+  implicit class LiteralStringExpression(str: String) extends ImplicitExpressionOperations {
+    def expr = Literal(str)
+  }
+
+  implicit class LiteralBooleanExpression(bool: Boolean) extends ImplicitExpressionOperations {
+    def expr = Literal(bool)
+  }
+
+  implicit def symbol2FieldExpression(sym: Symbol): Expression = UnresolvedFieldReference(sym.name)
+  implicit def int2Literal(i: Int): Expression = Literal(i)
+  implicit def long2Literal(l: Long): Expression = Literal(l)
+  implicit def double2Literal(d: Double): Expression = Literal(d)
+  implicit def float2Literal(d: Float): Expression = Literal(d)
+  implicit def string2Literal(str: String): Expression = Literal(str)
+  implicit def boolean2Literal(bool: Boolean): Expression = Literal(bool)
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala
new file mode 100644
index 0000000..e74651b
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/scala/table/package.scala
@@ -0,0 +1,105 @@
+/*
+ * 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.scala
+
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.table.{Row, Table}
+import org.apache.flink.streaming.api.scala.DataStream
+
+import scala.language.implicitConversions
+
+/**
+ * == Table API (Scala) ==
+ *
+ * Importing this package with:
+ *
+ * {{{
+ *   import org.apache.flink.api.scala.table._
+ * }}}
+ *
+ * imports implicit conversions for converting a [[DataSet]] or [[DataStream]] to a
+ * [[Table]]. This can be used to perform SQL-like queries on data. Please have
+ * a look at [[Table]] to see which operations are supported and
+ * [[org.apache.flink.api.scala.table.ImplicitExpressionOperations]] to see how an
+ * expression can be specified.
+ *
+ * When writing a query you can use Scala Symbols to refer to field names. One would
+ * refer to field `a` by writing `'a`. Sometimes it is necessary to manually confert a
+ * Scala literal to an Expression Literal, in those cases use `Literal`, as in `Literal(3)`.
+ *
+ * Example:
+ *
+ * {{{
+ *   import org.apache.flink.api.scala._
+ *   import org.apache.flink.api.scala.table._
+ *
+ *   val env = ExecutionEnvironment.getExecutionEnvironment
+ *   val input = env.fromElements(("Hello", 2), ("Hello", 5), ("Ciao", 3))
+ *   val result = input.as('word, 'count).groupBy('word).select('word, 'count.avg)
+ *   result.print()
+ *
+ *   env.execute()
+ * }}}
+ *
+ * A [[Table]] can be converted back to the underlying API
+ * representation using `as`:
+ *
+ * {{{
+ *   case class Word(word: String, count: Int)
+ *
+ *   val result = in.select(...).as('word, 'count)
+ *   val set = result.as[Word]
+ * }}}
+ */
+package object table extends ImplicitExpressionConversions {
+
+  implicit def table2TableConversions(table: Table): TableConversions = {
+    new TableConversions(table)
+  }
+
+  implicit def dataSet2DataSetConversions[T](set: DataSet[T]): DataSetConversions[T] = {
+    new DataSetConversions[T](set, set.getType.asInstanceOf[CompositeType[T]])
+  }
+
+  implicit def table2RowDataSet(
+      table: Table): DataSet[Row] = {
+    new ScalaBatchTranslator().translate[Row](table.operation)
+  }
+
+  implicit def rowDataSet2Table(
+      rowDataSet: DataSet[Row]): Table = {
+    rowDataSet.toTable
+  }
+
+  implicit def dataStream2DataSetConversions[T](
+      stream: DataStream[T]): DataStreamConversions[T] = {
+    new DataStreamConversions[T](
+      stream,
+      stream.getJavaStream.getType.asInstanceOf[CompositeType[T]])
+  }
+
+  implicit def table2RowDataStream(
+      table: Table): DataStream[Row] = {
+    new ScalaStreamingTranslator().translate[Row](table.operation)
+  }
+
+  implicit def rowDataStream2Table(
+      rowDataStream: DataStream[Row]): Table = {
+    rowDataStream.toTable
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/ExpressionException.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/ExpressionException.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/ExpressionException.scala
new file mode 100644
index 0000000..51c0a4d
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/ExpressionException.scala
@@ -0,0 +1,23 @@
+/*
+ * 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
+
+/**
+ * Exception for all errors occurring during expression evaluation.
+ */
+class ExpressionException(msg: String) extends RuntimeException(msg)

http://git-wip-us.apache.org/repos/asf/flink/blob/e9bf13d8/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Row.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Row.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Row.scala
new file mode 100644
index 0000000..e3baab3
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/Row.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
+
+/**
+ * This is used for executing Table API operations. We use manually generated
+ * TypeInfo to check the field types and create serializers and comparators.
+ */
+class Row(arity: Int) extends Product {
+
+  private val fields = new Array[Any](arity)
+
+  def productArity = fields.length
+
+  def productElement(i: Int): Any = fields(i)
+
+  def setField(i: Int, value: Any): Unit = fields(i) = value
+
+  def canEqual(that: Any) = false
+
+  override def toString = fields.mkString(",")
+
+}


Mime
View raw message