You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2017/08/23 08:15:16 UTC
[5/5] flink git commit: [FLINK-7337] [table] Refactor internal
handling of time indicator attributes.
[FLINK-7337] [table] Refactor internal handling of time indicator attributes.
- Expand phyiscal Row schema for time indicators.
- Refactor computation of logical schema of tables to import.
- Refactor operators to use time attribute in Row instead of StreamRecord timestamp.
This closes #4488.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/93d0ae4a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/93d0ae4a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/93d0ae4a
Branch: refs/heads/master
Commit: 93d0ae4a9f059da4bd2b720f7503da0f9c0a8c7c
Parents: 6ed5815
Author: Fabian Hueske <fh...@apache.org>
Authored: Fri Aug 4 02:20:56 2017 +0200
Committer: twalthr <tw...@apache.org>
Committed: Wed Aug 23 10:09:21 2017 +0200
----------------------------------------------------------------------
.../table/api/StreamTableEnvironment.scala | 155 ++++-
.../flink/table/api/TableEnvironment.scala | 8 +-
.../flink/table/calcite/FlinkTypeFactory.scala | 37 +-
.../calcite/RelTimeIndicatorConverter.scala | 67 +--
.../flink/table/codegen/CodeGenerator.scala | 78 ++-
.../table/codegen/calls/ScalarOperators.scala | 44 +-
.../table/functions/ProctimeSqlFunction.scala | 41 ++
.../TimeMaterializationSqlFunction.scala | 41 --
.../flink/table/plan/nodes/CommonCalc.scala | 6 +-
.../table/plan/nodes/CommonCorrelate.scala | 20 +-
.../plan/nodes/PhysicalTableSourceScan.scala | 4 +-
.../nodes/dataset/BatchTableSourceScan.scala | 4 +-
.../plan/nodes/datastream/DataStreamCalc.scala | 12 +-
.../nodes/datastream/DataStreamCorrelate.scala | 6 +-
.../datastream/DataStreamGroupAggregate.scala | 34 +-
.../DataStreamGroupWindowAggregate.scala | 84 +--
.../datastream/DataStreamOverAggregate.scala | 149 ++---
.../plan/nodes/datastream/DataStreamScan.scala | 2 +-
.../plan/nodes/datastream/DataStreamSort.scala | 20 +-
.../plan/nodes/datastream/DataStreamUnion.scala | 6 +-
.../nodes/datastream/DataStreamValues.scala | 12 +-
.../nodes/datastream/DataStreamWindowJoin.scala | 16 +-
.../plan/nodes/datastream/StreamScan.scala | 36 +-
.../datastream/StreamTableSourceScan.scala | 23 +-
.../logical/FlinkLogicalTableSourceScan.scala | 23 +-
.../datastream/DataStreamWindowJoinRule.scala | 2 +-
.../table/plan/schema/DataStreamTable.scala | 13 -
.../flink/table/plan/schema/FlinkTable.scala | 10 +-
.../flink/table/plan/schema/RowSchema.scala | 124 +---
.../plan/schema/StreamTableSourceTable.scala | 48 +-
.../runtime/CRowInputTupleOutputMapRunner.scala | 2 +
.../table/runtime/CRowOutputMapRunner.scala | 60 --
.../table/runtime/CRowOutputProcessRunner.scala | 72 +++
.../TimestampSetterProcessFunction.scala | 52 ++
...WrappingTimestampSetterProcessFunction.scala | 61 ++
.../table/runtime/aggregate/AggregateUtil.scala | 54 +-
...SetSessionWindowAggReduceGroupFunction.scala | 5 +-
...taSetSlideWindowAggReduceGroupFunction.scala | 5 +-
...TumbleTimeWindowAggReduceGroupFunction.scala | 2 +-
...rementalAggregateAllTimeWindowFunction.scala | 15 +-
...IncrementalAggregateTimeWindowFunction.scala | 19 +-
.../aggregate/ProcTimeBoundedRangeOver.scala | 4 +
.../aggregate/ProcTimeSortProcessFunction.scala | 8 +-
.../aggregate/RowTimeBoundedRangeOver.scala | 9 +-
.../aggregate/RowTimeBoundedRowsOver.scala | 9 +-
.../aggregate/RowTimeSortProcessFunction.scala | 20 +-
.../aggregate/RowTimeUnboundedOver.scala | 19 +-
.../table/runtime/aggregate/SortUtil.scala | 5 +-
.../aggregate/TimeWindowPropertyCollector.scala | 23 +-
.../table/runtime/join/WindowJoinUtil.scala | 11 +-
.../table/typeutils/TimeIndicatorTypeInfo.scala | 3 +
.../flink/table/api/stream/sql/SortTest.scala | 4 +-
.../api/stream/table/TableSourceTest.scala | 4 +-
.../plan/TimeIndicatorConversionTest.scala | 29 +-
.../table/runtime/harness/HarnessTestBase.scala | 22 +-
.../table/runtime/harness/JoinHarnessTest.scala | 4 +-
.../runtime/harness/NonWindowHarnessTest.scala | 4 +-
.../runtime/harness/OverWindowHarnessTest.scala | 586 ++++++++-----------
.../SortProcessFunctionHarnessTest.scala | 129 ++--
.../runtime/stream/table/TableSinkITCase.scala | 140 ++++-
60 files changed, 1362 insertions(+), 1143 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/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 7328b2a..c4e1450 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,10 +23,8 @@ 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
-import org.apache.calcite.rel.{RelNode, RelVisitor}
-import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode}
-import org.apache.calcite.sql.SqlKind
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField, RelDataTypeFieldImpl, RelRecordType}
+import org.apache.calcite.rel.RelNode
import org.apache.calcite.sql2rel.RelDecorrelator
import org.apache.calcite.tools.{RuleSet, RuleSets}
import org.apache.flink.api.common.functions.MapFunction
@@ -38,19 +36,19 @@ import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
-import org.apache.flink.table.calcite.RelTimeIndicatorConverter
+import org.apache.flink.table.calcite.{FlinkTypeFactory, RelTimeIndicatorConverter}
import org.apache.flink.table.explain.PlanJsonParser
import org.apache.flink.table.expressions._
import org.apache.flink.table.plan.nodes.FlinkConventions
-import org.apache.flink.table.plan.nodes.datastream.{DataStreamRel, UpdateAsRetractionTrait, _}
+import org.apache.flink.table.plan.nodes.datastream.{DataStreamRel, UpdateAsRetractionTrait}
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.types.{CRow, CRowTypeInfo}
-import org.apache.flink.table.runtime.{CRowInputJavaTupleOutputMapRunner, CRowInputMapRunner, CRowInputScalaTupleOutputMapRunner}
+import org.apache.flink.table.runtime.{CRowInputJavaTupleOutputMapRunner, CRowInputMapRunner, CRowInputScalaTupleOutputMapRunner, WrappingTimestampSetterProcessFunction}
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.TypeCheckUtils
+import org.apache.flink.table.typeutils.{TimeIndicatorTypeInfo, TypeCheckUtils}
import org.apache.flink.types.Row
import _root_.scala.collection.JavaConverters._
@@ -270,11 +268,11 @@ abstract class StreamTableEnvironment(
* valid Java class identifier.
*/
private def getConversionMapperWithChanges[OUT](
- physicalTypeInfo: TypeInformation[CRow],
- schema: RowSchema,
- requestedTypeInfo: TypeInformation[OUT],
- functionName: String):
- MapFunction[CRow, OUT] = {
+ physicalTypeInfo: TypeInformation[CRow],
+ schema: RowSchema,
+ requestedTypeInfo: TypeInformation[OUT],
+ functionName: String):
+ MapFunction[CRow, OUT] = {
requestedTypeInfo match {
@@ -356,9 +354,7 @@ abstract class StreamTableEnvironment(
val dataStreamTable = new DataStreamTable[T](
dataStream,
fieldIndexes,
- fieldNames,
- None,
- None
+ fieldNames
)
registerTableInternal(name, dataStreamTable)
}
@@ -393,12 +389,14 @@ abstract class StreamTableEnvironment(
s"But is: ${execEnv.getStreamTimeCharacteristic}")
}
+ // adjust field indexes and field names
+ val indexesWithIndicatorFields = adjustFieldIndexes(fieldIndexes, rowtime, proctime)
+ val namesWithIndicatorFields = adjustFieldNames(fieldNames, rowtime, proctime)
+
val dataStreamTable = new DataStreamTable[T](
dataStream,
- fieldIndexes,
- fieldNames,
- rowtime,
- proctime
+ indexesWithIndicatorFields,
+ namesWithIndicatorFields
)
registerTableInternal(name, dataStreamTable)
}
@@ -502,6 +500,63 @@ abstract class StreamTableEnvironment(
}
/**
+ * Injects markers for time indicator fields into the field indexes.
+ *
+ * @param fieldIndexes The field indexes into which the time indicators markers are injected.
+ * @param rowtime An optional rowtime indicator
+ * @param proctime An optional proctime indicator
+ * @return An adjusted array of field indexes.
+ */
+ private def adjustFieldIndexes(
+ fieldIndexes: Array[Int],
+ rowtime: Option[(Int, String)],
+ proctime: Option[(Int, String)]): Array[Int] = {
+
+ // inject rowtime field
+ val withRowtime = rowtime match {
+ case Some(rt) => fieldIndexes.patch(rt._1, Seq(TimeIndicatorTypeInfo.ROWTIME_MARKER), 0)
+ case _ => fieldIndexes
+ }
+
+ // inject proctime field
+ val withProctime = proctime match {
+ case Some(pt) => withRowtime.patch(pt._1, Seq(TimeIndicatorTypeInfo.PROCTIME_MARKER), 0)
+ case _ => withRowtime
+ }
+
+ withProctime
+ }
+
+ /**
+ * Injects names of time indicator fields into the list of field names.
+ *
+ * @param fieldNames The array of field names into which the time indicator field names are
+ * injected.
+ * @param rowtime An optional rowtime indicator
+ * @param proctime An optional proctime indicator
+ * @return An adjusted array of field names.
+ */
+ private def adjustFieldNames(
+ fieldNames: Array[String],
+ rowtime: Option[(Int, String)],
+ proctime: Option[(Int, String)]): Array[String] = {
+
+ // inject rowtime field
+ val withRowtime = rowtime match {
+ case Some(rt) => fieldNames.patch(rt._1, Seq(rowtime.get._2), 0)
+ case _ => fieldNames
+ }
+
+ // inject proctime field
+ val withProctime = proctime match {
+ case Some(pt) => withRowtime.patch(pt._1, Seq(proctime.get._2), 0)
+ case _ => withRowtime
+ }
+
+ withProctime
+ }
+
+ /**
* Returns the decoration rule set for this environment
* including a custom RuleSet configuration.
*/
@@ -632,10 +687,21 @@ abstract class StreamTableEnvironment(
val relNode = table.getRelNode
val dataStreamPlan = optimize(relNode, updatesAsRetraction)
- // we convert the logical row type to the output row type
- val convertedOutputType = RelTimeIndicatorConverter.convertOutputType(relNode)
-
- translate(dataStreamPlan, convertedOutputType, queryConfig, withChangeFlag)
+ // zip original field names with optimized field types
+ val fieldTypes = relNode.getRowType.getFieldList.asScala
+ .zip(dataStreamPlan.getRowType.getFieldList.asScala)
+ // get name of original plan and type of optimized plan
+ .map(x => (x._1.getName, x._2.getType))
+ // add field indexes
+ .zipWithIndex
+ // build new field types
+ .map(x => new RelDataTypeFieldImpl(x._1._1, x._2, x._1._2))
+
+ // build a record type from list of field types
+ val rowType = new RelRecordType(
+ fieldTypes.toList.asInstanceOf[List[RelDataTypeField]].asJava)
+
+ translate(dataStreamPlan, rowType, queryConfig, withChangeFlag)
}
/**
@@ -684,12 +750,41 @@ abstract class StreamTableEnvironment(
val rootParallelism = plan.getParallelism
- conversion match {
- case mapFunction: MapFunction[CRow, A] =>
- plan.map(mapFunction)
- .returns(tpe)
- .name(s"to: ${tpe.getTypeClass.getSimpleName}")
- .setParallelism(rootParallelism)
+ val rowtimeFields = logicalType.getFieldList.asScala
+ .filter(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
+
+ 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)
+
+ } 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.")
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/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 3bca156..b647c51 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
@@ -731,15 +731,15 @@ abstract class TableEnvironment(val config: TableConfig) {
// 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.physicalTypeInfo != inputTypeInfo) {
+ if (schema.typeInfo != inputTypeInfo) {
throw TableException(
s"The field types of physical and logical row types do not match. " +
- s"Physical type is [${schema.physicalTypeInfo}], 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.")
}
- val fieldTypes = schema.physicalFieldTypeInfo
- val fieldNames = schema.physicalFieldNames
+ val fieldTypes = schema.fieldTypeInfos
+ val fieldNames = schema.fieldNames
// validate requested type
if (requestedTypeInfo.getArity != fieldTypes.length) {
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
index dbefe20..637e8cc 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
@@ -172,45 +172,20 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
*
* @param fieldNames field names
* @param fieldTypes field types, every element is Flink's [[TypeInformation]]
- * @param rowtime optional system field to indicate event-time; the index determines the index
- * in the final record. If the index is smaller than the number of specified
- * fields, it shifts all following fields.
- * @param proctime optional system field to indicate processing-time; the index determines the
- * index in the final record. If the index is smaller than the number of
- * specified fields, it shifts all following fields.
* @return a struct type with the input fieldNames, input fieldTypes, and system fields
*/
def buildLogicalRowType(
fieldNames: Seq[String],
- fieldTypes: Seq[TypeInformation[_]],
- rowtime: Option[(Int, String)],
- proctime: Option[(Int, String)])
+ fieldTypes: Seq[TypeInformation[_]])
: RelDataType = {
val logicalRowTypeBuilder = builder
val fields = fieldNames.zip(fieldTypes)
-
- var totalNumberOfFields = fields.length
- if (rowtime.isDefined) {
- totalNumberOfFields += 1
- }
- if (proctime.isDefined) {
- totalNumberOfFields += 1
- }
-
- var addedTimeAttributes = 0
- for (i <- 0 until totalNumberOfFields) {
- if (rowtime.isDefined && rowtime.get._1 == i) {
- logicalRowTypeBuilder.add(rowtime.get._2, createRowtimeIndicatorType())
- addedTimeAttributes += 1
- } else if (proctime.isDefined && proctime.get._1 == i) {
- logicalRowTypeBuilder.add(proctime.get._2, createProctimeIndicatorType())
- addedTimeAttributes += 1
- } else {
- val field = fields(i - addedTimeAttributes)
- logicalRowTypeBuilder.add(field._1, createTypeFromTypeInfo(field._2, isNullable = true))
- }
- }
+ fields.foreach(f => {
+ // time indicators are not nullable
+ val nullable = !FlinkTypeFactory.isTimeIndicatorType(f._2)
+ logicalRowTypeBuilder.add(f._1, createTypeFromTypeInfo(f._2, nullable))
+ })
logicalRowTypeBuilder.build
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/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 eb14291..717a1af 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
@@ -18,7 +18,7 @@
package org.apache.flink.table.calcite
-import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFieldImpl, RelRecordType}
+import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core._
import org.apache.calcite.rel.logical._
import org.apache.calcite.rel.{RelNode, RelShuttle}
@@ -26,8 +26,8 @@ import org.apache.calcite.rex._
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.isTimeIndicatorType
-import org.apache.flink.table.functions.TimeMaterializationSqlFunction
+import org.apache.flink.table.calcite.FlinkTypeFactory.{isRowtimeIndicatorType, _}
+import org.apache.flink.table.functions.ProctimeSqlFunction
import org.apache.flink.table.plan.logical.rel.LogicalWindowAggregate
import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
@@ -242,9 +242,13 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder) extends RelShuttle {
case lp: LogicalProject =>
val projects = lp.getProjects.zipWithIndex.map { case (expr, idx) =>
if (isTimeIndicatorType(expr.getType) && refIndices.contains(idx)) {
- rexBuilder.makeCall(
- TimeMaterializationSqlFunction,
- expr)
+ if (isRowtimeIndicatorType(expr.getType)) {
+ // cast rowtime indicator to regular timestamp
+ rexBuilder.makeAbstractCast(timestamp, expr)
+ } else {
+ // generate proctime access
+ rexBuilder.makeCall(ProctimeSqlFunction, expr)
+ }
} else {
expr
}
@@ -259,9 +263,17 @@ class RelTimeIndicatorConverter(rexBuilder: RexBuilder) extends RelShuttle {
case _ =>
val projects = input.getRowType.getFieldList.map { field =>
if (isTimeIndicatorType(field.getType) && refIndices.contains(field.getIndex)) {
- rexBuilder.makeCall(
- TimeMaterializationSqlFunction,
- new RexInputRef(field.getIndex, field.getType))
+ if (isRowtimeIndicatorType(field.getType)) {
+ // cast rowtime indicator to regular timestamp
+ rexBuilder.makeAbstractCast(
+ timestamp,
+ new RexInputRef(field.getIndex, field.getType))
+ } else {
+ // generate proctime access
+ rexBuilder.makeCall(
+ ProctimeSqlFunction,
+ new RexInputRef(field.getIndex, field.getType))
+ }
} else {
new RexInputRef(field.getIndex, field.getType)
}
@@ -311,19 +323,19 @@ object RelTimeIndicatorConverter {
var needsConversion = false
- // materialize all remaining time indicators
+ // materialize remaining proctime indicators
val projects = convertedRoot.getRowType.getFieldList.map(field =>
- if (isTimeIndicatorType(field.getType)) {
+ if (isProctimeIndicatorType(field.getType)) {
needsConversion = true
rexBuilder.makeCall(
- TimeMaterializationSqlFunction,
+ ProctimeSqlFunction,
new RexInputRef(field.getIndex, field.getType))
} else {
new RexInputRef(field.getIndex, field.getType)
}
)
- // add final conversion
+ // add final conversion if necessary
if (needsConversion) {
LogicalProject.create(
convertedRoot,
@@ -334,27 +346,6 @@ object RelTimeIndicatorConverter {
}
}
- def convertOutputType(rootRel: RelNode): RelDataType = {
-
- val timestamp = rootRel
- .getCluster
- .getRexBuilder
- .getTypeFactory
- .asInstanceOf[FlinkTypeFactory]
- .createTypeFromTypeInfo(SqlTimeTypeInfo.TIMESTAMP, isNullable = false)
-
- // convert all time indicators types to timestamps
- val fields = rootRel.getRowType.getFieldList.map { field =>
- if (isTimeIndicatorType(field.getType)) {
- new RelDataTypeFieldImpl(field.getName, field.getIndex, timestamp)
- } else {
- field
- }
- }
-
- new RelRecordType(fields)
- }
-
/**
* Materializes time indicator accesses in an expression.
*
@@ -415,7 +406,13 @@ class RexTimeIndicatorMaterializer(
case _ =>
updatedCall.getOperands.map { o =>
if (isTimeIndicatorType(o.getType)) {
- rexBuilder.makeCall(TimeMaterializationSqlFunction, o)
+ if (isRowtimeIndicatorType(o.getType)) {
+ // cast rowtime indicator to regular timestamp
+ rexBuilder.makeAbstractCast(timestamp, o)
+ } else {
+ // generate proctime access
+ rexBuilder.makeCall(ProctimeSqlFunction, o)
+ }
} else {
o
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/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 28fea59..be55eac 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
@@ -37,11 +37,12 @@ import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.calcite.FlinkTypeFactory
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.codegen.calls.{BuiltInMethods, FunctionGenerator}
import org.apache.flink.table.functions.sql.ScalarSqlFunctions
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
-import org.apache.flink.table.functions.{FunctionContext, TimeMaterializationSqlFunction, UserDefinedFunction}
+import org.apache.flink.table.functions.{FunctionContext, ProctimeSqlFunction, UserDefinedFunction}
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
import org.apache.flink.table.typeutils.TypeCheckUtils._
import scala.collection.JavaConversions._
@@ -56,10 +57,11 @@ import scala.collection.mutable
* @param nullableInput input(s) can be null.
* @param input1 type information about the first input of the Function
* @param input2 type information about the second input if the Function is binary
- * @param input1FieldMapping additional mapping information for input1
- * (e.g. POJO types have no deterministic field order and some input fields might not be read)
- * @param input2FieldMapping additional mapping information for input2
- * (e.g. POJO types have no deterministic field order and some input fields might not be read)
+ * @param input1FieldMapping additional mapping information for input1.
+ * POJO types have no deterministic field order and some input fields might not be read.
+ * The input1FieldMapping is also used to inject time indicator attributes.
+ * @param input2FieldMapping additional mapping information for input2.
+ * POJO types have no deterministic field order and some input fields might not be read.
*/
abstract class CodeGenerator(
config: TableConfig,
@@ -245,16 +247,23 @@ abstract class CodeGenerator(
returnType: TypeInformation[_ <: Any],
resultFieldNames: Seq[String])
: GeneratedExpression = {
- val input1AccessExprs = input1Mapping.map { idx =>
- generateInputAccess(input1, input1Term, idx)
+
+ val input1AccessExprs = input1Mapping.map {
+ case TimeIndicatorTypeInfo.ROWTIME_MARKER =>
+ // attribute is a rowtime indicator. Access event-time timestamp in StreamRecord.
+ generateRowtimeAccess()
+ case TimeIndicatorTypeInfo.PROCTIME_MARKER =>
+ // attribute is proctime indicator.
+ // We use a null literal and generate a timestamp when we need it.
+ generateNullLiteral(TimeIndicatorTypeInfo.PROCTIME_INDICATOR)
+ case idx =>
+ // regular attribute. Access attribute in input data type.
+ generateInputAccess(input1, input1Term, idx)
}
val input2AccessExprs = input2 match {
case Some(ti) =>
- input2Mapping.map { idx =>
- generateInputAccess(ti, input2Term, idx)
- }.toSeq
-
+ input2Mapping.map(idx => generateInputAccess(ti, input2Term, idx)).toSeq
case None => Seq() // add nothing
}
@@ -724,10 +733,8 @@ abstract class CodeGenerator(
override def visitCall(call: RexCall): GeneratedExpression = {
// special case: time materialization
- if (call.getOperator == TimeMaterializationSqlFunction) {
- return generateRecordTimestamp(
- FlinkTypeFactory.isRowtimeIndicatorType(call.getOperands.get(0).getType)
- )
+ if (call.getOperator == ProctimeSqlFunction) {
+ return generateProctimeTimestamp()
}
val resultType = FlinkTypeFactory.toTypeInfo(call.getType)
@@ -967,10 +974,10 @@ abstract class CodeGenerator(
generateArrayElement(this, array)
case ScalarSqlFunctions.CONCAT =>
- generateConcat(BuiltInMethods.CONCAT, operands)
+ generateConcat(this.nullCheck, operands)
case ScalarSqlFunctions.CONCAT_WS =>
- generateConcat(BuiltInMethods.CONCAT_WS, operands)
+ generateConcatWs(operands)
// advanced scalar functions
case sqlOperator: SqlOperator =>
@@ -1216,8 +1223,14 @@ abstract class CodeGenerator(
|""".stripMargin
} else if (nullCheck) {
s"""
- |$resultTypeTerm $resultTerm = $unboxedFieldCode;
|boolean $nullTerm = $fieldTerm == null;
+ |$resultTypeTerm $resultTerm;
+ |if ($nullTerm) {
+ | $resultTerm = $defaultValue;
+ |}
+ |else {
+ | $resultTerm = $unboxedFieldCode;
+ |}
|""".stripMargin
} else {
s"""
@@ -1270,27 +1283,32 @@ abstract class CodeGenerator(
}
}
- private[flink] def generateRecordTimestamp(isEventTime: Boolean): GeneratedExpression = {
+ private[flink] def generateRowtimeAccess(): GeneratedExpression = {
val resultTerm = newName("result")
- val resultTypeTerm = primitiveTypeTermForTypeInfo(SqlTimeTypeInfo.TIMESTAMP)
+ val nullTerm = newName("isNull")
- val resultCode = if (isEventTime) {
+ val accessCode =
s"""
- |$resultTypeTerm $resultTerm;
- |if ($contextTerm.timestamp() == null) {
+ |Long $resultTerm = $contextTerm.timestamp();
+ |if ($resultTerm == null) {
| throw new RuntimeException("Rowtime timestamp is null. Please make sure that a proper " +
| "TimestampAssigner is defined and the stream environment uses the EventTime time " +
| "characteristic.");
|}
- |else {
- | $resultTerm = $contextTerm.timestamp();
- |}
- |""".stripMargin
- } else {
+ |boolean $nullTerm = false;
+ """.stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, accessCode, TimeIndicatorTypeInfo.ROWTIME_INDICATOR)
+ }
+
+ private[flink] def generateProctimeTimestamp(): GeneratedExpression = {
+ val resultTerm = newName("result")
+ val resultTypeTerm = primitiveTypeTermForTypeInfo(SqlTimeTypeInfo.TIMESTAMP)
+
+ val resultCode =
s"""
|$resultTypeTerm $resultTerm = $contextTerm.timerService().currentProcessingTime();
|""".stripMargin
- }
GeneratedExpression(resultTerm, NEVER_NULL, resultCode, SqlTimeTypeInfo.TIMESTAMP)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/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 1ab927d..01e9dff 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
@@ -17,8 +17,6 @@
*/
package org.apache.flink.table.codegen.calls
-import java.lang.reflect.Method
-
import org.apache.calcite.avatica.util.DateTimeUtils.MILLIS_PER_DAY
import org.apache.calcite.avatica.util.{DateTimeUtils, TimeUnitRange}
import org.apache.calcite.util.BuiltInMethod
@@ -1026,14 +1024,48 @@ object ScalarOperators {
}
def generateConcat(
- method: Method,
- operands: Seq[GeneratedExpression]): GeneratedExpression = {
+ nullCheck: Boolean,
+ operands: Seq[GeneratedExpression])
+ : GeneratedExpression = {
- generateCallIfArgsNotNull(false, STRING_TYPE_INFO, operands) {
- (terms) =>s"${qualifyMethod(method)}(${terms.mkString(", ")})"
+ generateCallIfArgsNotNull(nullCheck, STRING_TYPE_INFO, operands) {
+ (terms) =>s"${qualifyMethod(BuiltInMethods.CONCAT)}(${terms.mkString(", ")})"
}
}
+ def generateConcatWs(operands: Seq[GeneratedExpression]): GeneratedExpression = {
+
+ val resultTerm = newName("result")
+ val nullTerm = newName("isNull")
+ val defaultValue = primitiveDefaultValue(Types.STRING)
+
+ val tempTerms = operands.tail.map(_ => newName("temp"))
+
+ val operatorCode =
+ s"""
+ |${operands.map(_.code).mkString("\n")}
+ |
+ |String $resultTerm;
+ |boolean $nullTerm;
+ |if (${operands.head.nullTerm}) {
+ | $nullTerm = true;
+ | $resultTerm = $defaultValue;
+ |} else {
+ | ${operands.tail.zip(tempTerms).map {
+ case (o: GeneratedExpression, t: String) =>
+ s"String $t;\n" +
+ s" if (${o.nullTerm}) $t = null; else $t = ${o.resultTerm};"
+ }.mkString("\n")
+ }
+ | $nullTerm = false;
+ | $resultTerm = ${qualifyMethod(BuiltInMethods.CONCAT_WS)}
+ | (${operands.head.resultTerm}, ${tempTerms.mkString(", ")});
+ |}
+ |""".stripMargin
+
+ GeneratedExpression(resultTerm, nullTerm, operatorCode, Types.STRING)
+ }
+
def generateMapGet(
codeGenerator: CodeGenerator,
map: GeneratedExpression,
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/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
new file mode 100644
index 0000000..4fb0378
--- /dev/null
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/ProctimeSqlFunction.scala
@@ -0,0 +1,41 @@
+/*
+ * 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/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TimeMaterializationSqlFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TimeMaterializationSqlFunction.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TimeMaterializationSqlFunction.scala
deleted file mode 100644
index d875026..0000000
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/TimeMaterializationSqlFunction.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 time attribute to the metadata timestamp. After materialization
- * the result can be used in regular arithmetical calculations.
- */
-object TimeMaterializationSqlFunction
- extends SqlFunction(
- "TIME_MATERIALIZATION",
- 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/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCalc.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCalc.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCalc.scala
index 3e355ff..2f1871b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCalc.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCalc.scala
@@ -42,8 +42,8 @@ trait CommonCalc {
GeneratedFunction[T, Row] = {
val projection = generator.generateResultExpression(
- returnSchema.physicalTypeInfo,
- returnSchema.physicalFieldNames,
+ returnSchema.typeInfo,
+ returnSchema.fieldNames,
calcProjection)
// only projection
@@ -80,7 +80,7 @@ trait CommonCalc {
ruleDescription,
functionClass,
body,
- returnSchema.physicalTypeInfo)
+ returnSchema.typeInfo)
}
private[flink] def conditionToString(
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
index 96aaf3e..7c01fde 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonCorrelate.scala
@@ -53,12 +53,10 @@ trait CommonCorrelate {
functionClass: Class[T]):
GeneratedFunction[T, Row] = {
- val physicalRexCall = inputSchema.mapRexNode(rexCall)
-
val functionGenerator = new FunctionCodeGenerator(
config,
false,
- inputSchema.physicalTypeInfo,
+ inputSchema.typeInfo,
Some(udtfTypeInfo),
None,
pojoFieldMapping)
@@ -69,7 +67,7 @@ trait CommonCorrelate {
.addReusableConstructor(classOf[TableFunctionCollector[_]])
.head
- val call = functionGenerator.generateExpression(physicalRexCall)
+ val call = functionGenerator.generateExpression(rexCall)
var body =
s"""
|${call.resultTerm}.setCollector($collectorTerm);
@@ -90,8 +88,8 @@ trait CommonCorrelate {
}
val outerResultExpr = functionGenerator.generateResultExpression(
input1AccessExprs ++ input2NullExprs,
- returnSchema.physicalTypeInfo,
- returnSchema.physicalFieldNames)
+ returnSchema.typeInfo,
+ returnSchema.fieldNames)
body +=
s"""
|boolean hasOutput = $collectorTerm.isCollected();
@@ -108,7 +106,7 @@ trait CommonCorrelate {
ruleDescription,
functionClass,
body,
- returnSchema.physicalTypeInfo)
+ returnSchema.typeInfo)
}
/**
@@ -126,7 +124,7 @@ trait CommonCorrelate {
val generator = new CollectorCodeGenerator(
config,
false,
- inputSchema.physicalTypeInfo,
+ inputSchema.typeInfo,
Some(udtfTypeInfo),
None,
pojoFieldMapping)
@@ -135,8 +133,8 @@ trait CommonCorrelate {
val crossResultExpr = generator.generateResultExpression(
input1AccessExprs ++ input2AccessExprs,
- returnSchema.physicalTypeInfo,
- returnSchema.physicalFieldNames)
+ returnSchema.typeInfo,
+ returnSchema.fieldNames)
val collectorCode = if (condition.isEmpty) {
s"""
@@ -148,7 +146,7 @@ trait CommonCorrelate {
// adjust indicies of InputRefs to adhere to schema expected by generator
val changeInputRefIndexShuttle = new RexShuttle {
override def visitInputRef(inputRef: RexInputRef): RexNode = {
- new RexInputRef(inputSchema.physicalArity + inputRef.getIndex, inputRef.getType)
+ new RexInputRef(inputSchema.arity + inputRef.getIndex, inputRef.getType)
}
}
// Run generateExpression to add init statements (ScalarFunctions) of condition to generator.
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala
index dc7a0d6..5872d8c 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/PhysicalTableSourceScan.scala
@@ -39,9 +39,7 @@ abstract class PhysicalTableSourceScan(
val flinkTypeFactory = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
flinkTypeFactory.buildLogicalRowType(
TableEnvironment.getFieldNames(tableSource),
- TableEnvironment.getFieldTypes(tableSource.getReturnType),
- None,
- None)
+ TableEnvironment.getFieldTypes(tableSource.getReturnType))
}
override def explainTerms(pw: RelWriter): RelWriter = {
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
index fb291e4..74aac43 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/BatchTableSourceScan.scala
@@ -42,9 +42,7 @@ class BatchTableSourceScan(
val flinkTypeFactory = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
flinkTypeFactory.buildLogicalRowType(
TableEnvironment.getFieldNames(tableSource),
- TableEnvironment.getFieldTypes(tableSource.getReturnType),
- None,
- None)
+ TableEnvironment.getFieldTypes(tableSource.getReturnType))
}
override def computeSelfCost (planner: RelOptPlanner, metadata: RelMetadataQuery): RelOptCost = {
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCalc.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCalc.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCalc.scala
index 2e00330..45e6902 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCalc.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCalc.scala
@@ -52,7 +52,7 @@ class DataStreamCalc(
with CommonCalc
with DataStreamRel {
- override def deriveRowType(): RelDataType = schema.logicalType
+ override def deriveRowType(): RelDataType = schema.relDataType
override def copy(traitSet: RelTraitSet, child: RelNode, program: RexProgram): Calc = {
new DataStreamCalc(
@@ -100,7 +100,7 @@ class DataStreamCalc(
val condition = if (calcProgram.getCondition != null) {
val materializedCondition = RelTimeIndicatorConverter.convertExpression(
calcProgram.expandLocalRef(calcProgram.getCondition),
- inputSchema.logicalType,
+ inputSchema.relDataType,
cluster.getRexBuilder)
Some(materializedCondition)
} else {
@@ -110,12 +110,8 @@ class DataStreamCalc(
// filter out time attributes
val projection = calcProgram.getProjectList.asScala
.map(calcProgram.expandLocalRef)
- // time indicator fields must not be part of the code generation
- .filter(expr => !FlinkTypeFactory.isTimeIndicatorType(expr.getType))
- // update indices
- .map(expr => inputSchema.mapRexNode(expr))
- val generator = new FunctionCodeGenerator(config, false, inputSchema.physicalTypeInfo)
+ val generator = new FunctionCodeGenerator(config, false, inputSchema.typeInfo)
val genFunction = generateFunction(
generator,
@@ -132,7 +128,7 @@ class DataStreamCalc(
val processFunc = new CRowProcessRunner(
genFunction.name,
genFunction.code,
- CRowTypeInfo(schema.physicalTypeInfo))
+ CRowTypeInfo(schema.typeInfo))
inputDataStream
.process(processFunc)
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCorrelate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCorrelate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCorrelate.scala
index b7165cd..18ab2a3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCorrelate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamCorrelate.scala
@@ -50,7 +50,7 @@ class DataStreamCorrelate(
with CommonCorrelate
with DataStreamRel {
- override def deriveRowType() = schema.logicalType
+ override def deriveRowType() = schema.relDataType
override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
new DataStreamCorrelate(
@@ -78,7 +78,7 @@ class DataStreamCorrelate(
super.explainTerms(pw)
.item("invocation", scan.getCall)
.item("function", sqlFunction.getTableFunction.getClass.getCanonicalName)
- .item("rowType", schema.logicalType)
+ .item("rowType", schema.relDataType)
.item("joinType", joinType)
.itemIf("condition", condition.orNull, condition.isDefined)
}
@@ -130,7 +130,7 @@ class DataStreamCorrelate(
.process(processFunc)
// preserve input parallelism to ensure that acc and retract messages remain in order
.setParallelism(inputParallelism)
- .name(correlateOpName(rexCall, sqlFunction, schema.logicalType))
+ .name(correlateOpName(rexCall, sqlFunction, schema.relDataType))
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
index 12694fc..590d9be 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
@@ -59,7 +59,7 @@ class DataStreamGroupAggregate(
private val LOG = LoggerFactory.getLogger(this.getClass)
- override def deriveRowType() = schema.logicalType
+ override def deriveRowType() = schema.relDataType
override def needsUpdatesAsRetraction = true
@@ -83,20 +83,20 @@ class DataStreamGroupAggregate(
override def toString: String = {
s"Aggregate(${
if (!groupings.isEmpty) {
- s"groupBy: (${groupingToString(inputSchema.logicalType, groupings)}), "
+ s"groupBy: (${groupingToString(inputSchema.relDataType, groupings)}), "
} else {
""
}
}select:(${aggregationToString(
- inputSchema.logicalType, groupings, getRowType, namedAggregates, Nil)}))"
+ inputSchema.relDataType, groupings, getRowType, namedAggregates, Nil)}))"
}
override def explainTerms(pw: RelWriter): RelWriter = {
super.explainTerms(pw)
.itemIf("groupBy", groupingToString(
- inputSchema.logicalType, groupings), !groupings.isEmpty)
+ inputSchema.relDataType, groupings), !groupings.isEmpty)
.item("select", aggregationToString(
- inputSchema.logicalType, groupings, getRowType, namedAggregates, Nil))
+ inputSchema.relDataType, groupings, getRowType, namedAggregates, Nil))
}
override def translateToPlan(
@@ -112,37 +112,29 @@ class DataStreamGroupAggregate(
val inputDS = input.asInstanceOf[DataStreamRel].translateToPlan(tableEnv, queryConfig)
- val physicalNamedAggregates = namedAggregates.map { namedAggregate =>
- new CalcitePair[AggregateCall, String](
- inputSchema.mapAggregateCall(namedAggregate.left),
- namedAggregate.right)
- }
-
- val outRowType = CRowTypeInfo(schema.physicalTypeInfo)
+ val outRowType = CRowTypeInfo(schema.typeInfo)
val generator = new AggregationCodeGenerator(
tableEnv.getConfig,
false,
- inputSchema.physicalTypeInfo)
+ inputSchema.typeInfo)
val aggString = aggregationToString(
- inputSchema.logicalType,
+ inputSchema.relDataType,
groupings,
getRowType,
namedAggregates,
Nil)
- val keyedAggOpName = s"groupBy: (${groupingToString(inputSchema.logicalType, groupings)}), " +
+ val keyedAggOpName = s"groupBy: (${groupingToString(inputSchema.relDataType, groupings)}), " +
s"select: ($aggString)"
val nonKeyedAggOpName = s"select: ($aggString)"
- val physicalGrouping = groupings.map(inputSchema.mapIndex)
-
val processFunction = AggregateUtil.createGroupAggregateFunction(
generator,
- physicalNamedAggregates,
- inputSchema.logicalType,
- inputSchema.physicalFieldTypeInfo,
+ namedAggregates,
+ inputSchema.relDataType,
+ inputSchema.fieldTypeInfos,
groupings,
queryConfig,
DataStreamRetractionRules.isAccRetract(this),
@@ -150,7 +142,7 @@ class DataStreamGroupAggregate(
val result: DataStream[CRow] =
// grouped / keyed aggregation
- if (physicalGrouping.nonEmpty) {
+ if (groupings.nonEmpty) {
inputDS
.keyBy(groupings: _*)
.process(processFunction)
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/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 c4ffdb1..ac63be1 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
@@ -32,11 +32,13 @@ import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.codegen.AggregationCodeGenerator
import org.apache.flink.table.expressions.ExpressionUtils._
+import org.apache.flink.table.expressions.ResolvedFieldReference
import org.apache.flink.table.plan.logical._
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.aggregate.AggregateUtil._
import org.apache.flink.table.runtime.aggregate._
import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
@@ -58,7 +60,7 @@ class DataStreamGroupWindowAggregate(
private val LOG = LoggerFactory.getLogger(this.getClass)
- override def deriveRowType(): RelDataType = schema.logicalType
+ override def deriveRowType(): RelDataType = schema.relDataType
override def needsUpdatesAsRetraction = true
@@ -84,14 +86,14 @@ class DataStreamGroupWindowAggregate(
override def toString: String = {
s"Aggregate(${
if (!grouping.isEmpty) {
- s"groupBy: (${groupingToString(inputSchema.logicalType, grouping)}), "
+ s"groupBy: (${groupingToString(inputSchema.relDataType, grouping)}), "
} else {
""
}
}window: ($window), " +
s"select: (${
aggregationToString(
- inputSchema.logicalType,
+ inputSchema.relDataType,
grouping,
getRowType,
namedAggregates,
@@ -101,13 +103,13 @@ class DataStreamGroupWindowAggregate(
override def explainTerms(pw: RelWriter): RelWriter = {
super.explainTerms(pw)
- .itemIf("groupBy", groupingToString(inputSchema.logicalType, grouping), !grouping.isEmpty)
+ .itemIf("groupBy", groupingToString(inputSchema.relDataType, grouping), !grouping.isEmpty)
.item("window", window)
.item(
"select", aggregationToString(
- inputSchema.logicalType,
+ inputSchema.relDataType,
grouping,
- schema.logicalType,
+ schema.relDataType,
namedAggregates,
namedProperties))
}
@@ -118,14 +120,6 @@ class DataStreamGroupWindowAggregate(
val inputDS = input.asInstanceOf[DataStreamRel].translateToPlan(tableEnv, queryConfig)
- val physicalNamedAggregates = namedAggregates.map { namedAggregate =>
- new CalcitePair[AggregateCall, String](
- inputSchema.mapAggregateCall(namedAggregate.left),
- namedAggregate.right)
- }
- val physicalNamedProperties = namedProperties
- .filter(np => !FlinkTypeFactory.isTimeIndicatorType(np.property.resultType))
-
val inputIsAccRetract = DataStreamRetractionRules.isAccRetract(input)
if (inputIsAccRetract) {
@@ -148,16 +142,30 @@ class DataStreamGroupWindowAggregate(
"state size. You may specify a retention time of 0 to not clean up the state.")
}
- val outRowType = CRowTypeInfo(schema.physicalTypeInfo)
+ val timestampedInput = if (isRowtimeAttribute(window.timeAttribute)) {
+ // copy the window rowtime attribute into the StreamRecord timestamp field
+ val timeAttribute = window.timeAttribute.asInstanceOf[ResolvedFieldReference].name
+ val timeIdx = inputSchema.fieldNames.indexOf(timeAttribute)
+
+ inputDS
+ .process(
+ new TimestampSetterProcessFunction(timeIdx, CRowTypeInfo(inputSchema.typeInfo)))
+ .setParallelism(inputDS.getParallelism)
+ .name(s"time attribute: ($timeAttribute)")
+ } else {
+ inputDS
+ }
+
+ val outRowType = CRowTypeInfo(schema.typeInfo)
val aggString = aggregationToString(
- inputSchema.logicalType,
+ inputSchema.relDataType,
grouping,
- schema.logicalType,
+ schema.relDataType,
namedAggregates,
namedProperties)
- val keyedAggOpName = s"groupBy: (${groupingToString(inputSchema.logicalType, grouping)}), " +
+ val keyedAggOpName = s"groupBy: (${groupingToString(inputSchema.relDataType, grouping)}), " +
s"window: ($window), " +
s"select: ($aggString)"
val nonKeyedAggOpName = s"window: ($window), select: ($aggString)"
@@ -165,24 +173,22 @@ class DataStreamGroupWindowAggregate(
val generator = new AggregationCodeGenerator(
tableEnv.getConfig,
false,
- inputSchema.physicalTypeInfo)
+ inputSchema.typeInfo)
val needMerge = window match {
case SessionGroupWindow(_, _, _) => true
case _ => false
}
- val physicalGrouping = grouping.map(inputSchema.mapIndex)
-
// grouped / keyed aggregation
- if (physicalGrouping.length > 0) {
+ if (grouping.length > 0) {
val windowFunction = AggregateUtil.createAggregationGroupWindowFunction(
window,
- physicalGrouping.length,
- physicalNamedAggregates.size,
- schema.physicalArity,
- physicalNamedProperties)
+ grouping.length,
+ namedAggregates.size,
+ schema.arity,
+ namedProperties)
- val keyedStream = inputDS.keyBy(physicalGrouping: _*)
+ val keyedStream = timestampedInput.keyBy(grouping: _*)
val windowedStream =
createKeyedWindowedStream(queryConfig, window, keyedStream)
.asInstanceOf[WindowedStream[CRow, Tuple, DataStreamWindow]]
@@ -190,11 +196,11 @@ class DataStreamGroupWindowAggregate(
val (aggFunction, accumulatorRowType, aggResultRowType) =
AggregateUtil.createDataStreamAggregateFunction(
generator,
- physicalNamedAggregates,
- inputSchema.physicalType,
- inputSchema.physicalFieldTypeInfo,
- schema.physicalType,
- physicalGrouping,
+ namedAggregates,
+ inputSchema.relDataType,
+ inputSchema.fieldTypeInfos,
+ schema.relDataType,
+ grouping,
needMerge)
windowedStream
@@ -205,20 +211,20 @@ class DataStreamGroupWindowAggregate(
else {
val windowFunction = AggregateUtil.createAggregationAllWindowFunction(
window,
- schema.physicalArity,
- physicalNamedProperties)
+ schema.arity,
+ namedProperties)
val windowedStream =
- createNonKeyedWindowedStream(queryConfig, window, inputDS)
+ createNonKeyedWindowedStream(queryConfig, window, timestampedInput)
.asInstanceOf[AllWindowedStream[CRow, DataStreamWindow]]
val (aggFunction, accumulatorRowType, aggResultRowType) =
AggregateUtil.createDataStreamAggregateFunction(
generator,
- physicalNamedAggregates,
- inputSchema.physicalType,
- inputSchema.physicalFieldTypeInfo,
- schema.physicalType,
+ namedAggregates,
+ inputSchema.relDataType,
+ inputSchema.fieldTypeInfos,
+ schema.relDataType,
Array[Int](),
needMerge)
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
index 34a7fd8..7bf342a 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
@@ -50,7 +50,7 @@ class DataStreamOverAggregate(
with DataStreamRel {
private val LOG = LoggerFactory.getLogger(this.getClass)
- override def deriveRowType(): RelDataType = schema.logicalType
+ override def deriveRowType(): RelDataType = schema.relDataType
override def needsUpdatesAsRetraction = true
@@ -78,15 +78,15 @@ class DataStreamOverAggregate(
super.explainTerms(pw)
.itemIf("partitionBy",
- partitionToString(schema.logicalType, partitionKeys), partitionKeys.nonEmpty)
+ partitionToString(schema.relDataType, partitionKeys), partitionKeys.nonEmpty)
.item("orderBy",
- orderingToString(schema.logicalType, overWindow.orderKeys.getFieldCollations))
+ orderingToString(schema.relDataType, overWindow.orderKeys.getFieldCollations))
.itemIf("rows", windowRange(logicWindow, overWindow, inputNode), overWindow.isRows)
.itemIf("range", windowRange(logicWindow, overWindow, inputNode), !overWindow.isRows)
.item(
"select", aggregationToString(
- inputSchema.logicalType,
- schema.logicalType,
+ inputSchema.relDataType,
+ schema.relDataType,
namedAggregates))
}
@@ -134,67 +134,44 @@ class DataStreamOverAggregate(
val generator = new AggregationCodeGenerator(
tableEnv.getConfig,
false,
- inputSchema.physicalTypeInfo)
+ inputSchema.typeInfo)
- val timeType = schema.logicalType
+ val timeType = schema.relDataType
.getFieldList
.get(orderKey.getFieldIndex)
.getType
- timeType match {
- case _ if FlinkTypeFactory.isProctimeIndicatorType(timeType) =>
- // proc-time OVER window
- if (overWindow.lowerBound.isUnbounded && overWindow.upperBound.isCurrentRow) {
- // unbounded OVER window
- createUnboundedAndCurrentRowOverWindow(
- queryConfig,
- generator,
- inputDS,
- isRowTimeType = false,
- isRowsClause = overWindow.isRows)
- } else if (
- overWindow.lowerBound.isPreceding && !overWindow.lowerBound.isUnbounded &&
- overWindow.upperBound.isCurrentRow) {
-
- // bounded OVER window
- createBoundedAndCurrentRowOverWindow(
- queryConfig,
- generator,
- inputDS,
- isRowTimeType = false,
- isRowsClause = overWindow.isRows)
- } else {
- throw new TableException(
- "OVER RANGE FOLLOWING windows are not supported yet.")
- }
-
- case _ if FlinkTypeFactory.isRowtimeIndicatorType(timeType) =>
- // row-time OVER window
- if (overWindow.lowerBound.isPreceding &&
- overWindow.lowerBound.isUnbounded && overWindow.upperBound.isCurrentRow) {
- // unbounded OVER window
- createUnboundedAndCurrentRowOverWindow(
- queryConfig,
- generator,
- inputDS,
- isRowTimeType = true,
- isRowsClause = overWindow.isRows)
- } else if (overWindow.lowerBound.isPreceding && overWindow.upperBound.isCurrentRow) {
- // bounded OVER window
- createBoundedAndCurrentRowOverWindow(
- queryConfig,
- generator,
- inputDS,
- isRowTimeType = true,
- isRowsClause = overWindow.isRows)
- } else {
- throw new TableException(
- "OVER RANGE FOLLOWING windows are not supported yet.")
- }
-
- case _ =>
- throw new TableException(
- s"OVER windows can only be applied on time attributes.")
+ // identify window rowtime attribute
+ val rowTimeIdx: Option[Int] = if (FlinkTypeFactory.isRowtimeIndicatorType(timeType)) {
+ Some(orderKey.getFieldIndex)
+ } else if (FlinkTypeFactory.isProctimeIndicatorType(timeType)) {
+ None
+ } else {
+ throw new TableException(s"OVER windows can only be applied on time attributes.")
+ }
+
+ if (overWindow.lowerBound.isPreceding && overWindow.lowerBound.isUnbounded &&
+ overWindow.upperBound.isCurrentRow) {
+ // unbounded OVER window
+ createUnboundedAndCurrentRowOverWindow(
+ queryConfig,
+ generator,
+ inputDS,
+ rowTimeIdx,
+ isRowsClause = overWindow.isRows)
+ } else if (
+ overWindow.lowerBound.isPreceding && !overWindow.lowerBound.isUnbounded &&
+ overWindow.upperBound.isCurrentRow) {
+
+ // bounded OVER window
+ createBoundedAndCurrentRowOverWindow(
+ queryConfig,
+ generator,
+ inputDS,
+ rowTimeIdx,
+ isRowsClause = overWindow.isRows)
+ } else {
+ throw new TableException("OVER RANGE FOLLOWING windows are not supported yet.")
}
}
@@ -202,31 +179,26 @@ class DataStreamOverAggregate(
queryConfig: StreamQueryConfig,
generator: AggregationCodeGenerator,
inputDS: DataStream[CRow],
- isRowTimeType: Boolean,
+ rowTimeIdx: Option[Int],
isRowsClause: Boolean): DataStream[CRow] = {
val overWindow: Group = logicWindow.groups.get(0)
- val partitionKeys: Array[Int] = overWindow.keys.toArray.map(schema.mapIndex)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
- val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates.map {
- namedAggregate =>
- new CalcitePair[AggregateCall, String](
- schema.mapAggregateCall(namedAggregate.left),
- namedAggregate.right)
- }
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
// get the output types
- val returnTypeInfo = CRowTypeInfo(schema.physicalTypeInfo)
+ val returnTypeInfo = CRowTypeInfo(schema.typeInfo)
val processFunction = AggregateUtil.createUnboundedOverProcessFunction(
generator,
namedAggregates,
- inputSchema.physicalType,
- inputSchema.physicalTypeInfo,
- inputSchema.physicalFieldTypeInfo,
+ inputSchema.relDataType,
+ inputSchema.typeInfo,
+ inputSchema.fieldTypeInfos,
queryConfig,
- isRowTimeType,
+ rowTimeIdx,
partitionKeys.nonEmpty,
isRowsClause)
@@ -254,34 +226,29 @@ class DataStreamOverAggregate(
queryConfig: StreamQueryConfig,
generator: AggregationCodeGenerator,
inputDS: DataStream[CRow],
- isRowTimeType: Boolean,
+ rowTimeIdx: Option[Int],
isRowsClause: Boolean): DataStream[CRow] = {
val overWindow: Group = logicWindow.groups.get(0)
- val partitionKeys: Array[Int] = overWindow.keys.toArray.map(schema.mapIndex)
- val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates.map {
- namedAggregate =>
- new CalcitePair[AggregateCall, String](
- schema.mapAggregateCall(namedAggregate.left),
- namedAggregate.right)
- }
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
val precedingOffset =
getLowerBoundary(logicWindow, overWindow, getInput()) + (if (isRowsClause) 1 else 0)
// get the output types
- val returnTypeInfo = CRowTypeInfo(schema.physicalTypeInfo)
+ val returnTypeInfo = CRowTypeInfo(schema.typeInfo)
val processFunction = AggregateUtil.createBoundedOverProcessFunction(
generator,
namedAggregates,
- inputSchema.physicalType,
- inputSchema.physicalTypeInfo,
- inputSchema.physicalFieldTypeInfo,
+ inputSchema.relDataType,
+ inputSchema.typeInfo,
+ inputSchema.fieldTypeInfos,
precedingOffset,
queryConfig,
isRowsClause,
- isRowTimeType
+ rowTimeIdx
)
val result: DataStream[CRow] =
// partitioned aggregation
@@ -318,18 +285,18 @@ class DataStreamOverAggregate(
s"over: (${
if (!partitionKeys.isEmpty) {
- s"PARTITION BY: ${partitionToString(inputSchema.logicalType, partitionKeys)}, "
+ s"PARTITION BY: ${partitionToString(inputSchema.relDataType, partitionKeys)}, "
} else {
""
}
- }ORDER BY: ${orderingToString(inputSchema.logicalType,
+ }ORDER BY: ${orderingToString(inputSchema.relDataType,
overWindow.orderKeys.getFieldCollations)}, " +
s"${if (overWindow.isRows) "ROWS" else "RANGE"}" +
s"${windowRange(logicWindow, overWindow, inputNode)}, " +
s"select: (${
aggregationToString(
- inputSchema.logicalType,
- schema.logicalType,
+ inputSchema.relDataType,
+ schema.relDataType,
namedAggregates)
}))"
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamScan.scala
index 424c6a2..9352efb 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamScan.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamScan.scala
@@ -43,7 +43,7 @@ class DataStreamScan(
val dataStreamTable: DataStreamTable[Any] = getTable.unwrap(classOf[DataStreamTable[Any]])
- override def deriveRowType(): RelDataType = schema.logicalType
+ override def deriveRowType(): RelDataType = schema.relDataType
override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
new DataStreamScan(
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala
index a11e6c1..8f9942f 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamSort.scala
@@ -53,7 +53,7 @@ class DataStreamSort(
with CommonSort
with DataStreamRel {
- override def deriveRowType(): RelDataType = schema.logicalType
+ override def deriveRowType(): RelDataType = schema.relDataType
override def copy(
traitSet: RelTraitSet,
@@ -75,13 +75,13 @@ class DataStreamSort(
}
override def toString: String = {
- sortToString(schema.logicalType, sortCollation, sortOffset, sortFetch)
+ sortToString(schema.relDataType, sortCollation, sortOffset, sortFetch)
}
override def explainTerms(pw: RelWriter) : RelWriter = {
sortExplainTerms(
pw.input("input", getInput()),
- schema.logicalType,
+ schema.relDataType,
sortCollation,
sortOffset,
sortFetch)
@@ -94,7 +94,7 @@ class DataStreamSort(
val inputDS = input.asInstanceOf[DataStreamRel].translateToPlan(tableEnv, queryConfig)
// need to identify time between others order fields. Time needs to be first sort element
- val timeType = SortUtil.getFirstSortField(sortCollation, schema.logicalType).getType
+ val timeType = SortUtil.getFirstSortField(sortCollation, schema.relDataType).getType
// time ordering needs to be ascending
if (SortUtil.getFirstSortDirection(sortCollation) != Direction.ASCENDING) {
@@ -141,15 +141,15 @@ class DataStreamSort(
inputDS: DataStream[CRow],
execCfg: ExecutionConfig): DataStream[CRow] = {
- val returnTypeInfo = CRowTypeInfo(schema.physicalTypeInfo)
+ val returnTypeInfo = CRowTypeInfo(schema.typeInfo)
// if the order has secondary sorting fields in addition to the proctime
if (sortCollation.getFieldCollations.size() > 1) {
val processFunction = SortUtil.createProcTimeSortFunction(
sortCollation,
- inputSchema.logicalType,
- inputSchema.physicalTypeInfo,
+ inputSchema.relDataType,
+ inputSchema.typeInfo,
execCfg)
inputDS.keyBy(new NullByteKeySelector[CRow])
@@ -173,12 +173,12 @@ class DataStreamSort(
inputDS: DataStream[CRow],
execCfg: ExecutionConfig): DataStream[CRow] = {
- val returnTypeInfo = CRowTypeInfo(schema.physicalTypeInfo)
+ val returnTypeInfo = CRowTypeInfo(schema.typeInfo)
val processFunction = SortUtil.createRowTimeSortFunction(
sortCollation,
- inputSchema.logicalType,
- inputSchema.physicalTypeInfo,
+ inputSchema.relDataType,
+ inputSchema.typeInfo,
execCfg)
inputDS.keyBy(new NullByteKeySelector[CRow])
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamUnion.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamUnion.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamUnion.scala
index 6f4980a..7258ec8 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamUnion.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamUnion.scala
@@ -38,7 +38,7 @@ class DataStreamUnion(
extends BiRel(cluster, traitSet, leftNode, rightNode)
with DataStreamRel {
- override def deriveRowType() = schema.logicalType
+ override def deriveRowType() = schema.relDataType
override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
new DataStreamUnion(
@@ -55,7 +55,7 @@ class DataStreamUnion(
}
override def toString = {
- s"Union All(union: (${schema.logicalFieldNames.mkString(", ")}))"
+ s"Union All(union: (${schema.fieldNames.mkString(", ")}))"
}
override def translateToPlan(
@@ -68,6 +68,6 @@ class DataStreamUnion(
}
private def unionSelectionToString: String = {
- schema.logicalFieldNames.mkString(", ")
+ schema.fieldNames.mkString(", ")
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamValues.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamValues.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamValues.scala
index 1476681..1ef9107 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamValues.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamValues.scala
@@ -41,10 +41,10 @@ class DataStreamValues(
schema: RowSchema,
tuples: ImmutableList[ImmutableList[RexLiteral]],
ruleDescription: String)
- extends Values(cluster, schema.logicalType, tuples, traitSet)
+ extends Values(cluster, schema.relDataType, tuples, traitSet)
with DataStreamRel {
- override def deriveRowType() = schema.logicalType
+ override def deriveRowType() = schema.relDataType
override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
new DataStreamValues(
@@ -62,14 +62,14 @@ class DataStreamValues(
val config = tableEnv.getConfig
- val returnType = CRowTypeInfo(schema.physicalTypeInfo)
+ val returnType = CRowTypeInfo(schema.typeInfo)
val generator = new InputFormatCodeGenerator(config)
// generate code for every record
val generatedRecords = getTuples.asScala.map { r =>
generator.generateResultExpression(
- schema.physicalTypeInfo,
- schema.physicalFieldNames,
+ schema.typeInfo,
+ schema.fieldNames,
r.asScala)
}
@@ -77,7 +77,7 @@ class DataStreamValues(
val generatedFunction = generator.generateValuesInputFormat(
ruleDescription,
generatedRecords.map(_.code),
- schema.physicalTypeInfo)
+ schema.typeInfo)
val inputFormat = new CRowValuesInputFormat(
generatedFunction.name,
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala
index 987947c..f8015b3 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamWindowJoin.scala
@@ -54,7 +54,7 @@ class DataStreamWindowJoin(
with CommonJoin
with DataStreamRel {
- override def deriveRowType(): RelDataType = schema.logicalType
+ override def deriveRowType(): RelDataType = schema.relDataType
override def copy(traitSet: RelTraitSet, inputs: java.util.List[RelNode]): RelNode = {
new DataStreamWindowJoin(
@@ -76,7 +76,7 @@ class DataStreamWindowJoin(
override def toString: String = {
joinToString(
- schema.logicalType,
+ schema.relDataType,
joinCondition,
joinType,
getExpressionString)
@@ -85,7 +85,7 @@ class DataStreamWindowJoin(
override def explainTerms(pw: RelWriter): RelWriter = {
joinExplainTerms(
super.explainTerms(pw),
- schema.logicalType,
+ schema.relDataType,
joinCondition,
joinType,
getExpressionString)
@@ -117,8 +117,8 @@ class DataStreamWindowJoin(
WindowJoinUtil.generateJoinFunction(
config,
joinType,
- leftSchema.physicalTypeInfo,
- rightSchema.physicalTypeInfo,
+ leftSchema.typeInfo,
+ rightSchema.typeInfo,
schema,
remainCondition,
ruleDescription)
@@ -160,13 +160,13 @@ class DataStreamWindowJoin(
leftKeys: Array[Int],
rightKeys: Array[Int]): DataStream[CRow] = {
- val returnTypeInfo = CRowTypeInfo(schema.physicalTypeInfo)
+ val returnTypeInfo = CRowTypeInfo(schema.typeInfo)
val procInnerJoinFunc = new ProcTimeWindowInnerJoin(
leftLowerBound,
leftUpperBound,
- leftSchema.physicalTypeInfo,
- rightSchema.physicalTypeInfo,
+ leftSchema.typeInfo,
+ rightSchema.typeInfo,
joinFunctionName,
joinFunctionCode)
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamScan.scala
index 25e72fa..4aca856 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamScan.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamScan.scala
@@ -18,14 +18,15 @@
package org.apache.flink.table.plan.nodes.datastream
-import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.table.api.TableConfig
+import org.apache.flink.table.codegen.FunctionCodeGenerator
import org.apache.flink.table.plan.nodes.CommonScan
import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.types.Row
import org.apache.flink.table.plan.schema.FlinkTable
-import org.apache.flink.table.runtime.CRowOutputMapRunner
+import org.apache.flink.table.runtime.CRowOutputProcessRunner
import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
import scala.collection.JavaConverters._
@@ -40,29 +41,42 @@ trait StreamScan extends CommonScan[CRow] with DataStreamRel {
: DataStream[CRow] = {
val inputType = input.getType
- val internalType = CRowTypeInfo(schema.physicalTypeInfo)
+ val internalType = CRowTypeInfo(schema.typeInfo)
// conversion
if (needsConversion(input.getType, internalType)) {
- val function = generatedConversionFunction(
+ val generator = new FunctionCodeGenerator(
config,
- classOf[MapFunction[Any, Row]],
+ false,
inputType,
- schema.physicalTypeInfo,
- "DataStreamSourceConversion",
- schema.physicalFieldNames,
+ None,
Some(flinkTable.fieldIndexes))
- val mapFunc = new CRowOutputMapRunner(
+ val conversion = generator.generateConverterResultExpression(
+ schema.typeInfo,
+ schema.fieldNames)
+
+ val body =
+ s"""
+ |${conversion.code}
+ |${generator.collectorTerm}.collect(${conversion.resultTerm});
+ |""".stripMargin
+
+ val function = generator.generateFunction(
+ "DataStreamSourceConversion",
+ classOf[ProcessFunction[Any, Row]],
+ body,
+ schema.typeInfo)
+
+ val processFunc = new CRowOutputProcessRunner(
function.name,
function.code,
internalType)
val opName = s"from: (${getRowType.getFieldNames.asScala.toList.mkString(", ")})"
- // TODO we need a ProcessFunction here
- input.map(mapFunc).name(opName).returns(internalType)
+ input.process(processFunc).name(opName).returns(internalType)
}
// no conversion necessary, forward
else {
http://git-wip-us.apache.org/repos/asf/flink/blob/93d0ae4a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
index 72ecac5..663b276 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/StreamTableSourceScan.scala
@@ -30,6 +30,7 @@ import org.apache.flink.table.sources._
import org.apache.flink.table.plan.schema.TableSourceTable
import org.apache.flink.table.runtime.types.CRow
import org.apache.flink.table.sources.{StreamTableSource, TableSource}
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
/** Flink RelNode to read data from an external source defined by a [[StreamTableSource]]. */
class StreamTableSourceScan(
@@ -46,29 +47,29 @@ class StreamTableSourceScan(
val fieldNames = TableEnvironment.getFieldNames(tableSource).toList
val fieldTypes = TableEnvironment.getFieldTypes(tableSource.getReturnType).toList
- val fieldCnt = fieldNames.length
+ val fields = fieldNames.zip(fieldTypes)
- val rowtime = tableSource match {
+ val withRowtime = tableSource match {
case timeSource: DefinedRowtimeAttribute if timeSource.getRowtimeAttribute != null =>
val rowtimeAttribute = timeSource.getRowtimeAttribute
- Some((fieldCnt, rowtimeAttribute))
+ fields :+ (rowtimeAttribute, TimeIndicatorTypeInfo.ROWTIME_INDICATOR)
case _ =>
- None
+ fields
}
- val proctime = tableSource match {
+ val withProctime = tableSource match {
case timeSource: DefinedProctimeAttribute if timeSource.getProctimeAttribute != null =>
val proctimeAttribute = timeSource.getProctimeAttribute
- Some((fieldCnt + (if (rowtime.isDefined) 1 else 0), proctimeAttribute))
+ withRowtime :+ (proctimeAttribute, TimeIndicatorTypeInfo.PROCTIME_INDICATOR)
case _ =>
- None
+ withRowtime
}
+ val (fieldNamesWithIndicators, fieldTypesWithIndicators) = withProctime.unzip
+
flinkTypeFactory.buildLogicalRowType(
- fieldNames,
- fieldTypes,
- rowtime,
- proctime)
+ fieldNamesWithIndicators,
+ fieldTypesWithIndicators)
}
override def computeSelfCost (planner: RelOptPlanner, metadata: RelMetadataQuery): RelOptCost = {