Local (non-distributed) function in subquery is pushed down to workers and fails with "does not exist"

# Synposis

In our application, we have a function that computes aggregates of a distributed table. This function runs fine when invoked on its own, with the aggregation distributed across the cluster and collected by the coordinator.

We have a use case where this function is invoked in a subquery. In this case, the query fails with the error message shown below. It appears that Citus is attempting to push down the local (non distributed) function in the subquery to the workers, when it should have been run in the coordinator.

```
ERROR: function test.agg_on_distributed() does not exist
HINT: No function matches the given name and argument types. You might need to add explicit type casts.
CONTEXT: while executing command on 10.0.0.100:5432
```

(10.0.0.100 is the IP address of one of the worker nodes.)

# Environment

Cluster: Azure Database for PostgreSQL Hyperscale (Citus)
Nodes: 1 coordinator, 2 workers
PostgreSQL version: 11
Citus version: 9.5-1

# Reproducing the Issue

## Given

1. A reference table `reference`.
2. A distributed table `distributed`.
3. A function that computes an aggregate over the distributed table, `agg_on_distributed()`.

*The query used in this example of `agg_on_distributed()` is fairly straightforward and does not need to be put into a function. The query in the actual application is much more complex, involving dynamically generated SQL, so a function is needed. However, this simple example suffices to reproduce this issue.*

```sql
CREATE SCHEMA test;

-- Reference table
CREATE TABLE test.reference (
"label" text PRIMARY KEY,
"attr_1" boolean,
"attr2" integer);
SELECT create_reference_table('test.reference');

-- Distributed table
CREATE TABLE test.distributed (
"user" text,
"label" text REFERENCES test.reference ("label"),
PRIMARY KEY ("user", "label"));
SELECT create_distributed_table('test.distributed', 'user');

-- Not shown: Insert data into reference and distributed tables.

SELECT * FROM test.reference;

label | attr_1 | attr2
----------------------------+--------+-------
is_high_aa_activity | t | 20
is_young_adult | f | 10
is_not_repeat_user | t | 20
is_medium_aa_activity | t | 20
is_female | f | 10
is_10k_user | t | 20
is_7h_engager | t | 20
(7 rows)

SELECT * FROM test.distributed;

user | label
--------+----------------------------
2BR563 | is_10k_user
2BR563 | is_7h_engager
9SBEOM | is_10k_user
6OC6SV | is_not_repeat_user
6OC6SV | is_10k_user
09N2SY | is_medium_aa_activity
09N2SY | is_female
0AMZNY | is_young_adult
0AMZNY | is_high_aa_activity
0DPVOD | is_not_repeat_user
(10 rows)

-- Function that computes aggregates on the distributed table
CREATE OR REPLACE FUNCTION test.agg_on_distributed()
RETURNS TABLE ("label" text, "count" bigint)
STABLE
AS $$
BEGIN
-- Simple example shown here, but in the real application this is dynamic SQL.
RETURN QUERY
SELECT d."label", count(*) AS "count" FROM test.distributed d GROUP BY d."label";
END;
$$ language plpgsql;
```

## When invoked directly, Then it works

```sql
-- This works
SELECT * FROM test.agg_on_distributed();

label | count
----------------------------+-------
is_7h_engager | 1
is_not_repeat_user | 2
is_10k_user | 3
is_female | 1
is_high_aa_activity | 1
is_medium_aa_activity | 1
is_young_adult | 1
(7 rows)
```

## When invoked as part of a subquery, Then it fails

In our use case, the function is a subquery to a larger query that involves joins to reference tables.

```sql
-- This doesn't work
SELECT * FROM
(SELECT * FROM test.agg_on_distributed()) AS sub
INNER JOIN test.reference r ON (sub."label" = r."label");

ERROR: function test.agg_on_distributed() does not exist
HINT: No function matches the given name and argument types. You might need to add explicit type casts.
CONTEXT: while executing command on 10.0.0.100:5432

-- Get query plan to see what is happening
EXPLAIN SELECT * FROM
(SELECT * FROM test.agg_on_distributed()) AS sub
INNER JOIN test.reference r ON (sub."label" = r."label");

QUERY PLAN
----------------------------------------------------------------
Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=0 width=0)
Task Count: 1
Tasks Shown: All
-> Task
Error: Could not get remote plan.
(5 rows)
```

## When the aggregate query is hard coded in the subquery, Then it works

To troubleshoot, I hard coded the aggregate query into the subquery, and it works. So the problem only happens when the subquery is a function invocation, and not a `SELECT` statement.

```sql
-- Expected result
SELECT * FROM
(SELECT * FROM (
-- The exact same query as in the function.
SELECT d."label", count(*) AS "count" FROM test.distributed d GROUP BY d."label") AS q
) AS sub
INNER JOIN test.reference r ON (sub."label" = r."label");

label | count | label | attr_1 | attr2
----------------------------+-------+----------------------------+--------+-------
is_7h_engager | 1 | is_7h_engager | t | 20
is_not_repeat_user | 2 | is_not_repeat_user | t | 20
is_10k_user | 3 | is_10k_user | t | 20
is_female | 1 | is_female | f | 10
is_high_aa_activity | 1 | is_high_aa_activity | t | 20
is_medium_aa_activity | 1 | is_medium_aa_activity | t | 20
is_young_adult | 1 | is_young_adult | f | 10
(7 rows)

EXPLAIN SELECT * FROM
(SELECT * FROM (
-- The exact same query as in the function.
SELECT d."label", count(*) AS "count" FROM test.distributed d GROUP BY d."label") AS q
) AS sub
INNER JOIN test.reference r ON (sub."label" = r."label");

QUERY PLAN
------------------------------------------------------------------------------------------------------------------------
Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=0 width=0)
-> Distributed Subplan 8_1
-> HashAggregate (cost=500.00..503.50 rows=200 width=40)
Group Key: remote_scan.label
-> Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=100000 width=40)
Task Count: 32
Tasks Shown: One of 32
-> Task
Node: host=10.0.0.100 port=5432 dbname=citus
-> HashAggregate (cost=23.20..25.20 rows=200 width=40)
Group Key: label
-> Seq Scan on distributed_110264 d (cost=0.00..18.80 rows=880 width=32)
Task Count: 1
Tasks Shown: All
-> Task
Node: host=10.0.0.100 port=5432 dbname=citus
-> Hash Join (cost=38.13..50.76 rows=1000 width=77)
Hash Cond: (intermediate_result.label = r.label)
-> Function Scan on read_intermediate_result intermediate_result (cost=0.00..10.00 rows=1000 width=40)
-> Hash (cost=22.50..22.50 rows=1250 width=37)
-> Seq Scan on reference_110263 r (cost=0.00..22.50 rows=1250 width=37)
(21 rows)
```