Guosmilesmile opened a new pull request, #17280:
URL: https://github.com/apache/iceberg/pull/17280
## 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:
```sql
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:
```sql
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
```sql
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
```sql
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
```sql
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
```sql
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
```sql
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
