external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add local-dist table joins to arbitrary configs

test2
Sait Talha Nisanci 2021-10-13 12:26:54 +03:00 committed by SaitTalhaNisanci
parent 31c8f279ac
commit 7336c03c22
7 changed files with 1337 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/test/regress/citus_tests/config.py
View File

@ -257,6 +257,7 @@ class CitusUnusualExecutorConfig(CitusMXBaseClusterConfig):
# "citus.force_max_query_parallelization": "on", # "citus.force_max_query_parallelization": "on",
"citus.binary_worker_copy_format": False, "citus.binary_worker_copy_format": False,
"citus.enable_binary_protocol": False, "citus.enable_binary_protocol": False,
"citus.local_table_join_policy": "prefer-distributed",
} }
@ -290,7 +291,7 @@ class CitusUnusualQuerySettingsConfig(CitusMXBaseClusterConfig):
"citus.enable_single_hash_repartition_joins": True, "citus.enable_single_hash_repartition_joins": True,
"citus.recover_2pc_interval": "1s", "citus.recover_2pc_interval": "1s",
"citus.remote_task_check_interval": "1ms", "citus.remote_task_check_interval": "1ms",
"citus.values_materialization_threshold" : "0" "citus.values_materialization_threshold": "0",
} }

1
src/test/regress/create_schedule
View File

@ -1,3 +1,4 @@
test: create_test intermediate_result_pruning_create test: create_test intermediate_result_pruning_create
test: prepared_statements_create_load ch_benchmarks_create_load test: prepared_statements_create_load ch_benchmarks_create_load
test: dropped_columns_create_load distributed_planning_create_load test: dropped_columns_create_load distributed_planning_create_load
test: local_dist_join_load

906
src/test/regress/expected/local_dist_join.out Normal file
View File

