> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/sql/streaming/joins.md).

# Joins

e6 Ingestion Engine SQL supports various types of joins over streaming and batch data sources.

## Window joins

Window joins operate over bounded [time-oriented windows](/ingestion-engine/sql/streaming/windows.md) (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:

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

```sql
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 e6 Ingestion 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](/ingestion-engine/sql/streaming/stateful.md) 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:

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


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.e6data.com/ingestion-engine/sql/streaming/joins.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.