[Enhancement] Support right-outer/semi/anti and full-outer range-colocate joins (#76040)

xiangguangyxg · claude · web-flow · commit fb3daac5ff90 · 2026-07-09T01:45:30.000Z
Signed-off-by: xiangguangyxg &lt;xiangguangyxg@gmail.com&gt;
Co-authored-by: Claude Opus 4.8 &lt;noreply@anthropic.com&gt;
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/ChildOutputPropertyGuarantor.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/ChildOutputPropertyGuarantor.java
@@ -106,9 +106,15 @@ public boolean canColocateJoin(HashDistributionSpec leftLocalDistributionSpec,
      *     {@code HINT_JOIN_SKEW}, or {@code HINT_JOIN_BUCKET} disables
      *     colocate. Only {@code null} or {@code HINT_JOIN_COLOCATE} lets
      *     the range fast path proceed.
-     * <li>Join-type allowlist: INNER, LEFT OUTER, LEFT SEMI, LEFT ANTI.
-     *     Right-family and full-outer joins are rejected (NULL-key runtime
-     *     correctness validation deferred past P2).
+     * <li>Join-type allowlist: INNER, LEFT/RIGHT OUTER, LEFT/RIGHT SEMI,
+     *     LEFT/RIGHT ANTI, FULL OUTER. All equi-join families are colocate-safe
+     *     because a colocate range join is bucket-local: rows sharing a colocate
+     *     key (NULL included — NULL sorts into the first ColocateRange on every
+     *     table in the group) always land in the same bucket, so the unmatched
+     *     right rows a right/full-outer join must emit are produced by the right
+     *     scan node in the same fragment. CROSS, NULL-aware anti, and ASOF joins
+     *     stay rejected (no covering equijoin key / cross-bucket NULL semantics /
+     *     inequality match).
      * <li>{@code disable_colocate_join} session kill switch (shared with
      *     hash colocate).
      * <li>Structural {@link RangeDistributionSpec#canColocate}: same group,
@@ -139,10 +145,15 @@ private boolean canRangeColocateJoin(String hint,
                 || HintNode.HINT_JOIN_BUCKET.equals(hint)) {
             return false;
         }
-        if (joinType != JoinOperator.INNER_JOIN
-                && joinType != JoinOperator.LEFT_OUTER_JOIN
-                && joinType != JoinOperator.LEFT_SEMI_JOIN
-                && joinType != JoinOperator.LEFT_ANTI_JOIN) {
+        boolean supportedJoinType = joinType == JoinOperator.INNER_JOIN
+                || joinType == JoinOperator.LEFT_OUTER_JOIN
+                || joinType == JoinOperator.LEFT_SEMI_JOIN
+                || joinType == JoinOperator.LEFT_ANTI_JOIN
+                || joinType == JoinOperator.RIGHT_OUTER_JOIN
+                || joinType == JoinOperator.RIGHT_SEMI_JOIN
+                || joinType == JoinOperator.RIGHT_ANTI_JOIN
+                || joinType == JoinOperator.FULL_OUTER_JOIN;
+        if (!supportedJoinType) {
             return false;
         }
         if (ConnectContext.get().getSessionVariable().isDisableColocateJoin()) {
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/OutputPropertyDeriver.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/OutputPropertyDeriver.java
@@ -352,14 +352,27 @@ private PhysicalPropertySet visitPhysicalJoin(PhysicalJoinOperator node, Express
         List<DistributionCol> leftOnPredicateColumns = joinHelper.getLeftCols();
         List<DistributionCol> rightOnPredicateColumns = joinHelper.getRightCols();
 
-        // Range-colocate join: both children are RangeDistributionSpec.
-        // ChildOutputPropertyGuarantor.canRangeColocateJoin already rejected
-        // right-family / full-outer joins, so the join type here is safe.
+        // Range-colocate join: both children are RangeDistributionSpec. The
+        // dominated output side mirrors the hash-colocate mapping in
+        // computeColocateJoinOutputProperty: right-family joins (right outer /
+        // semi / anti) output the right-side range spec, full-outer outputs the
+        // null-relaxed left-side spec (either side may be NULL-padded), and
+        // everything else (inner / left outer / left semi / left anti) outputs
+        // the left-side spec.
         DistributionSpec leftRawSpec = leftChildOutputProperty.getDistributionProperty().getSpec();
         DistributionSpec rightRawSpec = rightChildOutputProperty.getDistributionProperty().getSpec();
         if (leftRawSpec instanceof RangeDistributionSpec && rightRawSpec instanceof RangeDistributionSpec) {
-            // Left-dominated output (inner / left-outer / left-semi / left-anti).
-            PhysicalPropertySet outputProperty = createPropertySetByDistribution(leftRawSpec);
+            JoinOperator joinType = node.getJoinType();
+            DistributionSpec dominatedRangeSpec;
+            if (joinType.isRightJoin()) {
+                dominatedRangeSpec = rightRawSpec;
+            } else if (joinType.isFullOuterJoin()) {
+                RangeDistributionSpec leftRange = (RangeDistributionSpec) leftRawSpec;
+                dominatedRangeSpec = leftRange.getNullRelaxSpec(leftRange.getEquivalentDescriptor());
+            } else {
+                dominatedRangeSpec = leftRawSpec;
+            }
+            PhysicalPropertySet outputProperty = createPropertySetByDistribution(dominatedRangeSpec);
             return updateEquivalentDescriptor(node, outputProperty,
                     leftOnPredicateColumns, rightOnPredicateColumns);
         }
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/base/DistributionProperty.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/base/DistributionProperty.java
@@ -104,8 +104,11 @@ public DistributionProperty getNullStrictProperty() {
                         hashDistributionSpec.getEquivDesc()), isCTERequired);
             }
         }
-        // RangeDistributionSpec always builds colocate columns with
-        // nullStrict=true (see RangeDistributionSpec constructor); no conversion needed.
+        // Non-hash specs (gather / broadcast / any / range) need no conversion. A
+        // RangeDistributionSpec in particular is scan-local and never a required
+        // distribution property (appendEnforcers rejects it, and this method only runs
+        // on context.getRequiredProperty()), so a range spec never reaches here even
+        // though a full-outer range-colocate join can produce a null-relaxed one.
         return this;
     }
 
diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/base/RangeDistributionSpec.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/base/RangeDistributionSpec.java
@@ -187,10 +187,10 @@ private boolean isSatisfyHashShuffle(HashDistributionSpec hashSpec) {
     }
 
     /**
-     * Null-relaxed variant for symmetry with {@code HashDistributionSpec}.
-     * Not invoked by the P2 join planner because full-outer range colocate
-     * is rejected by the join-type allowlist; retained for parallelism and
-     * future phases.
+     * Null-relaxed variant (colocate columns rebuilt with {@code nullStrict=false}),
+     * mirroring {@code HashDistributionSpec.getNullRelaxSpec}. Used to derive the
+     * output distribution of a full-outer range-colocate join, where either side
+     * may be NULL-padded.
      */
     public RangeDistributionSpec getNullRelaxSpec(EquivalentDescriptor descriptor) {
         List<DistributionCol> relaxed = colocateColumns.stream()
diff --git a/fe/fe-core/src/test/java/com/starrocks/planner/RangeColocateMixedJoinPlanTest.java b/fe/fe-core/src/test/java/com/starrocks/planner/RangeColocateMixedJoinPlanTest.java
@@ -208,6 +208,46 @@ public void t12_hashLocalRangeShuffleHint() throws Exception {
                 "INNER JOIN (PARTITIONED)");
     }
 
+    // ---------- Range × range colocate: right-family / full-outer join types ----------
+
+    /** T13: range×range RIGHT OUTER on the colocate key must remain colocate. */
+    @Test
+    public void t13_rangeRightOuterRangeStillColocate() throws Exception {
+        String fragmentPlan = plan("select count(*) from cd a right outer join cd2 b on a.k1 = b.k1");
+        Assertions.assertTrue(fragmentPlan.contains("colocate: true"),
+                () -> "range×range RIGHT OUTER on colocate key must remain colocate, plan was:\n" + fragmentPlan);
+    }
+
+    /** T14: range×range FULL OUTER on the colocate key must remain colocate. */
+    @Test
+    public void t14_rangeFullOuterRangeStillColocate() throws Exception {
+        String fragmentPlan = plan("select count(*) from cd a full outer join cd2 b on a.k1 = b.k1");
+        Assertions.assertTrue(fragmentPlan.contains("colocate: true"),
+                () -> "range×range FULL OUTER on colocate key must remain colocate, plan was:\n" + fragmentPlan);
+    }
+
+    /** T15: [shuffle] hint defeats range×range RIGHT OUTER colocate. */
+    @Test
+    public void t15_rangeRightOuterRangeShuffleHint() throws Exception {
+        String fragmentPlan = plan("select count(*) from cd a right outer join [shuffle] cd2 b on a.k1 = b.k1");
+        Assertions.assertFalse(fragmentPlan.contains("colocate: true"),
+                () -> "shuffle hint must defeat range×range RIGHT OUTER colocate, plan was:\n" + fragmentPlan);
+    }
+
+    /** T16: disable_colocate_join defeats range×range FULL OUTER colocate. */
+    @Test
+    public void t16_rangeFullOuterRangeDisableColocate() throws Exception {
+        Deencapsulation.setField(connectContext.getSessionVariable(), "disableColocateJoin", true);
+        try {
+            String fragmentPlan = plan("select count(*) from cd a full outer join cd2 b on a.k1 = b.k1");
+            Assertions.assertFalse(fragmentPlan.contains("colocate: true"),
+                    () -> "disable_colocate_join must defeat range×range FULL OUTER colocate, plan was:\n"
+                            + fragmentPlan);
+        } finally {
+            Deencapsulation.setField(connectContext.getSessionVariable(), "disableColocateJoin", false);
+        }
+    }
+
     // ---------- Optimizer-level invariant ----------
 
     /**
diff --git a/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/ChildOutputPropertyGuarantorRangeTest.java b/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/ChildOutputPropertyGuarantorRangeTest.java
@@ -40,8 +40,9 @@
  *
  * <p>Covers:
  * <ul>
- * <li>Join-type allowlist (INNER, LEFT OUTER, LEFT SEMI, LEFT ANTI) —
- *     RIGHT / FULL rejected.
+ * <li>Join-type allowlist accepts all equi-join families (INNER, LEFT/RIGHT
+ *     OUTER, LEFT/RIGHT SEMI, LEFT/RIGHT ANTI, FULL OUTER); CROSS and NULL-aware
+ *     left anti stay rejected.
  * <li>Position-preserving pairing accepts identity join keys.
  * <li>Position-preserving pairing rejects swapped-column joins.
  * <li>Partial shuffle coverage rejected.
@@ -92,13 +93,81 @@ private static boolean invokeGate(ChildOutputPropertyGuarantor guarantor,
                 hint, joinType, left, right, leftShuffle, rightShuffle);
     }
 
+    /**
+     * Installs the "same group, stable, non-empty" expectations shared by every
+     * accepts-* test: colocate is not session-disabled, both tables are in the
+     * same stable colocate group. Group ids 100L/200L match {@link #buildSide}.
+     */
+    private static void expectStableSameGroup(ConnectContext connectContext,
+                                              SessionVariable sessionVariable,
+                                              GlobalStateMgr globalStateMgr,
+                                              ColocateTableIndex colocateTableIndex,
+                                              ColocateTableIndex.GroupId groupId) {
+        new Expectations() {
+            {
+                ConnectContext.get();
+                result = connectContext;
+                connectContext.getSessionVariable();
+                result = sessionVariable;
+                sessionVariable.isDisableColocateJoin();
+                result = false;
+                GlobalStateMgr.getCurrentState();
+                result = globalStateMgr;
+                globalStateMgr.getColocateTableIndex();
+                result = colocateTableIndex;
+                colocateTableIndex.isSameGroup(100L, 200L);
+                result = true;
+                colocateTableIndex.getGroup(100L);
+                result = groupId;
+                colocateTableIndex.getGroup(200L);
+                result = groupId;
+                colocateTableIndex.isGroupUnstable(groupId);
+                result = false;
+            }
+        };
+    }
+
+    /**
+     * Asserts an identity-key join of {@code joinType} on a stable same-group
+     * range-colocate pair is accepted (colocate = (1,2), shuffle = (1,2)).
+     */
+    private void assertJoinTypeAccepted(TaskContext taskContext, ConnectContext connectContext,
+                                        SessionVariable sessionVariable, GlobalStateMgr globalStateMgr,
+                                        ColocateTableIndex colocateTableIndex, JoinOperator joinType) {
+        ColocateTableIndex.GroupId groupId = new ColocateTableIndex.GroupId(1L, 1L);
+        expectStableSameGroup(connectContext, sessionVariable, globalStateMgr, colocateTableIndex, groupId);
+        assertTrue(invokeGate(newGuarantor(taskContext), joinType,
+                buildSide(100L, 1, 2), buildSide(200L, 1, 2), cols(1, 2), cols(1, 2)));
+    }
+
     @Test
-    void rejectsRightOuterJoin(@Mocked TaskContext taskContext) {
-        // Join-type allowlist is the earliest gate; no ConnectContext access.
-        RangeDistributionSpec left = buildSide(100L, 1, 2);
-        RangeDistributionSpec right = buildSide(200L, 1, 2);
-        assertFalse(invokeGate(newGuarantor(taskContext), JoinOperator.RIGHT_OUTER_JOIN,
-                left, right, cols(1, 2), cols(1, 2)));
+    void acceptsRightOuterJoin(@Mocked TaskContext taskContext,
+                               @Mocked ConnectContext connectContext,
+                               @Mocked SessionVariable sessionVariable,
+                               @Mocked GlobalStateMgr globalStateMgr,
+                               @Mocked ColocateTableIndex colocateTableIndex) {
+        assertJoinTypeAccepted(taskContext, connectContext, sessionVariable, globalStateMgr,
+                colocateTableIndex, JoinOperator.RIGHT_OUTER_JOIN);
+    }
+
+    @Test
+    void acceptsRightSemiJoin(@Mocked TaskContext taskContext,
+                              @Mocked ConnectContext connectContext,
+                              @Mocked SessionVariable sessionVariable,
+                              @Mocked GlobalStateMgr globalStateMgr,
+                              @Mocked ColocateTableIndex colocateTableIndex) {
+        assertJoinTypeAccepted(taskContext, connectContext, sessionVariable, globalStateMgr,
+                colocateTableIndex, JoinOperator.RIGHT_SEMI_JOIN);
+    }
+
+    @Test
+    void acceptsRightAntiJoin(@Mocked TaskContext taskContext,
+                              @Mocked ConnectContext connectContext,
+                              @Mocked SessionVariable sessionVariable,
+                              @Mocked GlobalStateMgr globalStateMgr,
+                              @Mocked ColocateTableIndex colocateTableIndex) {
+        assertJoinTypeAccepted(taskContext, connectContext, sessionVariable, globalStateMgr,
+                colocateTableIndex, JoinOperator.RIGHT_ANTI_JOIN);
     }
 
     @Test
@@ -131,11 +200,33 @@ void rejectsBucketHint(@Mocked TaskContext taskContext) {
     }
 
     @Test
-    void rejectsFullOuterJoin(@Mocked TaskContext taskContext) {
-        // Join-type allowlist is the earliest gate; no ConnectContext access.
+    void acceptsFullOuterJoin(@Mocked TaskContext taskContext,
+                              @Mocked ConnectContext connectContext,
+                              @Mocked SessionVariable sessionVariable,
+                              @Mocked GlobalStateMgr globalStateMgr,
+                              @Mocked ColocateTableIndex colocateTableIndex) {
+        assertJoinTypeAccepted(taskContext, connectContext, sessionVariable, globalStateMgr,
+                colocateTableIndex, JoinOperator.FULL_OUTER_JOIN);
+    }
+
+    @Test
+    void rejectsCrossJoin(@Mocked TaskContext taskContext) {
+        // CROSS join stays rejected by the allowlist (earliest gate after hint;
+        // no ConnectContext access). It has no covering equijoin key anyway.
+        RangeDistributionSpec left = buildSide(100L, 1, 2);
+        RangeDistributionSpec right = buildSide(200L, 1, 2);
+        assertFalse(invokeGate(newGuarantor(taskContext), JoinOperator.CROSS_JOIN,
+                left, right, cols(1, 2), cols(1, 2)));
+    }
+
+    @Test
+    void rejectsNullAwareLeftAntiJoin(@Mocked TaskContext taskContext) {
+        // NULL-aware left anti (NOT IN with nullable) needs cross-bucket NULL
+        // knowledge, which bucket-local colocate execution cannot provide, so it
+        // stays rejected by the allowlist.
         RangeDistributionSpec left = buildSide(100L, 1, 2);
         RangeDistributionSpec right = buildSide(200L, 1, 2);
-        assertFalse(invokeGate(newGuarantor(taskContext), JoinOperator.FULL_OUTER_JOIN,
+        assertFalse(invokeGate(newGuarantor(taskContext), JoinOperator.NULL_AWARE_LEFT_ANTI_JOIN,
                 left, right, cols(1, 2), cols(1, 2)));
     }
 
diff --git a/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/OutputPropertyDeriverRangeTest.java b/fe/fe-core/src/test/java/com/starrocks/sql/optimizer/OutputPropertyDeriverRangeTest.java

Original file line number	Diff line number	Diff line change
`@@ -104,8 +104,11 @@ public DistributionProperty getNullStrictProperty() {`
`104`	`104`	`hashDistributionSpec.getEquivDesc()), isCTERequired);`
`105`	`105`	`}`
`106`	`106`	`}`
`107`		`- // RangeDistributionSpec always builds colocate columns with`
`108`		`- // nullStrict=true (see RangeDistributionSpec constructor); no conversion needed.`
	`107`	`+ // Non-hash specs (gather / broadcast / any / range) need no conversion. A`
	`108`	`+ // RangeDistributionSpec in particular is scan-local and never a required`
	`109`	`+ // distribution property (appendEnforcers rejects it, and this method only runs`
	`110`	`+ // on context.getRequiredProperty()), so a range spec never reaches here even`
	`111`	`+ // though a full-outer range-colocate join can produce a null-relaxed one.`
`109`	`112`	`return this;`
`110`	`113`	`}`
`111`	`114`