CREATE TABLE source
(
    id   int,
    name varchar,
    proc_time AS PROCTIME()
) WITH (
      'properties.bootstrap.servers' = 'flink01:9092,flink02:9092,flink03:9092',
      'connector' = 'kafka-x',
      'scan.parallelism' = '1',
      'format' = 'json',
      'topic' = 'tiezhu_in_one',
      'scan.startup.mode' = 'latest-offset'
      );

CREATE TABLE dim
(
    id   int,
    name varchar,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
      'connector' = 'mysql-x',
      'url' = 'jdbc:mysql://localhost:3306/tiezhu?useSSL=false',
      'table-name' = 'test_one_bk',
      'username' = 'root',
      'password' = 'password',
      'lookup.cache-type' = 'NONE'
      );

CREATE TABLE sink
(
    id               int,
    name_from_source varchar,
    name_from_dim    varchar
) WITH (
      'connector' = 'stream-x',
      'print' = 'true'
      );

INSERT INTO sink
SELECT source.id     as id,
       source.name   as name_from_source,
       dim_join.name as name_from_dim
FROM source
         LEFT JOIN dim
    FOR SYSTEM_TIME AS OF source.proc_time AS dim_join
                   ON source.id = dim_join.id;