Hi team, I’m wondering if I found a bug in Window TVF not correctly propagating

rowtime

in the

toDataStream

conversion, at least in batch execution mode. The core logic of what I’m doing is in the attached snippet. At a high level, the code extracts a field from a POJO in a DataStream, converts to the Table API, performs a Window TVF, and converts back to a DataStream<Tuple<>>. Then, that DataStream is converted back into a Table, where another Window TVF is applied, and the output is converted back to a different Tuple DataStream. I realize there’s no need to convert the first table back to DataStream, but the point is the

toDataStream

conversion isn’t, apparently, propagating the

rowtime

on the first Table. When this logic runs in batch execution mode, the error is:

org.apache.flink.util.FlinkRuntimeException: Could not find timestamp in DataStream API record. Make sure that timestamps have been assigned before and the event-time characteristic is enabled.

. From inserting some .shuffle calls to split the operators and narrow down the error source, it looks like the first phase of the 2nd window TVF is where it happens, i.e. this stage fails:

[40]:TableSourceScan(table=[[*anonymous_datastream_source$6*]], fields=[f0, f1, rowtime])
+- [41]:Calc(select=[f0, f1, CAST(rowtime AS TIMESTAMP_LTZ(3) *ROWTIME*) AS rowtime])
   +- [42]:WindowTableFunction(window=[HOP(time_col=[rowtime], size=[86400000 ms], slide=[1 h])])
      +- [43]:Calc(select=[f0, window_start, window_end, window_time AS rowtime, f1])
         +- [44]:LocalHashAggregate(groupBy=[f0, window_start, window_end, rowtime], select=[f0, window_start, window_end, rowtime, Partial_SUM(f1) AS sum$0])

The earlier stage looks like this:

[37]:HashAggregate(isMerge=[true], groupBy=[f0, window_start, window_end, rowtime], select=[f0, window_start, window_end, rowtime, Final_COUNT(count1$0) AS f3])
+- [38]:Calc(select=[f0, f3, rowtime])
   +- [39]:ConstraintEnforcer[NotNullEnforcer(fields=[f3, rowtime])]
      +- TableToDataSteam(type=ROW<`f0` INT, `f3` BIGINT NOT NULL, `rowtime` TIMESTAMP_LTZ(3) NOT NULL> NOT NULL, rowtime=false)

Not sure if it’s relevant, but I thought the

rowtime=false

was unusual. I expected the first Window TVF should propagate a time attribute. The documentation says

window_time

is a time attribute, and shows how it’s used in a subsequent TVF as the

rowtime

. So, I’d expect it to propagate via

toDataStream

, too.