@ -0,0 +1,906 @@
SET search_path TO local_dist_join_mixed;
-- very simple 1-1 Joins
SELECT count(*) FROM distributed JOIN local USING (id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(*) FROM distributed JOIN local ON (name = title);
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text);
count
---------------------------------------------------------------------
10201
(1 row)
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int);
count
---------------------------------------------------------------------
5050
(1 row)
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
count
---------------------------------------------------------------------
99
(1 row)
SELECT count(*) FROM distributed JOIN local USING (id) WHERE false;
count
---------------------------------------------------------------------
0
(1 row)
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 OR True;
count
---------------------------------------------------------------------
5050
(1 row)
SELECT count(*) FROM distributed d1 JOIN local ON (name::int + local.id > d1.id AND d1.id < local.title::int) WHERE d1.id = 1;
count
---------------------------------------------------------------------
99
(1 row)
SELECT count(*) FROM distributed JOIN local ON (hashtext(name) = hashtext(title));
count
---------------------------------------------------------------------
101
(1 row)
SELECT hashtext(local.id::text) FROM distributed JOIN local ON (hashtext(name) = hashtext(title)) ORDER BY 1 LIMIT 4;
hashtext
---------------------------------------------------------------------
-2114455578
-2097988278
-1997006946
-1985772843
(4 rows)
SELECT '' as "xxx", local.*, 'xxx' as "test" FROM distributed JOIN local ON (hashtext(name) = hashtext(title)) ORDER BY 1,2,3 LIMIT 4;
xxx | id | title | test
---------------------------------------------------------------------
| 0 | 0 | xxx
| 1 | 1 | xxx
| 2 | 2 | xxx
| 3 | 3 | xxx
(4 rows)
SELECT local.title, count(*) FROM distributed JOIN local USING (id) GROUP BY 1 ORDER BY 1, 2 DESC LIMIT 5;
title | count
---------------------------------------------------------------------
0 | 1
1 | 1
10 | 1
100 | 1
11 | 1
(5 rows)
SELECT distributed.id as id1, local.id as id2 FROM distributed JOIN local USING(id) ORDER BY distributed.id + local.id LIMIT 5;
id1 | id2
---------------------------------------------------------------------
0 | 0
1 | 1
2 | 2
3 | 3
4 | 4
(5 rows)
SELECT distributed.id as id1, local.id as id2, count(*) FROM distributed JOIN local USING(id) GROUP BY distributed.id, local.id ORDER BY 1,2 LIMIT 5;
id1 | id2 | count
---------------------------------------------------------------------
0 | 0 | 1
1 | 1 | 1
2 | 2 | 1
3 | 3 | 1
4 | 4 | 1
(5 rows)
-- basic subqueries that cannot be pulled up
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local USING (id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = title);
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text);
count
---------------------------------------------------------------------
10201
(1 row)
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int);
count
---------------------------------------------------------------------
5050
(1 row)
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
count
---------------------------------------------------------------------
99
(1 row)
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 AND false;
count
---------------------------------------------------------------------
0
(1 row)
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 OR true;
count
---------------------------------------------------------------------
5050
(1 row)
-- pull up subqueries as they are pretty simple, local table should be recursively planned
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local USING (id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = title);
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text);
count
---------------------------------------------------------------------
10201
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int);
count
---------------------------------------------------------------------
5050
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
count
---------------------------------------------------------------------
99
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 AND false;
count
---------------------------------------------------------------------
0
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 OR true;
count
---------------------------------------------------------------------
5050
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed WHERE id = 2) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
count
---------------------------------------------------------------------
0
(1 row)
SELECT count(*) FROM (SELECT * FROM distributed WHERE false) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
count
---------------------------------------------------------------------
0
(1 row)
-- final queries are pushdown queries
SELECT sum(d1.id + local.id) FROM distributed d1 JOIN local USING (id);
sum
---------------------------------------------------------------------
10100
(1 row)
SELECT sum(d1.id + local.id) OVER (PARTITION BY d1.id) FROM distributed d1 JOIN local USING (id) ORDER BY 1 DESC LIMIT 4;
sum
---------------------------------------------------------------------
200
198
196
194
(4 rows)
SELECT count(*) FROM distributed d1 JOIN local USING (id) LEFT JOIN distributed d2 USING (id) ORDER BY 1 DESC LIMIT 4;
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(DISTINCT d1.name::int * local.id) FROM distributed d1 JOIN local USING (id);
count
---------------------------------------------------------------------
101
(1 row)
-- final queries are router queries
SELECT sum(d1.id + local.id) FROM distributed d1 JOIN local USING (id) WHERE d1.id = 1;
sum
---------------------------------------------------------------------
2
(1 row)
SELECT sum(d1.id + local.id) OVER (PARTITION BY d1.id) FROM distributed d1 JOIN local USING (id) WHERE d1.id = 1 ORDER BY 1 DESC LIMIT 4;
sum
---------------------------------------------------------------------
2
(1 row)
SELECT count(*) FROM distributed d1 JOIN local USING (id) LEFT JOIN distributed d2 USING (id) WHERE d2.id = 1 ORDER BY 1 DESC LIMIT 4;
count
---------------------------------------------------------------------
1
(1 row)
-- final queries are pull to coordinator queries
SELECT sum(d1.id + local.id) OVER (PARTITION BY d1.id + local.id) FROM distributed d1 JOIN local USING (id) ORDER BY 1 DESC LIMIT 4;
sum
---------------------------------------------------------------------
200
198
196
194
(4 rows)
-- nested subqueries
SELECT
count(*)
FROM
(SELECT * FROM (SELECT * FROM distributed) as foo) as bar
JOIN
local
USING(id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
local
USING(id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
local
USING(id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id);
count
---------------------------------------------------------------------
101
(1 row)
-- TODO: Unnecessary recursive planning for local
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed LIMIT 1) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id);
count
---------------------------------------------------------------------
1
(1 row)
-- subqueries in WHERE clause
-- is not colocated, and the JOIN inside as well.
-- so should be recursively planned twice
SELECT
count(*)
FROM
distributed
WHERE
id > (SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id)
);
count
---------------------------------------------------------------------
0
(1 row)
-- two distributed tables are co-located and JOINed on distribution
-- key, so should be fine to pushdown
SELECT
count(*)
FROM
distributed d_upper
WHERE
(SELECT
bar.id
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed WHERE distributed.id = d_upper.id) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id)
) IS NOT NULL;
count
---------------------------------------------------------------------
101
(1 row)
SELECT
count(*)
FROM
distributed d_upper
WHERE
(SELECT
bar.id
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed WHERE distributed.id = d_upper.id) as foo) as bar
JOIN
local as foo
USING(id)
) IS NOT NULL;
count
---------------------------------------------------------------------
101
(1 row)
SELECT
count(*)
FROM
distributed d_upper
WHERE d_upper.id >
(SELECT
bar.id
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed WHERE distributed.id = d_upper.id) as foo) as bar
JOIN
local as foo
USING(id)
);
count
---------------------------------------------------------------------
0
(1 row)
-- subqueries in the target list
-- router, should work
select (SELECT local.id) FROM local, distributed WHERE distributed.id = 1 LIMIT 1;
id
---------------------------------------------------------------------
0
(1 row)
-- set operations
SELECT local.* FROM distributed JOIN local USING (id)
EXCEPT
SELECT local.* FROM distributed JOIN local USING (id);
id | title
---------------------------------------------------------------------
(0 rows)
SELECT distributed.* FROM distributed JOIN local USING (id)
EXCEPT
SELECT distributed.* FROM distributed JOIN local USING (id);
id | name | created_at
---------------------------------------------------------------------
(0 rows)
SELECT count(*) FROM
(
(SELECT * FROM (SELECT * FROM local) as f JOIN distributed USING (id))
UNION ALL
(SELECT * FROM (SELECT * FROM local) as f2 JOIN distributed USING (id))
) bar;
count
---------------------------------------------------------------------
202
(1 row)
SELECT count(*) FROM
(
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as fo)
UNION ALL
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as ba)
) bar;
count
---------------------------------------------------------------------
202
(1 row)
select count(DISTINCT id)
FROM
(
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as fo)
UNION ALL
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as ba)
) bar;
count
---------------------------------------------------------------------
101
(1 row)
-- lateral joins
SELECT COUNT(*) FROM local JOIN LATERAL (SELECT * FROM distributed WHERE local.id = distributed.id) as foo ON (true);
count
---------------------------------------------------------------------
101
(1 row)
SELECT COUNT(*) FROM local JOIN LATERAL (SELECT * FROM distributed WHERE local.id > distributed.id) as foo ON (true);
count
---------------------------------------------------------------------
5050
(1 row)
SELECT count(*) FROM distributed CROSS JOIN local;
count
---------------------------------------------------------------------
10201
(1 row)
SELECT count(*) FROM distributed CROSS JOIN local WHERE distributed.id = 1;
count
---------------------------------------------------------------------
101
(1 row)
-- w count(*) it works fine as PG ignores the inner tables
SELECT count(*) FROM distributed LEFT JOIN local USING (id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT count(*) FROM local LEFT JOIN distributed USING (id);
count
---------------------------------------------------------------------
101
(1 row)
SELECT id, name FROM distributed LEFT JOIN local USING (id) ORDER BY 1 LIMIT 1;
id | name
---------------------------------------------------------------------
0 | 0
(1 row)
SELECT
foo1.id
FROM
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo9,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo8,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo7,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo6,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo5,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo4,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo3,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo2,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo10,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo1
WHERE
foo1.id = foo9.id AND
foo1.id = foo8.id AND
foo1.id = foo7.id AND
foo1.id = foo6.id AND
foo1.id = foo5.id AND
foo1.id = foo4.id AND
foo1.id = foo3.id AND
foo1.id = foo2.id AND
foo1.id = foo10.id AND
foo1.id = foo1.id
ORDER BY 1;
id
---------------------------------------------------------------------
0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
(101 rows)
SELECT
foo1.id
FROM
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo1,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo2,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo3,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo4,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo5
WHERE
foo1.id = foo4.id AND
foo1.id = foo2.id AND
foo1.id = foo3.id AND
foo1.id = foo4.id AND
foo1.id = foo5.id
ORDER BY 1;
id
---------------------------------------------------------------------
0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
(101 rows)
SELECT
foo1.id
FROM
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 1) as foo1,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 2) as foo2,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 3) as foo3,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 4) as foo4,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 5) as foo5
WHERE
foo1.id = foo4.id AND
foo1.id = foo2.id AND
foo1.id = foo3.id AND
foo1.id = foo4.id AND
foo1.id = foo5.id
ORDER BY 1;
id
---------------------------------------------------------------------
(0 rows)
--https://github.com/citusdata/citus/issues/5384
-- SELECT
-- count(*)
-- FROM
-- distributed
-- JOIN LATERAL
-- (SELECT
-- *
-- FROM
-- local
-- JOIN
-- distributed d2
-- ON(true)
-- WHERE local.id = distributed.id AND d2.id = local.id) as foo
-- ON (true);
SELECT local.title, local.title FROM local JOIN distributed USING(id) ORDER BY 1,2 LIMIt 1;
title | title
---------------------------------------------------------------------
0 | 0
(1 row)
SELECT NULL FROM local JOIN distributed USING(id) ORDER BY 1 LIMIt 1;
?column?
---------------------------------------------------------------------
(1 row)
SELECT distributed.name, distributed.name, local.title, local.title FROM local JOIN distributed USING(id) ORDER BY 1,2,3,4 LIMIT 1;
name | name | title | title
---------------------------------------------------------------------
0 | 0 | 0 | 0
(1 row)
SELECT
COUNT(*)
FROM
local
JOIN
distributed
USING
(id)
JOIN
(SELECT id, NULL, NULL FROM distributed) foo
USING
(id);
count
---------------------------------------------------------------------
101
(1 row)
BEGIN;
SELECT COUNT(DISTINCT title) FROM local;
count
---------------------------------------------------------------------
101
(1 row)
UPDATE
local
SET
title = 'test'
FROM
distributed
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT title) FROM local;
count
---------------------------------------------------------------------
1
(1 row)
ROLLBACK;
BEGIN;
SELECT COUNT(DISTINCT name) FROM distributed;
count
---------------------------------------------------------------------
101
(1 row)
UPDATE
distributed
SET
name = 'test'
FROM
local
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT name) FROM distributed;
count
---------------------------------------------------------------------
1
(1 row)
ROLLBACK;
BEGIN;
SELECT COUNT(DISTINCT name) FROM distributed;
count
---------------------------------------------------------------------
101
(1 row)
UPDATE
distributed
SET
name = 'test'
FROM
local
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT name) FROM distributed;
count
---------------------------------------------------------------------
1
(1 row)
ROLLBACK;
BEGIN;
SELECT COUNT(DISTINCT name) FROM distributed;
count
---------------------------------------------------------------------
101
(1 row)
UPDATE
distributed
SET
name = 'test'
FROM
local
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT name) FROM distributed;
count
---------------------------------------------------------------------
1
(1 row)
ROLLBACK;

