# 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](../example-data.md#flink-sql-example-data) 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.
```sql
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](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-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.
```sql
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.
```sql
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
```sql
-- 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](../statements/explain.md#flink-sql-explain)
For queries with regular joins that grow state indefinitely, also consider:
* Setting an appropriate [state TTL](../statements/set.md#flink-sql-set-statement-config-options)
to limit state growth
* Using [interval joins](#flink-sql-interval-joins) when time bounds
are acceptable
* Using [lookup joins](#flink-sql-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](../../concepts/timely-stream-processing.md#flink-sql-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.
```sql
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](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-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](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-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](../statements/create-table.md#flink-sql-system-columns-rowtime) system column to get the
timestamp from an Apache Kafka® record. This is also used for the default
[watermark](../../../_glossary.md#term-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:
```sql
SELECT [column_list]
FROM table1 [AS ]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ rowtime } [AS ]
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.
```mysql
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.
#### 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](../../concepts/snapshot-queries.md#flink-sql-snapshot-queries).
To enrich data with a versioned table in batch mode, use a
[lookup join](#flink-sql-lookup-joins) instead, or run the query in
streaming mode. For more information, see
[Batch and Stream Processing](../../concepts/batch-and-stream-processing.md#flink-sql-batch-and-stream-processing).
## 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](../../concepts/external-tables.md#flink-external-tables), and the
lookup is expressed as a lateral join against one of three built-in
table-valued aggregation functions.
The following example runs a key lookup against an external `customers_ext`
table to enrich an `orders` stream:
```sql
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](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-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](../functions/search-functions.md#flink-sql-key-search-function): 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](../functions/search-functions.md#flink-sql-text-search-function): Full-text search
against an external database. Use this when the join condition is text
relevance rather than equality.
- [VECTOR_SEARCH_AGG](../functions/search-functions.md#flink-sql-vector-search-function): Semantic
similarity search using vector embeddings. Use this together with
[AI_EMBEDDING](../functions/model-inference-functions.md#flink-sql-ai-embedding-function) to embed the input
column inline.
For the underlying concept and the list of supported external systems for each
search type, see [External Tables](../../concepts/external-tables.md#flink-external-tables). For the full
function syntax and configuration options, see
[Search Functions](../functions/search-functions.md#flink-sql-search-functions). For provider-specific
configuration and end-to-end examples, see
[Search External Tables with Confluent Cloud for Apache Flink](../../../ai/external-tables/overview.md#ai-external-tables-overview).
## Array expansion
Returns a new row for each element in the given array.
```sql
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.
```sql
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](https://developer.confluent.io/courses/flink-sql/streaming-joins/)
- [Example Data Streams](../example-data.md#flink-sql-example-data): run the join queries
on this page against real, ready-to-use tables in the `examples` catalog
- [Flink SQL Queries](overview.md#flink-sql-queries)
- [Flink SQL Functions](../functions/overview.md#flink-sql-functions-overview)
- [Statements](../statements/overview.md#flink-sql-statements-overview)
#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).