Flink SQL: How to tag a column as 'rowtime' from a Avro based DataStream?

[Niels Basjes]

Hi,

Experimenting with the StreamTableEnvironment I build something like this:

DataStream<Tuple3<Long, String, Long>> letterStream = ...
tableEnv.registerDataStream("TestStream", letterStream, "EventTime.rowtime,
letter, counter");


Because the "EventTime" was tagged with ".rowtime" it is now being used as
the rowtime and has the DATETIME so I can do this

TUMBLE_START(eventTime, INTERVAL '1' MINUTE)


So far so good.

Working towards a more realistic scenario I have a source that produces a
stream of records that have been defined using Apache Avro.

So I have a Measurement.avdl that (among other things) contains something
like this:

record Measurement {
   /** The time (epoch in milliseconds since 1970-01-01 UTC) when the event
occurred */
    long                        timestamp;
    string                      letter;
    long                        pageviews;
}


Now because the registerDataStream call can also derive the schema from the
provided data I can do this:

DataStream<Measurement> inputStream = ...
tableEnv.registerDataStream("DataStream", inputStream);


This is very nice because any real schema is big (few hundred columns) and
changes over time.

Now In the SQL the timestamp is a BIGINT and not a DATETIME and as a
consequence I get this error

Cannot apply 'TUMBLE' to arguments of type 'TUMBLE(<BIGINT>, <INTERVAL
MINUTE>)'. Supported form(s): 'TUMBLE(<DATETIME>, <DATETIME_INTERVAL>)'


So far I have now yet figured how to make the system understand that the
timestamp column show be treated as the rowtime.
How do I do that?

-- 
Best regards / Met vriendelijke groeten,

Niels Basjes