Enrich a Stream with Reference Data in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® can enrich each row of a table with the current version of a reference table, such as a table of customers, products, or rates that changes far less often than the table you’re enriching. For example, an order carries a customer ID, but the report you want shows the customer’s tier and country.

This guide shows how to enrich a table of orders with the current customer details by using a LATERAL SNAPSHOT join. You create a small reference table, join an orders table against it, and then update the reference data to see the new values flow through.

Prerequisites

Access to Confluent Cloud.
A Kafka cluster to hold the tables you create in this guide.
A Flink compute pool in the same region as the cluster.

Why not a regular join?

A regular join keeps every row from both sides in state forever and revises earlier results whenever either side changes. For enrichment, that means state that grows without limit and a stream of corrections instead of one clean row per order.

A LATERAL SNAPSHOT join keeps the reference table in state, plus any orders that arrive while it first loads that table. It joins each order a single time, against the customer details that are current at that moment, and emits one row per order that it never revises.

Create the tables

Create a customers reference table keyed by customer ID and an orders table.

CREATE TABLE customers (
    customer_id INT,
    tier STRING,
    country STRING,
    PRIMARY KEY (customer_id) NOT ENFORCED
);

CREATE TABLE orders (
    order_id INT,
    customer_id INT,
    amount DECIMAL(10, 2)
);

Because customers has a primary key, each new row for a customer replaces the previous one. orders has no primary key, so it’s an append-only table, which is what the join needs on the side it enriches.

Add two customers:

INSERT INTO customers VALUES
    (1, 'gold', 'US'),
    (2, 'silver', 'UK');

Enrich the orders

Start the join. It needs two arguments:

The input argument names the reference table to look values up in, here customers.
The on_time argument names the column that records when Flink wrote each customer row, here $rowtime. The join uses it to apply changes to customers in the right order. Every table has this $rowtime system column, so you don’t need to add it yourself.

SELECT o.order_id, o.customer_id, o.amount, c.tier, c.country
FROM orders AS o
LEFT JOIN LATERAL SNAPSHOT(
    input => TABLE customers,
    on_time => DESCRIPTOR($rowtime),
    load_completed_idle_timeout => INTERVAL '1' MINUTE
) AS c
ON o.customer_id = c.customer_id;

The join first reads the whole customers table, then starts joining. It treats the table as fully read when the table’s watermark reaches the time you started the statement. Both of your rows are older than that, so the third argument, load_completed_idle_timeout, tells the join to stop waiting after the table’s watermark has stopped moving for one minute. To learn why you need it, see Why you need the idle timeout argument.

To skip the wait, insert another customer row instead. Its timestamp is later than the time you started the statement, so the watermark can move past that time and the join finishes loading. The watermark can trail the new row, so loading can take a little longer than the insert.

While the query runs, add three orders in a separate statement:

INSERT INTO orders VALUES
    (101, 1, 20.00),
    (102, 2, 35.50),
    (103, 3, 12.00);

About a minute after you started the query, the join finishes loading. It enriches the orders that arrived meanwhile and emits them together:

order_id  customer_id  amount  tier    country
========  ===========  ======  ======  =======
     101            1   20.00  gold    US
     102            2   35.50  silver  UK
     103            3   12.00  NULL    NULL

Order 103 is for a customer who isn’t in the reference table. The LEFT JOIN keeps the order and fills the customer columns with NULL. A plain JOIN would drop the order.

Update the reference data

Suppose customer 1 moves up to the platinum tier. Insert the new version of their row:

INSERT INTO customers VALUES (1, 'platinum', 'US');

The join doesn’t apply this change immediately. It first waits until it’s sure no earlier changes to customers are still on their way, which takes about 10 seconds here. Wait that long, then place another order for that customer:

INSERT INTO orders VALUES (104, 1, 50.00);

The new order picks up the new tier:

order_id  customer_id  amount  tier      country
========  ===========  ======  ========  =======
     104            1   50.00  platinum  US

Order 101 still shows gold, because the join never revises an order it has already emitted.

Behind the wait, the join holds each change until the customers watermark moves past it, so it applies changes in time order. When a table gets no new data, its watermark moves only after Flink detects that the table is idle. That takes about 10 seconds on a new statement and grows to as much as five minutes on one that has run for a while. The join keeps joining orders the whole time, against the version it already holds. This delay doesn’t depend on the load_completed_idle_timeout argument, which only decides when loading ends. For more on idleness detection, see Validate or disable idleness handling.

Timing decides which version an order sees. An order that the join processes before an update reaches it gets the earlier version, so the same data can give slightly different results from one run to the next.

Why you need the idle timeout argument

Before the join enriches any orders, it reads all the rows the reference table held when you started the statement. That way, the join never matches an order against a partly loaded table. To tell when it has read them all, the join watches the table’s watermark, a timestamp that moves forward as new rows arrive. After the watermark reaches the time you submitted the statement, the join treats the table as fully loaded.

Reference tables are often static or slowly updating. They receive their rows up front, then updates now and then. When no new rows arrive, the watermark stops at the time of the last row and never reaches the time you submitted the statement. The join keeps waiting to finish loading, holds every order, and returns nothing, while the statement still shows as running.

The load_completed_idle_timeout argument solves this. When the reference table’s watermark has stopped moving for the length of time you set, which usually means the table has stopped receiving new data, the join treats the table as fully loaded and starts joining. It isn’t set by default, so set it whenever your reference table might be static or slowly updating.

Choose the timeout value

Pick a value longer than any pause you expect while the join first reads the reference table. About a minute suits most reference tables.

Too short: If the first read pauses for longer than the timeout, the join starts early and joins some orders against a partly loaded table.
Too long: You wait longer for the first results.

A large reference table doesn’t need a longer timeout on its own. While the join is still reading rows, the watermark usually keeps moving, so the wait only starts after new data stops arriving.

If you’d rather name the exact point that completes the load, set the load_completed_time argument to a timestamp instead. The load finishes when the reference table’s watermark reaches that time. This suits reference tables that keep receiving updates, where the watermark reliably moves past the time you choose.

Troubleshoot missing results

If the join returns no results, check these in order.

Is the idle timeout argument set?: A static or slowly updating reference table without the load_completed_idle_timeout argument is the most common cause. Stop the statement and run it again with the argument set.
Is the join still loading?: Open the statement in the Query Profiler. If the join’s state keeps growing while nothing comes out and the reference table’s watermark has stopped moving, the join is still in its load phase.
Do the join keys match?: An inner JOIN drops orders that have no matching reference row. Switch to LEFT JOIN to see those orders with NULL values.