flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [44/50] [abbrv] flink git commit: [FLINK-3226] Implement a CodeGenerator for an efficient translation to DataSet programs
Date Fri, 12 Feb 2016 11:30:09 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateJoin.scala
deleted file mode 100644
index b706e6d..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateJoin.scala
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.codegen
-
-import java.io.StringReader
-
-import org.apache.flink.api.common.functions.FlatJoinFunction
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.table.TableConfig
-import org.apache.flink.api.table.codegen.Indenter._
-import org.apache.flink.api.table.expressions.{Expression, NopExpression}
-import org.slf4j.LoggerFactory
-
-/**
- * Code generator for assembling the result of a binary operation.
- */
-class GenerateJoin[L, R, O](
-    leftTypeInfo: CompositeType[L],
-    rightTypeInfo: CompositeType[R],
-    resultTypeInfo: CompositeType[O],
-    predicate: Expression,
-    outputFields: Seq[Expression],
-    cl: ClassLoader,
-    config: TableConfig)
-  extends GenerateResultAssembler[FlatJoinFunction[L, R, O]](
-    Seq(("in0", leftTypeInfo), ("in1", rightTypeInfo)),
-    cl = cl,
-    config) {
-
-  val LOG = LoggerFactory.getLogger(this.getClass)
-
-
-  override protected def generateInternal(): FlatJoinFunction[L, R, O] = {
-
-    val leftTpe = typeTermForTypeInfo(leftTypeInfo)
-    val rightTpe = typeTermForTypeInfo(rightTypeInfo)
-    val resultTpe = typeTermForTypeInfo(resultTypeInfo)
-
-
-    val resultCode = createResult(resultTypeInfo, outputFields, o => s"coll.collect($o);")
-
-    val generatedName = freshName("GeneratedJoin")
-
-
-    val code = predicate match {
-      case n: NopExpression =>
-        // Janino does not support generics, that's why we need
-        // manual casting here
-        if (nullCheck) {
-          j"""
-        public class $generatedName
-            implements org.apache.flink.api.common.functions.FlatFlatJoinFunction {
-
-          ${reuseCode(resultTypeInfo)}
-
-          public org.apache.flink.api.table.TableConfig config = null;
-          public $generatedName(org.apache.flink.api.table.TableConfig config) {
-            this.config = config;
-          }
-
-          public void join(Object _in0, Object _in1, org.apache.flink.util.Collector coll) {
-            $leftTpe in0 = ($leftTpe) _in0;
-            $rightTpe in1 = ($rightTpe) _in1;
-
-            $resultCode
-          }
-        }
-      """
-        } else {
-          j"""
-        public class $generatedName
-            implements org.apache.flink.api.common.functions.FlatJoinFunction {
-
-          ${reuseCode(resultTypeInfo)}
-
-          public org.apache.flink.api.table.TableConfig config = null;
-          public $generatedName(org.apache.flink.api.table.TableConfig config) {
-            this.config = config;
-          }
-
-          public void join(Object _in0, Object _in1, org.apache.flink.util.Collector coll) {
-            $leftTpe in0 = ($leftTpe) _in0;
-            $rightTpe in1 = ($rightTpe) _in1;
-
-            $resultCode
-          }
-        }
-      """
-        }
-
-      case _ =>
-        val pred = generateExpression(predicate)
-        // Janino does not support generics, that's why we need
-        // manual casting here
-        if (nullCheck) {
-          j"""
-        public class $generatedName
-            implements org.apache.flink.api.common.functions.FlatFlatJoinFunction {
-
-          ${reuseCode(resultTypeInfo)}
-
-          org.apache.flink.api.table.TableConfig config = null;
-
-          public $generatedName(org.apache.flink.api.table.TableConfig config) {
-            this.config = config;
-            ${reuseInitCode()}
-          }
-
-          public void join(Object _in0, Object _in1, org.apache.flink.util.Collector coll) {
-            $leftTpe in0 = ($leftTpe) _in0;
-            $rightTpe in1 = ($rightTpe) _in1;
-
-            ${pred.code}
-
-            if (${pred.nullTerm} && ${pred.resultTerm}) {
-              $resultCode
-            }
-          }
-        }
-      """
-        } else {
-          j"""
-        public class $generatedName
-            implements org.apache.flink.api.common.functions.FlatJoinFunction {
-
-          ${reuseCode(resultTypeInfo)}
-
-          org.apache.flink.api.table.TableConfig config = null;
-
-          public $generatedName(org.apache.flink.api.table.TableConfig config) {
-            this.config = config;
-            ${reuseInitCode()}
-          }
-
-          public void join(Object _in0, Object _in1, org.apache.flink.util.Collector coll) {
-            $leftTpe in0 = ($leftTpe) _in0;
-            $rightTpe in1 = ($rightTpe) _in1;
-
-            ${pred.code}
-
-            if (${pred.resultTerm}) {
-              $resultCode
-            }
-          }
-        }
-      """
-        }
-    }
-
-    LOG.debug(s"""Generated join:\n$code""")
-    compiler.cook(new StringReader(code))
-    val clazz = compiler.getClassLoader().loadClass(generatedName)
-    val constructor = clazz.getConstructor(classOf[TableConfig])
-    constructor.newInstance(config).asInstanceOf[FlatJoinFunction[L, R, O]]
-  }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateResultAssembler.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateResultAssembler.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateResultAssembler.scala
deleted file mode 100644
index 3916410..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateResultAssembler.scala
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.codegen
-
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo}
-import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
-import org.apache.flink.api.table.TableConfig
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.api.table.typeinfo.RowTypeInfo
-
-/**
- * Base class for unary and binary result assembler code generators.
- */
-abstract class GenerateResultAssembler[R](
-    inputs: Seq[(String, CompositeType[_])],
-    cl: ClassLoader,
-    config: TableConfig)
-  extends ExpressionCodeGenerator[R](inputs, cl = cl, config) {
-
-  def reuseCode[A](resultTypeInfo: CompositeType[A]) = {
-      val resultTpe = typeTermForTypeInfo(resultTypeInfo)
-      resultTypeInfo match {
-        case pj: PojoTypeInfo[_] =>
-          super.reuseMemberCode() +
-            s"$resultTpe out = new ${pj.getTypeClass.getCanonicalName}();"
-
-        case row: RowTypeInfo =>
-          super.reuseMemberCode() +
-            s"org.apache.flink.api.table.Row out =" +
-            s" new org.apache.flink.api.table.Row(${row.getArity});"
-
-        case _ => ""
-      }
-  }
-
-  def createResult[T](
-      resultTypeInfo: CompositeType[T],
-      outputFields: Seq[Expression],
-      result: String => String): String = {
-
-    val resultType = typeTermForTypeInfo(resultTypeInfo)
-
-    val fieldsCode = outputFields.map(generateExpression)
-
-    val block = resultTypeInfo match {
-      case ri: RowTypeInfo =>
-        val resultSetters: String = fieldsCode.zipWithIndex map {
-          case (fieldCode, i) =>
-            s"""
-              |${fieldCode.code}
-              |out.setField($i, ${fieldCode.resultTerm});
-            """.stripMargin
-        } mkString("\n")
-
-        s"""
-          |$resultSetters
-          |${result("out")}
-        """.stripMargin
-
-      case pj: PojoTypeInfo[_] =>
-        val resultSetters: String = fieldsCode.zip(outputFields) map {
-        case (fieldCode, expr) =>
-          val fieldName = expr.name
-          s"""
-            |${fieldCode.code}
-            |out.$fieldName = ${fieldCode.resultTerm};
-          """.stripMargin
-        } mkString("\n")
-
-        s"""
-          |$resultSetters
-          |${result("out")}
-        """.stripMargin
-
-      case tup: TupleTypeInfo[_] =>
-        val resultSetters: String = fieldsCode.zip(outputFields) map {
-          case (fieldCode, expr) =>
-            val fieldName = expr.name
-            s"""
-              |${fieldCode.code}
-              |out.$fieldName = ${fieldCode.resultTerm};
-            """.stripMargin
-        } mkString("\n")
-
-        s"""
-          |$resultSetters
-          |${result("out")}
-        """.stripMargin
-
-      case cc: CaseClassTypeInfo[_] =>
-        val fields: String = fieldsCode.map(_.code).mkString("\n")
-        val ctorParams: String = fieldsCode.map(_.resultTerm).mkString(",")
-
-        s"""
-          |$fields
-          |return new $resultType($ctorParams);
-        """.stripMargin
-    }
-
-    block
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateSelect.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateSelect.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateSelect.scala
deleted file mode 100644
index a75d15b..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GenerateSelect.scala
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.codegen
-
-import java.io.StringReader
-
-import org.apache.flink.api.common.functions.MapFunction
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.table.TableConfig
-import org.apache.flink.api.table.codegen.Indenter._
-import org.apache.flink.api.table.expressions.Expression
-import org.slf4j.LoggerFactory
-
-/**
- * Code generator for assembling the result of a select operation.
- */
-class GenerateSelect[I, O](
-    inputTypeInfo: CompositeType[I],
-    resultTypeInfo: CompositeType[O],
-    outputFields: Seq[Expression],
-    cl: ClassLoader,
-    config: TableConfig)
-  extends GenerateResultAssembler[MapFunction[I, O]](
-    Seq(("in0", inputTypeInfo)),
-    cl = cl,
-    config) {
-
-  val LOG = LoggerFactory.getLogger(this.getClass)
-
-  override protected def generateInternal(): MapFunction[I, O] = {
-
-    val inputTpe = typeTermForTypeInfo(inputTypeInfo)
-    val resultTpe = typeTermForTypeInfo(resultTypeInfo)
-
-    val resultCode = createResult(resultTypeInfo, outputFields, o => s"return $o;")
-
-    val generatedName = freshName("GeneratedSelect")
-
-    // Janino does not support generics, that's why we need
-    // manual casting here
-    val code =
-      j"""
-        public class $generatedName
-            implements org.apache.flink.api.common.functions.MapFunction<$inputTpe, $resultTpe> {
-
-          ${reuseCode(resultTypeInfo)}
-
-          org.apache.flink.api.table.TableConfig config = null;
-
-          public $generatedName(org.apache.flink.api.table.TableConfig config) {
-            this.config = config;
-            ${reuseInitCode()}
-          }
-
-          @Override
-          public Object map(Object _in0) {
-            $inputTpe in0 = ($inputTpe) _in0;
-            $resultCode
-          }
-        }
-      """
-
-    LOG.debug(s"""Generated select:\n$code""")
-    compiler.cook(new StringReader(code))
-    val clazz = compiler.getClassLoader().loadClass(generatedName)
-    val constructor = clazz.getConstructor(classOf[TableConfig])
-    constructor.newInstance(config).asInstanceOf[MapFunction[I, O]]
-  }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala
new file mode 100644
index 0000000..7c20e38
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedExpression.scala
@@ -0,0 +1,27 @@
+/*
+ * 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.codegen
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+
+case class GeneratedExpression(
+    resultTerm: String,
+    nullTerm: String,
+    code: String,
+    resultType: TypeInformation[_])

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.scala
new file mode 100644
index 0000000..575e7ab
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/GeneratedFunction.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.codegen
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+
+case class GeneratedFunction[T](name: String, returnType: TypeInformation[Any], code: String)

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/Indenter.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/Indenter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/Indenter.scala
index 1319f21..c7d9a2e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/Indenter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/Indenter.scala
@@ -18,16 +18,16 @@
 package org.apache.flink.api.table.codegen
 
 class IndentStringContext(sc: StringContext) {
-  def j(args: Any*):String = {
+  def j(args: Any*): String = {
     val sb = new StringBuilder()
     for ((s, a) <- sc.parts zip args) {
       sb append s
 
       val ind = getindent(s)
-      if (ind.size > 0) {
-        sb append a.toString().replaceAll("\n", "\n" + ind)
+      if (ind.nonEmpty) {
+        sb append a.toString.replaceAll("\n", "\n" + ind)
       } else {
-        sb append a.toString()
+        sb append a.toString
       }
     }
     if (sc.parts.size > args.size) {
@@ -50,5 +50,5 @@ class IndentStringContext(sc: StringContext) {
 }
 
 object Indenter {
-  implicit  def toISC(sc: StringContext) = new IndentStringContext(sc)
+  implicit def toISC(sc: StringContext): IndentStringContext = new IndentStringContext(sc)
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/OperatorCodeGen.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/OperatorCodeGen.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/OperatorCodeGen.scala
new file mode 100644
index 0000000..8402569
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/codegen/OperatorCodeGen.scala
@@ -0,0 +1,367 @@
+/*
+ * 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.codegen
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo.BOOLEAN_TYPE_INFO
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.table.codegen.CodeGenUtils._
+
+object OperatorCodeGen {
+
+   def generateArithmeticOperator(
+      operator: String,
+      nullCheck: Boolean,
+      resultType: TypeInformation[_],
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+    : GeneratedExpression = {
+    generateOperatorIfNotNull(nullCheck, resultType, left, right) {
+      (leftTerm, rightTerm) => s"$leftTerm $operator $rightTerm"
+    }
+  }
+
+  def generateUnaryArithmeticOperator(
+      operator: String,
+      nullCheck: Boolean,
+      resultType: TypeInformation[_],
+      operand: GeneratedExpression)
+    : GeneratedExpression = {
+    generateUnaryOperatorIfNotNull(nullCheck, resultType, operand) {
+      (operandTerm) => s"$operator($operandTerm)"
+    }
+  }
+
+  def generateEquals(
+      nullCheck: Boolean,
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+    : GeneratedExpression = {
+    generateOperatorIfNotNull(nullCheck, BOOLEAN_TYPE_INFO, left, right) {
+      if (isReference(left)) {
+        (leftTerm, rightTerm) => s"$leftTerm.equals($rightTerm)"
+      }
+      else if (isReference(right)) {
+        (leftTerm, rightTerm) => s"$rightTerm.equals($leftTerm)"
+      }
+      else {
+        (leftTerm, rightTerm) => s"$leftTerm == $rightTerm"
+      }
+    }
+  }
+
+  def generateNotEquals(
+      nullCheck: Boolean,
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+    : GeneratedExpression = {
+    generateOperatorIfNotNull(nullCheck, BOOLEAN_TYPE_INFO, left, right) {
+      if (isReference(left)) {
+        (leftTerm, rightTerm) => s"!($leftTerm.equals($rightTerm))"
+      }
+      else if (isReference(right)) {
+        (leftTerm, rightTerm) => s"!($rightTerm.equals($leftTerm))"
+      }
+      else {
+        (leftTerm, rightTerm) => s"$leftTerm != $rightTerm"
+      }
+    }
+  }
+
+  def generateComparison(
+      operator: String,
+      nullCheck: Boolean,
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+    : GeneratedExpression = {
+    generateOperatorIfNotNull(nullCheck, BOOLEAN_TYPE_INFO, left, right) {
+      if (isString(left) && isString(right)) {
+        (leftTerm, rightTerm) => s"$leftTerm.compareTo($rightTerm) $operator 0"
+      }
+      else if (isNumeric(left) && isNumeric(right)) {
+        (leftTerm, rightTerm) => s"$leftTerm $operator $rightTerm"
+      }
+      else {
+        throw new CodeGenException("Comparison is only supported for Strings and numeric types.")
+      }
+    }
+  }
+
+  def generateIsNull(
+      nullCheck: Boolean,
+      operand: GeneratedExpression)
+    : GeneratedExpression = {
+    val resultTerm = newName("result")
+    val nullTerm = newName("isNull")
+    val operatorCode = if (nullCheck) {
+      s"""
+        |${operand.code}
+        |boolean $resultTerm = ${operand.nullTerm};
+        |boolean $nullTerm = false;
+        |""".stripMargin
+    }
+    else if (!nullCheck && isReference(operand.resultType)) {
+      s"""
+        |${operand.code}
+        |boolean $resultTerm = ${operand.resultTerm} == null;
+        |boolean $nullTerm = false;
+        |""".stripMargin
+    }
+    else {
+      s"""
+        |${operand.code}
+        |boolean $resultTerm = false;
+        |""".stripMargin
+    }
+
+    GeneratedExpression(resultTerm, nullTerm, operatorCode, BOOLEAN_TYPE_INFO)
+  }
+
+  def generateIsNotNull(
+      nullCheck: Boolean,
+      operand: GeneratedExpression)
+    : GeneratedExpression = {
+    val resultTerm = newName("result")
+    val nullTerm = newName("isNull")
+    val operatorCode = if (nullCheck) {
+      s"""
+        |${operand.code}
+        |boolean $resultTerm = !${operand.nullTerm};
+        |boolean $nullTerm = false;
+        |""".stripMargin
+    }
+    else if (!nullCheck && isReference(operand.resultType)) {
+      s"""
+        |${operand.code}
+        |boolean $resultTerm = ${operand.resultTerm} != null;
+        |boolean $nullTerm = false;
+        |""".stripMargin
+    }
+    else {
+      s"""
+        |${operand.code}
+        |boolean $resultTerm = true;
+        |""".stripMargin
+    }
+
+    GeneratedExpression(resultTerm, nullTerm, operatorCode, BOOLEAN_TYPE_INFO)
+  }
+
+  def generateAnd(
+      nullCheck: Boolean,
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+    : GeneratedExpression = {
+    val resultTerm = newName("result")
+    val nullTerm = newName("isNull")
+
+    val operatorCode = if (nullCheck) {
+      // Three-valued logic:
+      // no Unknown -> Two-valued logic
+      // True && Unknown -> Unknown
+      // False && Unknown -> False
+      // Unknown && True -> Unknown
+      // Unknown && False -> False
+      // Unknown && Unknown -> Unknown
+      s"""
+        |${left.code}
+        |${right.code}
+        |boolean $resultTerm;
+        |boolean $nullTerm;
+        |if (!${left.nullTerm} && !${right.nullTerm}) {
+        |  $resultTerm = ${left.resultTerm} && ${right.resultTerm};
+        |  $nullTerm = false;
+        |}
+        |else if (!${left.nullTerm} && ${left.resultTerm} && ${right.nullTerm}) {
+        |  $resultTerm = false;
+        |  $nullTerm = true;
+        |}
+        |else if (!${left.nullTerm} && !${left.resultTerm} && ${right.nullTerm}) {
+        |  $resultTerm = false;
+        |  $nullTerm = false;
+        |}
+        |else if (${left.nullTerm} && !${right.nullTerm} && ${right.resultTerm}) {
+        |  $resultTerm = false;
+        |  $nullTerm = true;
+        |}
+        |else if (${left.nullTerm} && !${right.nullTerm} && !${right.resultTerm}) {
+        |  $resultTerm = false;
+        |  $nullTerm = false;
+        |}
+        |else {
+        |  $resultTerm = false;
+        |  $nullTerm = true;
+        |}
+        |""".stripMargin
+    }
+    else {
+      s"""
+        |${left.code}
+        |${right.code}
+        |boolean $resultTerm = ${left.resultTerm} && ${right.resultTerm};
+        |""".stripMargin
+    }
+
+    GeneratedExpression(resultTerm, nullTerm, operatorCode, BOOLEAN_TYPE_INFO)
+  }
+
+  def generateOr(
+      nullCheck: Boolean,
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+    : GeneratedExpression = {
+    val resultTerm = newName("result")
+    val nullTerm = newName("isNull")
+
+    val operatorCode = if (nullCheck) {
+      // Three-valued logic:
+      // no Unknown -> Two-valued logic
+      // True && Unknown -> True
+      // False && Unknown -> Unknown
+      // Unknown && True -> True
+      // Unknown && False -> Unknown
+      // Unknown && Unknown -> Unknown
+      s"""
+        |${left.code}
+        |${right.code}
+        |boolean $resultTerm;
+        |boolean $nullTerm;
+        |if (!${left.nullTerm} && !${right.nullTerm}) {
+        |  $resultTerm = ${left.resultTerm} || ${right.resultTerm};
+        |  $nullTerm = false;
+        |}
+        |else if (!${left.nullTerm} && ${left.resultTerm} && ${right.nullTerm}) {
+        |  $resultTerm = true;
+        |  $nullTerm = false;
+        |}
+        |else if (!${left.nullTerm} && !${left.resultTerm} && ${right.nullTerm}) {
+        |  $resultTerm = false;
+        |  $nullTerm = true;
+        |}
+        |else if (${left.nullTerm} && !${right.nullTerm} && ${right.resultTerm}) {
+        |  $resultTerm = true;
+        |  $nullTerm = false;
+        |}
+        |else if (${left.nullTerm} && !${right.nullTerm} && !${right.resultTerm}) {
+        |  $resultTerm = false;
+        |  $nullTerm = true;
+        |}
+        |else {
+        |  $resultTerm = false;
+        |  $nullTerm = true;
+        |}
+        |""".stripMargin
+    }
+    else {
+      s"""
+        |${left.code}
+        |${right.code}
+        |boolean $resultTerm = ${left.resultTerm} && ${right.resultTerm};
+        |""".stripMargin
+    }
+
+    GeneratedExpression(resultTerm, nullTerm, operatorCode, BOOLEAN_TYPE_INFO)
+  }
+
+  def generateNot(
+      nullCheck: Boolean,
+      operand: GeneratedExpression)
+    : GeneratedExpression = {
+    // Three-valued logic:
+    // no Unknown -> Two-valued logic
+    // Unknown -> Unknown
+    generateUnaryOperatorIfNotNull(nullCheck, BOOLEAN_TYPE_INFO, operand) {
+      (operandTerm) => s"!($operandTerm)"
+    }
+  }
+
+  // ----------------------------------------------------------------------------------------------
+
+  private def generateUnaryOperatorIfNotNull(
+      nullCheck: Boolean,
+      resultType: TypeInformation[_],
+      operand: GeneratedExpression)
+      (expr: (String) => String)
+    : GeneratedExpression = {
+    val resultTerm = newName("result")
+    val nullTerm = newName("isNull")
+    val resultTypeTerm = primitiveTypeTermForTypeInfo(resultType)
+    val defaultValue = primitiveDefaultValue(resultType)
+
+    val operatorCode = if (nullCheck) {
+      s"""
+        |${operand.code}
+        |$resultTypeTerm $resultTerm;
+        |boolean $nullTerm;
+        |if (!${operand.nullTerm}) {
+        |  $resultTerm = ${expr(operand.resultTerm)};
+        |  $nullTerm = false;
+        |}
+        |else {
+        |  $resultTerm = $defaultValue;
+        |  $nullTerm = true;
+        |}
+        |""".stripMargin
+    }
+    else {
+      s"""
+        |${operand.code}
+        |$resultTypeTerm $resultTerm = ${expr(operand.resultTerm)};
+        |""".stripMargin
+    }
+
+    GeneratedExpression(resultTerm, nullTerm, operatorCode, resultType)
+  }
+
+  private def generateOperatorIfNotNull(
+      nullCheck: Boolean,
+      resultType: TypeInformation[_],
+      left: GeneratedExpression,
+      right: GeneratedExpression)
+      (expr: (String, String) => String)
+    : GeneratedExpression = {
+    val resultTerm = newName("result")
+    val nullTerm = newName("isNull")
+    val resultTypeTerm = primitiveTypeTermForTypeInfo(resultType)
+    val defaultValue = primitiveDefaultValue(resultType)
+
+    val resultCode = if (nullCheck) {
+      s"""
+        |${left.code}
+        |${right.code}
+        |boolean $nullTerm = ${left.nullTerm} || ${right.nullTerm};
+        |$resultTypeTerm $resultTerm;
+        |if ($nullTerm) {
+        |  $resultTerm = $defaultValue;
+        |}
+        |else {
+        |  $resultTerm = ${expr(left.resultTerm, right.resultTerm)};
+        |}
+        |""".stripMargin
+    }
+    else {
+      s"""
+        |${left.code}
+        |${right.code}
+        |$resultTypeTerm $resultTerm = ${expr(left.resultTerm, right.resultTerm)};
+        |""".stripMargin
+    }
+
+    GeneratedExpression(resultTerm, nullTerm, resultCode, resultType)
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
index f909cab..85956a2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/expressions/literals.scala
@@ -23,8 +23,10 @@ import org.apache.flink.api.scala.table.ImplicitExpressionOperations
 
 object Literal {
   def apply(l: Any): Literal = l match {
-    case i:Int => Literal(i, BasicTypeInfo.INT_TYPE_INFO)
-    case l:Long => Literal(l, BasicTypeInfo.LONG_TYPE_INFO)
+    case i: Int => Literal(i, BasicTypeInfo.INT_TYPE_INFO)
+    case s: Short => Literal(s, BasicTypeInfo.SHORT_TYPE_INFO)
+    case b: Byte => Literal(b, BasicTypeInfo.BYTE_TYPE_INFO)
+    case l: Long => Literal(l, BasicTypeInfo.LONG_TYPE_INFO)
     case d: Double => Literal(d, BasicTypeInfo.DOUBLE_TYPE_INFO)
     case f: Float => Literal(f, BasicTypeInfo.FLOAT_TYPE_INFO)
     case str: String => Literal(str, BasicTypeInfo.STRING_TYPE_INFO)

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
index f6fe2e4..227b3e8 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/TypeConverter.scala
@@ -18,13 +18,19 @@
 
 package org.apache.flink.api.table.plan
 
+import org.apache.calcite.rel.`type`.RelDataType
 import org.apache.calcite.sql.`type`.SqlTypeName
 import org.apache.calcite.sql.`type`.SqlTypeName._
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo, GenericTypeInfo}
+import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation}
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.api.java.typeutils.TupleTypeInfo
 import org.apache.flink.api.java.typeutils.ValueTypeInfo._
-import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
+import org.apache.flink.api.table.typeinfo.RowTypeInfo
+import org.apache.flink.api.table.{Row, TableException}
+
+import scala.collection.JavaConversions._
 
 object TypeConverter {
 
@@ -51,7 +57,7 @@ object TypeConverter {
 //    case p: PojoTypeInfo[_] => STRUCTURED
 //    case g: GenericTypeInfo[_] => OTHER
     case _ => ??? // TODO more types
-    }
+  }
 
   def sqlTypeToTypeInfo(sqlType: SqlTypeName): TypeInformation[_] = sqlType match {
     case BOOLEAN => BOOLEAN_TYPE_INFO
@@ -63,7 +69,74 @@ object TypeConverter {
     case DOUBLE => DOUBLE_TYPE_INFO
     case VARCHAR | CHAR => STRING_TYPE_INFO
     case DATE => DATE_TYPE_INFO
-    case _ => ??? // TODO more types
+    case _ =>
+      println(sqlType)
+      ??? // TODO more types
+  }
+
+  def determineReturnType(
+      logicalRowType: RelDataType,
+      expectedPhysicalType: Option[TypeInformation[Any]],
+      nullable: Boolean,
+      useEfficientTypes: Boolean)
+    : TypeInformation[Any] = {
+    // convert to type information
+    val logicalFieldTypes = logicalRowType.getFieldList map { relDataType =>
+      TypeConverter.sqlTypeToTypeInfo(relDataType.getType.getSqlTypeName)
+    }
+
+    val returnType = expectedPhysicalType match {
+      // a certain physical type is expected (but not Row)
+      // check if expected physical type is compatible with logical field type
+      case Some(typeInfo) if typeInfo.getTypeClass != classOf[Row] =>
+        if (typeInfo.getArity != logicalFieldTypes.length) {
+          throw new TableException("Arity of result does not match expected type.")
+        }
+        typeInfo match {
+          case ct: CompositeType[_] =>
+            logicalFieldTypes.zipWithIndex foreach {
+              case (fieldTypeInfo, i) =>
+                val expectedTypeInfo = ct.getTypeAt(i)
+                if (fieldTypeInfo != expectedTypeInfo) {
+                  throw new TableException(s"Result field does not match expected type." +
+                    s"Expected: $expectedTypeInfo; Actual: $fieldTypeInfo")
+                }
+            }
+          case at: AtomicType[_] =>
+            val fieldTypeInfo = logicalFieldTypes.head
+            if (fieldTypeInfo != at) {
+              throw new TableException(s"Result field does not match expected type." +
+                s"Expected: $at; Actual: $fieldTypeInfo")
+            }
+
+          case _ =>
+            throw new TableException("Unsupported result type.")
+        }
+        typeInfo
+
+      // Row is expected, create the arity for it
+      case Some(typeInfo) if typeInfo.getTypeClass == classOf[Row] =>
+        new RowTypeInfo(logicalFieldTypes)
+
+      // no physical type
+      // determine type based on logical fields and configuration parameters
+      case None =>
+        // no need for efficient types -> use Row
+        // we cannot use efficient types if row arity > tuple arity or nullable
+        if (!useEfficientTypes || logicalFieldTypes.length > Tuple.MAX_ARITY || nullable) {
+          new RowTypeInfo(logicalFieldTypes)
+        }
+        // use efficient type tuple or atomic type
+        else {
+          if (logicalFieldTypes.length == 1) {
+            logicalFieldTypes.head
+          }
+          else {
+            new TupleTypeInfo[Any](logicalFieldTypes.toArray:_*)
+          }
+        }
     }
+    returnType.asInstanceOf[TypeInformation[Any]]
+  }
 
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
index ec5805a..00cf899 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetExchange.scala
@@ -18,12 +18,13 @@
 
 package org.apache.flink.api.table.plan.nodes.dataset
 
-import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
 import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
 import org.apache.flink.api.common.operators.base.PartitionOperatorBase.PartitionMethod
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.TableConfig
 
 /**
   * Flink RelNode which matches along with PartitionOperator.
@@ -57,7 +58,9 @@ class DataSetExchange(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def translateToPlan: DataSet[Any] = {
+  override def translateToPlan(
+      config: TableConfig, 
+      expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
     ???
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
index 913cca0..9744792 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetFlatMap.scala
@@ -18,12 +18,14 @@
 
 package org.apache.flink.api.table.plan.nodes.dataset
 
-import org.apache.calcite.plan.{RelOptCost, RelOptPlanner, RelTraitSet, RelOptCluster}
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
 import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
 import org.apache.flink.api.common.functions.FlatMapFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.TableConfig
+import org.apache.flink.api.table.plan.TypeConverter._
 
 /**
   * Flink RelNode which matches along with FlatMapOperator.
@@ -35,7 +37,7 @@ class DataSetFlatMap(
     input: RelNode,
     rowType: RelDataType,
     opName: String,
-    func: FlatMapFunction[Row, Row])
+    func: (TableConfig, TypeInformation[Any], TypeInformation[Any]) => FlatMapFunction[Any, Any])
   extends SingleRel(cluster, traitSet, input)
   with DataSetRel {
 
@@ -56,7 +58,17 @@ class DataSetFlatMap(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def translateToPlan: DataSet[Any] = {
-    ???
+  override def toString = opName
+
+  override def translateToPlan(config: TableConfig,
+      expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
+    val inputDataSet = input.asInstanceOf[DataSetRel].translateToPlan(config)
+    val returnType = determineReturnType(
+      getRowType,
+      expectedType,
+      config.getNullCheck,
+      config.getEfficientTypeUsage)
+    val flatMapFunc = func.apply(config, inputDataSet.getType, returnType)
+    inputDataSet.flatMap(flatMapFunc)
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
index 11bb160..ae76d29 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetGroupReduce.scala
@@ -22,8 +22,9 @@ import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
 import org.apache.calcite.rel.`type`.RelDataType
 import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
 import org.apache.flink.api.common.functions.GroupReduceFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.{TableConfig, Row}
 
 /**
   * Flink RelNode which matches along with ReduceGroupOperator.
@@ -57,7 +58,9 @@ class DataSetGroupReduce(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def translateToPlan: DataSet[Any] = {
+  override def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
     ???
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
index c20cdc5..de436be 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetJoin.scala
@@ -23,9 +23,10 @@ import org.apache.calcite.rel.`type`.RelDataType
 import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
 import org.apache.flink.api.common.functions.JoinFunction
 import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
 import org.apache.flink.api.java.operators.join.JoinType
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.{TableConfig, Row}
 
 /**
   * Flink RelNode which matches along with JoinOperator and its related operations.
@@ -67,7 +68,9 @@ class DataSetJoin(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def translateToPlan: DataSet[Any] = {
+  override def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
     ???
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
index be8bd9d..f4f8afb 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetMap.scala
@@ -18,12 +18,14 @@
 
 package org.apache.flink.api.table.plan.nodes.dataset
 
-import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
 import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
 import org.apache.flink.api.common.functions.MapFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.TableConfig
+import org.apache.flink.api.table.plan.TypeConverter.determineReturnType
 
 /**
   * Flink RelNode which matches along with MapOperator.
@@ -34,7 +36,7 @@ class DataSetMap(
     input: RelNode,
     rowType: RelDataType,
     opName: String,
-    func: MapFunction[Row, Row])
+    func: (TableConfig, TypeInformation[Any], TypeInformation[Any]) => MapFunction[Any, Any])
   extends SingleRel(cluster, traitSet, input)
   with DataSetRel {
 
@@ -55,9 +57,19 @@ class DataSetMap(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def toString() = opName
+  override def toString = opName
 
-  override def translateToPlan: DataSet[Any] = {
-    ???
+  override def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]])
+    : DataSet[Any] = {
+    val inputDataSet = input.asInstanceOf[DataSetRel].translateToPlan(config)
+    val returnType = determineReturnType(
+      getRowType,
+      expectedType,
+      config.getNullCheck,
+      config.getEfficientTypeUsage)
+    val mapFunc = func.apply(config, inputDataSet.getType, returnType)
+    inputDataSet.map(mapFunc)
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
index 567a91c..e6fc0f9 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetReduce.scala
@@ -22,8 +22,9 @@ import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
 import org.apache.calcite.rel.`type`.RelDataType
 import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
 import org.apache.flink.api.common.functions.ReduceFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.{TableConfig, Row}
 
 /**
   * Flink RelNode which matches along with ReduceOperator.
@@ -57,7 +58,9 @@ class DataSetReduce(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def translateToPlan: DataSet[Any] = {
+  override def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
     ???
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
index 20677b3..16a0ae3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetRel.scala
@@ -19,15 +19,19 @@
 package org.apache.flink.api.table.plan.nodes.dataset
 
 import org.apache.calcite.rel.RelNode
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.TableConfig
 
 trait DataSetRel extends RelNode {
 
   /**
     * Translate the FlinkRelNode into Flink operator.
     */
-  def translateToPlan: DataSet[Any]
+  def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]] = None)
+    : DataSet[Any]
 
 }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
