citusdata / citusdata/citus

Distributed and local table joins might end up with planning errors when distributed is choosen

Open
#6,659 1 comment 0 reactions 0 assignees View on GitHub
bug
Dominant language
C
Stars
12.8k
Forks
794
Avg merge
2d 14h
Merged PRs (30d)
31

Description

```SQL
create table local(id int);
create table distributed(id int);
select create_distributed_table('distributed','id');

set citus.local_table_join_policy TO 'prefer-distributed';

SELECT COUNT(*)
FROM
local
JOIN
distributed USING (id)
JOIN (SELECT id, NULL, NULL FROM distributed) foo USING (id);

DEBUG: Wrapping relation "distributed" to a subquery
DEBUG: generating subplan 54_1 for subquery SELECT id FROM public.distributed WHERE true
DEBUG: Plan 54 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((public.local JOIN (SELECT distributed_1.id FROM (SELECT intermediate_result.id FROM read_intermediate_result('54_1'::text, 'binary'::citus_copy_format) intermediate_result(id integer)) distributed_1) distributed USING (id)) JOIN (SELECT distributed_1.id, NULL::text AS "?column?", NULL::text AS "?column?" FROM public.distributed distributed_1) foo(id, "?column?", "?column?_1") USING (id))
DEBUG: Wrapping relation "local" to a subquery
DEBUG: generating subplan 54_1 for subquery SELECT id FROM public.local WHERE true
DEBUG: Plan 54 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (((SELECT local_1.id FROM (SELECT intermediate_result.id FROM read_intermediate_result('54_1'::text, 'binary'::citus_copy_format) intermediate_result(id integer)) local_1) local JOIN (SELECT distributed_1.id FROM (SELECT intermediate_result.id FROM read_intermediate_result('54_1'::text, 'binary'::citus_copy_format) intermediate_result(id integer)) distributed_1) distributed USING (id)) JOIN (SELECT distributed_1.id, NULL::text AS "?column?", NULL::text AS "?column?" FROM public.distributed distributed_1) foo(id, "?column?", "?column?_1") USING (id))
```

The problem is that we force to recursively plan distributed table, but then the second call to CreateDistributedPlan finds a join tree which is NOT ready yet, and does one more recursive planning due to the local table in the join tree. Thus, we end up with the same subPlanId, where the second one is overridden by the first one.

In the above example, we used `set citus.local_table_join_policy TO 'prefer-distributed';` which is a niche area. However, we can repro the same with "auto" mode, and primary keys on the distributed tables:

```SQL
set citus.local_table_join_policy TO 'auto';

create table local(id int);
create table distributed(id int PRIMARY KEY);
select create_distributed_table('distributed','id');

SELECT COUNT(*) FROM local JOIN distributed d1 USING (id) JOIN (SELECT id, NULL, NULL FROM distributed) foo USING (id) WHERE d1.id = 15;
DEBUG: Wrapping relation "distributed" "d1" to a subquery
DEBUG: generating subplan 58_1 for subquery SELECT id FROM public.distributed d1 WHERE (id OPERATOR(pg_catalog.=) 15)
DEBUG: Plan 58 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM ((public.local JOIN (SELECT d1_1.id FROM (SELECT intermediate_result.id FROM read_intermediate_result('58_1'::text, 'binary'::citus_copy_format) intermediate_result(id integer)) d1_1) d1 USING (id)) JOIN (SELECT distributed.id, NULL::text AS "?column?", NULL::text AS "?column?" FROM public.distributed) foo(id, "?column?", "?column?_1") USING (id)) WHERE (d1.id OPERATOR(pg_catalog.=) 15)
DEBUG: Wrapping relation "local" to a subquery
DEBUG: generating subplan 58_1 for subquery SELECT id FROM public.local WHERE (id OPERATOR(pg_catalog.=) 15)
DEBUG: Plan 58 query after replacing subqueries and CTEs: SELECT count(*) AS count FROM (((SELECT local_1.id FROM (SELECT intermediate_result.id FROM read_intermediate_result('58_1'::text, 'binary'::citus_copy_format) intermediate_result(id integer)) local_1) local JOIN (SELECT d1_1.id FROM (SELECT intermediate_result.id FROM read_intermediate_result('58_1'::text, 'binary'::citus_copy_format) intermediate_result(id integer)) d1_1) d1 USING (id)) JOIN (SELECT distributed.id, NULL::text AS "?column?", NULL::text AS "?column?" FROM public.distributed) foo(id, "?column?", "?column?_1") USING (id)) WHERE (d1.id OPERATOR(pg_catalog.=) 15)
```

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.