Limitations of Local Execution

In #2938 we've implemented the local exection logic. The basic idea is that if a query hits a worker node where the shard that the query hits reside, instead of connecting the localhost and do the execution over a connection, simlpy do the execution in the same session. This helps to avoid using extra connections, which are scrase resources in Postgres.

The local execution is especially very useful in the case of procedure/function pushdown. THe idea is that the coordinator delegates the function/procedure to the worker node where the data that the function touches reside. All the commands in the function avoids round-trips to localhost, simply executing everyting within the same session that does the planning.

This is valid as long as the node that is executing the query has both the metadata and the data. This is clearly the case for Citus MX. We're planning to expand this in two new ways (1) as we plan to replicate reference tables to the coordinator, we'd have both the data and the metadata for the reference tables on the coordinator (2) if a distributed query ends-up only with intermediate results, we can execute the final query on the coordinator locally, instead of going to a one of the worker nodes.

Given that we're planning to expand the usage, we should be awere of the limitations. All of the limitations are trade-offs for simple implementation.
Below, I'm listing the limitations that I remember:

- [ ] Performance within transaction blocks

If the queries inside a transaction block/function hits a single set of colocated shards, then the performance is great.

But, if we have something like the following, (or the same transaction block where the query starts with a modification to reference tables), the performance might not be great

```SQL
BEGIN;

-- hits a single shard, and we go through the local execution
DELETE FROM users_table WHERE user_id = 1;
LOG: executing the command locally: DELETE FROM public.users_table_102014 users_table WHERE (user_id OPERATOR(pg_catalog.=) 5)

-- hits multi shard. Thus, all the queries on the same worker needs to go through locally, preventing
-- parallelism.
SELECT count(*) FROM users_table;
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102008 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102011 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102014 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102017 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102020 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102023 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102026 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102029 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102032 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102035 users_table WHERE true
LOG: executing the command locally: SELECT count(*) AS count FROM users_table_102038 users_table WHERE true
..
DETAIL: on server onderkalaci@localhost:9701
LOG: issuing SELECT count(*) AS count FROM users_table_102028 users_table WHERE true
DETAIL: on server onderkalaci@localhost:9702
LOG: issuing SELECT count(*) AS count FROM users_table_102027 users_table WHERE true
DETAIL: on server onderkalaci@localhost:9701
LOG: issuing SELECT count(*) AS count FROM users_table_102034 users_table WHERE true
DETAIL: on server onderkalaci@localhost:9702
LOG: issuing SELECT count(*) AS count FROM users_table_102033 users_table WHERE true
DETAIL: on server onderkalaci@localhost:9701
LOG: issuing SELECT count(*) AS count FROM users_table_102036 users_table WHERE true
DETAIL: on server onderkalaci@localhost:9701

```

- [x] Unsupported features: COPY/DDL/TRUNCATE/create_distributed_table

Citus currently doesn't know how to handle COPY, DDLs and INSERT .. SELECT via coordinator locally. So, if a transaction block starts with a local execution, and then switches to one of the above, we'd error out.

```SQL
-- multi-shard queries do not switch to the sequential mode if they are
-- there is not local execution has happened in the transaction block.
BEGIN;
DELETE FROM users_table WHERE user_id WHERE user_id = 2;
LOG: executing the command locally: DELETE FROM public.users_table_102032 users_table WHERE (user_id OPERATOR(pg_catalog.=) 2)
copy users_table FROM '/tmp/data' WITH CSV;
ERROR: cannot execute command because a local execution has already been done in the transaction
DETAIL: Some parallel commands cannot be executed if a previous command has already been executed locally
HINT: Try re-running the transaction with "SET LOCAL citus.enable_local_execution TO OFF;"
```

- [ ] EXPLAIN

Explain doesn't give a clue about local execution:

```SQL
DELETE FROM users_table WHERE user_id = 2;
LOG: executing the command locally: DELETE FROM public.users_table_102032 users_table WHERE (user_id OPERATOR(pg_catalog.=) 2)

-- same query with explain
EXPLAIN DELETE FROM users_table WHERE user_id = 2;
│ Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=0 width=0) │
│ Task Count: 1 │
│ Tasks Shown: All │
│ -> Task │
│ Node: host=localhost port=9700 dbname=postgres │
│ -> Delete on users_table_102032 users_table (cost=0.00..25.88 rows=6 width=6) │
│ -> Seq Scan on users_table_102032 users_table (cost=0.00..25.88 rows=6 width=6) │
│ Filter: (user_id = 2) │
└─────────────────────────────────────────────────
```

- [ ] EXPLAIN ANALYZE

Explain analyze is not supported after local query execution in a transaction block

```SQL
BEGIN;

-- local execution
DELETE FROM users_table WHERE user_id = 2

EXPLAIN ANALYZE DELETE FROM users_table WHERE user_id = 2;
ERROR: cannot execute command because a local execution has already been done in the transaction
DETAIL: Some parallel commands cannot be executed if a previous command has already been executed locally
```

- [ ] Task assignment policy is ignored by local execution: Even if the task is assigned to `localhost:9701`, the task is executed locally (node executing the query is `localhost:9700`)

```SQL

SET client_min_messages TO DEBUG4;
set citus.log_remote_commands TO ON;
set citus.task_assignment_policy TO "round-robin";

SELECT count(*) FROM users_table_ref ;
DEBUG: Distributed planning for a fast-path router query
DEBUG: assigned task 0 to node localhost:9701
DEBUG: Creating router plan
DEBUG: Plan is router executable
LOG: executing the command locally: SELECT count(*) AS count FROM public.users_table_ref_102072 users_table_ref
```

- [x] #3084 The bug around composite types & prepared statements