index df5301d..033711b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSort.scala
@@ -18,11 +18,12 @@
 
 package org.apache.flink.api.table.plan.nodes.dataset
 
-import org.apache.calcite.plan.{RelTraitSet, RelOptCluster}
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
 import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.{RelWriter, RelNode, SingleRel}
+import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.TableConfig
 
 /**
   * Flink RelNode which matches along with SortPartitionOperator.
@@ -56,7 +57,9 @@ class DataSetSort(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def translateToPlan: DataSet[Any] = {
+  override def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
     ???
   }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
index 53067dc..33fe430 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetSource.scala
@@ -18,25 +18,18 @@
 
 package org.apache.flink.api.table.plan.nodes.dataset
 
-import java.lang.reflect.Field
-
 import org.apache.calcite.plan._
 import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.`type`.RelDataType
 import org.apache.calcite.rel.core.TableScan
-import org.apache.flink.api.common.functions.RichMapFunction
-import org.apache.flink.api.common.typeinfo.{TypeInformation, AtomicType}
+import org.apache.flink.api.common.functions.MapFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.java.tuple.Tuple
-import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo}
-import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
-import org.apache.flink.api.table.Row
-import org.apache.flink.api.table.plan.TypeConverter
+import org.apache.flink.api.table.TableConfig
+import org.apache.flink.api.table.codegen.CodeGenerator
+import org.apache.flink.api.table.plan.TypeConverter.determineReturnType
 import org.apache.flink.api.table.plan.schema.DataSetTable
-import org.apache.flink.api.table.typeinfo.RowTypeInfo
-import org.apache.flink.configuration.Configuration
-
-import scala.collection.JavaConverters._
+import org.apache.flink.api.table.runtime.MapRunner
 
 /**
   * Flink RelNode which matches along with DataSource.
@@ -62,132 +55,57 @@ class DataSetSource(
     )
   }
 
-  override def translateToPlan: DataSet[Any] = {
+  override def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]])
+    : DataSet[Any] = {
 
     val inputDataSet: DataSet[Any] = dataSetTable.dataSet
-
-    // extract Flink data types
-    val fieldTypes: Array[TypeInformation[_]] = rowType.getFieldList.asScala
-      .map(f => f.getType.getSqlTypeName)
-      .map(n => TypeConverter.sqlTypeToTypeInfo(n))
-      .toArray
-
-    val rowTypeInfo = new RowTypeInfo(fieldTypes, dataSetTable.fieldNames)
-
-    // convert input data set into row data set
-    inputDataSet.getType match {
-      case t: TupleTypeInfo[_] =>
-        val rowMapper = new TupleToRowMapper(dataSetTable.fieldIndexes)
-        inputDataSet.asInstanceOf[DataSet[Tuple]]
-          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
-
-      case c: CaseClassTypeInfo[_] =>
-        val rowMapper = new CaseClassToRowMapper(dataSetTable.fieldIndexes)
-        inputDataSet.asInstanceOf[DataSet[Product]]
-          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
-
-      case p: PojoTypeInfo[_] =>
-        // get pojo class
-        val typeClazz = p.getTypeClass.asInstanceOf[Class[Any]]
-        // get original field names
-        val origFieldNames = dataSetTable.fieldIndexes.map(i => p.getFieldNames()(i))
-
-        val rowMapper = new PojoToRowMapper(typeClazz, origFieldNames)
-        inputDataSet.asInstanceOf[DataSet[Any]]
-          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
-
-      case a: AtomicType[_] =>
-        val rowMapper = new AtomicToRowMapper
-        inputDataSet.asInstanceOf[DataSet[Any]]
-          .map(rowMapper).returns(rowTypeInfo).asInstanceOf[DataSet[Any]]
-    }
-  }
-
-}
-
-class TupleToRowMapper(val fromIndexes: Array[Int])
-  extends RichMapFunction[Tuple, Row]
-{
-
-  @transient var outR: Row = null
-
-  override def open(conf: Configuration): Unit = {
-    outR = new Row(fromIndexes.length)
-  }
-
-  override def map(v: Tuple): Row = {
-
-    var i = 0
-    while (i < fromIndexes.length) {
-      outR.setField(i, v.getField(fromIndexes(i)))
-      i += 1
-    }
-    outR
-  }
-}
-
-class CaseClassToRowMapper(val fromIndexes: Array[Int])
-  extends RichMapFunction[Product, Row]
-{
-
-  @transient var outR: Row = null
-
-  override def open(conf: Configuration): Unit = {
-    outR = new Row(fromIndexes.length)
-  }
-
-  override def map(v: Product): Row = {
-
-    var i = 0
-    while (i < fromIndexes.length) {
-      outR.setField(i, v.productElement(fromIndexes(i)))
-      i += 1
-    }
-    outR
-  }
-}
-
-class PojoToRowMapper(val inClazz: Class[Any], val fieldNames: Array[String])
-  extends RichMapFunction[Any, Row]
-{
-
-  @transient var outR: Row = null
-  @transient var fields: Array[Field] = null
-
-  override def open(conf: Configuration): Unit = {
-
-    fields = fieldNames.map { n =>
-      val f = inClazz.getField(n)
-      f.setAccessible(true)
-      f
-    }
-    outR = new Row(fieldNames.length)
-  }
-
-  override def map(v: Any): Row = {
-
-    var i = 0
-    while (i < fields.length) {
-      outR.setField(i, fields(i).get(v))
-      i += 1
+    val inputType = inputDataSet.getType
+
+    // special case:
+    // if efficient type usage is enabled and no expected type is set
+    // we can simply forward the DataSet to the next operator
+    expectedType match {
+      case None if config.getEfficientTypeUsage =>
+        inputDataSet
+
+      case _ =>
+        val determinedType = determineReturnType(
+          getRowType,
+          expectedType,
+          config.getNullCheck,
+          config.getEfficientTypeUsage)
+
+        // conversion
+        if (determinedType != inputType) {
+          val generator = new CodeGenerator(config, inputDataSet.getType)
+          val conversion = generator.generateConverterResultExpression(determinedType)
+
+          val body =
+            s"""
+              |${conversion.code}
+              |return ${conversion.resultTerm};
+              |""".stripMargin
+
+          val genFunction = generator.generateFunction(
+            "DataSetSourceConversion",
+            classOf[MapFunction[Any, Any]],
+            body,
+            determinedType)
+
+          val mapFunc = new MapRunner[Any, Any](
+            genFunction.name,
+            genFunction.code,
+            genFunction.returnType)
+
+          inputDataSet.map(mapFunc)
+        }
+        // no conversion necessary, forward
+        else {
+          inputDataSet
+        }
     }
-    outR
-  }
-}
-
-class AtomicToRowMapper()
-  extends RichMapFunction[Any, Row]
-{
-
-  @transient var outR: Row = null
-
-  override def open(conf: Configuration): Unit = {
-    outR = new Row(1)
   }
 
-  override def map(v: Any): Row = {
-
-    outR.setField(0, v)
-    outR
-  }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
index a510fc9..ebfd48a 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/nodes/dataset/DataSetUnion.scala
@@ -21,8 +21,9 @@ package org.apache.flink.api.table.plan.nodes.dataset
 import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
 import org.apache.calcite.rel.`type`.RelDataType
 import org.apache.calcite.rel.{RelWriter, BiRel, RelNode}
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.DataSet
-import org.apache.flink.api.table.Row
+import org.apache.flink.api.table.{TableConfig, Row}
 
 /**
 * Flink RelNode which matches along with UnionOperator.
@@ -55,7 +56,9 @@ class DataSetUnion(
     super.explainTerms(pw).item("name", opName)
   }
 
-  override def translateToPlan: DataSet[Any] = {
+  override def translateToPlan(
+      config: TableConfig,
+      expectedType: Option[TypeInformation[Any]]): DataSet[Any] = {
     ???
   }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
index 383c965..0ca153d 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetFilterRule.scala
@@ -21,8 +21,13 @@ package org.apache.flink.api.table.plan.rules.dataset
 import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
 import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.common.functions.FlatMapFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.table.TableConfig
+import org.apache.flink.api.table.codegen.CodeGenerator
 import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetFlatMap}
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkFilter, FlinkConvention}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkConvention, FlinkFilter}
+import org.apache.flink.api.table.runtime.FlatMapRunner
 
 class DataSetFilterRule
   extends ConverterRule(
@@ -37,13 +42,54 @@ class DataSetFilterRule
     val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
     val convInput: RelNode = RelOptRule.convert(filter.getInput, DataSetConvention.INSTANCE)
 
+    val func = (
+        config: TableConfig,
+        inputType: TypeInformation[Any],
+        returnType: TypeInformation[Any]) => {
+      val generator = new CodeGenerator(config, inputType)
+
+      val condition = generator.generateExpression(filter.getCondition)
+
+      // conversion
+      val body = if (inputType != returnType) {
+        val conversion = generator.generateConverterResultExpression(returnType)
+        s"""
+          |${condition.code}
+          |if (${condition.resultTerm}) {
+          |  ${conversion.code}
+          |  ${generator.collectorTerm}.collect(${conversion.resultTerm});
+          |}
+          |""".stripMargin
+      }
+      // no conversion
+      else {
+        s"""
+          |${condition.code}
+          |if (${condition.resultTerm}) {
+          |  ${generator.collectorTerm}.collect(${generator.input1Term});
+          |}
+          |""".stripMargin
+      }
+
+      val genFunction = generator.generateFunction(
+        description,
+        classOf[FlatMapFunction[Any, Any]],
+        body,
+        returnType)
+
+      new FlatMapRunner[Any, Any](
+        genFunction.name,
+        genFunction.code,
+        genFunction.returnType)
+    }
+
     new DataSetFlatMap(
       rel.getCluster,
       traitSet,
       convInput,
       rel.getRowType,
       filter.toString,
-      null)
+      func)
   }
 }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
index 7796d66..d747ba9 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/plan/rules/dataset/DataSetProjectRule.scala
@@ -21,8 +21,15 @@ package org.apache.flink.api.table.plan.rules.dataset
 import org.apache.calcite.plan.{RelOptRule, RelTraitSet}
 import org.apache.calcite.rel.RelNode
 import org.apache.calcite.rel.convert.ConverterRule
+import org.apache.flink.api.common.functions.MapFunction
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.table.TableConfig
+import org.apache.flink.api.table.codegen.CodeGenerator
 import org.apache.flink.api.table.plan.nodes.dataset.{DataSetConvention, DataSetMap}
-import org.apache.flink.api.table.plan.nodes.logical.{FlinkProject, FlinkConvention}
+import org.apache.flink.api.table.plan.nodes.logical.{FlinkConvention, FlinkProject}
+import org.apache.flink.api.table.runtime.MapRunner
+
+import scala.collection.JavaConversions._
 
 class DataSetProjectRule
   extends ConverterRule(
@@ -37,13 +44,40 @@ class DataSetProjectRule
     val traitSet: RelTraitSet = rel.getTraitSet.replace(DataSetConvention.INSTANCE)
     val convInput: RelNode = RelOptRule.convert(proj.getInput, DataSetConvention.INSTANCE)
 
+    val func = (
+        config: TableConfig,
+        inputType: TypeInformation[Any],
+        returnType: TypeInformation[Any]) => {
+      val generator = new CodeGenerator(config, inputType)
+
+      // projection and implicit type conversion
+      val projection = generator.generateResultExpression(returnType, proj.getProjects)
+
+      val body =
+        s"""
+          |${projection.code}
+          |return ${projection.resultTerm};
+          |""".stripMargin
+
+      val genFunction = generator.generateFunction(
+        description,
+        classOf[MapFunction[Any, Any]],
+        body,
+        returnType)
+
+      new MapRunner[Any, Any](
+        genFunction.name,
+        genFunction.code,
+        genFunction.returnType)
+    }
+
     new DataSetMap(
       rel.getCluster,
       traitSet,
       convInput,
       rel.getRowType,
       proj.toString,
-      null)
+      func)
   }
 }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionAggregateFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionAggregateFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionAggregateFunction.scala
deleted file mode 100644
index 38afc21..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionAggregateFunction.scala
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.runtime
-
-import org.apache.flink.api.table.Row
-import org.apache.flink.api.common.functions.{GroupReduceFunction, GroupCombineFunction, RichGroupReduceFunction}
-import org.apache.flink.api.java.aggregation.AggregationFunction
-import org.apache.flink.configuration.Configuration
-import org.apache.flink.util.Collector
-
-class ExpressionAggregateFunction(
-    private val fieldPositions: Seq[Int],
-    private val functions: Seq[AggregationFunction[Any]])
-  extends RichGroupReduceFunction[Row, Row]
-  with GroupCombineFunction[Row, Row]
-{
-
-  override def open(conf: Configuration): Unit = {
-    var i = 0
-    val len = functions.length
-    while (i < len) {
-      functions(i).initializeAggregate()
-      i += 1
-    }
-  }
-
-  override def reduce(in: java.lang.Iterable[Row], out: Collector[Row]): Unit = {
-
-    val fieldPositions = this.fieldPositions
-    val functions = this.functions
-
-    var current: Row = null
-
-    val values = in.iterator()
-    while (values.hasNext) {
-      current = values.next()
-
-      var i = 0
-      val len = functions.length
-      while (i < len) {
-        functions(i).aggregate(current.productElement(fieldPositions(i)))
-        i += 1
-      }
-    }
-
-    var i = 0
-    val len = functions.length
-    while (i < len) {
-      current.setField(fieldPositions(i), functions(i).getAggregate)
-      functions(i).initializeAggregate()
-      i += 1
-    }
-
-    out.collect(current)
-  }
-
-  override def combine(in: java.lang.Iterable[Row], out: Collector[Row]): Unit = {
-    reduce(in, out)
-  }
-
-}
-
-
-class NoExpressionAggregateFunction()
-  extends GroupReduceFunction[Row, Row]
-  with GroupCombineFunction[Row, Row]
-{
-
-  override def reduce(in: java.lang.Iterable[Row], out: Collector[Row]): Unit = {
-
-    var first: Row = null
-
-    val values = in.iterator()
-    if (values.hasNext) {
-      first = values.next()
-    }
-
-    out.collect(first)
-  }
-
-  override def combine(in: java.lang.Iterable[Row], out: Collector[Row]): Unit = {
-    reduce(in, out)
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionFilterFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionFilterFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionFilterFunction.scala
deleted file mode 100644
index 4e50272..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionFilterFunction.scala
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.runtime
-
-import org.apache.flink.api.common.functions.{FilterFunction, RichFilterFunction}
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.table.TableConfig
-import org.apache.flink.api.table.codegen.GenerateFilter
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.configuration.Configuration
-
-/**
- * Proxy function that takes an expression predicate. This is compiled
- * upon runtime and calls to [[filter()]] are forwarded to the compiled code.
- */
-class ExpressionFilterFunction[T](
-    predicate: Expression,
-    inputType: CompositeType[T],
-    config: TableConfig = TableConfig.DEFAULT) extends RichFilterFunction[T] {
-
-  var compiledFilter: FilterFunction[T] = null
-
-  override def open(c: Configuration): Unit = {
-    if (compiledFilter == null) {
-      val codegen = new GenerateFilter[T](
-        inputType,
-        predicate,
-        getRuntimeContext.getUserCodeClassLoader,
-        config)
-      compiledFilter = codegen.generate()
-    }
-  }
-
-  override def filter(in: T) = compiledFilter.filter(in)
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionJoinFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionJoinFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionJoinFunction.scala
deleted file mode 100644
index cf2c90f..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionJoinFunction.scala
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.runtime
-
-import org.apache.flink.api.common.functions.{FlatJoinFunction, RichFlatJoinFunction}
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.table.TableConfig
-import org.apache.flink.api.table.codegen.GenerateJoin
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.configuration.Configuration
-import org.apache.flink.util.Collector
-
-/**
- * Proxy function that takes an expression predicate and output fields. These are compiled
- * upon runtime and calls to [[join()]] are forwarded to the compiled code.
- */
-class ExpressionJoinFunction[L, R, O](
-    predicate: Expression,
-    leftType: CompositeType[L],
-    rightType: CompositeType[R],
-    resultType: CompositeType[O],
-    outputFields: Seq[Expression],
-    config: TableConfig = TableConfig.DEFAULT) extends RichFlatJoinFunction[L, R, O] {
-
-  var compiledJoin: FlatJoinFunction[L, R, O] = null
-
-  override def open(c: Configuration): Unit = {
-    val codegen = new GenerateJoin[L, R, O](
-      leftType,
-      rightType,
-      resultType,
-      predicate,
-      outputFields,
-      getRuntimeContext.getUserCodeClassLoader,
-      config)
-    compiledJoin = codegen.generate()
-  }
-
-  def join(left: L, right: R, out: Collector[O]) = {
-    compiledJoin.join(left, right, out)
-  }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionSelectFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionSelectFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionSelectFunction.scala
deleted file mode 100644
index ab7adb1..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/ExpressionSelectFunction.scala
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.runtime
-
-import org.apache.flink.api.table.TableConfig
-import org.apache.flink.api.table.expressions.Expression
-import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
-import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.table.codegen.GenerateSelect
-import org.apache.flink.configuration.Configuration
-
-/**
- * Proxy function that takes expressions. These are compiled
- * upon runtime and calls to [[map()]] are forwarded to the compiled code.
- */
-class ExpressionSelectFunction[I, O](
-     inputType: CompositeType[I],
-     resultType: CompositeType[O],
-     outputFields: Seq[Expression],
-     config: TableConfig = TableConfig.DEFAULT) extends RichMapFunction[I, O] {
-
-  var compiledSelect: MapFunction[I, O] = null
-
-  override def open(c: Configuration): Unit = {
-
-    if (compiledSelect == null) {
-      val resultCodegen = new GenerateSelect[I, O](
-        inputType,
-        resultType,
-        outputFields,
-        getRuntimeContext.getUserCodeClassLoader,
-        config)
-
-      compiledSelect = resultCodegen.generate()
-    }
-  }
-
-  def map(in: I): O = {
-    compiledSelect.map(in)
-  }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FlatMapRunner.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FlatMapRunner.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FlatMapRunner.scala
new file mode 100644
index 0000000..8a3482f
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FlatMapRunner.scala
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.runtime
+
+import org.apache.flink.api.common.functions.{FlatMapFunction, RichFlatMapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.util.Collector
+import org.slf4j.LoggerFactory
+
+class FlatMapRunner[IN, OUT](
+    name: String,
+    code: String,
+    @transient returnType: TypeInformation[OUT])
+  extends RichFlatMapFunction[IN, OUT]
+  with ResultTypeQueryable[OUT]
+  with FunctionCompiler[FlatMapFunction[IN, OUT]] {
+
+  val LOG = LoggerFactory.getLogger(this.getClass)
+
+  private var function: FlatMapFunction[IN, OUT] = null
+
+  override def open(parameters: Configuration): Unit = {
+    LOG.debug(s"Compiling FlatMapFunction: $name \n\n Code:\n$code")
+    val clazz = compile(getRuntimeContext.getUserCodeClassLoader, name, code)
+    LOG.debug("Instantiating FlatMapFunction.")
+    function = clazz.newInstance()
+  }
+
+  override def flatMap(in: IN, out: Collector[OUT]): Unit =
+    function.flatMap(in, out)
+
+  override def getProducedType: TypeInformation[OUT] = returnType
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FunctionCompiler.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FunctionCompiler.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FunctionCompiler.scala
new file mode 100644
index 0000000..9cd44d1
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/FunctionCompiler.scala
@@ -0,0 +1,35 @@
+/*
+ * 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.runtime
+
+import org.apache.flink.api.common.functions.Function
+import org.codehaus.commons.compiler.CompileException
+import org.codehaus.janino.SimpleCompiler
+
+trait FunctionCompiler[T <: Function] {
+
+  @throws(classOf[CompileException])
+  def compile(cl: ClassLoader, name: String, code: String): Class[T] = {
+    require(cl != null, "Classloader must not be null.")
+    val compiler = new SimpleCompiler()
+    compiler.setParentClassLoader(cl)
+    compiler.cook(code)
+    compiler.getClassLoader.loadClass(name).asInstanceOf[Class[T]]
+  }
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/MapRunner.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/MapRunner.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/MapRunner.scala
new file mode 100644
index 0000000..f64635b
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/MapRunner.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.table.runtime
+
+import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable
+import org.apache.flink.configuration.Configuration
+import org.slf4j.LoggerFactory
+
+class MapRunner[IN, OUT](
+    name: String,
+    code: String,
+    @transient returnType: TypeInformation[OUT])
+  extends RichMapFunction[IN, OUT]
+  with ResultTypeQueryable[OUT]
+  with FunctionCompiler[MapFunction[IN, OUT]] {
+
+  val LOG = LoggerFactory.getLogger(this.getClass)
+
+  private var function: MapFunction[IN, OUT] = null
+
+  override def open(parameters: Configuration): Unit = {
+    LOG.debug(s"Compiling MapFunction: $name \n\n Code:\n$code")
+    val clazz = compile(getRuntimeContext.getUserCodeClassLoader, name, code)
+    LOG.debug("Instantiating MapFunction.")
+    function = clazz.newInstance()
+  }
+
+  override def map(in: IN): OUT =
+    function.map(in)
+
+  override def getProducedType: TypeInformation[OUT] = returnType
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/package.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/package.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/package.scala
deleted file mode 100644
index a1bc4b7..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/runtime/package.scala
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table
-
-/**
- * The functions in this package are used transforming Table API operations to Java API operations.
- */
-package object runtime

http://git-wip-us.apache.org/repos/asf/flink/blob/a4ad9dd5/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RenameOperator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RenameOperator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RenameOperator.scala
deleted file mode 100644
index 3b5459b..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/api/table/typeinfo/RenameOperator.scala
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.api.table.typeinfo
-
-import org.apache.flink.api.common.operators.Operator
-import org.apache.flink.api.java.operators.SingleInputOperator
-import org.apache.flink.api.java.{DataSet => JavaDataSet}
-
-/**
- * This is a logical operator that can hold a [[RenamingProxyTypeInfo]] for renaming some
- * fields of a [[org.apache.flink.api.common.typeutils.CompositeType]]. At runtime this
- * disappears since the translation methods simply returns the input.
- */
-class RenameOperator[T](
-    input: JavaDataSet[T],
-    renamingTypeInformation: RenamingProxyTypeInfo[T])
-  extends SingleInputOperator[T, T, RenameOperator[T]](input, renamingTypeInformation) {
-
-  override protected def translateToDataFlow(
-      input: Operator[T]): Operator[T] = input
-}


Mime
View raw message