[Bug] [Lookup] ChunJun 1.12 dimension table cannot be associated with the latest data

[FlechazoW]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

flinksql 1.12 使用mysql维表（维表缓存类型为NONE）关联数据发现维表数据更新后，关联到的数据依旧是第一次关联的值，并不是最新值。

What you expected to happen

使用NONE缓存时，维表数据更新后，下一次数据获取的是最新值。

How to reproduce

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;

kafka 写入数据 id =1，更新mysql 数据

Anything else

No response

Version

master

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct