Guosmilesmile opened a new pull request, #18144:
URL: https://github.com/apache/iceberg/pull/18144
This PR adds **lookup join support for the Iceberg Flink table source**,
using a **full in-memory lookup cache**.
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
```
- **Pushed-down filter support**
- Existing source filters are reused when loading the cache.
- Join conditions such as:
```sql
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.
- **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
```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;
```
### Load the cache when the lookup function opens
```sql
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
```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.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 | `false` | 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.
--
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]
