Generate watermarks and set event time using expressions

[jbeisen]

Modify the periodic watermark generator to accept a physical expressions that it uses to calculate the watermark. The expression can be set via a virtual field in the create table statement, or it defaults to a max lateness expression. The operator is inserted via a user-defined logical node.

Add an expression to the projection in the SourceRewriter to set the event time if present.

Putting it all together, this query

CREATE TABLE input (
  price bigint,
  double_price bigint generated always as (price * 2) stored,
  my_time timestamp generated always as (cast('2020-01-01' as timestamp)) stored
) WITH (
  connector = 'filesystem',
  type = 'source',
  path = '/Users/jonaheisen/dev/data',
  format = 'parquet',
  'source.regex-pattern' = '.*\.parquet$',
  event_time_field = 'my_time',
  watermark_field = 'my_time'
);

select * from input;

generates this job graph