Join Queries in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables join data streams over Flink SQL dynamic tables.

Description

Flink supports complex and flexible join operations over dynamic tables. There are a number of different types of joins to account for the wide variety of semantics that queries can require.

By default, the order of joins is not optimized. Tables are joined in the order in which they are specified in the FROM clause.

You can tweak the performance of your join queries, by listing the tables with the lowest update frequency first and the tables with the highest update frequency last. Make sure to specify tables in an order that doesn’t yield a cross join (Cartesian product), which isn’t supported and would cause a query to fail.

Many of the tables used in the join examples on this page, like orders, correspond to real, ready-to-use tables in the example data streams catalog. You can run these join queries yourself against actual data.

Regular joins

Regular joins are the most generic type of join in which any new row, or changes to either side of the join, are visible and affect the whole join result.

For example, if there is a new record on the left side, it is joined with all of the previous and future records on the right side when the join fields are equal.

SELECT * FROM orders
INNER JOIN Product
ON orders.productId = Product.id

For streaming queries, the grammar of regular joins is the most flexible and enables any kind of updates (insert, update, delete) on the input table. But this operation has important implications: it requires keeping both sides of the join input in state forever, so the required state for computing the query result might grow indefinitely, depending on the number of distinct input rows of all input tables and intermediate join results.

Table types. A regular join accepts an append-only or updating table on both sides, and it produces an updating table, because a change to either side can change or remove an already-emitted result row.

INNER Equi-JOIN

Returns a simple Cartesian product restricted by the join condition.

Only equi-joins are supported — that is, joins that have at least one conjunctive condition with an equality predicate. Arbitrary cross or theta joins aren’t supported.

SELECT *
FROM orders
INNER JOIN Product
ON orders.product_id = Product.id

OUTER Equi-JOIN

Returns all rows in the qualified Cartesian product (that is, all combined rows that pass its join condition), plus one copy of each row in an outer table for which the join condition did not match with any row of the other table.

Flink supports LEFT, RIGHT, and FULL outer joins.

Only equi-joins are supported — that is, joins that have at least one conjunctive condition with an equality predicate. Arbitrary cross or theta joins aren’t supported.

SELECT *
FROM orders
LEFT JOIN Product
ON orders.product_id = Product.id

SELECT *
FROM orders
RIGHT JOIN Product
ON orders.product_id = Product.id

SELECT *
FROM orders
FULL OUTER JOIN Product
ON orders.product_id = Product.id

Multi-way join optimization

When joining three or more tables on a common join key, Flink can use a multi-way join operator instead of chaining multiple binary joins. This optimization reduces state by eliminating intermediate join results.

How it works

With traditional binary joins, joining tables A → B → C requires storing:

• Records from A and B

• Intermediate results of A JOIN B

• Records from C

The multi-way join operator stores only the input records from each table—no intermediate results. For complex join chains involving many tables, this can significantly reduce state size.

Enabling multi-way joins

To use the multi-way join operator, add the MULTI_JOIN hint to your query. The optimizer applies the optimization when all of the following conditions are met:

1. Three or more tables are being joined

2. Common partitioning key: All joins must share a common join key that allows records to be co-located

   -- Supported: all joins share the same key
   SELECT /*+ MULTI_JOIN(o, c, a) */ * FROM orders o
   JOIN customers c ON o.customer_id = c.id
   JOIN addresses a ON c.id = a.customer_id;
   
   -- NOT supported: different keys prevent co-location
   SELECT /*+ MULTI_JOIN(o, p, c) */ * FROM orders o
   JOIN products p ON o.product_id = p.id
   JOIN categories c ON p.category_id = c.id;  -- Different key chain
   

3. Supported join types: Only INNER JOIN and LEFT OUTER JOIN are currently supported

When multi-way joins might not help

The multi-way join operator is not a silver bullet. Consider that:

• High-cardinality inputs still result in large state, even without intermediate results

• High-cardinality common key results in considerable reprocessing and slow performance

• Infrequently matching joins can perform more computation than optimally ordered binary joins

• Queries that don’t meet the criteria silently fall back to binary joins—always verify with EXPLAIN

For queries with regular joins that grow state indefinitely, also consider:

• Setting an appropriate state TTL to limit state growth

• Using interval joins when time bounds are acceptable

• Using lookup joins to enrich a stream with data from an external system

Interval joins

An interval join returns a simple Cartesian product restricted by the join condition and a time constraint.

An interval join requires at least one equi-join predicate and a join condition that bounds the time on both sides. Two appropriate range predicates can define such a condition (<, <=, >=, >), a BETWEEN predicate, or a single equality predicate that compares time attributes of both input tables.

For example, the following query joins all orders with their corresponding shipments if the order was shipped four hours after the order was received.

