Reconsider switching to local execution in transactions
- Dominant language
- C
- Stars
- 12.8k
- Forks
- 794
- Avg merge
- 2d 14h
- Merged PRs (30d)
- 31
Description
Switching to local execution means that Citus does not use parallel connections to execute queries, but instead use the same session that is executing the distributed query. This method is very useful for single-shard queries as it avoids the need for an extra connection. However, for multi-shard queries, switching to local execution means that we lose the parallelism as all queries on the shards are executed locally over the same backend.
With #3714, Citus switches to local execution when the execution is in a transaction. The meaning of transaction is defined in [`IsMultiStatementTransaction ()`](https://github.com/citusdata/citus/blob/c79c6506b980c80595e08c46c27b241fa7a497b5/src/backend/distributed/transaction/transaction_management.c#L702-L736). Basically, the following fall into the definition of transaction:
- Explicit transaction blocks
- Any query that involves pull-push execution
- INSERT .. SELECT via coordinator
- Queries inside functions/procedures
This creates an unexpected behavior for single-node Citus, losing parallelism for many cases. As a user, I'd expect the following commands to use parallelism but they don't:
```SQL
WITH cte_1 AS (SELECT * FROM test LIMIT 5) SELECT * FROM cte_1;
DEBUG: CTE cte_1 is going to be inlined via distributed planning
DEBUG: push down of limit count: 5
DEBUG: generating subplan 68_1 for subquery SELECT x, y FROM "Single.Node".test LIMIT 5
DEBUG: Plan 68 query after replacing subqueries and CTEs: SELECT x, y FROM (SELECT intermediate_result.x, intermediate_result.y FROM read_intermediate_result('68_1'::text, 'binary'::citus_copy_format) intermediate_result(x integer, y integer)) cte_1
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630500 test WHERE true LIMIT '5'::bigint
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630501 test WHERE true LIMIT '5'::bigint
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630502 test WHERE true LIMIT '5'::bigint
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630503 test WHERE true LIMIT '5'::bigint
NOTICE: executing the command locally: SELECT x, y FROM (SELECT intermediate_result.x, intermediate_result.y FROM read_intermediate_result('68_1'::text, 'binary'::citus_copy_format) intermediate_result(x integer, y integer)) cte_1
x | y
---+---
1 | 2
5 | 6
1 | 1
5 | 5
8 | 8
(5 rows)
Time: 5.784 ms
```
Or,
```SQL
insert into test select * from test LIMIT 50;
DEBUG: LIMIT clauses are not allowed in distributed INSERT ... SELECT queries
DEBUG: push down of limit count: 50
DEBUG: Collecting INSERT ... SELECT results on coordinator
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630500 test WHERE true LIMIT '50'::bigint
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630501 test WHERE true LIMIT '50'::bigint
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630502 test WHERE true LIMIT '50'::bigint
NOTICE: executing the command locally: SELECT x, y FROM "Single.Node".test_90630503 test WHERE true LIMIT '50'::bigint
NOTICE: executing the copy locally for shard 90630500
INSERT 0 50
```
As far as I remember, the main reason for us to switch to local execution is to manage connections wisely. If we can get #4178, is there a reason not to try parallel connections at first?
With the current approach, we are being pessimistic, switching to local execution at the start. Instead, we could try to be optimistic and go ahead with the parallel execution. If newer transactions that require remote connections cannot get any, they could be able to switch back to local execution.
Contributor guide
Assessment
This issue has not been assessed yet.