Grouped execution is disabled for the entire fragment when a FULL/RIGHT join has a remote (repartitioned) build side, even when the probe subtree is a colocated bucketed join

[kaikalur]

Summary

When a FULL OUTER JOIN (or RIGHT JOIN) has a build (right) side that arrives via a remote repartition exchange, the planner marks the whole fragment as not grouped-execution-capable. This is overly conservative: if the probe side is a colocated join over co-bucketed tables, that sub-join could run with grouped (dynamic-lifespan) execution while only the outer join crosses a shuffle. Today the presence of the outer join collapses the entire fragment to UNGROUPED_EXECUTION, giving up grouped execution over the already-bucketed scans.

How to reproduce

Using the Hive connector with three tables — two co-bucketed on the join key, one not co-bucketed:

CREATE TABLE b1 WITH (bucket_count = 16, bucketed_by = ARRAY['k']) AS
  SELECT orderkey AS k FROM tpch.tiny.orders;
CREATE TABLE b2 WITH (bucket_count = 16, bucketed_by = ARRAY['k']) AS
  SELECT orderkey AS k FROM tpch.tiny.orders;
CREATE TABLE t3_unbucketed AS
  SELECT orderkey AS k FROM tpch.tiny.orders;

Session properties: colocated_join = true, grouped_execution = true, join_distribution_type = PARTITIONED.

EXPLAIN (TYPE DISTRIBUTED)
SELECT count(*)
FROM b1 JOIN b2 ON b1.k = b2.k
FULL OUTER JOIN t3_unbucketed t3 ON b1.k = t3.k;

Observed behavior

The fragment holding (b1 JOIN b2) FULL JOIN t3 runs UNGROUPED_EXECUTION; every bucketed scan shows grouped = false. Only t3 is repartitioned (a single remote exchange into the bucketed partitioning). Condensed plan:

Fragment 1 [hive:buckets=16 ...]   Stage Execution Strategy: UNGROUPED_EXECUTION
  Aggregate(PARTIAL)
    FullJoin  (b1.k = t3.k)
      InnerJoin (b1.k = b2.k)
        Scan b1                 grouped = false
        LocalExchange[HASH] <- Scan b2   grouped = false
      LocalExchange[HASH]
      RemoteSource[2]  <──────── Fragment 2: Scan t3_unbucketed, repartitioned to hive:buckets=16 [k]

For contrast, when t3 is co-bucketed on k with the same bucket count, there is no remote source and the whole tree runs DYNAMIC_LIFESPAN_SCHEDULE_GROUPED_EXECUTION.

Expected behavior

b1 JOIN b2 is a colocated join over co-bucketed tables and could execute with grouped (dynamic-lifespan) execution. Only the FULL OUTER JOIN to the non-co-bucketed t3 requires a shuffle. Ideally the colocated sub-join stays in a grouped fragment and only the outer join crosses a shuffle boundary, instead of the whole fragment falling back to ungrouped execution.

Root cause

GroupedExecutionTagger.visitJoin (presto-main-base, com.facebook.presto.sql.planner.GroupedExecutionTagger) returns notCapable() for a RIGHT/FULL join whenever the right side is not currentNodeCapable (i.e. it comes from a remote exchange), because the LookupOuterOperator can only emit unmatched build rows after all probe input is seen, which conflicts with per-lifespan grouped execution. That verdict propagates up and makes the whole fragment ungrouped, discarding the capability of the colocated probe-side sub-join.

The code already carries a TODO for exactly this case:

// TODO:
// The RJoin can still execute as grouped if there is no subsequent operator that depends
// on the RJoin being executed in a grouped manner. However, this is not currently implemented.
// Support for this scenario is already implemented in the execution side.

Note: grouped_execution_when_capable does not help here — it gates on properties.isCurrentNodeCapable(), which is already false for this fragment, so the switch never fires (verified: the distributed plan is byte-for-byte identical with and without it).

Proposed direction

Split the outer join off so the colocated bucketed sub-join can run in its own grouped (dynamic-lifespan) fragment, and only the FULL/RIGHT outer join runs ungrouped past a shuffle boundary — rather than forcing the entire fragment ungrouped. The execution side reportedly already supports the grouped R/FULL-join lifecycle when nothing downstream depends on grouping; the missing piece is on the planner/fragmenter side.

Environment

• Presto: master (behavior is in GroupedExecutionTagger, connector-agnostic; reproduced with the Hive connector).