[flink] Implement Fluss full-cache lookup join provider

[loserwang1024]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Currently, Fluss Flink Connector supports lookup join through per-record lookup RPCs, optionally with Flink's partial lookup cache. In this mode, the connector does not preload dimension table data. Each cache miss triggers a lookup request to Fluss TabletServer.

This works for small traffic or low-cardinality lookup workloads, but it can become expensive in production when lookup QPS is high or the dimension table is relatively stable.

This issue aims to implement a Fluss full-cache lookup join provider, so lookup data can be loaded into local cache before serving lookup requests.

For many dimension table scenarios, users prefer full-cache lookup join:

SELECT s.order_id, s.user_id, d.user_name, d.level
FROM order_stream AS s
LEFT JOIN user_dim FOR SYSTEM_TIME AS OF s.proc_time AS d
ON s.user_id = d.user_id;

If user_dim is fully cached locally, lookup requests can be served without per-record RPC to Fluss, reducing latency and TabletServer pressure.

Solution

No response

Anything else?

No response

Willingness to contribute

• I'm willing to submit a PR!