Skip to main content

Lookup Joins

Lookup Joins are a type of join in streaming queries. It is used to enrich a table with data that is queried from Paimon. The join requires one table to have a processing time attribute and the other table to be backed by a lookup source connector.

Paimon supports lookup joins on tables with primary keys and append tables in Flink. The following example illustrates this feature.

Choose a Lookup Strategy​

SituationApproachConstraint
Dimension data fits in each lookup taskNormal lookupEach task maintains a local copy by default.
Dimension updates arrive after the main streamRetry lookupA miss can delay processing; retries are bounded by the configured attempts.
Retries block unrelated recordsAsync retryUnordered output applies to append streams; CDC inputs have additional ordering constraints.
A full local copy is too largeShuffle lookupFlink 2.0+, fixed buckets, and join keys containing all bucket keys.
Only the newest partition is neededDynamic partitionConfigure partition selection and refresh interval.
A shared query service is runningQuery ServiceRequires a fixed-bucket primary-key table.

Normal lookup tasks each hold a full local table copy; shuffle lookup routes records by bucket to tasks holding subsets.

Prepare​

This example uses Flink's built-in DataGen connector and a small Paimon dimension table. Complete Installation first. Run the SQL in a new session, or use different table names if they already exist.

Create and Populate the Dimension Table​

CREATE CATALOG my_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'file:/tmp/paimon-lookup'
);
USE CATALOG my_catalog;

SET 'execution.runtime-mode' = 'batch';
SET 'table.dml-sync' = 'true';
SET 'parallelism.default' = '1';

CREATE TABLE customers (
    id INT PRIMARY KEY NOT ENFORCED,
    name STRING,
    country STRING,
    zip STRING
) WITH ('bucket' = '4');

INSERT INTO customers VALUES
    (1, 'Alice', 'US', '10001'),
    (2, 'Bob', 'DE', '10115'),
    (3, 'Chen', 'CN', '200000');

Wait for the insert to finish before starting the lookup query. In a production pipeline, a separate streaming writer can keep the dimension table up to date. Use a warehouse on shared storage for a distributed cluster.

Create the Incoming Stream​

SET 'execution.runtime-mode' = 'streaming';

CREATE TEMPORARY TABLE orders (
    order_id INT,
    total INT,
    customer_id INT,
    proc_time AS PROCTIME()
) WITH (
    'connector' = 'datagen',
    'rows-per-second' = '1',
    'fields.order_id.kind' = 'sequence',
    'fields.order_id.start' = '1',
    'fields.order_id.end' = '3',
    'fields.customer_id.kind' = 'sequence',
    'fields.customer_id.start' = '1',
    'fields.customer_id.end' = '3',
    'fields.total.min' = '10',
    'fields.total.max' = '100'
);

The source emits three orders and finishes. The proc_time column provides the processing-time attribute required by the lookup join. Replace this temporary table with your Kafka or other streaming source when adapting the example.

Normal Lookup​

Use customers as the lookup side of the join. Each order is enriched using the dimension data available to the lookup operator when the record is processed. This is a processing-time lookup, not a historical read of the dimension table at the order's event time.

-- enrich each order with customer information
SELECT o.order_id, o.total, c.country, c.zip
FROM orders AS o
JOIN customers
FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;

The example returns one enriched row for each of the three orders. An inner JOIN drops an order when no customer matches; use LEFT JOIN to retain it with null dimension fields.

Retry Lookup​

Use a delayed retry when an order arrives before its customer is visible to the lookup operator. Flink 1.16+ supports a delayed retry strategy. The example retries a lookup miss with a one-second delay and at most 600 attempts. Choose the retry budget for the expected dimension-data delay.

-- enrich each order with customer information
SELECT /*+ LOOKUP(
    'table' = 'c',
    'retry-predicate' = 'lookup_miss',
    'retry-strategy' = 'fixed_delay',
    'fixed-delay' = '1s',
    'max-attempts' = '600'
) */
o.order_id, o.total, c.country, c.zip
FROM orders AS o
JOIN customers
FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;