39
src/test/regress/expected/local_dist_join_load.out Normal file
View File

@ -0,0 +1,39 @@
CREATE SCHEMA local_dist_join_mixed;
SET search_path TO local_dist_join_mixed;
CREATE TABLE distributed (key int, id bigserial PRIMARY KEY,
name text,
created_at timestamptz DEFAULT now(), b int);
CREATE TABLE reference (a int, id bigserial PRIMARY KEY,
title text, b int);
CREATE TABLE local (key int, id bigserial PRIMARY KEY, key2 int,
title text, key3 int);
-- drop columns so that we test the correctness in different scenarios.
ALTER TABLE local DROP column key;
ALTER TABLE local DROP column key2;
ALTER TABLE local DROP column key3;
ALTER TABLE distributed DROP column key;
ALTER TABLE reference DROP column b;
-- these above restrictions brought us to the following schema
SELECT create_reference_table('reference');
create_reference_table
---------------------------------------------------------------------
(1 row)
SELECT create_distributed_table('distributed', 'id');
create_distributed_table
---------------------------------------------------------------------
(1 row)
SELECT citus_add_local_table_to_metadata('local');
citus_add_local_table_to_metadata
---------------------------------------------------------------------
(1 row)
ALTER TABLE distributed DROP column b;
ALTER TABLE reference DROP column a;
INSERT INTO distributed SELECT i, i::text, now() FROM generate_series(0,100)i;
INSERT INTO reference SELECT i, i::text FROM generate_series(0,100)i;
INSERT INTO local SELECT i, i::text FROM generate_series(0,100)i;

