Joins

Window joins

Window joins operate over bounded time-oriented windows (one of TUMBLE, HOP, or SESSION). Each subquery being joined must be computed over the exact same window, and all matching elements within that window (according to the type of join) will be returned. For example, a join over TUMBLE(interval '1 minute') will be computed once a minute, and return all matching elements within that minute. Window completions are triggered by the arrival of watermarks. Window joins default to INNER joins, but you may also specify LEFT, RIGHT, or FULL before the JOIN keyword. A full example of a window join looks like this:

CREATE TABLE page_views (
  event_time TIMESTAMP,
  user_id TEXT,
  page_url TEXT
) WITH (
    connector = 'kafka',
    bootstrap_servers = 'broker:9092',
    topic = 'page_views',
    format = 'json',
    type = 'source'
);

CREATE TABLE ad_clicks (
  event_time TIMESTAMP,
  user_id TEXT,
  ad_id TEXT
) WITH (
    connector = 'kafka',
    bootstrap_servers = 'broker:9092',
    topic = 'ad_clicks',
    format = 'json',
    type = 'source'
);

SELECT
  pv.window,
  page_views,
  ad_clicks
FROM (
  SELECT TUMBLE(INTERVAL '1 minute') AS window, COUNT(DISTINCT user_id) as page_views
  FROM page_views
  GROUP BY 1
) pv
INNER JOIN (
  SELECT TUMBLE(INTERVAL '1 minute') AS window, COUNT(DISTINCT user_id) as ad_clicks
  FROM ad_clicks
  GROUP BY 1
) ac
ON pv.window = ac.window;

Example

Given the following input data: page_views:

event_time	user_id	page_url
2024-01-01 10:00:15	alice	/home
2024-01-01 10:00:30	bob	/products
2024-01-01 10:00:45	alice	/checkout
2024-01-01 10:01:10	charlie	/home
2024-01-01 10:01:40	bob	/products

ad_clicks:

event_time	user_id	ad_id
2024-01-01 10:00:20	alice	ad_1
2024-01-01 10:00:35	bob	ad_2
2024-01-01 10:01:50	charlie	ad_3

The TUMBLE(INTERVAL '1 minute') window divides the timeline into fixed 1-minute intervals. For the window [10:00:00, 10:01:00), page_views contains events from alice and bob (2 distinct users), and ad_clicks contains events from alice and bob (2 distinct users). For the window [10:01:00, 10:02:00), page_views contains events from charlie and bob (2 distinct users), and ad_clicks contains an event from charlie (1 distinct user). Output:

window	page_views	ad_clicks
2024-01-01 10:00:00 - 10:01:00	2	2
2024-01-01 10:01:00 - 10:02:00	2	1

Note that it is not currently possible to reaggregate the results of a windowed join.

Updating joins

Unlike window joins, updating joins do not require inputs to be time-windowed. These behave like standard SQL joins, but are computed incrementally and produce updating output (a stream of delete, append, and update results), which must be sent to an update-compatible sink (for example, using the debezium_json format). An example updating join looks like this:

CREATE TABLE users (
  user_id TEXT PRIMARY KEY,
  user_name TEXT
) WITH (
    connector = 'kafka',
    bootstrap_servers = 'broker:9092',
    topic = 'users',
    format = 'json',
    type = 'source'
);

CREATE TABLE user_actions (
  event_time TIMESTAMP,
  user_id TEXT,
  action TEXT
) WITH (
    connector = 'kafka',
    bootstrap_servers = 'broker:9092',
    topic = 'user_actions',
    format = 'json',
    type = 'source'
);

SELECT u.user_id, u.user_name, a.action, a.event_time
FROM users u
JOIN user_actions a
ON u.user_id = a.user_id;

Example

Given the following input data: users:

user_id	user_name
u1	Alice
u2	Bob

user_actions (stream):