Async Retry Lookup​

Synchronous retries delay subsequent records in the same lookup task. Enable asynchronous lookup and allow unordered output to let other completed lookups proceed while a miss is retried.

-- enrich each order with customer information
SELECT /*+ LOOKUP(
    'table' = 'c',
    'retry-predicate' = 'lookup_miss',
    'output-mode' = 'allow_unordered',
    'retry-strategy' = 'fixed_delay',
    'fixed-delay' = '1s',
    'max-attempts' = '600'
) */
o.order_id, o.total, c.country, c.zip
FROM orders AS o
JOIN customers /*+ OPTIONS('lookup.async'='true', 'lookup.async-thread-number'='16') */
FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;
info

Flink only allows unordered lookup output for append streams. For a CDC input, ordering constraints can still delay subsequent records. An audit-log table exposes changes as append records, but the query must then interpret the row kinds explicitly; it is not a transparent replacement for a CDC stream.

Large Scale Lookup (Fixed Bucket)​

By default, each Flink subtask would store a whole copy of the lookup table. If the amount of data in customers (lookup table) is too large for a single subtask, you can enable the shuffle lookup optimization as follows (For Flink 2.0+ and fixed-bucket Paimon table). This optimization enables sending data of the same bucket to designated subtask(s), so each Flink subtask would only need to store a part of the whole data.

-- enrich each order with customer information
SELECT /*+ LOOKUP('table'='c', 'shuffle'='true') */
o.order_id, o.total, c.country, c.zip
FROM orders AS o
JOIN customers
FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;

This requires Flink 2.0+, a fixed-bucket dimension table, and join keys containing all bucket keys. The customers table above uses four fixed buckets. Choose the bucket count for your workload; see Data Distribution.

Dynamic Partition​

When each partition contains a complete dimension snapshot, use max_pt() to select the partition with the largest partition value. The lookup operator periodically refreshes the selection. The following example uses a separate table from the unpartitioned customers above.

Create Paimon Partitioned Table

CREATE TABLE partitioned_customers (
  id INT,
  name STRING,
  country STRING,
  zip STRING,
  dt STRING,
  PRIMARY KEY (id, dt) NOT ENFORCED
) PARTITIONED BY (dt);

Lookup Join

SELECT o.order_id, o.total, c.country, c.zip
FROM orders AS o
JOIN partitioned_customers /*+ OPTIONS(
    'scan.partitions' = 'max_pt()',
    'lookup.dynamic-partition.refresh-interval' = '1 h'
) */
FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;

The Lookup node will automatically refresh the latest partition and query the data of the latest partition.

The option scan.partitions can also specify fixed partitions in the form of key1=value1,key2=value2. Multiple partitions should be separated by semicolon (;). When specifying fixed partitions, this option can also be used in batch joins.

The option scan.partitions can also specify max_pt() for parent partition in the form of key1=max_pt(),key2=max_pt(). All subpartitions for the latest parent partition will be loaded. For example, if partition keys is 'year', 'day', 'hh', you can specify year=max_pt(), it will find the latest partition for year and load all its subpartitions for lookup. Only supports partitions to be specified hierarchically. For example, if setting year=max_pt(),hh=max_pt(), hh=max_pt() makes no sense.

Query Service​

A query service runs as a separate Flink streaming job. It requires a primary-key table with fixed buckets, such as the customers table in this example. When a query service is available, lookup joins prioritize it. Allocate enough cluster resources for both the service and the lookup job.

SET 'execution.runtime-mode' = 'streaming';
SET 'table.dml-sync' = 'false';
CALL sys.query_service('default.customers', 1);

Verify the Result​

If a lookup returns no match, first query the dimension table in batch mode and check the join keys. If the data exists, check the lookup's selected partitions and refresh settings. Retrying cannot fix an incorrect key or a partition excluded by the query.

See Troubleshooting for the full diagnostic sequence.