INSERT INTO (SELECT...) improvement proposal
- Dominant language
- C
- Stars
- 12.8k
- Forks
- 794
- Avg merge
- 2d 14h
- Merged PRs (30d)
- 31
Description
Consider the following setup: here we have two tables distributed by `user_id` and the third one, distributed by `install_date`.
```sql
CREATE TABLE t1_user_id_distributed
(
user_id TEXT,
install_date timestamptz,
some_value NUMERIC
);
SELECT create_distributed_table('t1_user_id_distributed', 'user_id');
CREATE TABLE t2_user_id_distributed
(
user_id TEXT,
some_value NUMERIC
);
SELECT create_distributed_table('t2_user_id_distributed', 'user_id');
CREATE TABLE t3_instal_date_distributed
(
install_date timestamptz,
some_aggregated_value NUMERIC
);
SELECT create_distributed_table('t3_instal_date_distributed', 'install_date');
```
Now, consider the following SQL that fills `t3` with aggregated data from the join of `t1` and `t2`. As can be seen, it results in `INSERT/SELECT method: pull to coordinator`:
```sql
EXPLAIN
INSERT INTO t3_instal_date_distributed
(SELECT t1.install_date, avg(t1.some_value) + avg(t2.some_value)
FROM t1_user_id_distributed t1
LEFT JOIN t2_user_id_distributed t2 USING (user_id)
GROUP BY t1.install_date);
-- Output
-- Custom Scan (Citus INSERT ... SELECT) (cost=0.00..0.00 rows=0 width=0)
-- INSERT/SELECT method: pull to coordinator
-- -> HashAggregate (cost=1250.00..1255.50 rows=200 width=40)
-- Group Key: remote_scan.install_date
-- -> Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=100000 width=88)
-- ...
```
But theoretically, it is possible to
1. Gather the result of join repartitioned by `install_date`
2. GROUP BY the repartitioned result and insert into target table
It can be demonstrated by intermediate "temporary" distributed table:
```sql
BEGIN ;
CREATE TABLE distributed_temp_table
(
user_id TEXT,
install_date timestamptz,
some_value_1 NUMERIC,
some_value_2 NUMERIC
);
SELECT create_distributed_table('distributed_temp_table', 'install_date');
EXPLAIN
INSERT INTO distributed_temp_table
(SELECT t1.user_id, t1.install_date, t1.some_value, t2.some_value
FROM t1_user_id_distributed t1
LEFT JOIN t2_user_id_distributed t2 USING (user_id));
-- Output
-- Custom Scan (Citus INSERT ... SELECT) (cost=0.00..0.00 rows=0 width=0)
-- INSERT/SELECT method: repartition
-- -> Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=100000 width=104)
-- Task Count: 32
-- Tasks Shown: One of 32
-- ...
EXPLAIN
INSERT INTO t3_instal_date_distributed
(SELECT temp_table.install_date, avg(some_value_1) + avg(some_value_2)
FROM distributed_temp_table temp_table
GROUP BY temp_table.install_date);
-- Output
-- Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=0 width=0)
-- Task Count: 32
-- Tasks Shown: One of 32
-- -> Task
-- Node: host=citus-repartition-plan-example_worker_1 port=5432 dbname=postgres
-- -> Insert on t3_instal_date_distributed_102104 citus_table_alias (cost=20.65..26.15 rows=0 width=0)
-- -> HashAggregate (cost=20.65..24.15 rows=200 width=40)
-- Group Key: temp_table.install_date
-- -> Seq Scan on distributed_temp_table_102200 temp_table (cost=0.00..16.10 rows=607 width=72)
-- Filter: (install_date IS NOT NULL)
DROP TABLE distributed_temp_table;
END ;
```
Currently, we employ the last approach as a workaround and were able to gain x5 boost in performance: the "refresh" time went from 40 minutes to 5.
Contributor guide
Assessment
This issue has not been assessed yet.