citusdata / citusdata/citus

Perform simple non-co-located joins via pull-push

Open
#4,156 2 comments 0 reactions 0 assignees View on GitHub
performance
Dominant language
C
Stars
12.8k
Forks
794
Avg merge
2d 14h
Merged PRs (30d)
31

Description

Consider the following schema:
```sql
CREATE TABLE todo (todo_id int primary key, project_id int, deadline date, description text);
CREATE TABLE projects (project_id int primary key, name text);
INSERT INTO projects VALUES (1,'Citus development');
INSERT INTO todo VALUES (1, 1, '2020-09-11', 'Write GitHub issue on non-co-located joins');
CREATE INDEX todo_project_idx ON todo (project_id);
CREATE INDEX todo_desc_idx ON todo (description);
```
The application may issue lookup queries such as
```sql
SELECT * FROM todo JOIN projects USING (project_id) WHERE todo_id = 1;
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 2.720 ms

SELECT * FROM todo JOIN projects USING (project_id) WHERE project_id = 1;
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 2.523 ms

SELECT * FROM todo JOIN projects USING (project_id) WHERE description LIKE 'Write%';
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 2.709 ms
```

Perhaps the most appropriate way to distribute the tables is to make projects a reference table, but this only works well if the table is small and infrequently updated.
```sql
SELECT create_distributed_table('todo', 'todo_id');
SELECT create_reference_table('projects');

SELECT * FROM todo JOIN projects USING (project_id) WHERE todo_id = 1;
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 4.449 ms

SELECT * FROM todo JOIN projects USING (project_id) WHERE project_id = 1;
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 19.589 ms

SELECT * FROM todo JOIN projects USING (project_id) WHERE description LIKE 'Write%';
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 26.445 ms
```

If we took a naive approach and distributed both tables by primary key our queries all require re-partitioning:
```sql
SELECT create_distributed_table('todo', 'todo_id');
SELECT create_distributed_table('projects', 'project_id');

SELECT * FROM todo JOIN projects USING (project_id) WHERE todo_id = 1;
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 901.202 ms

SELECT * FROM todo JOIN projects USING (project_id) WHERE project_id = 1;
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 1486.482 ms (00:01.486)

SELECT * FROM todo JOIN projects USING (project_id) WHERE description LIKE 'Write%';
┌────────────┬─────────┬────────────┬────────────────────────────────────────────┬───────────────────┐
│ project_id │ todo_id │ deadline │ description │ name │
├────────────┼─────────┼────────────┼────────────────────────────────────────────┼───────────────────┤
│ 1 │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │ Citus development │
└────────────┴─────────┴────────────┴────────────────────────────────────────────┴───────────────────┘
(1 row)

Time: 1707.952 ms (00:01.708)
```

Conversely, if we rewrite it to a semantically equivalent queries that uses pull-push, we get the same result 50-100x faster:
```sql
WITH t1 AS (SELECT * FROM todo WHERE todo_id = 1)
SELECT * FROM projects JOIN t1 USING (project_id);
┌────────────┬───────────────────┬─────────┬────────────┬────────────────────────────────────────────┐
│ project_id │ name │ todo_id │ deadline │ description │
├────────────┼───────────────────┼─────────┼────────────┼────────────────────────────────────────────┤
│ 1 │ Citus development │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │
└────────────┴───────────────────┴─────────┴────────────┴────────────────────────────────────────────┘
(1 row)

Time: 24.253 ms

WITH p1 AS (SELECT * FROM projects WHERE project_id = 1)
SELECT * FROM p1 JOIN todo USING (project_id);
┌────────────┬───────────────────┬─────────┬────────────┬────────────────────────────────────────────┐
│ project_id │ name │ todo_id │ deadline │ description │
├────────────┼───────────────────┼─────────┼────────────┼────────────────────────────────────────────┤
│ 1 │ Citus development │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │
└────────────┴───────────────────┴─────────┴────────────┴────────────────────────────────────────────┘
(1 row)

Time: 20.884 ms

WITH t1 AS (SELECT * FROM todo WHERE description LIKE 'Write%')
SELECT * FROM projects JOIN t1 USING (project_id);
┌────────────┬───────────────────┬─────────┬────────────┬────────────────────────────────────────────┐
│ project_id │ name │ todo_id │ deadline │ description │
├────────────┼───────────────────┼─────────┼────────────┼────────────────────────────────────────────┤
│ 1 │ Citus development │ 1 │ 2020-09-11 │ Write GitHub issue on non-co-located joins │
└────────────┴───────────────────┴─────────┴────────────┴────────────────────────────────────────────┘
(1 row)

Time: 35.700 ms
```

While using a more optimized data model is still preferable, the 50-100x speedup can make the difference between an application remaining functional and breaking down completely. Hence it seems desirable to use pull-push for such joins. I also believe that the execution can be optimized to <10ms.

To decide when to use pull-push, we may need some sense of cost, since re-partitioning can still be preferable if the filters match most of the table.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.