flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From twal...@apache.org
Subject [2/5] flink git commit: [FLINK-7337] [table] Efficient handling of rowtime timestamps
Date Wed, 23 Aug 2017 08:15:13 GMT
[FLINK-7337] [table] Efficient handling of rowtime timestamps

Use Long instead of a SQL Timestamp to represent timestamps internally

This closes #4532.


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

Branch: refs/heads/master
Commit: 47944b1bb23136ae498971b3765a0d3fe6bf2f18
Parents: 93d0ae4
Author: twalthr <twalthr@apache.org>
Authored: Sat Aug 12 13:51:42 2017 +0200
Committer: twalthr <twalthr@apache.org>
Committed: Wed Aug 23 10:09:21 2017 +0200

----------------------------------------------------------------------
 .../flink/table/api/BatchTableEnvironment.scala |  36 +-
 .../table/api/StreamTableEnvironment.scala      | 245 +++++------
 .../flink/table/api/TableEnvironment.scala      |  73 ++--
 .../calcite/RelTimeIndicatorConverter.scala     |   2 +-
 .../codegen/AggregationCodeGenerator.scala      |  25 +-
 .../flink/table/codegen/CodeGenUtils.scala      |  29 +-
 .../flink/table/codegen/CodeGenerator.scala     |  21 +-
 .../table/codegen/calls/ScalarOperators.scala   |   7 +-
 .../table/functions/ProctimeSqlFunction.scala   |  41 --
 .../functions/sql/ProctimeSqlFunction.scala     |  43 ++
 .../utils/UserDefinedFunctionUtils.scala        |   6 +-
 .../DataStreamGroupWindowAggregate.scala        |   4 +-
 .../table/runtime/CRowInputMapRunner.scala      |  57 ---
 .../runtime/CRowInputTupleOutputMapRunner.scala |  94 ----
 .../flink/table/runtime/CRowMapRunner.scala     |  57 +++
 .../runtime/OutputRowtimeProcessFunction.scala  |  58 +++
 .../table/runtime/RowtimeProcessFunction.scala  |  49 +++
 .../TimestampSetterProcessFunction.scala        |  52 ---
 ...WrappingTimestampSetterProcessFunction.scala |  61 ---
 .../table/runtime/aggregate/AggregateUtil.scala |  33 +-
 .../aggregate/RowTimeBoundedRangeOver.scala     |   4 +-
 .../aggregate/RowTimeBoundedRowsOver.scala      |  13 +-
 .../aggregate/RowTimeSortProcessFunction.scala  |  15 +-
 .../aggregate/RowTimeUnboundedOver.scala        |  17 +-
 .../aggregate/TimeWindowPropertyCollector.scala |   2 +-
 .../conversion/CRowToJavaTupleMapFunction.scala |  40 ++
 .../conversion/CRowToJavaTupleMapRunner.scala   |  64 +++
 .../conversion/CRowToRowMapFunction.scala       |  32 ++
 .../CRowToScalaTupleMapFunction.scala           |  33 ++
 .../conversion/CRowToScalaTupleMapRunner.scala  |  56 +++
 .../table/typeutils/TimeIndicatorTypeInfo.scala |  13 +-
 .../runtime/harness/OverWindowHarnessTest.scala | 436 +++++++++----------
 .../SortProcessFunctionHarnessTest.scala        |  49 +--
 .../runtime/stream/TimeAttributesITCase.scala   |  10 +-
 .../runtime/stream/table/TableSinkITCase.scala  |   4 +-
 35 files changed, 952 insertions(+), 829 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
index 78667a2..a9d60dd 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
@@ -153,28 +153,22 @@ abstract class BatchTableEnvironment(
       physicalTypeInfo: TypeInformation[IN],
       schema: RowSchema,
       requestedTypeInfo: TypeInformation[OUT],
-      functionName: String):
-    Option[MapFunction[IN, OUT]] = {
-
-    if (requestedTypeInfo.getTypeClass == classOf[Row]) {
-      // Row to Row, no conversion needed
-      None
-    } else {
-      // some type that is neither Row or CRow
-
-      val converterFunction = generateRowConverterFunction[OUT](
-        physicalTypeInfo.asInstanceOf[TypeInformation[Row]],
-        schema,
-        requestedTypeInfo,
-        functionName
-      )
-
-      val mapFunction = new MapRunner[IN, OUT](
-        converterFunction.name,
-        converterFunction.code,
-        converterFunction.returnType)
+      functionName: String)
+    : Option[MapFunction[IN, OUT]] = {
+
+    val converterFunction = generateRowConverterFunction[OUT](
+      physicalTypeInfo.asInstanceOf[TypeInformation[Row]],
+      schema,
+      requestedTypeInfo,
+      functionName
+    )
 
-      Some(mapFunction)
+    // add a runner if we need conversion
+    converterFunction.map { func =>
+      new MapRunner[IN, OUT](
+          func.name,
+          func.code,
+          func.returnType)
     }
   }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
index c4e1450..8d8cebb 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
@@ -23,15 +23,15 @@ import _root_.java.util.concurrent.atomic.AtomicInteger
 
 import org.apache.calcite.plan.RelOptUtil
 import org.apache.calcite.plan.hep.HepMatchOrder
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField, RelDataTypeFieldImpl, RelRecordType}
 import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField, RelDataTypeFieldImpl, RelRecordType}
 import org.apache.calcite.sql2rel.RelDecorrelator
 import org.apache.calcite.tools.{RuleSet, RuleSets}
 import org.apache.flink.api.common.functions.MapFunction
-import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation}
+import org.apache.flink.api.common.typeinfo.{AtomicType, SqlTimeTypeInfo, TypeInformation}
 import org.apache.flink.api.common.typeutils.CompositeType
 import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2}
-import org.apache.flink.api.java.typeutils.{PojoTypeInfo, TupleTypeInfo}
+import org.apache.flink.api.java.typeutils.{PojoTypeInfo, RowTypeInfo, TupleTypeInfo}
 import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
 import org.apache.flink.streaming.api.TimeCharacteristic
 import org.apache.flink.streaming.api.datastream.DataStream
