INSERT/SELECT with repartition triggering an out of disk error

**Settings**

PostgreSQL version: **14**

Citus version: **11.1**

Coordinator node: **4 vCores / 16 GiB RAM, 512 GiB storage**

Worker nodes: **4 nodes, 16 vCores / 512 GiB RAM, 4096 GiB storage**

**Problem description**

Running an INSERT/SELECT statement to fill a distributed table on schema B with data from another distributed table on schema A triggers an out of disk error, even though there's more than enough disk on workers to store the results of the query. The error can be seen next:

```
ERROR: could not write to file "base/pgsql_tmp/pgsql_tmp4110.60": No space left on device
CONTEXT: while executing command on :5432
```

**How to reproduce**

Execute the following SQL statements:

```
CREATE SCHEMA disk_source;

CREATE TABLE disk_source.requests (
id bigint,
metadata jsonb,
request_time timestamp without time zone NOT NULL,
department_id bigint NOT NULL
)
PARTITION BY RANGE (request_time);

SELECT create_time_partitions(
table_name := 'disk_source.requests',
partition_interval := '1 day',
start_from := now() - '4 years'::interval,
end_at := now()
);

SELECT create_distributed_table('disk_source.requests', 'department_id');
```

This will create the source table. Next, we must feed it with data. For this, create a VM geographically close to the cluster and setup a screen session to execute the next statement: (This will take some hours to finish and will create ~6TB of data, spread evenly on all 4 worker nodes)

```
INSERT INTO disk_source.requests
SELECT
generate_series(1, 1000000000)::bigint as id,
'{"guid":"049f2dcf-0046-48ea-b2dc-f593dcbad3d8","isActive":false,"balance":"$1,563.29","picture":"http://placehold.it/32x32","age":37,"eyeColor":"blue","name":"Eula Simmons","email":"eulasimmons@quility.com","phone":"+1 (909) 123123-3289","address":"Address test 123","about":"Lorem cillum pariatur esse ullamco fugiat officia eu nostrud ex nostrud.","registered":"2015-01-15T10:30:53 +02:00","latitude":-72.598788,"longitude":107.073131}' as metadata,
now() - justify_hours(random() * (interval '4 years')) as request_time,
floor(random() * 10000 + 1)::bigint as department_id;
```

After the transaction finishes, execute the next SQL statements:

```
CREATE SCHEMA disk_target;

CREATE TABLE disk_target.requests (
id bigint,
metadata jsonb,
request_time timestamp without time zone NOT NULL,
department_id bigint NOT NULL,
another_dist_id bigint NOT NULL
)
PARTITION BY RANGE (request_time);

SELECT create_time_partitions(
table_name := 'disk_target.requests',
partition_interval := '1 day',
start_from := now() - '4 years'::interval,
end_at := now()
);

SELECT create_distributed_table('disk_target.requests', 'another_dist_id');
```

Using another screen session inside the VM, execute the next statement:

```
INSERT INTO disk_target.requests
SELECT
dsr.*,
floor(random() * 10000 + 1)::bigint as another_dist_id
FROM disk_source.requests dsr;
```

It will take some hours to fail, and we believe the reason is the amount of storage used by the job cache directory, used to store intermediate results generated in a repartition operation. For this operation we've seen the directory disk usage grow to more than 2TB on each worker node.