Flink: Support Lookup join

[Guosmilesmile]

Summary

This PR adds lookup join support for Iceberg Flink table source, including both partial lookup cache, and full lookup cache with optional periodic refresh.

The implementation enables Iceberg tables to be used as temporal lookup join dimensions in Flink SQL.

Supported Features

• Lookup join against Iceberg table source

  • Supports Flink SQL temporal lookup join syntax:

    LEFT JOIN iceberg_catalog.`db`.`dim_table`
      FOR SYSTEM_TIME AS OF o.proc_time AS u
    ON o.user_id = u.user_id

• Lookup key filtering

  • Lookup keys are converted into Iceberg filter expressions.
  • NULL lookup keys are handled correctly with IS NULL semantics.

• Pushed-down filter support

  • Existing source filters are reused by lookup readers.
  • Join conditions such as:

    ON o.user_id = u.user_id AND u.city = 'beijing'

    are applied together with the lookup key condition.

• Partial lookup cache

  • Supports Flink built-in partial lookup cache options.
  • Uses DefaultLookupCache and PartialCachingLookupProvider.

• Full lookup cache

  • Loads the full projected Iceberg table into memory on first lookup.

• Periodic refresh for full lookup cache

  • Full cache can be refreshed periodically using a simple background timer.
  • If a scheduled refresh fails, the previous cache remains available.

How to Use

Basic lookup join

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users` FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

Lookup join with partial cache

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users`
/*+ OPTIONS(
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '10000'
) */
FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

Lookup join with full cache

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users`
/*+ OPTIONS(
  'lookup.cache' = 'FULL'
) */
FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

or

CREATE TABLE dim_users (
  user_id BIGINT,
  name STRING,
  city STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = 'iceberg_catalog',
  'catalog-type' = 'hadoop',
  'warehouse' = '/path/to/warehouse',
  'catalog-database' = 'db',
  'catalog-table' = 'users',

  'lookup.cache' = 'FULL',
  'lookup.full-cache.periodic-reload.interval' = '10 min'
);

Full cache with periodic refresh

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users`
/*+ OPTIONS(
  'lookup.cache' = 'FULL',
  'lookup.full-cache.periodic-reload.interval' = '10 min'
) */
FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

Behavior by cache mode:

• lookup.cache = PARTIAL

  • Uses Flink DefaultLookupCache.
  • Lookup results are cached per key.

• lookup.cache = FULL

  • Uses Iceberg full-cache lookup function.
  • Full table data is loaded lazily on the first lookup.
  • If lookup.full-cache.periodic-reload.interval is configured, the cache is reloaded periodically.

[pvary]

@mxm, @swapna267: What do you think? Would this be a feature which is needed for the source?

[Guosmilesmile]

In our use case, dimension tables are synced from MySQL to Iceberg. Currently, we do lookup joins directly against MySQL. But this creates too many concurrent connections to MySQL, putting a lot of pressure on it.

Since the same data already exists in Iceberg, we could do lookup joins directly against Iceberg instead.In our case, the dimension tables are small, so a full-cache (or partial cache) based lookup join should work well enough.

For larger dimension tables where full caching isn't feasible, we may need indexing features that are being discussed for Iceberg V4. This can be explored later.

[swapna267]

Thanks @Guosmilesmile .

Yeah , i see some scenarios where this feature can be useful on source in our use cases when CDC is enabled on MySQL tables.

I am not fully aware of how Lookup source interacts with sources in general. I have concerns about the table size.
In brief, as i understand full-cache is loading all data into memory on each TM right ? Can you include how the Flink Dag looks in this case.

Looking to understand,

1. What is the behavior of lookup joins for stores like Mysql ? Is the cache built incrementally, by doing a key lookup ?
2. Lookup keys are converted into Iceberg filter expressions. => Can this be predetermined ? If number of keys is too large, can there be issues with filter expressions ?

[Guosmilesmile]

@swapna267

1. DAG：In lookup mode it is not materialized as a standalone source operator — Flink Planner fuses it into the LookupJoin operator as a LookupFunction.
2. We don't build IN-lists. Because LookupFunction.lookup is single-row, each expression is a single conjunction of equalities. There is no IN (v1..vN) or deep OR tree being constructed, so the "filter expression too large" concern doesn't apply to the current code path