Flink: Support Lookup Join using full in-memory lookup cache

[Guosmilesmile]

This PR adds lookup join support for the Iceberg Flink table source, using a full in-memory lookup cache.

Part of #18142

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

• Pushed-down filter support

  • Existing source filters are reused when loading the cache.
  • Join conditions such as:

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

    are applied together with the lookup key condition.

• Full in-memory lookup cache

  • The whole projected dimension table is loaded into a cache on the TaskManager heap, and every lookup is served from it, never falling back to the table.
  • A dimension key may match multiple rows; all matching rows are returned and joined.
  • Only lookup.cache=FULL is accepted; NONE and PARTIAL are rejected, because an Iceberg table cannot be point-looked-up effectively.

• Configurable load strategy

  • lookup.full-cache.eager-load selects when the cache is loaded:
    • false (default): on the first lookup. Simple, but that probe row is blocked for the duration of the load.
    • true: in open(), so no probe row is blocked, and a load that fails fails the job at startup instead of mid-stream.

• Metrics

  • Reported under the icebergLookupCache group: cacheHit and cacheMiss counters, and snapshotId and cachedRows gauges.

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;

Load the cache when the lookup function opens

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users`
/*+ OPTIONS('lookup.full-cache.eager-load' = 'true') */
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.full-cache.eager-load' = 'true'
);

Options

Option	Required	Default	Description
lookup.cache	No	-	Only FULL is accepted. NONE and PARTIAL are rejected, because an Iceberg table cannot be point-looked-up.
lookup.full-cache.eager-load	No	true	Whether to load the cache in open(), instead of on the first lookup.

Both can be set per join with an OPTIONS hint, or in the table DDL WITH clause.

Notes

• Memory only. The cache lives on the TaskManager heap, so this targets dimension tables that fit comfortably there. A disk-backed backend is left for a follow-up — keeping it out of this PR avoids adding native code to the runtime jar and any conflict with the RocksDB copy Flink already ships for its RocksDB state backend.
• Loaded once, no refresh. The cache is loaded when the lookup function starts serving and is then kept for the lifetime of the job; there is no background reload. The dimension table should be populated before the join starts.

[Guosmilesmile]

For snapshot pinning, I was thinking we could resolve the snapshot at job submission time and pass the snapshot ID to every TaskManager, so that all caches use the same snapshot. But there is a problem with this approach.

The problem is failover. Flink doesn't re-run the planner when a task restarts, so the pinned snapshot ID would remain unchanged. If that snapshot has expired by then, the cache load would fail with Cannot find snapshot with ID ... after the restart, and retries would keep failing until the job is resubmitted.

Periodic refresh doesn't help here, because the failure happens during the initial cache load. We also don't have a way to persist and update the snapshot ID from the connector side, since a lookup join doesn't have checkpointed state.

So while pinning the snapshot would ensure consistency across subtasks, it introduces a failover problem.

Additionally, dimension tables do not change frequently.

For this PR, I therefore use the latest snapshot at the moment and just added logging and a metric to report the snapshot ID being used.If you have a better idea, we'd be happy to take a look.

[swapna267]

why not use ReadOptions and pass when creating the IcebergFullCachingLookupFunction, which will keep it consistent across task restarts or job restarts.

At table creation time,

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.full-cache.eager-load' = 'true',
  'snapshot-id' = '121334'
);

OR per query

SELECT o.order_id, o.user_id, u.name, u.city
FROM orders AS o
LEFT JOIN iceberg_catalog.`db`.`users`
/*+ OPTIONS(
     'lookup.full-cache.eager-load' = 'true',
     'snapshot-id' = '121334'
   ) */
FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

There could be other strategies to simplify , instead of user specifying exact snapshot id also like latest or specific tag or as-of-timestamp

[Guosmilesmile]

@swapna267 Showing a specific snapshot ID is one approach, but it doesn't solve the problem of that snapshot expiring. On top of that, most users won't bother specifying a particular snapshot anyway.

[swapna267]

Yes it doesn't need to be particular snapshot Id. Instead it could be Latest_Snapshot or a particular Branch/Tag. I was referring to resolving it on Job Submission time instead of on TaskManagers. But i just realized you already mentioned that option.

Instead of being inconsistent with lookup result across TaskManagers, prefer to have all TM's load from same Snapshot Id. And fail loudly, incase of an expired snapshot or a Tag.

[Guosmilesmile]

@swapna267 I agree that resolving the snapshot ID at Job Submission time works well when the cache is not refreshed, with an explicit failure if the snapshot has expired.

However, with periodic refresh, a long-running job may encounter a failover after the original snapshot has expired. Resolving the snapshot ID at Job Submission would require manual intervention to restart the job, which doesn't seem ideal for the periodic refresh use case.

I also noticed that other connectors supporting lookup joins generally establish their connections independently on each TaskManager.

So I don't think resolving the snapshot ID at the Job Submission level is a good fit for this use case.

This is also why I'm still leaning toward using the latest snapshot rather than pinning to a specific snapshot.

[Guosmilesmile]

@pvary @mxm @talatuyarer If you get a chance, please take a look and let me know what you think. I'd really appreciate it.

[swapna267]

Thanks @Guosmilesmile . Yes I agree, this wouldn't make sense for Periodic Refresh .
With periodic refresh, falling back to Latest makes sense.

As this PR's scope was limited to one time load with no periodic refresh, i was recommending that.