SELECT *
FROM orders o, Shipments s
WHERE o.id = s.order_id
AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time

The following predicates are examples of valid interval join conditions:

• ltime = rtime

• ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE

• ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND

For streaming queries, compared to the regular join, interval join only supports append-only tables with time attributes. Because time attributes increase quasi-monotonically, Flink can remove old values from its state without affecting the correctness of the result.

Table types. An interval join requires an append-only table on both sides, and it produces an append-only table. Unlike a regular join, an interval join can’t consume an updating table on either side.

Temporal joins

A temporal join joins one table with another table that is updated over time. This join is made possible by linking both tables using a time attribute, which allows the join to consider the historical changes in the table. When viewing the table at a specific point in time, the join becomes a time-versioned join.

In a temporal join, the join condition is based on a time attribute, and the join result includes all rows that satisfy the temporal relationship. A common use case for temporal joins is analyzing financial data, which often includes information that changes over time, such as stock prices, interest rates, and exchange rates.

Table types. The left (probe) side of a temporal join can be an append-only or updating table. The right (versioned) side must be an updating table with a primary key, so that Flink can look up the value of a key as of a point in time. The join produces the same table type as the left side.

Event-time temporal joins

Event-time temporal joins are used to join two or more tables based on a common event time. Event time is the time at which an event occurred, which is typically embedded in the data itself. With Confluent Cloud for Apache Flink, you can use the $rowtime system column to get the timestamp from an Apache Kafka® record. This is also used for the default watermark strategy in Confluent Cloud.

Temporal joins take an arbitrary table (left input/probe side) and correlate each row to the corresponding row’s relevant version in the versioned table (right input/build side). Flink uses the SQL syntax of FOR SYSTEM_TIME AS OF to perform this operation from the SQL:2011 standard.

The syntax of a temporal join is as follows:

SELECT [column_list]
FROM table1 [AS <alias1>]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ rowtime } [AS <alias2>]
ON table1.column-name1 = table2.column-name1

With an event-time attribute, you can retrieve the value of a key as it was at some point in the past. This enables joining the two tables at a common point in time. The versioned table stores all versions, identified by time, since the last watermark.

For example, suppose you have a table of orders, each with prices in different currencies. To properly normalize this table to a single currency, such as USD, each order needs to be joined with the proper currency conversion rate from the point in time when the order was placed.

CREATE TABLE orders (
    order_id    STRING,
    price       DECIMAL(32,2),
    currency    STRING
);

CREATE TABLE currency_rates (
    currency STRING,
    conversion_rate DECIMAL(32, 2),
    PRIMARY KEY(currency) NOT ENFORCED
);

SELECT
     orders.order_id,
     orders.price,
     orders.currency,
     currency_rates.conversion_rate
FROM orders
LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.`$rowtime`
ON orders.currency = currency_rates.currency;

The event-time temporal join requires the primary key contained in the equivalence condition of the temporal join condition. In this example, the primary key currency_rates.currency in the currency_rates table is constrained in the condition orders.currency = currency_rates.currency expression.

With temporal joins, there’s some indeterminate amount of latency involved. In the example with orders and currency_rates, when enriching a particular order, an event-time temporal join waits until the watermark on the currency-rate stream reaches the timestamp of that order, because only then is it reasonable to be confident that the result of the join is being produced with complete knowledge of the relevant exchange-rate data.

Event-time temporal joins can’t guarantee perfectly correct results. Despite having waited for the watermark, the most relevant exchange-rate record could still be late, in which case the join uses an earlier version of the exchange rate.

If the enrichment stream has infrequent updates, this causes problems, because of the behavior of watermarking on idle streams. The operator, like any operator with two input streams, normally waits for the watermarks on both incoming streams to reach the desired timestamp before taking action.

If your reference table updates rarely and you want its current version rather than the version valid at each row’s event time, use a LATERAL SNAPSHOT join instead.

Note

Temporal joins on regular tables, using FOR SYSTEM_TIME AS OF, are supported only in streaming mode. Flink rejects this join type in batch mode, which includes snapshot queries. To enrich data with a versioned table in batch mode, use a lookup join instead, or run the query in streaming mode. For more information, see Batch and Stream Processing.

LATERAL SNAPSHOT joins

A LATERAL SNAPSHOT join joins each row of a table, the outer table, with the current version of a reference table. It works like a lookup against a table that stays up to date. For example, it can tag each order with the customer’s current tier or with the current exchange rate.

To write one, use the built-in LATERAL SNAPSHOT process table function. A LATERAL SNAPSHOT join covers some of the same use cases as a temporal join, but it has different syntax, input requirements, and semantics.

Important

If your reference table is static or updates slowly, set the load_completed_idle_timeout argument when you write the join. Without it, the join can wait and accumulate state indefinitely without returning results. For details, see Static or slowly updating reference tables.

