[improvement](fe) Convert inner join to semi join when right side is an existence filter

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java

@@ -56,6 +56,7 @@
 import org.apache.doris.nereids.rules.rewrite.CollectPredicateOnScan;
 import org.apache.doris.nereids.rules.rewrite.ColumnPruning;
 import org.apache.doris.nereids.rules.rewrite.ConstantPropagation;
+import org.apache.doris.nereids.rules.rewrite.ConvertInnerJoinToSemiJoin;
 import org.apache.doris.nereids.rules.rewrite.ConvertInnerOrCrossJoin;
 import org.apache.doris.nereids.rules.rewrite.ConvertOuterJoinToAntiJoin;
 import org.apache.doris.nereids.rules.rewrite.CountDistinctRewrite;
@@ -670,7 +671,8 @@ public class Rewriter extends AbstractBatchJobExecutor {
                 topic("eliminate join according unique or foreign key",
                     cascadesContext -> cascadesContext.rewritePlanContainsTypes(LogicalJoin.class),
                     bottomUp(new EliminateJoinByFK()),
-                    topDown(new EliminateJoinByUnique())
+                    topDown(new EliminateJoinByUnique()),
+                    topDown(new ConvertInnerJoinToSemiJoin())
                 ),
                 topic("join skew salting rewrite",
                         topDown(new SaltJoin())),

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java

@@ -306,6 +306,7 @@ public enum RuleType {
     ELIMINATE_GROUP_BY(RuleTypeClass.REWRITE),
     ELIMINATE_JOIN_BY_UK(RuleTypeClass.REWRITE),
     ELIMINATE_JOIN_BY_FK(RuleTypeClass.REWRITE),
+    CONVERT_INNER_JOIN_TO_SEMI_JOIN(RuleTypeClass.REWRITE),
     ELIMINATE_GROUP_BY_KEY(RuleTypeClass.REWRITE),
     ELIMINATE_GROUP_BY_KEY_BY_UNIFORM(RuleTypeClass.REWRITE),
     ELIMINATE_ORDER_BY_KEY(RuleTypeClass.REWRITE),

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/ConvertInnerJoinToSemiJoin.java

@@ -0,0 +1,130 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.Project;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Convert an inner join to a left semi join when the inner join is only used as an
+ * existence filter. Three conditions must be satisfied at the same time:
+ *
+ * 1. The right side columns of the join do not leak: every column referenced above the
+ *    join comes from the left side, i.e. the right side is only used in the join
+ *    conditions. (the "existence filter" property)
+ * 2. All join conditions are equal conjuncts: hashJoinConjuncts is not empty and
+ *    otherJoinConjuncts is empty, so the join is a pure equi-join.
+ * 3. There is a deduplication guarantee above the join: the aggregate that consumes the
+ *    join output is a DISTINCT-like aggregate, i.e. its group-by keys cover exactly its
+ *    output columns. Otherwise, in bag semantics, the row multiplication of an inner
+ *    join (a left row matching N right rows produces N copies) would change the result
+ *    after the conversion, because a semi join never multiplies rows.
+ *
+ * Example:
+ * <pre>
+ *   select distinct a1.* from a1, a5
+ *   where a1.lot_id = a5.lot_id and a1.ope_no = a5.ope_no and ...
+ *   ======>
+ *   select distinct a1.* from a1 left semi join a5
+ *   on a1.lot_id = a5.lot_id and a1.ope_no = a5.ope_no and ...
+ * </pre>
+ *
+ * The conversion avoids row multiplication (the output row count stays the left side
+ * cardinality instead of being multiplied by the average number of right side matches),
+ * and lets the right side be scanned/broadcast with only the join key columns.
+ */
+public class ConvertInnerJoinToSemiJoin implements RewriteRuleFactory {
+    @Override
+    public List<Rule> buildRules() {
+        return ImmutableList.of(
+                // Aggregate -> InnerJoin
+                logicalAggregate(innerLogicalJoin()
+                        .when(this::canConvertToSemiJoin))
+                        .when(this::isDistinctLikeAggregate)
+                        .thenApply(ctx -> convert(ctx.root, ctx.root.child()))
+                        .toRule(RuleType.CONVERT_INNER_JOIN_TO_SEMI_JOIN),
+                // Aggregate -> Project -> InnerJoin, where the project is a pure slot projection
+                logicalAggregate(logicalProject(innerLogicalJoin()
+                        .when(this::canConvertToSemiJoin))
+                        .when(Project::isAllSlots))
+                        .when(this::isDistinctLikeAggregate)
+                        .thenApply(ctx -> convert(ctx.root, ctx.root.child(), ctx.root.child().child()))
+                        .toRule(RuleType.CONVERT_INNER_JOIN_TO_SEMI_JOIN)
+        );
+    }
+
+    /**
+     * Condition 2: the join is a pure equi-join (hash conjuncts exist and no other
+     * conjuncts), and it is not a mark join.
+     */
+    private boolean canConvertToSemiJoin(LogicalJoin<?, ?> join) {
+        return !join.isMarkJoin()
+                && !join.getHashJoinConjuncts().isEmpty()
+                && join.getOtherJoinConjuncts().isEmpty();
+    }
+
+    /**
+     * Condition 3: the aggregate is a DISTINCT-like aggregate, i.e. its group-by keys
+     * cover exactly its output columns, so it collapses duplicate rows and the row
+     * multiplicity change of inner-join -> semi-join does not affect the final result.
+     */
+    private boolean isDistinctLikeAggregate(LogicalAggregate<?> agg) {
+        Set<ExprId> groupBySlotIds = agg.getGroupByExpressions().stream()
+                .filter(Slot.class::isInstance)
+                .map(expr -> ((Slot) expr).getExprId())
+                .collect(Collectors.toSet());
+        Set<ExprId> outputSlotIds = agg.getOutput().stream()
+                .map(Slot::getExprId)
+                .collect(Collectors.toSet());
+        return groupBySlotIds.equals(outputSlotIds);
+    }
+
+    /** Aggregate -> Join */
+    private Plan convert(LogicalAggregate<?> agg, LogicalJoin<?, ?> join) {
+        // Condition 1: the right side columns do not leak above the join.
+        if (!join.left().getOutputSet().containsAll(agg.getInputSlots())) {
+            return agg;
+        }
+        return agg.withChildren(join.withJoinType(JoinType.LEFT_SEMI_JOIN));
+    }
+
+    /** Aggregate -> Project -> Join */
+    private Plan convert(LogicalAggregate<?> agg, LogicalProject<?> project, LogicalJoin<?, ?> join) {
+        // Condition 1: the right side columns do not leak above the join.
+        // The project is a pure slot projection, so checking the project's input slots
+        // covers every column consumed above the join.
+        if (!join.left().getOutputSet().containsAll(project.getInputSlots())) {
+            return agg;
+        }
+        return agg.withChildren(project.withChildren(join.withJoinType(JoinType.LEFT_SEMI_JOIN)));
+    }
+}

fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/ConvertInnerJoinToSemiJoinTest.java

@@ -0,0 +1,119 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite;
+
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Test;
+
+class ConvertInnerJoinToSemiJoinTest extends TestWithFeService implements MemoPatternMatchSupported {
+    @Override
+    protected void runBeforeAll() throws Exception {
+        createDatabase("test");
+        connectContext.setDatabase("test");
+        connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
+        createTables(
+                "CREATE TABLE IF NOT EXISTS t1 (\n"
+                        + "    id1 int not null,\n"
+                        + "    v1 int not null\n"
+                        + ")\n"
+                        + "DUPLICATE KEY(id1)\n"
+                        + "DISTRIBUTED BY HASH(id1) BUCKETS 10\n"
+                        + "PROPERTIES (\"replication_num\" = \"1\")\n",
+                "CREATE TABLE IF NOT EXISTS t2 (\n"
+                        + "    id2 int not null,\n"
+                        + "    v2 int not null\n"
+                        + ")\n"
+                        + "DUPLICATE KEY(id2)\n"
+                        + "DISTRIBUTED BY HASH(id2) BUCKETS 10\n"
+                        + "PROPERTIES (\"replication_num\" = \"1\")\n");
+    }
+
+    // select distinct t1.id1 from t1 join t2 on t1.id1 = t2.id2
+    // t2's columns are not used above the join, the join keys are equal conjuncts,
+    // and the DISTINCT aggregate dedups the output:
+    // inner join -> left semi join
+    @Test
+    void testConvertInnerJoinToSemiJoin() throws Exception {
+        PlanChecker.from(connectContext)
+                .analyze("select distinct t1.id1 from t1 join t2 on t1.id1 = t2.id2")
+                .rewrite()
+                .anyMatches(logicalJoin().when(j -> j.getJoinType() == JoinType.LEFT_SEMI_JOIN))
+                .nonMatch(logicalJoin().when(j -> j.getJoinType() == JoinType.INNER_JOIN))
+                .printlnTree();
+    }
+
+    // t2.id2 is projected above the join, so the right side columns leak:
+    // keep inner join
+    @Test
+    void testNotConvertWhenRightSideColumnsUsed() throws Exception {
+        PlanChecker.from(connectContext)
+                .analyze("select distinct t1.id1, t2.id2 from t1 join t2 on t1.id1 = t2.id2")
+                .rewrite()
+                .nonMatch(logicalJoin().when(j -> j.getJoinType() == JoinType.LEFT_SEMI_JOIN))
+                .printlnTree();
+    }
+
+    // no DISTINCT (or group-by) dedup guarantee above the join:
+    // in bag semantics inner join row multiplication matters, keep inner join
+    @Test
+    void testNotConvertWithoutDistinct() throws Exception {
+        PlanChecker.from(connectContext)
+                .analyze("select t1.id1 from t1 join t2 on t1.id1 = t2.id2")
+                .rewrite()
+                .nonMatch(logicalJoin().when(j -> j.getJoinType() == JoinType.LEFT_SEMI_JOIN))
+                .printlnTree();
+    }
+
+    // non-equi join condition goes into otherJoinConjuncts:
+    // not a pure equi-join, keep inner join
+    @Test
+    void testNotConvertWithNonEquiCondition() throws Exception {
+        PlanChecker.from(connectContext)
+                .analyze("select distinct t1.id1 from t1 join t2 on t1.id1 > t2.id2")
+                .rewrite()
+                .nonMatch(logicalJoin().when(j -> j.getJoinType() == JoinType.LEFT_SEMI_JOIN))
+                .printlnTree();
+    }
+
+    // aggregation with aggregate functions must NOT be converted:
+    // count(*) observes the input row multiplicity, changing it would be wrong
+    @Test
+    void testNotConvertWithAggregateFunction() throws Exception {
+        PlanChecker.from(connectContext)
+                .analyze("select t1.id1, count(*) from t1 join t2 on t1.id1 = t2.id2 group by t1.id1")
+                .rewrite()
+                .nonMatch(logicalJoin().when(j -> j.getJoinType() == JoinType.LEFT_SEMI_JOIN))
+                .printlnTree();
+    }
+
+    // <=> (NullSafeEqual) is an EqualPredicate: FindHashConditionForJoin extracts it into
+    // hashJoinConjuncts, so the inner join can also be converted to left semi join
+    @Test
+    void testConvertWithNullSafeEqual() throws Exception {
+        PlanChecker.from(connectContext)
+                .analyze("select distinct t1.id1 from t1 join t2 on t1.id1 <=> t2.id2")
+                .rewrite()
+                .anyMatches(logicalJoin().when(j -> j.getJoinType() == JoinType.LEFT_SEMI_JOIN))
+                .nonMatch(logicalJoin().when(j -> j.getJoinType() == JoinType.INNER_JOIN))
+                .printlnTree();
+    }
+}