Github user twalthr commented on a diff in the pull request: https://github.com/apache/flink/pull/4894#discussion_r146584051 --- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala --- @@ -246,40 +245,31 @@ abstract class CodeGenerator( */ def generateConverterResultExpression( returnType: TypeInformation[_ <: Any], - resultFieldNames: Seq[String]) + resultFieldNames: Seq[String], + rowtimeExpression: Option[RexNode] = None) : GeneratedExpression = { val input1AccessExprs = input1Mapping.map { - case TimeIndicatorTypeInfo.ROWTIME_MARKER => - // attribute is a rowtime indicator. Access event-time timestamp in StreamRecord. - generateRowtimeAccess() - case TimeIndicatorTypeInfo.PROCTIME_MARKER => + case TimeIndicatorTypeInfo.ROWTIME_STREAM_MARKER | + TimeIndicatorTypeInfo.ROWTIME_BATCH_MARKER => + // attribute is a rowtime indicator. + if (rowtimeExpression.isDefined) { --- End diff -- We could use pattern matching here.
---