The following example adds the current exchange rate to a table of payments.

-- outer table: payments
-- payment_id | currency | amount
-- -----------+----------+-------
--          1 | EUR      |     10
--          2 | EUR      |     10
--          3 | USD      |     20

-- reference table: rates, primary key currency
-- currency | rate | $rowtime
-- ---------+------+---------
-- EUR      | 1.1  |    10:00
-- USD      | 1.0  |    10:00
-- EUR      | 1.2  |    10:30

SELECT p.payment_id, p.currency, p.amount, r.rate
FROM payments AS p
JOIN LATERAL SNAPSHOT(
    input => TABLE rates,
    on_time => DESCRIPTOR($rowtime),
    load_completed_idle_timeout => INTERVAL '1' MINUTE
) AS r
ON p.currency = r.currency;

Because rates has a primary key on currency, the 10:30 row replaces the earlier EUR rate. Suppose you submit the statement at 10:15. The 10:30 row hasn’t arrived yet, so rates has no rows newer than 10:00. The join finishes loading after the one-minute idle timeout. Flink processes payment 1 before the 10:30 rate update reaches the join and payment 2 after it:

payment_id  currency  amount  rate
==========  ========  ======  ====
         1  EUR           10  1.1
         2  EUR           10  1.2
         3  USD           20  1.0

Each payment gets the rate that is current when Flink processes it. If Flink had processed payment 2 before the 10:30 update reached the join, the join would have used the 1.1 rate. Because the result depends on timing, running the same query again can give slightly different results. The join never revises a row it has already emitted, so its output only ever adds rows.

When to use a LATERAL SNAPSHOT join

Use a LATERAL SNAPSHOT join when you want each row enriched with the current reference data as fast as possible. It favors low latency over exact alignment in time. The join matches each row with the version it holds when it processes the row. That version can be older or newer than the one that was valid at the row’s event time. It fits the cases where other joins struggle.

• The reference table updates rarely. An event-time temporal join waits for new data on both inputs, so a reference table that stops changing stalls it. A LATERAL SNAPSHOT join keeps producing results.

• You need low latency. After the join loads the reference table, it joins each row as soon as it arrives.

• The reference table has no primary key. Neither input needs a primary key. Without one, the join doesn’t replace reference rows, so each row of the outer table joins with every reference row that has its join key.

When results must be exactly the same every time you run the query, for example for audits or replays, use an event-time temporal join instead. It joins each row with the version of the reference data that was valid at the row’s event time. To enrich with data held outside Kafka, use a lookup join.

Load and join phases

The join never matches a row against a partly loaded reference table. It runs in two phases.

1. Load phase. The join reads the reference table into state and holds any rows of the outer table that arrive meanwhile. It emits nothing yet.

2. Join phase. The join matches the held rows against the loaded reference table and emits them. From then on, it matches each new row as it arrives. The join keeps applying changes to the reference table, so later rows see the current version.

The join tracks how far it has read the reference table with the table’s watermark, a timestamp that moves forward as new rows arrive. By default, the load phase ends when the watermark reaches the time you submitted the statement.

Static or slowly updating reference tables

Reference data such as customer profiles, product catalogs, or currency rates often arrives in one batch and then changes rarely. The watermark only moves forward when new rows arrive, so it stops short of the time you submitted the statement. The join stays in the load phase, collects all rows of the outer table, and returns no results. The statement still shows as running.

To finish the load phase for these tables, set the load_completed_idle_timeout argument. It tells the join how long to wait after the reference table’s watermark stops moving, which usually means the table has stopped receiving new data. After that time passes, the join treats the reference table as fully loaded and starts joining.

Choose a value longer than any pause you expect while the join first reads the reference table. About a minute suits most reference tables. While the join is still reading rows, the watermark usually keeps moving, so a long first read doesn’t end the load phase early on its own. If the value is too short and the read pauses for longer than it, the join can start before it has loaded the whole reference table. If it’s too long, you wait longer for the first results.

Changes to a static or slowly updating table also take a while to reach the join. The join applies each change in time order, after the reference table’s watermark passes it. On a table with no new data, the watermark moves when Flink detects that the table is idle. That takes about 10 seconds on a new statement and up to five minutes on one that has run for a while. This delay doesn’t depend on the load_completed_idle_timeout argument, which only decides when the load phase ends. For how idleness detection works, see Validate or disable idleness handling.

For a step-by-step example, see Enrich a Stream with Reference Data in Confluent Cloud for Apache Flink.

Syntax

In most cases, you can use the following form as is. Replace the table names and the join condition.

