title: how to auto generate watermark in flink sql table?
body: <p>I am testing flink cep sql and my watermark is define as row time, my table is a kafka table. As watermark depends on minimal of all the kafka partitions, so every new message has to wait kafka partition align, then cep trigger results.</p>
  <p>My kafka table(topic has 3 partitions) is defined as</p>
  <pre><code>create table test_table(
      agent_id String, room_id String, 
      create_time Bigint, 
      call_type String, 
      application_id String, 
      connect_time Bigint, 
      row_time as to_timestamp_ltz(create_time, 3), 
      WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
  )
  </code></pre>
  <p>Here is my cep sql</p>
  <pre><code>select * from test_table  match_recognize (
     partition by agent_id,room_id,call_type 
     order by row_time
     measures  
         last(BF.create_time) as create_time, 
         last(AF.connect_time) as connect_time 
     one row per match after match SKIP PAST LAST ROW 
     pattern (BF+ AF) WITHIN INTERVAL '1' HOUR 
     define 
         BF as BF.connect_time = 0,
         AF as AF.connect_time &gt; 0 and BF.room_id = AF.room_id and BF.call_type = AF.call_type 
  ) as T ;
  </code></pre>
  <p>The cep sql trigger result is right but is always late cause every partition needs align watermark. How can I get the newest result <strong>immediately</strong> or <strong>auto generate watermark in flink sql table</strong>?</p>