Guosmilesmile commented on code in PR #18144:
URL: https://github.com/apache/iceberg/pull/18144#discussion_r4089297349
##########
docs/docs/flink-configuration.md:
##########
@@ -226,3 +226,22 @@ builder.
| table.exec.iceberg.fetch-batch-record-count | 2048
| Target number of records per fetch batch in the Iceberg source
reader.
|
| table.exec.iceberg.worker-pool-size | max(2, available cpu)
| Size of the worker pool used to plan or scan manifests. Defaults to the
shared Iceberg worker pool size, which is controlled by the
`iceberg.worker.num-threads` system property.
|
| table.exec.iceberg.use-v2-sink | false
| Use the SinkV2 based `IcebergSink` implementation, see [Sink V2
based implementation](flink-writes.md#sink-v2-based-implementation).
|
+
+### Lookup options
+
+Flink lookup joins cache the whole projected dimension table in memory, and
the cache is kept in
+memory for the lifetime of the job, so the dimension table should be populated
before the join
+starts. See [Lookup Join](flink-queries.md#lookup-join) for details. These
options are set as table
+options in the DDL, or per query with the `OPTIONS` hint:
+
+```sql
+SELECT o.order_id, u.name
Review Comment:
Make sense.
##########
docs/docs/flink-queries.md:
##########
@@ -92,6 +92,33 @@ SELECT * FROM table /*+ OPTIONS('tag'='t1') */;
SELECT * FROM table /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s',
'start-tag'='t1', 'end-tag'='t2') */;
```
+### Lookup Join
+
+Iceberg supports Flink lookup join, which enriches a stream with data from an
Iceberg dimension table:
+
+```sql
+-- The OPTIONS hint used in this section requires dynamic table options, which
are disabled by default.
+SET table.dynamic-table-options.enabled=true;
+
+SELECT o.order_id, o.user_id, u.name, u.city
+FROM orders AS o
+LEFT JOIN iceberg_catalog.db.user_dim
+ FOR SYSTEM_TIME AS OF o.proc_time AS u
+ ON o.user_id = u.user_id;
+```
+
+Iceberg implements lookup join with a full cache: the whole projected
dimension table is loaded into the cache, and every lookup is served from it
without falling back to the table. The full cache is held in memory on the
TaskManager heap, so lookup join targets dimension tables that fit comfortably
there.
+
+`lookup.full-cache.eager-load` decides whether the job blocks at startup or on
the first lookup. With the default `true`, the lookup function loads the cache
when it is opened, so the job blocks during deployment, before it processes any
data, and fails at startup if the dimension table cannot be read; with `false`,
the load is deferred to the first lookup, so the data flow blocks only when the
first probe row arrives, and the subtasks of the join can end up on different
snapshots of the dimension table.
+
+There is no background refresh: each subtask keeps the snapshot it loaded, so
subtasks can serve different snapshots of the dimension table. Populate the
dimension table before the join starts.
+
+The lookup options are:
+
+| Option | Default | Description
|
+| ------------------------------ | ------- |
--------------------------------------------------------------------------------------------------
|
+| `lookup.full-cache.eager-load` | `true` | Whether to load the full cache
when the lookup function is opened, instead of on the first lookup. |
+
Review Comment:
Ok, do it now
--
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]