355
src/test/regress/sql/local_dist_join.sql Normal file
View File

@ -0,0 +1,355 @@
SET search_path TO local_dist_join_mixed;
-- very simple 1-1 Joins
SELECT count(*) FROM distributed JOIN local USING (id);
SELECT count(*) FROM distributed JOIN local ON (name = title);
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text);
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int);
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
SELECT count(*) FROM distributed JOIN local USING (id) WHERE false;
SELECT count(*) FROM distributed d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 OR True;
SELECT count(*) FROM distributed d1 JOIN local ON (name::int + local.id > d1.id AND d1.id < local.title::int) WHERE d1.id = 1;
SELECT count(*) FROM distributed JOIN local ON (hashtext(name) = hashtext(title));
SELECT hashtext(local.id::text) FROM distributed JOIN local ON (hashtext(name) = hashtext(title)) ORDER BY 1 LIMIT 4;
SELECT '' as "xxx", local.*, 'xxx' as "test" FROM distributed JOIN local ON (hashtext(name) = hashtext(title)) ORDER BY 1,2,3 LIMIT 4;
SELECT local.title, count(*) FROM distributed JOIN local USING (id) GROUP BY 1 ORDER BY 1, 2 DESC LIMIT 5;
SELECT distributed.id as id1, local.id as id2 FROM distributed JOIN local USING(id) ORDER BY distributed.id + local.id LIMIT 5;
SELECT distributed.id as id1, local.id as id2, count(*) FROM distributed JOIN local USING(id) GROUP BY distributed.id, local.id ORDER BY 1,2 LIMIT 5;
-- basic subqueries that cannot be pulled up
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local USING (id);
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = title);
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text);
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int);
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 AND false;
SELECT count(*) FROM (SELECT *, random() FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 OR true;
-- pull up subqueries as they are pretty simple, local table should be recursively planned
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local USING (id);
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = title);
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text);
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int);
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 AND false;
SELECT count(*) FROM (SELECT * FROM distributed) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1 OR true;
SELECT count(*) FROM (SELECT * FROM distributed WHERE id = 2) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
SELECT count(*) FROM (SELECT * FROM distributed WHERE false) as d1 JOIN local ON (name = d1.id::text AND d1.id < local.title::int) WHERE d1.id = 1;
-- final queries are pushdown queries
SELECT sum(d1.id + local.id) FROM distributed d1 JOIN local USING (id);
SELECT sum(d1.id + local.id) OVER (PARTITION BY d1.id) FROM distributed d1 JOIN local USING (id) ORDER BY 1 DESC LIMIT 4;
SELECT count(*) FROM distributed d1 JOIN local USING (id) LEFT JOIN distributed d2 USING (id) ORDER BY 1 DESC LIMIT 4;
SELECT count(DISTINCT d1.name::int * local.id) FROM distributed d1 JOIN local USING (id);
-- final queries are router queries
SELECT sum(d1.id + local.id) FROM distributed d1 JOIN local USING (id) WHERE d1.id = 1;
SELECT sum(d1.id + local.id) OVER (PARTITION BY d1.id) FROM distributed d1 JOIN local USING (id) WHERE d1.id = 1 ORDER BY 1 DESC LIMIT 4;
SELECT count(*) FROM distributed d1 JOIN local USING (id) LEFT JOIN distributed d2 USING (id) WHERE d2.id = 1 ORDER BY 1 DESC LIMIT 4;
-- final queries are pull to coordinator queries
SELECT sum(d1.id + local.id) OVER (PARTITION BY d1.id + local.id) FROM distributed d1 JOIN local USING (id) ORDER BY 1 DESC LIMIT 4;
-- nested subqueries
SELECT
count(*)
FROM
(SELECT * FROM (SELECT * FROM distributed) as foo) as bar
JOIN
local
USING(id);
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
local
USING(id);
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
local
USING(id);
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id);
-- TODO: Unnecessary recursive planning for local
SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed LIMIT 1) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id);
-- subqueries in WHERE clause
-- is not colocated, and the JOIN inside as well.
-- so should be recursively planned twice
SELECT
count(*)
FROM
distributed
WHERE
id > (SELECT
count(*)
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id)
);
-- two distributed tables are co-located and JOINed on distribution
-- key, so should be fine to pushdown
SELECT
count(*)
FROM
distributed d_upper
WHERE
(SELECT
bar.id
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed WHERE distributed.id = d_upper.id) as foo) as bar
JOIN
(SELECT *, random() FROM (SELECT *,random() FROM local) as foo2) as bar2
USING(id)
) IS NOT NULL;
SELECT
count(*)
FROM
distributed d_upper
WHERE
(SELECT
bar.id
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed WHERE distributed.id = d_upper.id) as foo) as bar
JOIN
local as foo
USING(id)
) IS NOT NULL;
SELECT
count(*)
FROM
distributed d_upper
WHERE d_upper.id >
(SELECT
bar.id
FROM
(SELECT *, random() FROM (SELECT *, random() FROM distributed WHERE distributed.id = d_upper.id) as foo) as bar
JOIN
local as foo
USING(id)
);
-- subqueries in the target list
-- router, should work
select (SELECT local.id) FROM local, distributed WHERE distributed.id = 1 LIMIT 1;
-- set operations
SELECT local.* FROM distributed JOIN local USING (id)
EXCEPT
SELECT local.* FROM distributed JOIN local USING (id);
SELECT distributed.* FROM distributed JOIN local USING (id)
EXCEPT
SELECT distributed.* FROM distributed JOIN local USING (id);
SELECT count(*) FROM
(
(SELECT * FROM (SELECT * FROM local) as f JOIN distributed USING (id))
UNION ALL
(SELECT * FROM (SELECT * FROM local) as f2 JOIN distributed USING (id))
) bar;
SELECT count(*) FROM
(
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as fo)
UNION ALL
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as ba)
) bar;
select count(DISTINCT id)
FROM
(
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as fo)
UNION ALL
(SELECT * FROM (SELECT distributed.* FROM local JOIN distributed USING (id)) as ba)
) bar;
-- lateral joins
SELECT COUNT(*) FROM local JOIN LATERAL (SELECT * FROM distributed WHERE local.id = distributed.id) as foo ON (true);
SELECT COUNT(*) FROM local JOIN LATERAL (SELECT * FROM distributed WHERE local.id > distributed.id) as foo ON (true);
SELECT count(*) FROM distributed CROSS JOIN local;
SELECT count(*) FROM distributed CROSS JOIN local WHERE distributed.id = 1;
-- w count(*) it works fine as PG ignores the inner tables
SELECT count(*) FROM distributed LEFT JOIN local USING (id);
SELECT count(*) FROM local LEFT JOIN distributed USING (id);
SELECT id, name FROM distributed LEFT JOIN local USING (id) ORDER BY 1 LIMIT 1;
SELECT
foo1.id
FROM
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo9,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo8,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo7,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo6,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo5,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo4,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo3,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo2,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo10,
(SELECT local.id, local.title FROM local, distributed WHERE local.id = distributed.id ) as foo1
WHERE
foo1.id = foo9.id AND
foo1.id = foo8.id AND
foo1.id = foo7.id AND
foo1.id = foo6.id AND
foo1.id = foo5.id AND
foo1.id = foo4.id AND
foo1.id = foo3.id AND
foo1.id = foo2.id AND
foo1.id = foo10.id AND
foo1.id = foo1.id
ORDER BY 1;
SELECT
foo1.id
FROM
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo1,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo2,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo3,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo4,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id ) as foo5
WHERE
foo1.id = foo4.id AND
foo1.id = foo2.id AND
foo1.id = foo3.id AND
foo1.id = foo4.id AND
foo1.id = foo5.id
ORDER BY 1;
SELECT
foo1.id
FROM
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 1) as foo1,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 2) as foo2,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 3) as foo3,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 4) as foo4,
(SELECT local.id FROM distributed, local WHERE local.id = distributed.id AND distributed.id = 5) as foo5
WHERE
foo1.id = foo4.id AND
foo1.id = foo2.id AND
foo1.id = foo3.id AND
foo1.id = foo4.id AND
foo1.id = foo5.id
ORDER BY 1;
--https://github.com/citusdata/citus/issues/5384
-- SELECT
-- count(*)
-- FROM
-- distributed
-- JOIN LATERAL
-- (SELECT
-- *
-- FROM
-- local
-- JOIN
-- distributed d2
-- ON(true)
-- WHERE local.id = distributed.id AND d2.id = local.id) as foo
-- ON (true);
SELECT local.title, local.title FROM local JOIN distributed USING(id) ORDER BY 1,2 LIMIt 1;
SELECT NULL FROM local JOIN distributed USING(id) ORDER BY 1 LIMIt 1;
SELECT distributed.name, distributed.name, local.title, local.title FROM local JOIN distributed USING(id) ORDER BY 1,2,3,4 LIMIT 1;
SELECT
COUNT(*)
FROM
local
JOIN
distributed
USING
(id)
JOIN
(SELECT id, NULL, NULL FROM distributed) foo
USING
(id);
BEGIN;
SELECT COUNT(DISTINCT title) FROM local;
UPDATE
local
SET
title = 'test'
FROM
distributed
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT title) FROM local;
ROLLBACK;
BEGIN;
SELECT COUNT(DISTINCT name) FROM distributed;
UPDATE
distributed
SET
name = 'test'
FROM
local
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT name) FROM distributed;
ROLLBACK;
BEGIN;
SELECT COUNT(DISTINCT name) FROM distributed;
UPDATE
distributed
SET
name = 'test'
FROM
local
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT name) FROM distributed;
ROLLBACK;
BEGIN;
SELECT COUNT(DISTINCT name) FROM distributed;
UPDATE
distributed
SET
name = 'test'
FROM
local
WHERE
distributed.id = local.id;
SELECT COUNT(DISTINCT name) FROM distributed;
ROLLBACK;