@@ -44,12 +44,12 @@ import org.apache.flink.table.plan.nodes.datastream.{DataStreamRel, UpdateAsRetr
 import org.apache.flink.table.plan.rules.FlinkRuleSets
 import org.apache.flink.table.plan.schema.{DataStreamTable, RowSchema, StreamTableSourceTable}
 import org.apache.flink.table.plan.util.UpdatingPlanChecker
+import org.apache.flink.table.runtime.conversion._
 import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
-import org.apache.flink.table.runtime.{CRowInputJavaTupleOutputMapRunner, CRowInputMapRunner, CRowInputScalaTupleOutputMapRunner, WrappingTimestampSetterProcessFunction}
+import org.apache.flink.table.runtime.{CRowMapRunner, OutputRowtimeProcessFunction}
 import org.apache.flink.table.sinks.{AppendStreamTableSink, RetractStreamTableSink, TableSink, UpsertStreamTableSink}
 import org.apache.flink.table.sources.{DefinedRowtimeAttribute, StreamTableSource, TableSource}
 import org.apache.flink.table.typeutils.{TimeIndicatorTypeInfo, TypeCheckUtils}
-import org.apache.flink.types.Row
 
 import _root_.scala.collection.JavaConverters._
 
@@ -223,38 +223,33 @@ abstract class StreamTableEnvironment(
   /**
     * Creates a final converter that maps the internal row type to external type.
     *
-    * @param physicalTypeInfo the input of the sink
+    * @param inputTypeInfo the input of the sink
     * @param schema the input schema with correct field names (esp. for POJO field mapping)
     * @param requestedTypeInfo the output type of the sink
     * @param functionName name of the map function. Must not be unique but has to be a
     *                     valid Java class identifier.
     */
-  protected def getConversionMapper[IN, OUT](
-      physicalTypeInfo: TypeInformation[IN],
+  protected def getConversionMapper[OUT](
+      inputTypeInfo: TypeInformation[CRow],
       schema: RowSchema,
       requestedTypeInfo: TypeInformation[OUT],
-      functionName: String):
-    MapFunction[IN, OUT] = {
-
-    if (requestedTypeInfo.getTypeClass == classOf[Row]) {
-      // CRow to Row, only needs to be unwrapped
-      new MapFunction[CRow, Row] {
-        override def map(value: CRow): Row = value.row
-      }.asInstanceOf[MapFunction[IN, OUT]]
-    } else {
-      // Some type that is neither CRow nor Row
-      val converterFunction = generateRowConverterFunction[OUT](
-        physicalTypeInfo.asInstanceOf[CRowTypeInfo].rowType,
-        schema,
-        requestedTypeInfo,
-        functionName
-      )
+      functionName: String)
+    : MapFunction[CRow, OUT] = {
+
+    val converterFunction = generateRowConverterFunction[OUT](
+      inputTypeInfo.asInstanceOf[CRowTypeInfo].rowType,
+      schema,
+      requestedTypeInfo,
+      functionName
+    )
 
-      new CRowInputMapRunner[OUT](
-        converterFunction.name,
-        converterFunction.code,
-        converterFunction.returnType)
-        .asInstanceOf[MapFunction[IN, OUT]]
+    converterFunction match {
+
+      case Some(func) =>
+        new CRowMapRunner[OUT](func.name, func.code, func.returnType)
+
+      case _ =>
+        new CRowToRowMapFunction().asInstanceOf[MapFunction[CRow, OUT]]
     }
   }
 
@@ -271,71 +266,62 @@ abstract class StreamTableEnvironment(
       physicalTypeInfo: TypeInformation[CRow],
       schema: RowSchema,
       requestedTypeInfo: TypeInformation[OUT],
-      functionName: String):
-    MapFunction[CRow, OUT] = {
-
-    requestedTypeInfo match {
-
-      // Scala tuple
-      case t: CaseClassTypeInfo[_]
-        if t.getTypeClass == classOf[(_, _)] && t.getTypeAt(0) == Types.BOOLEAN =>
-
-        val reqType = t.getTypeAt(1).asInstanceOf[TypeInformation[Any]]
-        if (reqType.getTypeClass == classOf[Row]) {
-          // Requested type is Row. Just rewrap CRow in Tuple2
-          new MapFunction[CRow, (Boolean, Row)] {
-            override def map(cRow: CRow): (Boolean, Row) = {
-              (cRow.change, cRow.row)
-            }
-          }.asInstanceOf[MapFunction[CRow, OUT]]
-        } else {
-          // Use a map function to convert Row into requested type and wrap result in Tuple2
-          val converterFunction = generateRowConverterFunction(
-            physicalTypeInfo.asInstanceOf[CRowTypeInfo].rowType,
-            schema,
-            reqType,
-            functionName
-          )
-
-          new CRowInputScalaTupleOutputMapRunner(
-            converterFunction.name,
-            converterFunction.code,
-            requestedTypeInfo.asInstanceOf[TypeInformation[(Boolean, Any)]])
-            .asInstanceOf[MapFunction[CRow, OUT]]
+      functionName: String)
+    : MapFunction[CRow, OUT] = requestedTypeInfo match {
 
-        }
+    // Scala tuple
+    case t: CaseClassTypeInfo[_]
+      if t.getTypeClass == classOf[(_, _)] && t.getTypeAt(0) == Types.BOOLEAN =>
 
-      // Java tuple
-      case t: TupleTypeInfo[_]
-        if t.getTypeClass == classOf[JTuple2[_, _]] && t.getTypeAt(0) == Types.BOOLEAN =>
-
-        val reqType = t.getTypeAt(1).asInstanceOf[TypeInformation[Any]]
-        if (reqType.getTypeClass == classOf[Row]) {
-          // Requested type is Row. Just rewrap CRow in Tuple2
-          new MapFunction[CRow, JTuple2[JBool, Row]] {
-            val outT = new JTuple2(true.asInstanceOf[JBool], null.asInstanceOf[Row])
-            override def map(cRow: CRow): JTuple2[JBool, Row] = {
-              outT.f0 = cRow.change
-              outT.f1 = cRow.row
-              outT
-            }
-          }.asInstanceOf[MapFunction[CRow, OUT]]
-        } else {
-          // Use a map function to convert Row into requested type and wrap result in Tuple2
-          val converterFunction = generateRowConverterFunction(
-            physicalTypeInfo.asInstanceOf[CRowTypeInfo].rowType,
-            schema,
-            reqType,
-            functionName
-          )
-
-          new CRowInputJavaTupleOutputMapRunner(
-            converterFunction.name,
-            converterFunction.code,
-            requestedTypeInfo.asInstanceOf[TypeInformation[JTuple2[JBool, Any]]])
-            .asInstanceOf[MapFunction[CRow, OUT]]
-        }
-    }
+      val reqType = t.getTypeAt[Any](1)
+
+      // convert Row into requested type and wrap result in Tuple2
+      val converterFunction = generateRowConverterFunction(
+        physicalTypeInfo.asInstanceOf[CRowTypeInfo].rowType,
+        schema,
+        reqType,
+        functionName
+      )
+
+      converterFunction match {
+
+        case Some(func) =>
+          new CRowToScalaTupleMapRunner(
+            func.name,
+            func.code,
+            requestedTypeInfo.asInstanceOf[TypeInformation[(Boolean, Any)]]
+          ).asInstanceOf[MapFunction[CRow, OUT]]
+
+        case _ =>
+          new CRowToScalaTupleMapFunction().asInstanceOf[MapFunction[CRow, OUT]]
+      }
+
+    // Java tuple
+    case t: TupleTypeInfo[_]
+      if t.getTypeClass == classOf[JTuple2[_, _]] && t.getTypeAt(0) == Types.BOOLEAN =>
+
+      val reqType = t.getTypeAt[Any](1)
+
+      // convert Row into requested type and wrap result in Tuple2
+      val converterFunction = generateRowConverterFunction(
+        physicalTypeInfo.asInstanceOf[CRowTypeInfo].rowType,
+        schema,
+        reqType,
+        functionName
+      )
+
+      converterFunction match {
+
+        case Some(func) =>
+          new CRowToJavaTupleMapRunner(
+            func.name,
+            func.code,
+            requestedTypeInfo.asInstanceOf[TypeInformation[JTuple2[JBool, Any]]]
+          ).asInstanceOf[MapFunction[CRow, OUT]]
+
+        case _ =>
+          new CRowToJavaTupleMapFunction().asInstanceOf[MapFunction[CRow, OUT]]
+      }
   }
 
   /**
@@ -733,16 +719,42 @@ abstract class StreamTableEnvironment(
     // get CRow plan
     val plan: DataStream[CRow] = translateToCRow(logicalPlan, queryConfig)
 
+    val rowtimeFields = logicalType
+      .getFieldList.asScala
+      .filter(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
+
+    // convert the input type for the conversion mapper
+    // the input will be changed in the OutputRowtimeProcessFunction later
+    val convType = if (rowtimeFields.size > 1) {
+      throw new TableException(
+        s"Found more than one rowtime field: [${rowtimeFields.map(_.getName).mkString(", ")}] in " +
+          s"the table that should be converted to a DataStream.\n" +
+          s"Please select the rowtime field that should be used as event-time timestamp for the " +
+          s"DataStream by casting all other fields to TIMESTAMP.")
+    } else if (rowtimeFields.size == 1) {
+      val origRowType = plan.getType.asInstanceOf[CRowTypeInfo].rowType
+      val convFieldTypes = origRowType.getFieldTypes.map { t =>
+        if (FlinkTypeFactory.isRowtimeIndicatorType(t)) {
+          SqlTimeTypeInfo.TIMESTAMP
+        } else {
+          t
+        }
+      }
+      CRowTypeInfo(new RowTypeInfo(convFieldTypes, origRowType.getFieldNames))
+    } else {
+      plan.getType
+    }
+
     // convert CRow to output type
-    val conversion = if (withChangeFlag) {
+    val conversion: MapFunction[CRow, A] = if (withChangeFlag) {
       getConversionMapperWithChanges(
-        plan.getType,
+        convType,
         new RowSchema(logicalType),
         tpe,
         "DataStreamSinkConversion")
     } else {
       getConversionMapper(
-        plan.getType,
+        convType,
         new RowSchema(logicalType),
         tpe,
         "DataStreamSinkConversion")
@@ -750,42 +762,19 @@ abstract class StreamTableEnvironment(
 
     val rootParallelism = plan.getParallelism
 
-    val rowtimeFields = logicalType.getFieldList.asScala
-      .filter(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
-
-    if (rowtimeFields.isEmpty) {
+    val withRowtime = if (rowtimeFields.isEmpty) {
       // no rowtime field to set
-      conversion match {
-        case mapFunction: MapFunction[CRow, A] =>
-          plan.map(mapFunction)
-            .returns(tpe)
-            .name(s"to: ${tpe.getTypeClass.getSimpleName}")
-            .setParallelism(rootParallelism)
-      }
-    } else if (rowtimeFields.size == 1) {
-      // set the only rowtime field as event-time timestamp for DataStream
-      val mapFunction = conversion match {
-        case mapFunction: MapFunction[CRow, A] => mapFunction
-        case _ => new MapFunction[CRow, A] {
-          override def map(cRow: CRow): A = cRow.asInstanceOf[A]
-        }
-      }
-
-      plan.process(
-        new WrappingTimestampSetterProcessFunction[A](
-          mapFunction,
-          rowtimeFields.head.getIndex))
-        .returns(tpe)
-        .name(s"to: ${tpe.getTypeClass.getSimpleName}")
-        .setParallelism(rootParallelism)
-
+      plan.map(conversion)
     } else {
-      throw new TableException(
-        s"Found more than one rowtime field: [${rowtimeFields.map(_.getName).mkString(", ")}] in " +
-          s"the table that should be converted to a DataStream.\n" +
-          s"Please select the rowtime field that should be used as event-time timestamp for the " +
-          s"DataStream by casting all other fields to TIMESTAMP.")
+      // set the only rowtime field as event-time timestamp for DataStream
+      // and convert it to SQL timestamp
+      plan.process(new OutputRowtimeProcessFunction[A](conversion, rowtimeFields.head.getIndex))
     }
+
+    withRowtime
+      .returns(tpe)
+      .name(s"to: ${tpe.getTypeClass.getSimpleName}")
+      .setParallelism(rootParallelism)
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index b647c51..2e9e18f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -38,7 +38,7 @@ import org.apache.calcite.tools._
 import org.apache.flink.api.common.functions.MapFunction
 import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation}
 import org.apache.flink.api.common.typeutils.CompositeType
-import org.apache.flink.api.java.typeutils._
+import org.apache.flink.api.java.typeutils.{RowTypeInfo, _}
 import org.apache.flink.api.java.{ExecutionEnvironment => JavaBatchExecEnv}
 import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
 import org.apache.flink.api.scala.{ExecutionEnvironment => ScalaBatchExecEnv}
@@ -48,26 +48,23 @@ import org.apache.flink.table.api.java.{BatchTableEnvironment => JavaBatchTableE
 import org.apache.flink.table.api.scala.{BatchTableEnvironment => ScalaBatchTableEnv, StreamTableEnvironment => ScalaStreamTableEnv}
 import org.apache.flink.table.calcite.{FlinkPlannerImpl, FlinkRelBuilder, FlinkTypeFactory, FlinkTypeSystem}
 import org.apache.flink.table.catalog.{ExternalCatalog, ExternalCatalogSchema}
-import org.apache.flink.table.codegen.{FunctionCodeGenerator, ExpressionReducer, GeneratedFunction}
-import org.apache.flink.table.expressions.{Alias, Expression, UnresolvedFieldReference}
-import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
-import org.apache.flink.table.functions.AggregateFunction
-import org.apache.flink.table.expressions._
-import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{checkForInstantiation, checkNotSingleton, createScalarSqlFunction, createTableSqlFunctions}
-import org.apache.flink.table.functions.{ScalarFunction, TableFunction}
+import org.apache.flink.table.codegen.{ExpressionReducer, FunctionCodeGenerator, GeneratedFunction}
+import org.apache.flink.table.expressions.{Alias, Expression, UnresolvedFieldReference, _}
+import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{checkForInstantiation, checkNotSingleton, createScalarSqlFunction, createTableSqlFunctions, _}
+import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction}
 import org.apache.flink.table.plan.cost.DataSetCostFactory
 import org.apache.flink.table.plan.logical.{CatalogNode, LogicalRelNode}
 import org.apache.flink.table.plan.rules.FlinkRuleSets
 import org.apache.flink.table.plan.schema.{RelTable, RowSchema}
 import org.apache.flink.table.sinks.TableSink
 import org.apache.flink.table.sources.{DefinedFieldNames, TableSource}
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
 import org.apache.flink.table.validate.FunctionCatalog
 import org.apache.flink.types.Row
-import org.apache.flink.api.java.typeutils.RowTypeInfo
 
-import _root_.scala.collection.JavaConverters._
-import _root_.scala.collection.mutable.HashMap
 import _root_.scala.annotation.varargs
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
 
 /**
   * The abstract base class for batch and stream TableEnvironments.
@@ -108,10 +105,10 @@ abstract class TableEnvironment(val config: TableConfig) {
   private[flink] val attrNameCntr: AtomicInteger = new AtomicInteger(0)
 
   // registered external catalog names -> catalog
-  private val externalCatalogs = new HashMap[String, ExternalCatalog]
+  private val externalCatalogs = new mutable.HashMap[String, ExternalCatalog]
 
   /** Returns the table config to define the runtime behavior of the Table API. */
-  def getConfig = config
+  def getConfig: TableConfig = config
 
   /**
     * Returns the operator table for this environment including a custom Calcite configuration.
@@ -692,7 +689,7 @@ abstract class TableEnvironment(val config: TableConfig) {
           case _ => throw new TableException(
             "Field reference expression or alias on field expression expected.")
         }
-      case r: RowTypeInfo => {
+      case r: RowTypeInfo =>
         exprs.zipWithIndex flatMap {
           case (UnresolvedFieldReference(name), idx) =>
             Some((idx, name))
@@ -707,8 +704,7 @@ abstract class TableEnvironment(val config: TableConfig) {
           case _ => throw new TableException(
             "Field reference expression or alias on field expression expected.")
         }
-        
-      }
+
       case tpe => throw new TableException(
         s"Source of type $tpe cannot be converted into Table.")
     }
@@ -719,33 +715,47 @@ abstract class TableEnvironment(val config: TableConfig) {
       throw new TableException("Field name can not be '*'.")
     }
 
-    (fieldNames.toArray, fieldIndexes.toArray)
+    (fieldNames.toArray, fieldIndexes.toArray) // build fails in Scala 2.10 if not converted
   }
 
   protected def generateRowConverterFunction[OUT](
       inputTypeInfo: TypeInformation[Row],
       schema: RowSchema,
       requestedTypeInfo: TypeInformation[OUT],
-      functionName: String):
-    GeneratedFunction[MapFunction[Row, OUT], OUT] = {
+      functionName: String)
+    : Option[GeneratedFunction[MapFunction[Row, OUT], OUT]] = {
 
     // validate that at least the field types of physical and logical type match
     // we do that here to make sure that plan translation was correct
     if (schema.typeInfo != inputTypeInfo) {
       throw TableException(
         s"The field types of physical and logical row types do not match. " +
-        s"Physical type is [${schema.typeInfo}], Logical type is [${inputTypeInfo}]. " +
+        s"Physical type is [${schema.typeInfo}], Logical type is [$inputTypeInfo]. " +
         s"This is a bug and should not happen. Please file an issue.")
     }
 
+    // generic row needs no conversion
+    if (requestedTypeInfo.isInstanceOf[GenericTypeInfo[_]] &&
+        requestedTypeInfo.getTypeClass == classOf[Row]) {
+      return None
+    }
+
     val fieldTypes = schema.fieldTypeInfos
     val fieldNames = schema.fieldNames
 
-    // validate requested type
+    // check for valid type info
     if (requestedTypeInfo.getArity != fieldTypes.length) {
       throw new TableException(
-        s"Arity[${fieldTypes.length}] of result[${fieldTypes}] does not match " +
-        s"the number[${requestedTypeInfo.getArity}] of requested type[${requestedTypeInfo}].")
+        s"Arity [${fieldTypes.length}] of result [$fieldTypes] does not match " +
+        s"the number[${requestedTypeInfo.getArity}] of requested type [$requestedTypeInfo].")
+    }
+
+    // check requested types
+
+    def validateFieldType(fieldType: TypeInformation[_]): Unit = fieldType match {
+      case _: TimeIndicatorTypeInfo =>
+        throw new TableException("The time indicator type is an internal type only.")
+      case _ => // ok
     }
 
     requestedTypeInfo match {
@@ -758,9 +768,10 @@ abstract class TableEnvironment(val config: TableConfig) {
               throw new TableException(s"POJO does not define field name: $fName")
             }
             val requestedTypeInfo = pt.getTypeAt(pojoIdx)
+            validateFieldType(requestedTypeInfo)
             if (fType != requestedTypeInfo) {
               throw new TableException(s"Result field does not match requested type. " +
-                s"requested: $requestedTypeInfo; Actual: $fType")
+                s"Requested: $requestedTypeInfo; Actual: $fType")
             }
         }
 
@@ -769,6 +780,7 @@ abstract class TableEnvironment(val config: TableConfig) {
         fieldTypes.zipWithIndex foreach {
           case (fieldTypeInfo, i) =>
             val requestedTypeInfo = tt.getTypeAt(i)
+            validateFieldType(requestedTypeInfo)
             if (fieldTypeInfo != requestedTypeInfo) {
               throw new TableException(s"Result field does not match requested type. " +
                 s"Requested: $requestedTypeInfo; Actual: $fieldTypeInfo")
@@ -781,10 +793,11 @@ abstract class TableEnvironment(val config: TableConfig) {
           throw new TableException(s"Requested result type is an atomic type but " +
             s"result[$fieldTypes] has more or less than a single field.")
         }
-        val fieldTypeInfo = fieldTypes.head
-        if (fieldTypeInfo != at) {
+        val requestedTypeInfo = fieldTypes.head
+        validateFieldType(requestedTypeInfo)
+        if (requestedTypeInfo != at) {
           throw new TableException(s"Result field does not match requested type. " +
-            s"Requested: $at; Actual: $fieldTypeInfo")
+            s"Requested: $at; Actual: $requestedTypeInfo")
         }
 
       case _ =>
@@ -809,11 +822,13 @@ abstract class TableEnvironment(val config: TableConfig) {
          |return ${conversion.resultTerm};
          |""".stripMargin
 
-    generator.generateFunction(
+    val generated = generator.generateFunction(
       functionName,
       classOf[MapFunction[Row, OUT]],
       body,
       requestedTypeInfo)
+
+    Some(generated)
   }
 }
 
@@ -972,7 +987,7 @@ object TableEnvironment {
     validateType(inputType)
 
     inputType match {
-      case t: CompositeType[_] => 0.until(t.getArity).map(t.getTypeAt(_)).toArray
+      case t: CompositeType[_] => 0.until(t.getArity).map(i => t.getTypeAt(i)).toArray
       case a: AtomicType[_] => Array(a.asInstanceOf[TypeInformation[_]])
       case tpe =>
         throw new TableException(s"Currently only CompositeType and AtomicType are supported.")

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
index 717a1af..1f88737 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/RelTimeIndicatorConverter.scala
@@ -27,7 +27,7 @@ import org.apache.calcite.sql.fun.SqlStdOperatorTable
 import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo
 import org.apache.flink.table.api.{TableException, ValidationException}
 import org.apache.flink.table.calcite.FlinkTypeFactory.{isRowtimeIndicatorType, _}
-import org.apache.flink.table.functions.ProctimeSqlFunction
+import org.apache.flink.table.functions.sql.ProctimeSqlFunction
 import org.apache.flink.table.plan.logical.rel.LogicalWindowAggregate
 import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
 

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala
index 680eb44..25527cc 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala
@@ -25,6 +25,7 @@ import org.apache.flink.table.api.TableConfig
 import org.apache.flink.table.codegen.Indenter.toISC
 import org.apache.flink.table.codegen.CodeGenUtils.newName
 import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
 import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{getUserDefinedMethod, signatureToString}
 import org.apache.flink.table.runtime.aggregate.{GeneratedAggregations, SingleElementIterable}
 
@@ -47,7 +48,6 @@ class AggregationCodeGenerator(
     *
     * @param name        Class name of the function.
     *                    Does not need to be unique but has to be a valid Java class identifier.
-    * @param generator   The code generator instance
     * @param physicalInputTypes Physical input row types
     * @param aggregates  All aggregate functions
     * @param aggFields   Indexes of the input fields for all aggregate functions
@@ -68,7 +68,6 @@ class AggregationCodeGenerator(
     */
   def generateAggregations(
     name: String,
-    generator: CodeGenerator,
     physicalInputTypes: Seq[TypeInformation[_]],
     aggregates: Array[AggregateFunction[_ <: Any, _ <: Any]],
     aggFields: Array[Array[Int]],
@@ -86,28 +85,29 @@ class AggregationCodeGenerator(
     // get unique function name
     val funcName = newName(name)
     // register UDAGGs
-    val aggs = aggregates.map(a => generator.addReusableFunction(a))
+    val aggs = aggregates.map(a => addReusableFunction(a))
     // get java types of accumulators
     val accTypeClasses = aggregates.map { a =>
       a.getClass.getMethod("createAccumulator").getReturnType
     }
     val accTypes = accTypeClasses.map(_.getCanonicalName)
 
-    // get java classes of input fields
-    val javaClasses = physicalInputTypes.map(t => t.getTypeClass)
     // get parameter lists for aggregation functions
-    val parameters = aggFields.map { inFields =>
+    val parametersCode = aggFields.map { inFields =>
       val fields = for (f <- inFields) yield
-        s"(${javaClasses(f).getCanonicalName}) input.getField($f)"
+        s"(${CodeGenUtils.boxedTypeTermForTypeInfo(physicalInputTypes(f))}) input.getField($f)"
       fields.mkString(", ")
     }
-    val methodSignaturesList = aggFields.map {
-      inFields => for (f <- inFields) yield javaClasses(f)
+
+    // get method signatures
+    val classes = UserDefinedFunctionUtils.typeInfoToClass(physicalInputTypes)
+    val methodSignaturesList = aggFields.map { inFields =>
+      inFields.map(classes(_))
     }
 
     // check and validate the needed methods
     aggregates.zipWithIndex.map {
-      case (a, i) => {
+      case (a, i) =>
         getUserDefinedMethod(a, "accumulate", Array(accTypeClasses(i)) ++ methodSignaturesList(i))
         .getOrElse(
           throw new CodeGenException(
@@ -159,7 +159,6 @@ class AggregationCodeGenerator(
                 s"aggregate ${a.getClass.getCanonicalName}'.")
           )
         }
-      }
     }
 
     def genSetAggregationResults: String = {
@@ -208,7 +207,7 @@ class AggregationCodeGenerator(
           j"""
              |    ${aggs(i)}.accumulate(
              |      ((${accTypes(i)}) accs.getField($i)),
-             |      ${parameters(i)});""".stripMargin
+             |      ${parametersCode(i)});""".stripMargin
       }.mkString("\n")
 
       j"""$sig {
@@ -229,7 +228,7 @@ class AggregationCodeGenerator(
           j"""
              |    ${aggs(i)}.retract(
              |      ((${accTypes(i)}) accs.getField($i)),
-             |      ${parameters(i)});""".stripMargin
+             |      ${parametersCode(i)});""".stripMargin
       }.mkString("\n")
 
       if (needRetract) {

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenUtils.scala
index 1d8c926..161f9a3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenUtils.scala
@@ -28,7 +28,7 @@ import org.apache.flink.api.common.typeinfo.{FractionalTypeInfo, SqlTimeTypeInfo
 import org.apache.flink.api.common.typeutils.CompositeType
 import org.apache.flink.api.java.typeutils.{PojoTypeInfo, RowTypeInfo, TupleTypeInfo, TypeExtractor}
 import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
-import org.apache.flink.table.typeutils.{TimeIntervalTypeInfo, TypeCheckUtils}
+import org.apache.flink.table.typeutils.{TimeIndicatorTypeInfo, TimeIntervalTypeInfo, TypeCheckUtils}
 
 object CodeGenUtils {
 
@@ -90,6 +90,9 @@ object CodeGenUtils {
     case BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO => "boolean[]"
     case CHAR_PRIMITIVE_ARRAY_TYPE_INFO => "char[]"
 
+    // time indicators are represented as Long even if they seem to be Timestamp
+    case _: TimeIndicatorTypeInfo => "java.lang.Long"
+
     case _ =>
       tpe.getTypeClass.getCanonicalName
   }
@@ -123,8 +126,10 @@ object CodeGenUtils {
   def qualifyEnum(enum: Enum[_]): String =
     enum.getClass.getCanonicalName + "." + enum.name()
 
-  def internalToTimePointCode(resultType: TypeInformation[_], resultTerm: String) =
+  def internalToTimePointCode(resultType: TypeInformation[_], resultTerm: String): String =
     resultType match {
+      case _: TimeIndicatorTypeInfo =>
+        resultTerm // time indicators are not modified
       case SqlTimeTypeInfo.DATE =>
         s"${qualifyMethod(BuiltInMethod.INTERNAL_TO_DATE.method)}($resultTerm)"
       case SqlTimeTypeInfo.TIME =>
@@ -133,7 +138,7 @@ object CodeGenUtils {
         s"${qualifyMethod(BuiltInMethod.INTERNAL_TO_TIMESTAMP.method)}($resultTerm)"
     }
 
-  def timePointToInternalCode(resultType: TypeInformation[_], resultTerm: String) =
+  def timePointToInternalCode(resultType: TypeInformation[_], resultTerm: String): String =
     resultType match {
       case SqlTimeTypeInfo.DATE =>
         s"${qualifyMethod(BuiltInMethod.DATE_TO_INT.method)}($resultTerm)"
@@ -157,43 +162,43 @@ object CodeGenUtils {
 
   // ----------------------------------------------------------------------------------------------
 
-  def requireNumeric(genExpr: GeneratedExpression) =
+  def requireNumeric(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isNumeric(genExpr.resultType)) {
       throw new CodeGenException("Numeric expression type expected, but was " +
         s"'${genExpr.resultType}'.")
     }
 
-  def requireComparable(genExpr: GeneratedExpression) =
+  def requireComparable(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isComparable(genExpr.resultType)) {
       throw new CodeGenException(s"Comparable type expected, but was '${genExpr.resultType}'.")
     }
 
-  def requireString(genExpr: GeneratedExpression) =
+  def requireString(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isString(genExpr.resultType)) {
       throw new CodeGenException("String expression type expected.")
     }
 
-  def requireBoolean(genExpr: GeneratedExpression) =
+  def requireBoolean(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isBoolean(genExpr.resultType)) {
       throw new CodeGenException("Boolean expression type expected.")
     }
 
-  def requireTemporal(genExpr: GeneratedExpression) =
+  def requireTemporal(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isTemporal(genExpr.resultType)) {
       throw new CodeGenException("Temporal expression type expected.")
     }
 
-  def requireTimeInterval(genExpr: GeneratedExpression) =
+  def requireTimeInterval(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isTimeInterval(genExpr.resultType)) {
       throw new CodeGenException("Interval expression type expected.")
     }
 
-  def requireArray(genExpr: GeneratedExpression) =
+  def requireArray(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isArray(genExpr.resultType)) {
       throw new CodeGenException("Array expression type expected.")
     }
 
-  def requireInteger(genExpr: GeneratedExpression) =
+  def requireInteger(genExpr: GeneratedExpression): Unit =
     if (!TypeCheckUtils.isInteger(genExpr.resultType)) {
       throw new CodeGenException("Integer expression type expected.")
     }
@@ -243,7 +248,7 @@ object CodeGenUtils {
         val fieldName = pt.getFieldNames()(index)
         getFieldAccessor(pt.getTypeClass, fieldName)
 
-      case _ => throw new CodeGenException(s"Unsupported composite type: '${compType}'")
+      case _ => throw new CodeGenException(s"Unsupported composite type: '$compType'")
     }
   }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
index be55eac..946c6cd 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
@@ -39,9 +39,9 @@ import org.apache.flink.table.codegen.CodeGenUtils._
 import org.apache.flink.table.codegen.GeneratedExpression.{NEVER_NULL, NO_CODE}
 import org.apache.flink.table.codegen.calls.FunctionGenerator
 import org.apache.flink.table.codegen.calls.ScalarOperators._
-import org.apache.flink.table.functions.sql.ScalarSqlFunctions
+import org.apache.flink.table.functions.sql.{ProctimeSqlFunction, ScalarSqlFunctions}
 import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
-import org.apache.flink.table.functions.{FunctionContext, ProctimeSqlFunction, UserDefinedFunction}
+import org.apache.flink.table.functions.{FunctionContext, UserDefinedFunction}
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
 import org.apache.flink.table.typeutils.TypeCheckUtils._
 
@@ -328,13 +328,13 @@ abstract class CodeGenerator(
     // initial type check
     if (returnType.getArity != fieldExprs.length) {
       throw new CodeGenException(
-        s"Arity[${returnType.getArity}] of result type[$returnType] does not match " +
-        s"number[${fieldExprs.length}] of expressions[$fieldExprs].")
+        s"Arity [${returnType.getArity}] of result type [$returnType] does not match " +
+        s"number [${fieldExprs.length}] of expressions [$fieldExprs].")
     }
     if (resultFieldNames.length != fieldExprs.length) {
       throw new CodeGenException(
-        s"Arity[${resultFieldNames.length}] of result field names[$resultFieldNames] does not " +
-        s"match number[${fieldExprs.length}] of expressions[$fieldExprs].")
+        s"Arity [${resultFieldNames.length}] of result field names [$resultFieldNames] does not " +
+        s"match number [${fieldExprs.length}] of expressions [$fieldExprs].")
     }
     // type check
     returnType match {
@@ -342,8 +342,8 @@ abstract class CodeGenerator(
         fieldExprs.zipWithIndex foreach {
           case (fieldExpr, i) if fieldExpr.resultType != pt.getTypeAt(resultFieldNames(i)) =>
             throw new CodeGenException(
-              s"Incompatible types of expression and result type. Expression[$fieldExpr] type is " +
-              s"[${fieldExpr.resultType}], result type is [${pt.getTypeAt(resultFieldNames(i))}]")
+              s"Incompatible types of expression and result type. Expression [$fieldExpr] type is" +
+              s" [${fieldExpr.resultType}], result type is [${pt.getTypeAt(resultFieldNames(i))}]")
 
           case _ => // ok
         }
@@ -359,7 +359,7 @@ abstract class CodeGenerator(
 
       case at: AtomicType[_] if at != fieldExprs.head.resultType =>
         throw new CodeGenException(
-          s"Incompatible types of expression and result type. Expression[${fieldExprs.head}] " +
+          s"Incompatible types of expression and result type. Expression [${fieldExprs.head}] " +
           s"type is [${fieldExprs.head.resultType}], result type is [$at]")
 
       case _ => // ok
@@ -1303,11 +1303,10 @@ abstract class CodeGenerator(
 
   private[flink] def generateProctimeTimestamp(): GeneratedExpression = {
     val resultTerm = newName("result")
-    val resultTypeTerm = primitiveTypeTermForTypeInfo(SqlTimeTypeInfo.TIMESTAMP)
 
     val resultCode =
       s"""
-        |$resultTypeTerm $resultTerm = $contextTerm.timerService().currentProcessingTime();
+        |long $resultTerm = $contextTerm.timerService().currentProcessingTime();
         |""".stripMargin
     GeneratedExpression(resultTerm, NEVER_NULL, resultCode, SqlTimeTypeInfo.TIMESTAMP)
   }

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
index 01e9dff..7de7aca 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
@@ -26,7 +26,7 @@ import org.apache.flink.api.java.typeutils.{MapTypeInfo, ObjectArrayTypeInfo}
 import org.apache.flink.table.codegen.CodeGenUtils._
 import org.apache.flink.table.codegen.calls.CallGenerator.generateCallIfArgsNotNull
 import org.apache.flink.table.codegen.{CodeGenException, CodeGenerator, GeneratedExpression}
-import org.apache.flink.table.typeutils.{TimeIntervalTypeInfo, TypeCoercion}
+import org.apache.flink.table.typeutils.{TimeIndicatorTypeInfo, TimeIntervalTypeInfo, TypeCoercion}
 import org.apache.flink.table.typeutils.TypeCheckUtils._
 
 object ScalarOperators {
@@ -543,6 +543,11 @@ object ScalarOperators {
       operand: GeneratedExpression,
       targetType: TypeInformation[_])
     : GeneratedExpression = (operand.resultType, targetType) match {
+
+    // special case: cast from TimeIndicatorTypeInfo to SqlTimeTypeInfo
+    case (ti: TimeIndicatorTypeInfo, SqlTimeTypeInfo.TIMESTAMP) =>
+      operand.copy(resultType = SqlTimeTypeInfo.TIMESTAMP) // just replace the TypeInformation
+
     // identity casting
     case (fromTp, toTp) if fromTp == toTp =>
       operand

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/ProctimeSqlFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/ProctimeSqlFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/ProctimeSqlFunction.scala
deleted file mode 100644
index 4fb0378..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/ProctimeSqlFunction.scala
+++ /dev/null
@@ -1,41 +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.table.functions
-
-import org.apache.calcite.sql._
-import org.apache.calcite.sql.`type`._
-import org.apache.calcite.sql.validate.SqlMonotonicity
-
-/**
-  * Function that materializes a processing time attribute.
-  * After materialization the result can be used in regular arithmetical calculations.
-  */
-object ProctimeSqlFunction
-  extends SqlFunction(
-    "PROCTIME",
-    SqlKind.OTHER_FUNCTION,
-    ReturnTypes.explicit(SqlTypeName.TIMESTAMP),
-    InferTypes.RETURN_TYPE,
-    OperandTypes.family(SqlTypeFamily.TIMESTAMP),
-    SqlFunctionCategory.SYSTEM) {
-
-  override def getSyntax: SqlSyntax = SqlSyntax.FUNCTION
-
-  override def getMonotonicity(call: SqlOperatorBinding): SqlMonotonicity =
-    SqlMonotonicity.INCREASING
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ProctimeSqlFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ProctimeSqlFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ProctimeSqlFunction.scala
new file mode 100644
index 0000000..f30ad2f
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/sql/ProctimeSqlFunction.scala
@@ -0,0 +1,43 @@
+/*
+ * 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.table.functions.sql
+
+import org.apache.calcite.sql._
+import org.apache.calcite.sql.`type`._
+import org.apache.calcite.sql.validate.SqlMonotonicity
+
+/**
+  * Function that materializes a processing time attribute.
+  * After materialization the result can be used in regular arithmetical calculations.
+  */
+object ProctimeSqlFunction
+  extends SqlFunction(
+    "PROCTIME",
+    SqlKind.OTHER_FUNCTION,
+    ReturnTypes.explicit(SqlTypeName.TIMESTAMP),
+    InferTypes.RETURN_TYPE,
+    OperandTypes.family(SqlTypeFamily.TIMESTAMP),
+    SqlFunctionCategory.SYSTEM) {
+
+  override def getSyntax: SqlSyntax = SqlSyntax.FUNCTION
+
+  override def getMonotonicity(call: SqlOperatorBinding): SqlMonotonicity =
+    SqlMonotonicity.INCREASING
+
+  override def isDeterministic: Boolean = false
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
index 47469d1..b44c28e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
@@ -175,11 +175,11 @@ object UserDefinedFunctionUtils {
         }
       }) {
       throw new ValidationException(
-        s"Scala-style variable arguments in '${methodName}' methods are not supported. Please " +
+        s"Scala-style variable arguments in '$methodName' methods are not supported. Please " +
           s"add a @scala.annotation.varargs annotation.")
     } else if (found.length > 1) {
       throw new ValidationException(
-        s"Found multiple '${methodName}' methods which match the signature.")
+        s"Found multiple '$methodName' methods which match the signature.")
     }
     found.headOption
   }
@@ -218,7 +218,7 @@ object UserDefinedFunctionUtils {
     if (methods.isEmpty) {
       throw new ValidationException(
         s"Function class '${function.getClass.getCanonicalName}' does not implement at least " +
-          s"one method named '${methodName}' which is public, not abstract and " +
+          s"one method named '$methodName' which is public, not abstract and " +
           s"(in case of table functions) not static.")
     }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
index ac63be1..0cf86f7 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
@@ -38,7 +38,7 @@ import org.apache.flink.table.plan.nodes.CommonAggregate
 import org.apache.flink.table.plan.schema.RowSchema
 import org.apache.flink.table.plan.nodes.datastream.DataStreamGroupWindowAggregate._
 import org.apache.flink.table.plan.rules.datastream.DataStreamRetractionRules
-import org.apache.flink.table.runtime.TimestampSetterProcessFunction
+import org.apache.flink.table.runtime.RowtimeProcessFunction
 import org.apache.flink.table.runtime.aggregate.AggregateUtil._
 import org.apache.flink.table.runtime.aggregate._
 import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
@@ -149,7 +149,7 @@ class DataStreamGroupWindowAggregate(
 
       inputDS
         .process(
-          new TimestampSetterProcessFunction(timeIdx, CRowTypeInfo(inputSchema.typeInfo)))
+          new RowtimeProcessFunction(timeIdx, CRowTypeInfo(inputSchema.typeInfo)))
         .setParallelism(inputDS.getParallelism)
         .name(s"time attribute: ($timeAttribute)")
     } else {

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputMapRunner.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputMapRunner.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputMapRunner.scala
deleted file mode 100644
index 109c6e1..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputMapRunner.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.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.apache.flink.table.codegen.Compiler
-import org.apache.flink.table.runtime.types.CRow
-import org.apache.flink.types.Row
-import org.slf4j.LoggerFactory
-
-/**
-  * MapRunner with [[CRow]] input.
-  */
-class CRowInputMapRunner[OUT](
-    name: String,
-    code: String,
-    @transient var returnType: TypeInformation[OUT])
-  extends RichMapFunction[CRow, OUT]
-  with ResultTypeQueryable[OUT]
-  with Compiler[MapFunction[Row, OUT]] {
-
-  val LOG = LoggerFactory.getLogger(this.getClass)
-
-  private var function: MapFunction[Row, OUT] = _
-
-  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: CRow): OUT = {
-    function.map(in.row)
-  }
-
-  override def getProducedType: TypeInformation[OUT] = returnType
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputTupleOutputMapRunner.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputTupleOutputMapRunner.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputTupleOutputMapRunner.scala
deleted file mode 100644
index 6b3aa44..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowInputTupleOutputMapRunner.scala
+++ /dev/null
@@ -1,94 +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.table.runtime
-
-import java.lang.{Boolean => JBool}
-import java.sql.Timestamp
-
-import org.apache.calcite.runtime.SqlFunctions
-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.apache.flink.table.codegen.Compiler
-import org.apache.flink.table.runtime.types.CRow
-import org.apache.flink.types.Row
-import org.slf4j.LoggerFactory
-import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2}
-
-/**
-  * Convert [[CRow]] to a [[JTuple2]]
-  */
-class CRowInputJavaTupleOutputMapRunner(
-    name: String,
-    code: String,
-    @transient var returnType: TypeInformation[JTuple2[JBool, Any]])
-  extends RichMapFunction[CRow, Any]
-          with ResultTypeQueryable[JTuple2[JBool, Any]]
-          with Compiler[MapFunction[Row, Any]] {
-
-  val LOG = LoggerFactory.getLogger(this.getClass)
-
-  private var function: MapFunction[Row, Any] = _
-  private var tupleWrapper: JTuple2[JBool, Any] = _
-
-  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()
-    tupleWrapper = new JTuple2[JBool, Any]()
-  }
-
-  override def map(in: CRow): JTuple2[JBool, Any] = {
-    tupleWrapper.f0 = in.change
-    tupleWrapper.f1 = function.map(in.row)
-    tupleWrapper
-  }
-
-  override def getProducedType: TypeInformation[JTuple2[JBool, Any]] = returnType
-}
-
-/**
-  * Convert [[CRow]] to a [[Tuple2]]
-  */
-class CRowInputScalaTupleOutputMapRunner(
-  name: String,
-  code: String,
-  @transient var returnType: TypeInformation[(Boolean, Any)])
-  extends RichMapFunction[CRow, (Boolean, Any)]
-    with ResultTypeQueryable[(Boolean, Any)]
-    with Compiler[MapFunction[Row, Any]] {
-
-  val LOG = LoggerFactory.getLogger(this.getClass)
-
-  private var function: MapFunction[Row, Any] = _
-
-  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: CRow): (Boolean, Any) =
-    (in.change, function.map(in.row))
-
-  override def getProducedType: TypeInformation[(Boolean, Any)] = returnType
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowMapRunner.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowMapRunner.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowMapRunner.scala
new file mode 100644
index 0000000..9ed9188
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/CRowMapRunner.scala
@@ -0,0 +1,57 @@
+/*
+ * 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.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.apache.flink.table.codegen.Compiler
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.types.Row
+import org.slf4j.LoggerFactory
+
+/**
+  * MapRunner with [[CRow]] input.
+  */
+class CRowMapRunner[OUT](
+    name: String,
+    code: String,
+    @transient var returnType: TypeInformation[OUT])
+  extends RichMapFunction[CRow, OUT]
+  with ResultTypeQueryable[OUT]
+  with Compiler[MapFunction[Row, OUT]] {
+
+  val LOG = LoggerFactory.getLogger(this.getClass)
+
+  private var function: MapFunction[Row, OUT] = _
+
+  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: CRow): OUT = {
+    function.map(in.row)
+  }
+
+  override def getProducedType: TypeInformation[OUT] = returnType
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/OutputRowtimeProcessFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/OutputRowtimeProcessFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/OutputRowtimeProcessFunction.scala
new file mode 100644
index 0000000..3eaeea3
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/OutputRowtimeProcessFunction.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.table.runtime
+
+import org.apache.calcite.runtime.SqlFunctions
+import org.apache.flink.api.common.functions.MapFunction
+import org.apache.flink.api.common.functions.util.FunctionUtils
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.flink.streaming.api.operators.TimestampedCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.util.Collector
+
+/**
+  * Wraps a ProcessFunction and sets a Timestamp field of a CRow as
+  * [[org.apache.flink.streaming.runtime.streamrecord.StreamRecord]] timestamp.
+  */
+class OutputRowtimeProcessFunction[OUT](
+    function: MapFunction[CRow, OUT],
+    rowtimeIdx: Int)
+  extends ProcessFunction[CRow, OUT] {
+
+  override def open(parameters: Configuration): Unit = {
+    FunctionUtils.setFunctionRuntimeContext(function, getRuntimeContext)
+    FunctionUtils.openFunction(function, parameters)
+  }
+
+  override def processElement(
+      in: CRow,
+      ctx: ProcessFunction[CRow, OUT]#Context,
+      out: Collector[OUT]): Unit = {
+
+    val timestamp = in.row.getField(rowtimeIdx).asInstanceOf[Long]
+    out.asInstanceOf[TimestampedCollector[_]].setAbsoluteTimestamp(timestamp)
+
+    val convertedTimestamp = SqlFunctions.internalToTimestamp(timestamp)
+    in.row.setField(rowtimeIdx, convertedTimestamp)
+
+    out.collect(function.map(in))
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/RowtimeProcessFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/RowtimeProcessFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/RowtimeProcessFunction.scala
new file mode 100644
index 0000000..e192b07
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/RowtimeProcessFunction.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.table.runtime
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable
+import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.flink.streaming.api.operators.TimestampedCollector
+import org.apache.flink.table.runtime.types.CRow
+import org.apache.flink.util.Collector
+
+/**
+  * ProcessFunction to copy a timestamp from a [[org.apache.flink.types.Row]] field into the
+  * [[org.apache.flink.streaming.runtime.streamrecord.StreamRecord]].
+  */
+class RowtimeProcessFunction(
+    val rowtimeIdx: Int,
+    @transient var returnType: TypeInformation[CRow])
+  extends ProcessFunction[CRow, CRow]
+  with ResultTypeQueryable[CRow] {
+
+  override def processElement(
+      in: CRow,
+      ctx: ProcessFunction[CRow, CRow]#Context,
+      out: Collector[CRow]): Unit = {
+
+    val timestamp = in.row.getField(rowtimeIdx).asInstanceOf[Long]
+    out.asInstanceOf[TimestampedCollector[CRow]].setAbsoluteTimestamp(timestamp)
+    out.collect(in)
+  }
+
+  override def getProducedType: TypeInformation[CRow] = returnType
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/TimestampSetterProcessFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/TimestampSetterProcessFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/TimestampSetterProcessFunction.scala
deleted file mode 100644
index 00961f0..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/TimestampSetterProcessFunction.scala
+++ /dev/null
@@ -1,52 +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.table.runtime
-
-import java.sql.Timestamp
-
-import org.apache.calcite.runtime.SqlFunctions
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.java.typeutils.ResultTypeQueryable
-import org.apache.flink.streaming.api.functions.ProcessFunction
-import org.apache.flink.streaming.api.operators.TimestampedCollector
-import org.apache.flink.table.runtime.types.CRow
-import org.apache.flink.util.Collector
-
-/**
-  * ProcessFunction to copy a timestamp from a [[org.apache.flink.types.Row]] field into the
-  * [[org.apache.flink.streaming.runtime.streamrecord.StreamRecord]].
-  */
-class TimestampSetterProcessFunction(
-    val rowtimeIdx: Int,
-    @transient var returnType: TypeInformation[CRow])
-  extends ProcessFunction[CRow, CRow]
-  with ResultTypeQueryable[CRow] {
-
-  override def processElement(
-      in: CRow,
-      ctx: ProcessFunction[CRow, CRow]#Context,
-      out: Collector[CRow]): Unit = {
-
-    val timestamp = SqlFunctions.toLong(in.row.getField(rowtimeIdx).asInstanceOf[Timestamp])
-    out.asInstanceOf[TimestampedCollector[CRow]].setAbsoluteTimestamp(timestamp)
-    out.collect(in)
-  }
-
-  override def getProducedType: TypeInformation[CRow] = returnType
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/WrappingTimestampSetterProcessFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/WrappingTimestampSetterProcessFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/WrappingTimestampSetterProcessFunction.scala
deleted file mode 100644
index 8f12c30..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/WrappingTimestampSetterProcessFunction.scala
+++ /dev/null
@@ -1,61 +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.table.runtime
-
-import java.sql.Timestamp
-
-import org.apache.calcite.runtime.SqlFunctions
-import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
-import org.apache.flink.configuration.Configuration
-import org.apache.flink.streaming.api.functions.ProcessFunction
-import org.apache.flink.streaming.api.operators.TimestampedCollector
-import org.apache.flink.table.runtime.types.CRow
-import org.apache.flink.util.Collector
-
-/**
-  * Wraps a ProcessFunction and sets a Timestamp field of a CRow as
-  * [[org.apache.flink.streaming.runtime.streamrecord.StreamRecord]] timestamp.
-  */
-class WrappingTimestampSetterProcessFunction[OUT](
-    function: MapFunction[CRow, OUT],
-    rowtimeIdx: Int)
-  extends ProcessFunction[CRow, OUT] {
-
-  override def open(parameters: Configuration): Unit = {
-    super.open(parameters)
-    function match {
-      case f: RichMapFunction[_, _] =>
-        f.setRuntimeContext(getRuntimeContext)
-        f.open(parameters)
-      case _ =>
-    }
-  }
-
-  override def processElement(
-      in: CRow,
-      ctx: ProcessFunction[CRow, OUT]#Context,
-      out: Collector[OUT]): Unit = {
-
-    val timestamp = SqlFunctions.toLong(in.row.getField(rowtimeIdx).asInstanceOf[Timestamp])
-    out.asInstanceOf[TimestampedCollector[_]].setAbsoluteTimestamp(timestamp)
-
-    out.collect(function.map(in))
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
index 52105e3..6304dc4 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
@@ -96,7 +96,6 @@ object AggregateUtil {
 
     val genFunction = generator.generateAggregations(
       "UnboundedProcessingOverAggregateHelper",
-      generator,
       inputFieldTypeInfo,
       aggregates,
       aggFields,
@@ -175,7 +174,6 @@ object AggregateUtil {
 
     val genFunction = generator.generateAggregations(
       "NonWindowedAggregationHelper",
-      generator,
       inputFieldTypes,
       aggregates,
       aggFields,
@@ -240,7 +238,6 @@ object AggregateUtil {
 
     val genFunction = generator.generateAggregations(
       "BoundedOverAggregateHelper",
-      generator,
       inputFieldTypeInfo,
       aggregates,
       aggFields,
@@ -372,7 +369,6 @@ object AggregateUtil {
 
     val genFunction = generator.generateAggregations(
       "DataSetAggregatePrepareMapHelper",
-      generator,
       inputFieldTypeInfo,
       aggregates,
       aggFieldIndexes,
@@ -451,7 +447,6 @@ object AggregateUtil {
         // sliding time-window for partial aggregations
         val genFunction = generator.generateAggregations(
           "DataSetAggregatePrepareMapHelper",
-          generator,
           physicalInputTypes,
           aggregates,
           aggFieldIndexes,
@@ -555,7 +550,6 @@ object AggregateUtil {
 
     val genPreAggFunction = generator.generateAggregations(
       "GroupingWindowAggregateHelper",
-      generator,
       physicalInputTypes,
       aggregates,
       aggFieldIndexes,
@@ -572,7 +566,6 @@ object AggregateUtil {
 
     val genFinalAggFunction = generator.generateAggregations(
       "GroupingWindowAggregateHelper",
-      generator,
       physicalInputTypes,
       aggregates,
       aggFieldIndexes,
@@ -714,7 +707,6 @@ object AggregateUtil {
 
         val genFunction = generator.generateAggregations(
           "GroupingWindowAggregateHelper",
-          generator,
           physicalInputTypes,
           aggregates,
           aggFieldIndexes,
@@ -789,7 +781,6 @@ object AggregateUtil {
 
         val genFunction = generator.generateAggregations(
           "GroupingWindowAggregateHelper",
-          generator,
           physicalInputTypes,
           aggregates,
           aggFieldIndexes,
@@ -870,7 +861,6 @@ object AggregateUtil {
 
       val genPreAggFunction = generator.generateAggregations(
         "DataSetAggregatePrepareMapHelper",
-        generator,
         inputFieldTypeInfo,
         aggregates,
         aggInFields,
@@ -897,7 +887,6 @@ object AggregateUtil {
 
       val genFinalAggFunction = generator.generateAggregations(
         "DataSetAggregateFinalHelper",
-        generator,
         inputFieldTypeInfo,
         aggregates,
         aggInFields,
@@ -921,7 +910,6 @@ object AggregateUtil {
     else {
       val genFunction = generator.generateAggregations(
         "DataSetAggregateHelper",
-        generator,
         inputFieldTypeInfo,
         aggregates,
         aggInFields,
@@ -1019,7 +1007,6 @@ object AggregateUtil {
 
     val genFunction = generator.generateAggregations(
       "GroupingWindowAggregateHelper",
-      generator,
       inputFieldTypeInfo,
       aggregates,
       aggFields,
@@ -1214,7 +1201,7 @@ object AggregateUtil {
               case DECIMAL =>
                 new DecimalSumWithRetractAggFunction
               case sqlType: SqlTypeName =>
-                throw new TableException(s"Sum aggregate does no support type: '${sqlType}'")
+                throw new TableException(s"Sum aggregate does no support type: '$sqlType'")
             }
           } else {
             aggregates(index) = sqlTypeName match {
@@ -1233,7 +1220,7 @@ object AggregateUtil {
               case DECIMAL =>
                 new DecimalSumAggFunction
               case sqlType: SqlTypeName =>
-                throw new TableException(s"Sum aggregate does no support type: '${sqlType}'")
+                throw new TableException(s"Sum aggregate does no support type: '$sqlType'")
             }
           }
 
@@ -1255,7 +1242,7 @@ object AggregateUtil {
               case DECIMAL =>
                 new DecimalSum0WithRetractAggFunction
               case sqlType: SqlTypeName =>
-                throw new TableException(s"Sum0 aggregate does no support type: '${sqlType}'")
+                throw new TableException(s"Sum0 aggregate does no support type: '$sqlType'")
             }
           } else {
             aggregates(index) = sqlTypeName match {
@@ -1274,7 +1261,7 @@ object AggregateUtil {
               case DECIMAL =>
                 new DecimalSum0AggFunction
               case sqlType: SqlTypeName =>
-                throw new TableException(s"Sum0 aggregate does no support type: '${sqlType}'")
+                throw new TableException(s"Sum0 aggregate does no support type: '$sqlType'")
             }
           }
 
@@ -1295,7 +1282,7 @@ object AggregateUtil {
             case DECIMAL =>
               new DecimalAvgAggFunction
             case sqlType: SqlTypeName =>
-              throw new TableException(s"Avg aggregate does no support type: '${sqlType}'")
+              throw new TableException(s"Avg aggregate does no support type: '$sqlType'")
           }
 
         case sqlMinMaxFunction: SqlMinMaxAggFunction =>
@@ -1322,7 +1309,7 @@ object AggregateUtil {
                   new StringMinWithRetractAggFunction
                 case sqlType: SqlTypeName =>
                   throw new TableException(
-                    s"Min with retract aggregate does no support type: '${sqlType}'")
+                    s"Min with retract aggregate does no support type: '$sqlType'")
               }
             } else {
               sqlTypeName match {
@@ -1345,7 +1332,7 @@ object AggregateUtil {
                 case VARCHAR | CHAR =>
                   new StringMinAggFunction
                 case sqlType: SqlTypeName =>
-                  throw new TableException(s"Min aggregate does no support type: '${sqlType}'")
+                  throw new TableException(s"Min aggregate does no support type: '$sqlType'")
               }
             }
           } else {
@@ -1371,7 +1358,7 @@ object AggregateUtil {
                   new StringMaxWithRetractAggFunction
                 case sqlType: SqlTypeName =>
                   throw new TableException(
-                    s"Max with retract aggregate does no support type: '${sqlType}'")
+                    s"Max with retract aggregate does no support type: '$sqlType'")
               }
             } else {
               sqlTypeName match {
@@ -1394,7 +1381,7 @@ object AggregateUtil {
                 case VARCHAR | CHAR =>
                   new StringMaxAggFunction
                 case sqlType: SqlTypeName =>
-                  throw new TableException(s"Max aggregate does no support type: '${sqlType}'")
+                  throw new TableException(s"Max aggregate does no support type: '$sqlType'")
               }
             }
           }
@@ -1463,7 +1450,7 @@ object AggregateUtil {
           relDataType.head.getIndex
         } else {
           throw TableException(
-            s"Encountered more than one time attribute with the same name: '${relDataType}'")
+            s"Encountered more than one time attribute with the same name: '$relDataType'")
         }
       case e => throw TableException(
         "The time attribute of window in batch environment should be " +

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRangeOver.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRangeOver.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRangeOver.scala
index ceb986d..8a0d682 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRangeOver.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRangeOver.scala
@@ -17,10 +17,8 @@
  */
 package org.apache.flink.table.runtime.aggregate
 
-import java.sql.Timestamp
 import java.util.{ArrayList => JArrayList, List => JList}
 
-import org.apache.calcite.runtime.SqlFunctions
 import org.apache.flink.api.common.state._
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
 import org.apache.flink.api.java.typeutils.{ListTypeInfo, RowTypeInfo}
@@ -118,7 +116,7 @@ class RowTimeBoundedRangeOver(
     registerProcessingCleanupTimer(ctx, ctx.timerService().currentProcessingTime())
 
     // triggering timestamp for trigger calculation
-    val triggeringTs = SqlFunctions.toLong(input.getField(rowTimeIdx).asInstanceOf[Timestamp])
+    val triggeringTs = input.getField(rowTimeIdx).asInstanceOf[Long]
 
     val lastTriggeringTs = lastTriggeringTsState.value
 

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRowsOver.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRowsOver.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRowsOver.scala
index 678a3b7..ba65846 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRowsOver.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeBoundedRowsOver.scala
@@ -17,11 +17,9 @@
  */
 package org.apache.flink.table.runtime.aggregate
 
-import java.sql.Timestamp
 import java.util
 import java.util.{List => JList}
 
-import org.apache.calcite.runtime.SqlFunctions
 import org.apache.flink.api.common.state._
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
 import org.apache.flink.api.java.typeutils.{ListTypeInfo, RowTypeInfo}
@@ -29,11 +27,11 @@ import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.functions.ProcessFunction
 import org.apache.flink.streaming.api.operators.TimestampedCollector
 import org.apache.flink.table.api.StreamQueryConfig
-import org.apache.flink.types.Row
-import org.apache.flink.util.{Collector, Preconditions}
 import org.apache.flink.table.codegen.{Compiler, GeneratedAggregationsFunction}
 import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
-import org.slf4j.LoggerFactory
+import org.apache.flink.types.Row
+import org.apache.flink.util.{Collector, Preconditions}
+import org.slf4j.{Logger, LoggerFactory}
 
 /**
  * Process Function for ROWS clause event-time bounded OVER window
@@ -56,6 +54,8 @@ class RowTimeBoundedRowsOver(
   Preconditions.checkNotNull(aggregationStateType)
   Preconditions.checkNotNull(precedingOffset)
 
+  val LOG: Logger = LoggerFactory.getLogger(this.getClass)
+
   private var output: CRow = _
 
   // the state which keeps the last triggering timestamp
@@ -73,7 +73,6 @@ class RowTimeBoundedRowsOver(
   // to this time stamp.
   private var dataState: MapState[Long, JList[Row]] = _
 
-  val LOG = LoggerFactory.getLogger(this.getClass)
   private var function: GeneratedAggregations = _
 
   override def open(config: Configuration) {
@@ -127,7 +126,7 @@ class RowTimeBoundedRowsOver(
     registerProcessingCleanupTimer(ctx, ctx.timerService().currentProcessingTime())
 
     // triggering timestamp for trigger calculation
-    val triggeringTs = SqlFunctions.toLong(input.getField(rowTimeIdx).asInstanceOf[Timestamp])
+    val triggeringTs = input.getField(rowTimeIdx).asInstanceOf[Long]
 
     val lastTriggeringTs = lastTriggeringTsState.value
     // check if the data is expired, if not, save the data and register event time timer

http://git-wip-us.apache.org/repos/asf/flink/blob/47944b1b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeSortProcessFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeSortProcessFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeSortProcessFunction.scala
index fd58678..0d69355 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeSortProcessFunction.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/RowTimeSortProcessFunction.scala
@@ -17,24 +17,17 @@
  */
 package org.apache.flink.table.runtime.aggregate
 
-import java.sql.Timestamp
+import java.util.{Collections, ArrayList => JArrayList, List => JList}
 
-import org.apache.flink.api.common.state.ValueState
-import org.apache.flink.api.common.state.ValueStateDescriptor
-import org.apache.flink.api.common.state.MapState
-import org.apache.flink.api.common.state.MapStateDescriptor
+import org.apache.flink.api.common.state.{MapState, MapStateDescriptor, ValueState, ValueStateDescriptor}
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
 import org.apache.flink.api.java.typeutils.ListTypeInfo
 import org.apache.flink.configuration.Configuration
 import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.flink.streaming.api.operators.TimestampedCollector
 import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
 import org.apache.flink.types.Row
 import org.apache.flink.util.{Collector, Preconditions}
-import java.util.Collections
-import java.util.{ArrayList => JArrayList, List => JList}
-
-import org.apache.calcite.runtime.SqlFunctions
-import org.apache.flink.streaming.api.operators.TimestampedCollector
 
 /**
  * ProcessFunction to sort on event-time and possibly addtional secondary sort attributes.
@@ -90,7 +83,7 @@ class RowTimeSortProcessFunction(
     val input = inputC.row
     
     // timestamp of the processed row
-    val rowtime = SqlFunctions.toLong(input.getField(rowtimeIdx).asInstanceOf[Timestamp])
+    val rowtime = input.getField(rowtimeIdx).asInstanceOf[Long]
 
     val lastTriggeringTs = lastTriggeringTsState.value
 


Mime
View raw message