SELECT column_list
FROM outer_table
LEFT JOIN LATERAL SNAPSHOT(
    input => TABLE reference_table,
    on_time => DESCRIPTOR($rowtime),
    load_completed_idle_timeout => INTERVAL '1' MINUTE
) AS alias
ON outer_table.column = alias.column

Use LEFT JOIN to keep every row of the outer table, with NULL values when there’s no match, or JOIN to keep only the rows that match. The join supports only these two join types. Include at least one equality condition in the ON clause so that the join can run in parallel. The ON clause can include other conditions too.

Argument	Type	Description
input	TABLE	Required. The reference table.
on_time	DESCRIPTOR	Required. The reference table’s time column, which sets the order in which the join applies its changes. Use DESCRIPTOR($rowtime), which works for any table because every table has the $rowtime system column. To use a different column, it must be a TIMESTAMP or TIMESTAMP_LTZ column with a precision of 3 or less and a watermark.
load_completed_idle_timeout	INTERVAL	How long to wait after the reference table’s watermark stops moving before the load phase ends. Set it for static or slowly updating reference tables. Not set by default.

Advanced arguments. Most joins don’t need these.

Argument	Type	Description
load_completed_time	TIMESTAMP_LTZ(3)	The point in the reference table’s time at which the load phase ends, instead of the time you submit the statement. It must be a constant.
state_ttl	INTERVAL	How long Flink keeps the state for a join key after the last row with that key from either table. Applies only in the join phase. Defaults to the statement’s sql.state-ttl setting.

Table types. The outer table must be an append-only table. The reference table can be append-only or updating. The join produces an append-only table that keeps the outer table’s time column, so you can still use it in windows. The time column from the reference table becomes a regular timestamp column in the output, so you can’t use it in windows.

State. Flink keeps the whole reference table in state, partitioned by the join keys in the equality conditions of the ON clause. During the load phase, it also holds the rows of the outer table that arrive. Several reference rows can share a join key. Each row of the outer table joins with all of them. Use state_ttl to drop the state for join keys that neither table has used for a while. After that, rows of the outer table with that key find no match until the reference row changes again.

Check the load phase. If the join returns no results, open the statement in the Query Profiler. A join still in the load phase shows its state growing as it holds rows of the outer table, while the reference table’s watermark has stopped moving.

Lookup joins

Lookup joins enrich a stream with data from an external system, such as a key-value store, full-text index, or vector database. Unlike a temporal join on a regular table, a lookup join is supported in both streaming and batch mode, including snapshot queries. The external system is exposed as a read-only external table, and the lookup is expressed as a lateral join against one of three built-in table-valued aggregation functions.

If the reference data is already in a Kafka topic, a LATERAL SNAPSHOT join keeps it in Flink state instead of querying an external system for each row.

The following example runs a key lookup against an external customers_ext table to enrich an orders stream:

SELECT o.*, lookup.*
FROM orders AS o,
     LATERAL TABLE(KEY_SEARCH_AGG(customers_ext, DESCRIPTOR(o.customer_id), id))
     AS lookup

Table types. The querying side of a lookup join can be an append-only or updating table, and the join produces the same table type. The external table on the lookup side isn’t a Flink SQL changelog, so it doesn’t add its own changelog-mode requirement.

Confluent Cloud for Apache Flink supports three search functions for lookup joins:

• KEY_SEARCH_AGG: Exact key lookup against an external database. Use this for joining a stream with a reference table keyed by a unique identifier.

• TEXT_SEARCH_AGG: Full-text search against an external database. Use this when the join condition is text relevance rather than equality.

• VECTOR_SEARCH_AGG: Semantic similarity search using vector embeddings. Use this together with AI_EMBEDDING to embed the input column inline.

For the underlying concept and the list of supported external systems for each search type, see External Tables. For the full function syntax and configuration options, see Search Functions. For provider-specific configuration and end-to-end examples, see Search External Tables with Confluent Cloud for Apache Flink.

Array expansion

Returns a new row for each element in the given array.

SELECT order_id, tag
FROM orders CROSS JOIN UNNEST(tags) AS t (tag)

To return the position of each element in the array, use WITH ORDINALITY. The position is 1-indexed, and you name the position column in the alias list.

SELECT order_id, tag, tag_position
FROM orders CROSS JOIN UNNEST(tags) WITH ORDINALITY AS t (tag, tag_position)

WITH ORDINALITY supports only CROSS JOIN. The LEFT JOIN variant of UNNEST ... WITH ORDINALITY isn’t supported.

Related content

• Confluent Developer: Temporal Joins Explained

• Enrich a Stream with Reference Data: a step-by-step LATERAL SNAPSHOT join, including how to handle a reference table that rarely changes

• Example Data Streams: run the join queries on this page against real, ready-to-use tables in the examples catalog

• Flink SQL Queries

• Flink SQL Functions

• Statements

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.