The efficiency of query execution in a distributed DBMS is determined by how many operations can be executed on nodes that hold the actual data. For Shardman, a lot of effort is devoted to pushing down join operations. When the planner finds a relation that is accessible via a foreign data wrapper (FDW), it creates ForeignPath to access it. Later, when it examines a join of two relations and both of them are available via ForeignPath from the same foreign server, it can consider pushing down this join to the server and generating a so-called ForeignJoinPath. The planner can fail to do it if the join type is not supported, if filters attached to the relation should be applied locally, or if the relation scan result contains fields that cannot be evaluated on the remote server. An example of a currently unsupported join type is anti-join. Local filters attached to the relation should be applied locally when remote execution can lead to a different result or if the postgres_fdw module cannot create SQL expressions to apply some of the filters. An example of fields that cannot be evaluated on a remote server are attributes of semi-join inner relation that are not accessible via an outer relation. If the foreign_join_fast_path configuration parameter is set to on (which is the default value), the Shardman planner stops searching for other join strategies of two relations once it finds a foreign join possible for them. When the postgres_fdw.enforce_foreign_join configuration parameter is set to on (which is also the default), the cost of a foreign join is estimated so as to be always less than the cost of a local join.

When several sharded tables are joined on a sharding key, a partitionwise join can be possible. This means that instead of joining original tables, we can join their matching partitions. Partitionwise join currently applies only when the join conditions include all the partition keys, which must be of the same data type and have exactly matching sets of child partitions. Partitionwise join is crucial to the efficient query execution as it allows pushing down joins of table partitions. Evidently, to push down a join of several partitions, these partitions should reside on the same node. This is usually the case when sharded tables are created with the same num_parts parameter. However, for a rebalance process to move the corresponding partitions to the same nodes, sharded tables should be marked as colocated when created (see Section 7.1.1). Partitionwise join is enabled with the enable_partitionwise_join configuration parameter, which is turned on by default in Shardman.

When a sharded table is joined to a plain global table, asymmetric partitionwise join is possible. This means that instead of joining original tables, we can join each partition of the sharded table with the global table. This makes it possible to push down a join of sharded table partitions - with a global table to the foreign server.

After planning joins, the planner considers paths for post-join operations, such as aggregations, limiting, sorting and grouping. Not all such operations reach FDW pushdown logic. For example, currently partitioning efficiently prevents the LIMIT clause from being pushed down. There are two efficient strategies for executing aggregates on remote nodes. The first one is a partitionwise aggregation — when a GROUP BY clause includes a partitioning key, the aggregate can be pushed down together with the GROUP BY clause (this behavior is controlled by the enable_partitionwise_aggregate configuration parameter, which is turned on by default in Shardman). Alternatively, the planner can decide to execute partial aggregation on each partition of a sharded table and then combine the results. In Shardman, such a partial aggregate can be pushed down if the partial aggregate efficiently matches the main aggregate. For example, partial sum() aggregate can always be pushed down, but avg() cannot. Also the planner refuses pushing down partial aggregates if they contain additional clauses, such as ORDER BY or DISTINCT, or if the statement has the HAVING clause.

Generally, subqueries cannot be pushed down to other cluster nodes. However, Shardman uses two approaches to alleviate this limitation.

The first is subquery unnesting. In PostgreSQL, non-correlated subqueries can be transformed into semi-joins. In the following example, ANY subquery on non-partitioned tables is transformed to Hash Semi Join:

EXPLAIN (COSTS OFF) SELECT * FROM pgbench_branches WHERE bid = ANY (SELECT bid FROM pgbench_tellers);
                        QUERY PLAN                         
-----------------------------------------------------------
 Hash Semi Join
   Hash Cond: (pgbench_branches.bid = pgbench_tellers.bid)
   ->  Seq Scan on pgbench_branches
   ->  Hash
         ->  Seq Scan on pgbench_tellers

When optimize_correlated_subqueries is on (which is the default), Shardman planner also tries to convert correlated subqueries (i.e., subqueries that reference upper-level relations) into semi-joins. This optimization works for IN and = operators. The transformation has some restrictions. For example, it is not considered if a subquery contains aggregates or references upper-level relations from outside of a WHERE clause. This optimization allows transforming more complex subqueries into semi-joins, like in the following example:

EXPLAIN (COSTS OFF) SELECT * FROM pgbench_branches WHERE bid = ANY (SELECT bid FROM pgbench_tellers WHERE tbalance = bbalance);
                                                       QUERY PLAN                                                       
------------------------------------------------------------------------------------------------------------------------
 Hash Semi Join
   Hash Cond: ((pgbench_branches.bid = pgbench_tellers.bid) AND (pgbench_branches.bbalance = pgbench_tellers.tbalance))
   ->  Seq Scan on pgbench_branches
   ->  Hash
         ->  Seq Scan on pgbench_tellers
(5 rows)

After applying subquery unnesting, semi-join can be pushed down for execution to a remote node.

The second approach is to push down the entire subquery. This is possible when the optimizer has already figured out that the subquery references only partitions from the same foreign server as the upper-level query and corresponding foreign scans do not have local conditions. The optimization is controlled by postgres_fdw.subplan_pushdown (which is off by default). When a decision to push down a subquery is made by postgres_fdw, it has to deparse this subquery. A subquery that contains plan nodes for which deparsing is not implemented will not be pushed down. An example of a subquery pushdown looks as follows:

EXPLAIN (VERBOSE ON, COSTS OFF)
SELECT * FROM pgbench_accounts a WHERE a.bid=90 AND abalance =
    (SELECT min(tbalance) FROM pgbench_tellers t WHERE t.bid=90 and a.bid=t.bid);
                                     QUERY PLAN                                                                                                                   
--------------------------------------------------------------------------------------
 Foreign Scan on public.pgbench_accounts_5_fdw a
   Output: a.aid, a.bid, a.abalance, a.filler
   Remote SQL: SELECT aid, bid, abalance, filler FROM public.pgbench_accounts_5 r2 WHERE ((r2.bid = 90)) AND ((r2.abalance = ((SELECT min(sp0_2.tbalance) FROM public.pgbench_tellers_5 sp0_2 WHERE ((sp0_2.bid = 90)) AND ((r2.bid = 90))))))
   Transport: Silk
   SubPlan 1
     ->  Finalize Aggregate
           Output: min(t.tbalance)
           ->  Foreign Scan
                 Output: (PARTIAL min(t.tbalance))
                 Relations: Aggregate on (public.pgbench_tellers_5_fdw t)
                 Remote SQL: SELECT min(tbalance) FROM public.pgbench_tellers_5 WHERE ((bid = 90)) AND (($1::integer = 90))
                 Transport: Silk

Note that in the plan above there are no references to SubPlan 1.