<a id="flink-enrich-stream-reference-data"></a>

# 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](../reference/queries/joins.md#flink-sql-lateral-snapshot-joins). 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](../reference/queries/joins.md#flink-sql-regular-joins) 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.

```sql
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:

```sql
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](../reference/statements/create-table.md#flink-sql-system-columns-rowtime) system column, so you
  don’t need to add it yourself.

```mysql
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](#flink-enrich-reference-data-timeout).

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:

```sql
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:

```none
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:

```sql
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:

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

The new order picks up the new tier:

```none
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](../operate-and-deploy/best-practices.md#flink-sql-best-practices-for-statements-idleness).

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.

<a id="flink-enrich-reference-data-timeout"></a>

## 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](../concepts/timely-stream-processing.md#flink-sql-event-time-and-watermarks), 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](profile-query.md#flink-sql-profile-query). 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.

## Related content

- [LATERAL SNAPSHOT joins](../reference/queries/joins.md#flink-sql-lateral-snapshot-joins)
- [Event-time temporal joins](../reference/queries/joins.md#flink-sql-event-time-temporal-joins)
- [Lookup joins](../reference/queries/joins.md#flink-sql-lookup-joins)
- [Time and Watermarks](../concepts/timely-stream-processing.md#flink-sql-timely-stream-processing)
- [Profile a Query](profile-query.md#flink-sql-profile-query)
- [Resolve Statement Issues](resolve-common-query-problems.md#flink-sql-statement-problems)