Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/4635#discussion_r139633811
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/schema/StreamTableSourceTable.scala
---
@@ -28,48 +29,113 @@ import
org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
class StreamTableSourceTable[T](
override val tableSource: TableSource[T],
override val statistic: FlinkStatistic = FlinkStatistic.UNKNOWN)
- extends TableSourceTable[T](tableSource, statistic) {
-
+ extends TableSourceTable[T](
+ tableSource,
+ StreamTableSourceTable.adjustFieldIndexes(tableSource),
+ StreamTableSourceTable.adjustFieldNames(tableSource),
+ statistic) {
override def getRowType(typeFactory: RelDataTypeFactory): RelDataType = {
+ val fieldTypes = StreamTableSourceTable.adjustFieldTypes(tableSource)
+
val flinkTypeFactory = typeFactory.asInstanceOf[FlinkTypeFactory]
+ flinkTypeFactory.buildLogicalRowType(
+ this.fieldNames,
+ fieldTypes)
+ }
- val fieldNames = TableEnvironment.getFieldNames(tableSource).toList
- val fieldTypes =
TableEnvironment.getFieldTypes(tableSource.getReturnType).toList
+}
- val fields = fieldNames.zip(fieldTypes)
+object StreamTableSourceTable {
+
+ private def adjustFieldIndexes(tableSource: TableSource[_]): Array[Int]
= {
+ val (rowtime, proctime) = getTimeIndicators(tableSource)
+
+ val original = TableEnvironment.getFieldIndices(tableSource)
+
+ // append rowtime marker
+ val withRowtime = if (rowtime.isDefined) {
+ original :+ TimeIndicatorTypeInfo.ROWTIME_MARKER
--- End diff --
That's how rowtime fields are handled at the moment. They are always
appended at the end of the row.
There is a JIRA to use existing fields as rowtime fields
(https://issues.apache.org/jira/browse/FLINK-7446) but this has not been
implemented yet. I'm currently working on a PR for that.
---