33
src/test/regress/sql/local_dist_join_load.sql Normal file
View File

@ -0,0 +1,33 @@
CREATE SCHEMA local_dist_join_mixed;
SET search_path TO local_dist_join_mixed;
CREATE TABLE distributed (key int, id bigserial PRIMARY KEY,
name text,
created_at timestamptz DEFAULT now(), b int);
CREATE TABLE reference (a int, id bigserial PRIMARY KEY,
title text, b int);
CREATE TABLE local (key int, id bigserial PRIMARY KEY, key2 int,
title text, key3 int);
-- drop columns so that we test the correctness in different scenarios.
ALTER TABLE local DROP column key;
ALTER TABLE local DROP column key2;
ALTER TABLE local DROP column key3;
ALTER TABLE distributed DROP column key;
ALTER TABLE reference DROP column b;
-- these above restrictions brought us to the following schema
SELECT create_reference_table('reference');
SELECT create_distributed_table('distributed', 'id');
SELECT citus_add_local_table_to_metadata('local');
ALTER TABLE distributed DROP column b;
ALTER TABLE reference DROP column a;
INSERT INTO distributed SELECT i, i::text, now() FROM generate_series(0,100)i;
INSERT INTO reference SELECT i, i::text FROM generate_series(0,100)i;
INSERT INTO local SELECT i, i::text FROM generate_series(0,100)i;

1
src/test/regress/sql_schedule
View File

@ -5,3 +5,4 @@ test: ch_benchmarks_1 ch_benchmarks_2 ch_benchmarks_3
test: ch_benchmarks_4 ch_benchmarks_5 ch_benchmarks_6 test: ch_benchmarks_4 ch_benchmarks_5 ch_benchmarks_6
test: intermediate_result_pruning_queries_1 intermediate_result_pruning_queries_2 test: intermediate_result_pruning_queries_1 intermediate_result_pruning_queries_2
test: dropped_columns_1 distributed_planning test: dropped_columns_1 distributed_planning
test: local_dist_join