event_time	user_id	action
2024-01-01 10:00:15	u1	login
2024-01-01 10:00:30	u2	login
2024-01-01 10:00:45	u1	purchase
2024-01-01 10:01:10	u3	login

Each event in user_actions is joined against the users table. Events for u1 match “Alice” and events for u2 match “Bob”. The event for u3 has no matching user record, so it is dropped by the INNER join. Output:

user_id	user_name	action	event_time
u1	Alice	login	2024-01-01 10:00:15
u2	Bob	login	2024-01-01 10:00:30
u1	Alice	purchase	2024-01-01 10:00:45

As updating joins are not time-bounded, there is nothing that bounds the state size; in principle they must remember all data the pipeline has seen. This is not usually practical, so by default Ingest Engine uses a TTL of 1 day after which join state is discarded. This may be configured via SET updating_ttl. See the updating docs for more details.

Lookup joins

Lookup joins allow you to enrich streams by referencing external or static data stored in external systems (e.g., Redis, relational databases). The right side of a lookup join is a special kind of table, called a TEMPORARY TABLE, which is unmaterialized and backed by the external system. Each record that comes into the join from the left (stream) side causes a query to the underlying system. For example, you may have a set of detailed customer records stored in an RDBMS that you would like to use to enrich an analytics stream. Currently, Redis is supported as a lookup connector. An example lookup join looks like this:

CREATE TEMPORARY TABLE customers (
    -- For Redis lookup tables, it's required that there be a single
    -- METADATA FROM 'key' marked as PRIMARY KEY, as Redis only supports
    -- efficient lookups by key
    customer_id TEXT METADATA FROM 'key' PRIMARY KEY,
    name TEXT,
    plan TEXT
) with (
    connector = 'redis',
    address = 'redis://localhost:6379',
    format = 'json',
    'lookup.cache.max_bytes' = 1000000,
    'lookup.cache.ttl' = interval '5 seconds'
);

CREATE TABLE events (
    event_id TEXT,
    timestamp TIMESTAMP,
    customer_id TEXT,
    event_type TEXT
) WITH (
    connector = 'kafka',
    topic = 'events',
    type = 'source',
    format = 'json',
    bootstrap_servers = 'broker:9092'
);

SELECT  e.event_id,  e.timestamp,  c.name, c.plan
FROM  events e
LEFT JOIN customers c
-- you may use SQL expressions like concat to generate the exact key
-- format in Redis
ON concat('customer.', e.customer_id) = c.customer_id
WHERE c.plan = 'Premium';

Example

The following is a conceptual illustration of lookup join behavior. The actual output depends on the state of the external system (Redis) at the time each event is processed. events (input stream):

event_id	timestamp	customer_id	event_type
e1	2024-01-01 10:00:15	c123	page_view
e2	2024-01-01 10:00:30	c456	purchase
e3	2024-01-01 10:00:45	c789	page_view

Redis state (at query time):

customer.c123 → {"name": "Alice", "plan": "Premium"}
customer.c456 → {"name": "Bob", "plan": "Premium"}
customer.c789 → {"name": "Charlie", "plan": "Free"}

Each event triggers a lookup against Redis using the key concat('customer.', customer_id). The WHERE c.plan = 'Premium' filter then removes any rows where the customer is not on the Premium plan, so Charlie’s event is excluded. Output (after WHERE c.plan = 'Premium'):

event_id	timestamp	name	plan
e1	2024-01-01 10:00:15	Alice	Premium
e2	2024-01-01 10:00:30	Bob	Premium

The lookup.cache.max_bytes and lookup.cache.ttl are optional arguments that control the behavior of the built-in cache, which avoids the need to query the same keys over and over again. Lookup joins can be either INNER (the default) or LEFT.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Window joins

Example

Updating joins

Example

Lookup joins

Example

​Window joins

​Example

​Updating joins

​Example

​Lookup joins

​Example

Window joins

Example

Updating joins

Example

Lookup joins

Example