[Feat](Nereids) Add leading and ordered hint (#22057)

Add leading hint and ordered hint. Usage:
select /*+ ordered / * from a join b on xxx; which will limit join order to original order
select /+ leading ({b a}) */ from a join b on xxx; which will change join order to b join a.

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4

@@ -232,6 +232,7 @@ LAST: 'LAST';
 LATERAL: 'LATERAL';
 LAZY: 'LAZY';
 LEADING: 'LEADING';
+LEFT_BRACE: '{';
 LEFT: 'LEFT';
 LIKE: 'LIKE';
 ILIKE: 'ILIKE';
@@ -272,6 +273,7 @@ OPTION: 'OPTION';
 OPTIONS: 'OPTIONS';
 OR: 'OR';
 ORDER: 'ORDER';
+ORDERED: 'ORDERED';
 OUT: 'OUT';
 OUTER: 'OUTER';
 OUTFILE: 'OUTFILE';
@@ -317,6 +319,7 @@ RESTRICT: 'RESTRICT';
 RESTRICTIVE: 'RESTRICTIVE';
 REVOKE: 'REVOKE';
 REWRITTEN: 'REWRITTEN';
+RIGHT_BRACE: '}';
 RIGHT: 'RIGHT';
 // original optimizer only support REGEXP, the new optimizer should be consistent with it
 RLIKE: 'RLIKE';
@@ -442,6 +445,11 @@ STRING_LITERAL
     | 'R"'(~'"')* '"'
     ;
 
+LEADING_STRING
+    : LEFT_BRACE
+    | RIGHT_BRACE
+    ;
+
 BIGINT_LITERAL
     : DIGIT+ 'L'
     ;

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -65,6 +65,7 @@ propertiesStatment
 identifierOrText
     : errorCapturingIdentifier
     | STRING_LITERAL
+    | LEADING_STRING
     ;
 
 userIdentify
@@ -199,7 +200,7 @@ havingClause
 selectHint: HINT_START hintStatements+=hintStatement (COMMA? hintStatements+=hintStatement)* HINT_END;
 
 hintStatement
-    : hintName=identifier LEFT_PAREN parameters+=hintAssignment (COMMA parameters+=hintAssignment)* RIGHT_PAREN
+    : hintName=identifier (LEFT_PAREN parameters+=hintAssignment (COMMA? parameters+=hintAssignment)* RIGHT_PAREN)?
     ;
 
 hintAssignment
@@ -645,6 +646,7 @@ nonReserved
     | LAST
     | LAZY
     | LEADING
+    | LEFT_BRACE
     | LIKE
     | ILIKE
     | LIMIT
@@ -676,6 +678,7 @@ nonReserved
     | OPTIONS
     | OR
     | ORDER
+    | ORDERED
     | OUT
     | OUTER
     | OUTPUTFORMAT
@@ -718,6 +721,7 @@ nonReserved
     | RESTRICTIVE
     | REVOKE
     | REWRITTEN
+    | RIGHT_BRACE
     | RLIKE
     | ROLE
     | ROLES

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

@@ -29,6 +29,7 @@ import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.glue.LogicalPlanAdapter;
 import org.apache.doris.nereids.glue.translator.PhysicalPlanTranslator;
 import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
+import org.apache.doris.nereids.hint.Hint;
 import org.apache.doris.nereids.jobs.executor.Optimizer;
 import org.apache.doris.nereids.jobs.executor.Rewriter;
 import org.apache.doris.nereids.memo.Group;
@@ -69,6 +70,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -347,26 +349,61 @@ public class NereidsPlanner extends Planner {
         return executor;
     }
 
+    /**
+     * getting hints explain string, which specified by enumerate and show in lists
+     * @param hintMap hint map recorded in statement context
+     * @return explain string shows using of hint
+     */
+    public String getHintExplainString(Map<String, Hint> hintMap) {
+        String used = "";
+        String unUsed = "";
+        String syntaxError = "";
+        for (Map.Entry<String, Hint> entry : hintMap.entrySet()) {
+            switch (entry.getValue().getStatus()) {
+                case UNUSED:
+                    unUsed = unUsed + " " + entry.getValue().getExplainString();
+                    break;
+                case SYNTAX_ERROR:
+                    syntaxError = syntaxError + " " + entry.getValue().getExplainString()
+                        + " Msg:" + entry.getValue().getErrorMessage();
+                    break;
+                case SUCCESS:
+                    used = used + " " + entry.getValue().getExplainString();
+                    break;
+                default:
+                    break;
+            }
+        }
+        return "\nUsed:" + used + "\nUnUsed:" + unUsed + "\nSyntaxError:" + syntaxError;
+    }
+
     @Override
     public String getExplainString(ExplainOptions explainOptions) {
         ExplainLevel explainLevel = getExplainLevel(explainOptions);
+        String plan = "";
         switch (explainLevel) {
             case PARSED_PLAN:
-                return parsedPlan.treeString();
+                plan = parsedPlan.treeString();
+                break;
             case ANALYZED_PLAN:
-                return analyzedPlan.treeString();
+                plan = analyzedPlan.treeString();
+                break;
             case REWRITTEN_PLAN:
-                return rewrittenPlan.treeString();
+                plan = rewrittenPlan.treeString();
+                break;
             case OPTIMIZED_PLAN:
-                return "cost = " + cost + "\n" + optimizedPlan.treeString();
+                plan = "cost = " + cost + "\n" + optimizedPlan.treeString();
+                break;
             case SHAPE_PLAN:
-                return optimizedPlan.shape("");
+                plan = optimizedPlan.shape("");
+                break;
             case MEMO_PLAN:
-                return cascadesContext.getMemo().toString()
+                plan = cascadesContext.getMemo().toString()
                     + "\n\n========== OPTIMIZED PLAN ==========\n"
                     + optimizedPlan.treeString();
+                break;
             case ALL_PLAN:
-                return "========== PARSED PLAN ==========\n"
+                plan = "========== PARSED PLAN ==========\n"
                         + parsedPlan.treeString() + "\n\n"
                         + "========== ANALYZED PLAN ==========\n"
                         + analyzedPlan.treeString() + "\n\n"
@@ -374,9 +411,15 @@ public class NereidsPlanner extends Planner {
                         + rewrittenPlan.treeString() + "\n\n"
                         + "========== OPTIMIZED PLAN ==========\n"
                         + optimizedPlan.treeString();
+                break;
             default:
-                return super.getExplainString(explainOptions);
+                plan = super.getExplainString(explainOptions);
         }
+        if (!statementContext.getHintMap().isEmpty()) {
+            String hint = getHintExplainString(cascadesContext.getStatementContext().getHintMap());
+            return plan + hint;
+        }
+        return plan;
     }
 
     @Override

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

@@ -21,6 +21,7 @@ import org.apache.doris.analysis.StatementBase;
 import org.apache.doris.catalog.View;
 import org.apache.doris.common.IdGenerator;
 import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.hint.Hint;
 import org.apache.doris.nereids.memo.Group;
 import org.apache.doris.nereids.rules.analysis.ColumnAliasGenerator;
 import org.apache.doris.nereids.trees.expressions.CTEId;
@@ -72,6 +73,8 @@ public class StatementContext {
     private boolean isDpHyp = false;
     private boolean isOtherJoinReorder = false;
 
+    private boolean isLeadingJoin = false;
+
     private final IdGenerator<ExprId> exprIdGenerator = ExprId.createGenerator();
     private final IdGenerator<ObjectId> objectIdGenerator = ObjectId.createGenerator();
     private final IdGenerator<RelationId> relationIdGenerator = RelationId.createGenerator();
@@ -85,6 +88,7 @@ public class StatementContext {
     private final Map<CTEId, List<Pair<Map<Slot, Slot>, Group>>> cteIdToConsumerGroup = new HashMap<>();
     private final Map<CTEId, LogicalPlan> rewrittenCtePlan = new HashMap<>();
     private final Set<View> views = Sets.newHashSet();
+    private final Map<String, Hint> hintMap = Maps.newLinkedHashMap();
 
     public StatementContext() {
         this.connectContext = ConnectContext.get();
@@ -143,6 +147,14 @@ public class StatementContext {
         isDpHyp = dpHyp;
     }
 
+    public boolean isLeadingJoin() {
+        return isLeadingJoin;
+    }
+
+    public void setLeadingJoin(boolean leadingJoin) {
+        isLeadingJoin = leadingJoin;
+    }
+
     public boolean isOtherJoinReorder() {
         return isOtherJoinReorder;
     }
@@ -181,6 +193,10 @@ public class StatementContext {
         return supplier.get();
     }
 
+    public Map<String, Hint> getHintMap() {
+        return hintMap;
+    }
+
     public ColumnAliasGenerator getColumnAliasGenerator() {
         return columnAliasGenerator == null
             ? columnAliasGenerator = new ColumnAliasGenerator()

fe/fe-core/src/main/java/org/apache/doris/nereids/hint/Hint.java

@@ -0,0 +1,85 @@
+// 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.hint;
+
+import java.util.Objects;
+
+/**
+ * select hint.
+ * e.g. set_var(query_timeout='1800', exec_mem_limit='2147483648')
+ */
+public class Hint {
+    // e.g. set_var
+    private String hintName;
+
+    private HintStatus status;
+
+    private String errorMessage = "";
+
+    /**
+     * hint status which need to show in explain when it is not used or have syntax error
+     */
+    public enum HintStatus {
+        UNUSED,
+        SYNTAX_ERROR,
+        SUCCESS
+    }
+
+    public Hint(String hintName) {
+        this.hintName = Objects.requireNonNull(hintName, "hintName can not be null");
+        this.status = HintStatus.UNUSED;
+    }
+
+    public void setHintName(String hintName) {
+        this.hintName = hintName;
+    }
+
+    public void setStatus(HintStatus status) {
+        this.status = status;
+    }
+
+    public HintStatus getStatus() {
+        return status;
+    }
+
+    public boolean isSuccess() {
+        return getStatus() == HintStatus.SUCCESS;
+    }
+
+    public boolean isSyntaxError() {
+        return getStatus() == HintStatus.SYNTAX_ERROR;
+    }
+
+    public String getHintName() {
+        return hintName;
+    }
+
+    public String getErrorMessage() {
+        return errorMessage;
+    }
+
+    public void setErrorMessage(String errorMessage) {
+        this.errorMessage = errorMessage;
+    }
+
+    public String getExplainString() {
+        StringBuilder out = new StringBuilder();
+        out.append("\nHint:\n");
+        return out.toString();
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/hint/JoinConstraint.java

@@ -0,0 +1,88 @@
+// 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.hint;
+
+import org.apache.doris.nereids.trees.plans.JoinType;
+
+/**
+ * Join constraint which helps for leading to construct outer join , semi join and anti join
+ */
+public class JoinConstraint {
+    private final Long minLeftHand;
+    private final Long minRightHand;
+
+    private final Long leftHand;
+    private final Long rightHand;
+
+    private final JoinType joinType;
+
+    private final boolean lhsStrict;
+
+    private boolean isReversed;
+
+    /**
+     * join constraints which means restriction to some logical not equivalence of left join and semi join
+     * @param minLeftHand minimal left hand table bitmap which needed for a special join
+     * @param minRightHand minimal right hand table bitmap which needed for a special join
+     * @param leftHand left hand table bitmap below current join
+     * @param rightHand right hand table bitmap below current join
+     * @param joinType join type, here we only have full outer join, left join, semi join and anti join
+     * @param lhsStrict is left hand side strict
+     */
+    public JoinConstraint(Long minLeftHand, Long minRightHand, Long leftHand, Long rightHand,
+                            JoinType joinType, boolean lhsStrict) {
+        this.minLeftHand = minLeftHand;
+        this.minRightHand = minRightHand;
+        this.leftHand = leftHand;
+        this.rightHand = rightHand;
+        this.joinType = joinType;
+        this.lhsStrict = lhsStrict;
+    }
+
+    public JoinType getJoinType() {
+        return joinType;
+    }
+
+    public Long getLeftHand() {
+        return leftHand;
+    }
+
+    public Long getRightHand() {
+        return rightHand;
+    }
+
+    public Long getMinLeftHand() {
+        return minLeftHand;
+    }
+
+    public Long getMinRightHand() {
+        return minRightHand;
+    }
+
+    public boolean isLhsStrict() {
+        return lhsStrict;
+    }
+
+    public void setReversed(boolean reversed) {
+        isReversed = reversed;
+    }
+
+    public boolean isReversed() {
+        return isReversed;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/hint/LeadingHint.java

@@ -0,0 +1,503 @@
+// 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.hint;
+
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.jobs.joinorder.hypergraph.bitmap.LongBitmap;
+import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.plans.JoinHint;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.util.JoinUtils;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.Stack;
+
+/**
+ * select hint.
+ * e.g. set_var(query_timeout='1800', exec_mem_limit='2147483648')
+ */
+public class LeadingHint extends Hint {
+    private String originalString = "";
+    private final List<String> tablelist = new ArrayList<>();
+    private final List<Integer> levellist = new ArrayList<>();
+
+    private final Map<RelationId, LogicalPlan> relationIdToScanMap = Maps.newLinkedHashMap();
+
+    private final List<Pair<RelationId, String>> relationIdAndTableName = new ArrayList<>();
+
+    private final Map<ExprId, String> exprIdToTableNameMap = Maps.newLinkedHashMap();
+
+    private final List<Pair<Long, Expression>> filters = new ArrayList<>();
+
+    private final List<JoinConstraint> joinConstraintList = new ArrayList<>();
+
+    private Long innerJoinBitmap = 0L;
+
+    public LeadingHint(String hintName) {
+        super(hintName);
+    }
+
+    /**
+     * Leading hint data structure before using
+     * @param hintName Leading
+     * @param parameters table name mixed with left and right brace
+     */
+    public LeadingHint(String hintName, List<String> parameters, String originalString) {
+        super(hintName);
+        this.originalString = originalString;
+        int level = 0;
+        for (String parameter : parameters) {
+            if (parameter.equals("{")) {
+                ++level;
+            } else if (parameter.equals("}")) {
+                level--;
+            } else {
+                tablelist.add(parameter);
+                levellist.add(level);
+            }
+        }
+    }
+
+    public List<String> getTablelist() {
+        return tablelist;
+    }
+
+    public List<Integer> getLevellist() {
+        return levellist;
+    }
+
+    public Map<RelationId, LogicalPlan> getRelationIdToScanMap() {
+        return relationIdToScanMap;
+    }
+
+    @Override
+    public String getExplainString() {
+        StringBuilder out = new StringBuilder();
+        out.append(originalString);
+        return out.toString();
+    }
+
+    /**
+     * Get logical plan by table name recorded in leading hint. if can not get, means leading has syntax error
+     * or need to update. So return null should be deal with when call
+     * @param name table name
+     * @return logical plan recorded when binding
+     */
+    public LogicalPlan getLogicalPlanByName(String name) {
+        RelationId id = findRelationIdAndTableName(name);
+        if (id == null) {
+            this.setStatus(HintStatus.SYNTAX_ERROR);
+            this.setErrorMessage("can not find table: " + name);
+            return null;
+        }
+        return relationIdToScanMap.get(id);
+    }
+
+    /**
+     * putting pair into list, if relation id already exist update table name
+     * @param relationIdTableNamePair pair of relation id and table name to be inserted
+     */
+    public void putRelationIdAndTableName(Pair<RelationId, String> relationIdTableNamePair) {
+        boolean isUpdate = false;
+        for (Pair<RelationId, String> pair : relationIdAndTableName) {
+            if (pair.first.equals(relationIdTableNamePair.first)) {
+                pair.second = relationIdTableNamePair.second;
+                isUpdate = true;
+            }
+        }
+        if (!isUpdate) {
+            relationIdAndTableName.add(relationIdTableNamePair);
+        }
+    }
+
+    /**
+     * putting pair into list, if relation id already exist update table name
+     * @param relationIdTableNamePair pair of relation id and table name to be inserted
+     */
+    public void updateRelationIdByTableName(Pair<RelationId, String> relationIdTableNamePair) {
+        boolean isUpdate = false;
+        for (Pair<RelationId, String> pair : relationIdAndTableName) {
+            if (pair.second.equals(relationIdTableNamePair.second)) {
+                pair.first = relationIdTableNamePair.first;
+                isUpdate = true;
+            }
+        }
+        if (!isUpdate) {
+            relationIdAndTableName.add(relationIdTableNamePair);
+        }
+    }
+
+    /**
+     * find relation id and table name pair, relation id is unique, but table name is not
+     * @param name table name
+     * @return relation id
+     */
+    public RelationId findRelationIdAndTableName(String name) {
+        for (Pair<RelationId, String> pair : relationIdAndTableName) {
+            if (pair.second.equals(name)) {
+                return pair.first;
+            }
+        }
+        return null;
+    }
+
+    private boolean hasSameName() {
+        Set<String> tableSet = Sets.newHashSet();
+        for (String table : tablelist) {
+            if (!tableSet.add(table)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    public Map<ExprId, String> getExprIdToTableNameMap() {
+        return exprIdToTableNameMap;
+    }
+
+    public List<Pair<Long, Expression>> getFilters() {
+        return filters;
+    }
+
+    public List<JoinConstraint> getJoinConstraintList() {
+        return joinConstraintList;
+    }
+
+    public Long getInnerJoinBitmap() {
+        return innerJoinBitmap;
+    }
+
+    public void setInnerJoinBitmap(Long innerJoinBitmap) {
+        this.innerJoinBitmap = innerJoinBitmap;
+    }
+
+    /**
+     * try to get join constraint, if can not get, it means join is inner join,
+     * @param joinTableBitmap table bitmap below this join
+     * @param leftTableBitmap table bitmap below right child
+     * @param rightTableBitmap table bitmap below right child
+     * @return boolean value used for judging whether the join is legal, and should this join need to reverse
+     */
+    public Pair<JoinConstraint, Boolean> getJoinConstraint(Long joinTableBitmap, Long leftTableBitmap,
+                                                           Long rightTableBitmap) {
+        boolean reversed = false;
+        boolean mustBeLeftjoin = false;
+
+        JoinConstraint matchedJoinConstraint = null;
+
+        for (JoinConstraint joinConstraint : joinConstraintList) {
+            if (!LongBitmap.isOverlap(joinConstraint.getMinRightHand(), joinTableBitmap)) {
+                continue;
+            }
+
+            if (LongBitmap.isSubset(joinTableBitmap, joinConstraint.getMinRightHand())) {
+                continue;
+            }
+
+            if (LongBitmap.isSubset(joinConstraint.getMinLeftHand(), leftTableBitmap)
+                    && LongBitmap.isSubset(joinConstraint.getMinRightHand(), leftTableBitmap)) {
+                continue;
+            }
+
+            if (LongBitmap.isSubset(joinConstraint.getMinLeftHand(), rightTableBitmap)
+                    && LongBitmap.isSubset(joinConstraint.getMinRightHand(), rightTableBitmap)) {
+                continue;
+            }
+
+            if (joinConstraint.getJoinType().isSemiJoin()) {
+                if (LongBitmap.isSubset(joinConstraint.getRightHand(), leftTableBitmap)
+                        && !LongBitmap.isSubset(joinConstraint.getRightHand(), leftTableBitmap)) {
+                    continue;
+                }
+                if (LongBitmap.isSubset(joinConstraint.getRightHand(), rightTableBitmap)
+                        && !joinConstraint.getRightHand().equals(rightTableBitmap)) {
+                    continue;
+                }
+            }
+
+            if (LongBitmap.isSubset(joinConstraint.getMinLeftHand(), leftTableBitmap)
+                    && LongBitmap.isSubset(joinConstraint.getMinRightHand(), rightTableBitmap)) {
+                if (matchedJoinConstraint != null) {
+                    return Pair.of(null, false);
+                }
+                matchedJoinConstraint = joinConstraint;
+                reversed = false;
+            } else if (LongBitmap.isSubset(joinConstraint.getMinLeftHand(), rightTableBitmap)
+                    && LongBitmap.isSubset(joinConstraint.getMinRightHand(), leftTableBitmap)) {
+                if (matchedJoinConstraint != null) {
+                    return Pair.of(null, false);
+                }
+                matchedJoinConstraint = joinConstraint;
+                reversed = true;
+            } else if (joinConstraint.getJoinType().isSemiJoin()
+                    && joinConstraint.getRightHand().equals(rightTableBitmap)) {
+                if (matchedJoinConstraint != null) {
+                    return Pair.of(null, false);
+                }
+                matchedJoinConstraint = joinConstraint;
+                reversed = false;
+            } else if (joinConstraint.getJoinType().isSemiJoin()
+                    && joinConstraint.getRightHand().equals(leftTableBitmap)) {
+                /* Reversed semijoin case */
+                if (matchedJoinConstraint != null) {
+                    return Pair.of(null, false);
+                }
+                matchedJoinConstraint = joinConstraint;
+                reversed = true;
+            } else {
+                if (LongBitmap.isOverlap(leftTableBitmap, joinConstraint.getMinRightHand())
+                        && LongBitmap.isOverlap(rightTableBitmap, joinConstraint.getMinRightHand())) {
+                    continue;
+                }
+
+                if (!joinConstraint.getJoinType().isLeftJoin()
+                        || LongBitmap.isOverlap(joinTableBitmap, joinConstraint.getMinLeftHand())) {
+                    return Pair.of(null, false);
+                }
+
+                mustBeLeftjoin = true;
+            }
+        }
+        if (mustBeLeftjoin && (matchedJoinConstraint == null || !matchedJoinConstraint.getJoinType().isLeftJoin()
+                || !matchedJoinConstraint.isLhsStrict())) {
+            return Pair.of(null, false);
+        }
+        // this means inner join
+        if (matchedJoinConstraint == null) {
+            return Pair.of(null, true);
+        }
+        matchedJoinConstraint.setReversed(reversed);
+        return Pair.of(matchedJoinConstraint, true);
+    }
+
+    /**
+     * Try to get join type of two random logical scan or join node table bitmap
+     * @param left left side table bitmap
+     * @param right right side table bitmap
+     * @return join type or failure
+     */
+    public JoinType computeJoinType(Long left, Long right, List<Expression> conditions) {
+        Pair<JoinConstraint, Boolean> joinConstraintBooleanPair
+                = getJoinConstraint(LongBitmap.or(left, right), left, right);
+        if (!joinConstraintBooleanPair.second) {
+            this.setStatus(HintStatus.UNUSED);
+        } else if (joinConstraintBooleanPair.first == null) {
+            if (conditions.isEmpty()) {
+                return JoinType.CROSS_JOIN;
+            }
+            return JoinType.INNER_JOIN;
+        } else {
+            JoinConstraint joinConstraint = joinConstraintBooleanPair.first;
+            if (joinConstraint.isReversed()) {
+                return joinConstraint.getJoinType().swap();
+            } else {
+                return joinConstraint.getJoinType();
+            }
+        }
+        if (conditions.isEmpty()) {
+            return JoinType.CROSS_JOIN;
+        }
+        return JoinType.INNER_JOIN;
+    }
+
+    /**
+     * using leading to generate plan, it could be failed, if failed set leading status to unused or syntax error
+     * @return plan
+     */
+    public Plan generateLeadingJoinPlan() {
+        this.setStatus(HintStatus.SUCCESS);
+        Stack<Pair<Integer, LogicalPlan>> stack = new Stack<>();
+        int index = 0;
+        LogicalPlan logicalPlan = getLogicalPlanByName(getTablelist().get(index));
+        if (logicalPlan == null) {
+            return null;
+        }
+        logicalPlan = makeFilterPlanIfExist(getFilters(), logicalPlan);
+        assert (logicalPlan != null);
+        stack.push(Pair.of(getLevellist().get(index), logicalPlan));
+        int stackTopLevel = getLevellist().get(index++);
+        while (index < getTablelist().size()) {
+            int currentLevel = getLevellist().get(index);
+            if (currentLevel == stackTopLevel) {
+                // should return error if can not found table
+                logicalPlan = getLogicalPlanByName(getTablelist().get(index++));
+                if (logicalPlan == null) {
+                    return null;
+                }
+                logicalPlan = makeFilterPlanIfExist(getFilters(), logicalPlan);
+                Pair<Integer, LogicalPlan> newStackTop = stack.peek();
+                while (!(stack.isEmpty() || stackTopLevel != newStackTop.first)) {
+                    // check join is legal and get join type
+                    newStackTop = stack.pop();
+                    List<Expression> conditions = getJoinConditions(
+                            getFilters(), newStackTop.second, logicalPlan);
+                    Pair<List<Expression>, List<Expression>> pair = JoinUtils.extractExpressionForHashTable(
+                            newStackTop.second.getOutput(), logicalPlan.getOutput(), conditions);
+                    JoinType joinType = computeJoinType(getBitmap(newStackTop.second),
+                            getBitmap(logicalPlan), conditions);
+                    if (!this.isSuccess()) {
+                        return null;
+                    }
+                    // get joinType
+                    LogicalJoin logicalJoin = new LogicalJoin<>(joinType, pair.first,
+                            pair.second,
+                            JoinHint.NONE,
+                            Optional.empty(),
+                            newStackTop.second,
+                            logicalPlan);
+                    logicalJoin.setBitmap(LongBitmap.or(getBitmap(newStackTop.second), getBitmap(logicalPlan)));
+                    if (stackTopLevel > 0) {
+                        stackTopLevel--;
+                    }
+                    if (!stack.isEmpty()) {
+                        newStackTop = stack.peek();
+                    }
+                    logicalPlan = logicalJoin;
+                }
+                stack.push(Pair.of(stackTopLevel, logicalPlan));
+            } else {
+                // push
+                logicalPlan = getLogicalPlanByName(getTablelist().get(index++));
+                if (logicalPlan == null) {
+                    return null;
+                }
+                logicalPlan = makeFilterPlanIfExist(getFilters(), logicalPlan);
+                stack.push(Pair.of(currentLevel, logicalPlan));
+                stackTopLevel = currentLevel;
+            }
+        }
+
+        LogicalJoin finalJoin = (LogicalJoin) stack.pop().second;
+        // we want all filters been remove
+        if (!getFilters().isEmpty()) {
+            List<Expression> conditions = getLastConditions(getFilters());
+            Pair<List<Expression>, List<Expression>> pair = JoinUtils.extractExpressionForHashTable(
+                    finalJoin.left().getOutput(), finalJoin.right().getOutput(), conditions);
+            finalJoin = new LogicalJoin<>(finalJoin.getJoinType(), pair.first,
+                pair.second,
+                JoinHint.NONE,
+                Optional.empty(),
+                finalJoin.left(),
+                finalJoin.right());
+        }
+        if (finalJoin != null) {
+            this.setStatus(HintStatus.SUCCESS);
+        }
+        return finalJoin;
+    }
+
+    private List<Expression> getJoinConditions(List<Pair<Long, Expression>> filters,
+                                               LogicalPlan left, LogicalPlan right) {
+        List<Expression> joinConditions = new ArrayList<>();
+        for (int i = filters.size() - 1; i >= 0; i--) {
+            Pair<Long, Expression> filterPair = filters.get(i);
+            Long tablesBitMap = LongBitmap.or(getBitmap(left), getBitmap(right));
+            // left one is smaller set
+            if (LongBitmap.isSubset(filterPair.first, tablesBitMap)) {
+                joinConditions.add(filterPair.second);
+                filters.remove(i);
+            }
+        }
+        return joinConditions;
+    }
+
+    private List<Expression> getLastConditions(List<Pair<Long, Expression>> filters) {
+        List<Expression> joinConditions = new ArrayList<>();
+        for (int i = filters.size() - 1; i >= 0; i--) {
+            Pair<Long, Expression> filterPair = filters.get(i);
+            joinConditions.add(filterPair.second);
+            filters.remove(i);
+        }
+        return joinConditions;
+    }
+
+    private LogicalPlan makeFilterPlanIfExist(List<Pair<Long, Expression>> filters, LogicalPlan scan) {
+        Set<Expression> newConjuncts = new HashSet<>();
+        for (int i = filters.size() - 1; i >= 0; i--) {
+            Pair<Long, Expression> filterPair = filters.get(i);
+            if (LongBitmap.isSubset(filterPair.first, getBitmap(scan))) {
+                newConjuncts.add(filterPair.second);
+                filters.remove(i);
+            }
+        }
+        if (newConjuncts.isEmpty()) {
+            return scan;
+        } else {
+            return new LogicalFilter<>(newConjuncts, scan);
+        }
+    }
+
+    private Long getBitmap(LogicalPlan root) {
+        if (root instanceof LogicalJoin) {
+            return ((LogicalJoin) root).getBitmap();
+        } else if (root instanceof LogicalRelation) {
+            return LongBitmap.set(0L, (((LogicalRelation) root).getRelationId().asInt()));
+        } else if (root instanceof LogicalFilter) {
+            return getBitmap((LogicalPlan) root.child(0));
+        } else if (root instanceof LogicalProject) {
+            return getBitmap((LogicalPlan) root.child(0));
+        } else {
+            return null;
+        }
+    }
+
+    /**
+     * get leading containing tables which means leading wants to combine tables into joins
+     * @return long value represent tables we included
+     */
+    public Long getLeadingTableBitmap(List<TableIf> tables) {
+        Long totalBitmap = 0L;
+        if (hasSameName()) {
+            this.setStatus(HintStatus.SYNTAX_ERROR);
+            this.setErrorMessage("duplicated table");
+            return totalBitmap;
+        }
+        if (getTablelist().size() != tables.size()) {
+            this.setStatus(HintStatus.SYNTAX_ERROR);
+            this.setErrorMessage("tables should be same as join tables");
+            return totalBitmap;
+        }
+        for (int index = 0; index < getTablelist().size(); index++) {
+            RelationId id = findRelationIdAndTableName(getTablelist().get(index));
+            if (id == null) {
+                this.setStatus(HintStatus.SYNTAX_ERROR);
+                this.setErrorMessage("can not find table: " + getTablelist().get(index));
+                return totalBitmap;
+            }
+            totalBitmap = LongBitmap.set(totalBitmap, id.asInt());
+        }
+        return totalBitmap;
+    }
+}

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

@@ -19,6 +19,8 @@ package org.apache.doris.nereids.jobs.executor;
 
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.jobs.rewrite.RewriteJob;
+import org.apache.doris.nereids.processor.pre.EliminateLogicalSelectHint;
+import org.apache.doris.nereids.rules.RuleType;
 import org.apache.doris.nereids.rules.analysis.AdjustAggregateNullableForEmptySet;
 import org.apache.doris.nereids.rules.analysis.AnalyzeCTE;
 import org.apache.doris.nereids.rules.analysis.BindExpression;
@@ -79,6 +81,8 @@ public class Analyzer extends AbstractBatchJobExecutor {
 
     private static List<RewriteJob> buildAnalyzeJobs(Optional<CustomTableResolver> customTableResolver) {
         return jobs(
+            // we should eliminate hint after "Subquery unnesting" because some hint maybe exist in the CTE or subquery.
+            custom(RuleType.ELIMINATE_HINT, EliminateLogicalSelectHint::new),
             topDown(new AnalyzeCTE()),
             bottomUp(
                 new BindRelation(customTableResolver),

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

@@ -68,7 +68,6 @@ public class Optimizer {
         cascadesContext.getJobScheduler().executeJobPool(cascadesContext);
     }
 
-    // DependsRules: EnsureProjectOnTopJoin.class
     private void dpHypOptimize() {
         Group root = cascadesContext.getMemo().getRoot();
         // Due to EnsureProjectOnTopJoin, root group can't be Join Group, so DPHyp doesn't change the root group

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

@@ -20,7 +20,6 @@ package org.apache.doris.nereids.jobs.executor;
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.jobs.rewrite.CostBasedRewriteJob;
 import org.apache.doris.nereids.jobs.rewrite.RewriteJob;
-import org.apache.doris.nereids.processor.pre.EliminateLogicalSelectHint;
 import org.apache.doris.nereids.rules.RuleSet;
 import org.apache.doris.nereids.rules.RuleType;
 import org.apache.doris.nereids.rules.analysis.AdjustAggregateNullableForEmptySet;
@@ -42,6 +41,7 @@ import org.apache.doris.nereids.rules.rewrite.CheckDataTypes;
 import org.apache.doris.nereids.rules.rewrite.CheckMatchExpression;
 import org.apache.doris.nereids.rules.rewrite.CheckMultiDistinct;
 import org.apache.doris.nereids.rules.rewrite.CollectFilterAboveConsumer;
+import org.apache.doris.nereids.rules.rewrite.CollectJoinConstraint;
 import org.apache.doris.nereids.rules.rewrite.CollectProjectAboveConsumer;
 import org.apache.doris.nereids.rules.rewrite.ColumnPruning;
 import org.apache.doris.nereids.rules.rewrite.ConvertInnerOrCrossJoin;
@@ -69,6 +69,7 @@ import org.apache.doris.nereids.rules.rewrite.InferFilterNotNull;
 import org.apache.doris.nereids.rules.rewrite.InferJoinNotNull;
 import org.apache.doris.nereids.rules.rewrite.InferPredicates;
 import org.apache.doris.nereids.rules.rewrite.InferSetOperatorDistinct;
+import org.apache.doris.nereids.rules.rewrite.LeadingJoin;
 import org.apache.doris.nereids.rules.rewrite.MergeFilters;
 import org.apache.doris.nereids.rules.rewrite.MergeOneRowRelationIntoUnion;
 import org.apache.doris.nereids.rules.rewrite.MergeProjects;
@@ -158,8 +159,6 @@ public class Rewriter extends AbstractBatchJobExecutor {
                             new ApplyToJoin()
                     )
             ),
-            // we should eliminate hint after "Subquery unnesting" because some hint maybe exist in the CTE or subquery.
-            custom(RuleType.ELIMINATE_HINT, EliminateLogicalSelectHint::new),
             topic("Eliminate optimization",
                     bottomUp(
                             new EliminateLimit(),
@@ -222,6 +221,15 @@ public class Rewriter extends AbstractBatchJobExecutor {
                     bottomUp(new EliminateNotNull()),
                     topDown(new ConvertInnerOrCrossJoin())
             ),
+            topic("LEADING JOIN",
+                bottomUp(
+                    new CollectJoinConstraint()
+                ),
+                custom(RuleType.LEADING_JOIN, LeadingJoin::new),
+                bottomUp(
+                    new ExpressionRewrite(CheckLegalityAfterRewrite.INSTANCE)
+                )
+            ),
             topic("Column pruning and infer predicate",
                     custom(RuleType.COLUMN_PRUNING, ColumnPruning::new),
                     custom(RuleType.INFER_PREDICATES, InferPredicates::new),

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/bitmap/LongBitmap.java

@@ -17,7 +17,10 @@
 
 package org.apache.doris.nereids.jobs.joinorder.hypergraph.bitmap;
 
+import org.apache.doris.nereids.trees.plans.RelationId;
+
 import java.util.BitSet;
+import java.util.Set;
 
 /**
  * This is helper class for some bitmap operation
@@ -147,6 +150,14 @@ public class LongBitmap {
         return Long.numberOfTrailingZeros(bitmap);
     }
 
+    public static Long computeTableBitmap(Set<RelationId> relationIdSet) {
+        Long totalBitmap = 0L;
+        for (RelationId id : relationIdSet) {
+            totalBitmap = LongBitmap.set(totalBitmap, (id.asInt()));
+        }
+        return totalBitmap;
+    }
+
     public static String toString(long bitmap) {
         long[] longs = {bitmap};
         BitSet bitSet = BitSet.valueOf(longs);

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -133,6 +133,8 @@ import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.exceptions.ParseException;
 import org.apache.doris.nereids.properties.OrderKey;
 import org.apache.doris.nereids.properties.SelectHint;
+import org.apache.doris.nereids.properties.SelectHintLeading;
+import org.apache.doris.nereids.properties.SelectHintSetVar;
 import org.apache.doris.nereids.trees.expressions.Add;
 import org.apache.doris.nereids.trees.expressions.And;
 import org.apache.doris.nereids.trees.expressions.BitAnd;
@@ -1711,20 +1713,34 @@ public class LogicalPlanBuilder extends DorisParserBaseVisitor<Object> {
         Map<String, SelectHint> hints = Maps.newLinkedHashMap();
         for (HintStatementContext hintStatement : hintContext.hintStatements) {
             String hintName = hintStatement.hintName.getText().toLowerCase(Locale.ROOT);
-            Map<String, Optional<String>> parameters = Maps.newLinkedHashMap();
-            for (HintAssignmentContext kv : hintStatement.parameters) {
-                String parameterName = visitIdentifierOrText(kv.key);
-                Optional<String> value = Optional.empty();
-                if (kv.constantValue != null) {
-                    Literal literal = (Literal) visit(kv.constantValue);
-                    value = Optional.ofNullable(literal.toLegacyLiteral().getStringValue());
-                } else if (kv.identifierValue != null) {
-                    // maybe we should throw exception when the identifierValue is quoted identifier
-                    value = Optional.ofNullable(kv.identifierValue.getText());
-                }
-                parameters.put(parameterName, value);
+            switch (hintName) {
+                case "set_var":
+                    Map<String, Optional<String>> parameters = Maps.newLinkedHashMap();
+                    for (HintAssignmentContext kv : hintStatement.parameters) {
+                        String parameterName = visitIdentifierOrText(kv.key);
+                        Optional<String> value = Optional.empty();
+                        if (kv.constantValue != null) {
+                            Literal literal = (Literal) visit(kv.constantValue);
+                            value = Optional.ofNullable(literal.toLegacyLiteral().getStringValue());
+                        } else if (kv.identifierValue != null) {
+                            // maybe we should throw exception when the identifierValue is quoted identifier
+                            value = Optional.ofNullable(kv.identifierValue.getText());
+                        }
+                        parameters.put(parameterName, value);
+                    }
+                    hints.put(hintName, new SelectHintSetVar(hintName, parameters));
+                    break;
+                case "leading":
+                    List<String> leadingParameters = new ArrayList<String>();
+                    for (HintAssignmentContext kv : hintStatement.parameters) {
+                        String parameterName = visitIdentifierOrText(kv.key);
+                        leadingParameters.add(parameterName);
+                    }
+                    hints.put(hintName, new SelectHintLeading(hintName, leadingParameters));
+                    break;
+                default:
+                    break;
             }
-            hints.put(hintName, new SelectHint(hintName, parameters));
         }
         return new LogicalSelectHint<>(hints, logicalPlan);
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/pre/EliminateLogicalSelectHint.java

@@ -21,12 +21,17 @@ import org.apache.doris.analysis.SetVar;
 import org.apache.doris.analysis.StringLiteral;
 import org.apache.doris.nereids.StatementContext;
 import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.hint.Hint;
+import org.apache.doris.nereids.hint.LeadingHint;
 import org.apache.doris.nereids.jobs.JobContext;
 import org.apache.doris.nereids.properties.SelectHint;
+import org.apache.doris.nereids.properties.SelectHintLeading;
+import org.apache.doris.nereids.properties.SelectHintSetVar;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSelectHint;
 import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.qe.VariableMgr;
 
@@ -54,7 +59,11 @@ public class EliminateLogicalSelectHint extends PlanPreprocessor implements Cust
         for (Entry<String, SelectHint> hint : selectHintPlan.getHints().entrySet()) {
             String hintName = hint.getKey();
             if (hintName.equalsIgnoreCase("SET_VAR")) {
-                setVar(hint.getValue(), context);
+                setVar((SelectHintSetVar) hint.getValue(), context);
+            } else if (hintName.equalsIgnoreCase("ORDERED")) {
+                ConnectContext.get().getSessionVariable().setDisableJoinReorder(true);
+            } else if (hintName.equalsIgnoreCase("LEADING")) {
+                extractLeading((SelectHintLeading) hint.getValue(), context);
             } else {
                 logger.warn("Can not process select hint '{}' and skip it", hint.getKey());
             }
@@ -63,7 +72,7 @@ public class EliminateLogicalSelectHint extends PlanPreprocessor implements Cust
         return (LogicalPlan) selectHintPlan.child();
     }
 
-    private void setVar(SelectHint selectHint, StatementContext context) {
+    private void setVar(SelectHintSetVar selectHint, StatementContext context) {
         SessionVariable sessionVariable = context.getConnectContext().getSessionVariable();
         // set temporary session value, and then revert value in the 'finally block' of StmtExecutor#execute
         sessionVariable.setIsSingleSetVar(true);
@@ -91,4 +100,16 @@ public class EliminateLogicalSelectHint extends PlanPreprocessor implements Cust
             throw new AnalysisException("The nereids is disabled in this sql, fallback to original planner");
         }
     }
+
+    private void extractLeading(SelectHintLeading selectHint, StatementContext context) {
+        LeadingHint hint = new LeadingHint("Leading", selectHint.getParameters(), selectHint.toString());
+        if (context.getHintMap().get("Leading") != null) {
+            hint.setStatus(Hint.HintStatus.SYNTAX_ERROR);
+            hint.setErrorMessage("can only have one leading clause");
+        }
+        context.getHintMap().put("Leading", hint);
+        context.setLeadingJoin(true);
+        assert (selectHint != null);
+        assert (context != null);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/properties/SelectHint.java

@@ -17,10 +17,7 @@
 
 package org.apache.doris.nereids.properties;
 
-import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
-import java.util.stream.Collectors;
 
 /**
  * select hint.
@@ -28,34 +25,17 @@ import java.util.stream.Collectors;
  */
 public class SelectHint {
     // e.g. set_var
-    private final String hintName;
-    // e.g. query_timeout='1800', exec_mem_limit='2147483648'
-    private final Map<String, Optional<String>> parameters;
+    private String hintName;
 
-    public SelectHint(String hintName, Map<String, Optional<String>> parameters) {
+    public SelectHint(String hintName) {
         this.hintName = Objects.requireNonNull(hintName, "hintName can not be null");
-        this.parameters = Objects.requireNonNull(parameters, "parameters can not be null");
+    }
+
+    public void setHintName(String hintName) {
+        this.hintName = hintName;
     }
 
     public String getHintName() {
         return hintName;
     }
-
-    public Map<String, Optional<String>> getParameters() {
-        return parameters;
-    }
-
-    @Override
-    public String toString() {
-        String kvString = parameters
-                .entrySet()
-                .stream()
-                .map(kv ->
-                    kv.getValue().isPresent()
-                        ? kv.getKey() + "='" + kv.getValue().get() + "'"
-                        : kv.getKey()
-                )
-                .collect(Collectors.joining(", "));
-        return hintName + "(" + kvString + ")";
-    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/properties/SelectHintLeading.java

@@ -0,0 +1,47 @@
+// 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.properties;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * select hint.
+ * e.g. set_var(query_timeout='1800', exec_mem_limit='2147483648')
+ */
+public class SelectHintLeading extends SelectHint {
+    // e.g. query_timeout='1800', exec_mem_limit='2147483648'
+    private final List<String> parameters;
+
+    public SelectHintLeading(String hintName, List<String> parameters) {
+        super(hintName);
+        this.parameters = parameters;
+    }
+
+    public List<String> getParameters() {
+        return parameters;
+    }
+
+    @Override
+    public String toString() {
+        String leadingString = parameters
+                .stream()
+                .collect(Collectors.joining(" "));
+        return super.getHintName() + "(" + leadingString + ")";
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/properties/SelectHintSetVar.java

@@ -0,0 +1,54 @@
+// 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.properties;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * select hint.
+ * e.g. set_var(query_timeout='1800', exec_mem_limit='2147483648')
+ */
+public class SelectHintSetVar extends SelectHint {
+    // e.g. query_timeout='1800', exec_mem_limit='2147483648'
+    private final Map<String, Optional<String>> parameters;
+
+    public SelectHintSetVar(String hintName, Map<String, Optional<String>> parameters) {
+        super(hintName);
+        this.parameters = parameters;
+    }
+
+    public Map<String, Optional<String>> getParameters() {
+        return parameters;
+    }
+
+    @Override
+    public String toString() {
+        String kvString = parameters
+                .entrySet()
+                .stream()
+                .map(kv ->
+                        kv.getValue().isPresent()
+                        ? kv.getKey() + "='" + kv.getValue().get() + "'"
+                        : kv.getKey()
+                )
+                .collect(Collectors.joining(", "));
+        return super.getHintName() + "(" + kvString + ")";
+    }
+}

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

@@ -254,8 +254,13 @@ public enum RuleType {
     CTE_INLINE(RuleTypeClass.REWRITE),
     REWRITE_CTE_CHILDREN(RuleTypeClass.REWRITE),
     COLLECT_FILTER_ON_CONSUMER(RuleTypeClass.REWRITE),
+
+    COLLECT_FILTER(RuleTypeClass.REWRITE),
+    COLLECT_JOIN_CONSTRAINT(RuleTypeClass.REWRITE),
     COLLECT_PROJECT_ABOVE_CONSUMER(RuleTypeClass.REWRITE),
     COLLECT_PROJECT_ABOVE_FILTER_CONSUMER(RuleTypeClass.REWRITE),
+
+    LEADING_JOIN(RuleTypeClass.REWRITE),
     REWRITE_SENTINEL(RuleTypeClass.REWRITE),
 
     // topn opts

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

@@ -26,6 +26,7 @@ import org.apache.doris.catalog.external.EsExternalTable;
 import org.apache.doris.catalog.external.ExternalTable;
 import org.apache.doris.catalog.external.HMSExternalTable;
 import org.apache.doris.common.Config;
+import org.apache.doris.common.Pair;
 import org.apache.doris.common.util.Util;
 import org.apache.doris.nereids.CTEContext;
 import org.apache.doris.nereids.CascadesContext;
@@ -33,6 +34,7 @@ import org.apache.doris.nereids.analyzer.Unbound;
 import org.apache.doris.nereids.analyzer.UnboundRelation;
 import org.apache.doris.nereids.analyzer.UnboundResultSink;
 import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.hint.LeadingHint;
 import org.apache.doris.nereids.parser.NereidsParser;
 import org.apache.doris.nereids.pattern.MatchingContext;
 import org.apache.doris.nereids.properties.LogicalProperties;
@@ -125,8 +127,15 @@ public class BindRelation extends OneAnalysisRuleFactory {
         if (cteContext != null) {
             Optional<LogicalPlan> analyzedCte = cteContext.getAnalyzedCTEPlan(tableName);
             if (analyzedCte.isPresent()) {
-                return new LogicalCTEConsumer(unboundRelation.getRelationId(),
+                LogicalCTEConsumer consumer = new LogicalCTEConsumer(unboundRelation.getRelationId(),
                         cteContext.getCteId(), tableName, analyzedCte.get());
+                if (cascadesContext.getStatementContext().isLeadingJoin()) {
+                    LeadingHint leading = (LeadingHint) cascadesContext.getStatementContext()
+                            .getHintMap().get("Leading");
+                    leading.putRelationIdAndTableName(Pair.of(consumer.getRelationId(), tableName));
+                    leading.getRelationIdToScanMap().put(consumer.getRelationId(), consumer);
+                }
+                return consumer;
             }
         }
         List<String> tableQualifier = RelationUtil.getQualifierName(cascadesContext.getConnectContext(),
@@ -147,7 +156,13 @@ public class BindRelation extends OneAnalysisRuleFactory {
         }
 
         // TODO: should generate different Scan sub class according to table's type
-        return getLogicalPlan(table, unboundRelation, tableQualifier, cascadesContext);
+        LogicalPlan scan = getLogicalPlan(table, unboundRelation, tableQualifier, cascadesContext);
+        if (cascadesContext.getStatementContext().isLeadingJoin()) {
+            LeadingHint leading = (LeadingHint) cascadesContext.getStatementContext().getHintMap().get("Leading");
+            leading.putRelationIdAndTableName(Pair.of(unboundRelation.getRelationId(), tableName));
+            leading.getRelationIdToScanMap().put(unboundRelation.getRelationId(), scan);
+        }
+        return scan;
     }
 
     private LogicalPlan bind(CascadesContext cascadesContext, UnboundRelation unboundRelation) {

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

@@ -17,10 +17,15 @@
 
 package org.apache.doris.nereids.rules.analysis;
 
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.hint.Hint;
+import org.apache.doris.nereids.hint.LeadingHint;
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
 import org.apache.doris.nereids.rules.rewrite.OneRewriteRuleFactory;
+import org.apache.doris.nereids.trees.plans.RelationId;
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
 
 import com.google.common.collect.ImmutableList;
 
@@ -34,8 +39,29 @@ public class LogicalSubQueryAliasToLogicalProject extends OneRewriteRuleFactory
     @Override
     public Rule build() {
         return RuleType.LOGICAL_SUB_QUERY_ALIAS_TO_LOGICAL_PROJECT.build(
-                logicalSubQueryAlias().then(
-                        alias -> new LogicalProject<>(ImmutableList.copyOf(alias.getOutput()), alias.child()))
+                logicalSubQueryAlias().thenApply(ctx -> {
+                    LogicalProject project = new LogicalProject<>(
+                            ImmutableList.copyOf(ctx.root.getOutput()), ctx.root.child());
+                    if (ctx.cascadesContext.getStatementContext().isLeadingJoin()) {
+                        String aliasName = ctx.root.getAlias();
+                        LeadingHint leading = (LeadingHint) ctx.cascadesContext.getStatementContext()
+                                .getHintMap().get("Leading");
+                        if (!(project.child() instanceof LogicalRelation)) {
+                            if (leading.getTablelist().contains(aliasName)) {
+                                leading.setStatus(Hint.HintStatus.SYNTAX_ERROR);
+                                leading.setErrorMessage("Leading alias can only be table name alias");
+                            }
+                        } else {
+                            RelationId id = leading.findRelationIdAndTableName(aliasName);
+                            if (id == null) {
+                                id = ((LogicalRelation) project.child()).getRelationId();
+                            }
+                            leading.putRelationIdAndTableName(Pair.of(id, aliasName));
+                            leading.getRelationIdToScanMap().put(id, project);
+                        }
+                    }
+                    return project;
+                })
         );
     }
 }

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

@@ -22,6 +22,7 @@ import org.apache.doris.nereids.trees.copier.DeepCopierContext;
 import org.apache.doris.nereids.trees.copier.LogicalPlanDeepCopier;
 import org.apache.doris.nereids.trees.expressions.Alias;
 import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.plans.Plan;
@@ -37,6 +38,7 @@ import org.apache.doris.qe.ConnectContext;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 
+import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -101,8 +103,10 @@ public class CTEInline extends DefaultPlanRewriter<LogicalCTEProducer<?>> implem
             for (Slot consumerSlot : cteConsumer.getOutput()) {
                 Slot producerSlot = cteConsumer.getProducerSlot(consumerSlot);
                 ExprId inlineExprId = deepCopierContext.exprIdReplaceMap.get(producerSlot.getExprId());
-                Alias alias = new Alias(consumerSlot.getExprId(), producerSlot.withExprId(inlineExprId),
-                        consumerSlot.getName());
+                List<Expression> childrenExprs = new ArrayList<>();
+                childrenExprs.add(producerSlot.withExprId(inlineExprId));
+                Alias alias = new Alias(consumerSlot.getExprId(), childrenExprs,
+                        consumerSlot.getName(), producerSlot.getQualifier());
                 projects.add(alias);
             }
             return new LogicalProject<>(projects, inlinedPlan);

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

@@ -0,0 +1,218 @@
+// 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.common.Pair;
+import org.apache.doris.nereids.hint.Hint;
+import org.apache.doris.nereids.hint.JoinConstraint;
+import org.apache.doris.nereids.hint.LeadingHint;
+import org.apache.doris.nereids.jobs.joinorder.hypergraph.bitmap.LongBitmap;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * CollectJoinConstraint
+ */
+public class CollectJoinConstraint implements RewriteRuleFactory {
+
+    @Override
+    public List<Rule> buildRules() {
+        return ImmutableList.of(
+            logicalRelation().thenApply(ctx -> {
+                LeadingHint leading = (LeadingHint) ctx.cascadesContext
+                        .getStatementContext().getHintMap().get("Leading");
+                if (leading == null) {
+                    return ctx.root;
+                } else if (leading.isSyntaxError()) {
+                    return ctx.root;
+                }
+                return ctx.root;
+            }).toRule(RuleType.COLLECT_JOIN_CONSTRAINT),
+
+            logicalJoin().thenApply(ctx -> {
+                LeadingHint leading = (LeadingHint) ctx.cascadesContext
+                        .getStatementContext().getHintMap().get("Leading");
+                if (leading == null) {
+                    return ctx.root;
+                }
+                LogicalJoin join = ctx.root;
+                List<Expression> expressions = join.getHashJoinConjuncts();
+                Long totalFilterBitMap = 0L;
+                Long nonNullableSlotBitMap = 0L;
+                for (Expression expression : expressions) {
+                    Long nonNullable = calSlotsTableBitMap(leading, expression.getInputSlots(), true);
+                    nonNullableSlotBitMap = LongBitmap.or(nonNullableSlotBitMap, nonNullable);
+                    Long filterBitMap = calSlotsTableBitMap(leading, expression.getInputSlots(), false);
+                    totalFilterBitMap = LongBitmap.or(totalFilterBitMap, filterBitMap);
+                    leading.getFilters().add(Pair.of(filterBitMap, expression));
+                }
+                expressions = join.getOtherJoinConjuncts();
+                for (Expression expression : expressions) {
+                    Long nonNullable = calSlotsTableBitMap(leading, expression.getInputSlots(), true);
+                    nonNullableSlotBitMap = LongBitmap.or(nonNullableSlotBitMap, nonNullable);
+                    Long filterBitMap = calSlotsTableBitMap(leading, expression.getInputSlots(), false);
+                    totalFilterBitMap = LongBitmap.or(totalFilterBitMap, filterBitMap);
+                    leading.getFilters().add(Pair.of(filterBitMap, expression));
+                }
+                Long leftHand = LongBitmap.computeTableBitmap(join.left().getInputRelations());
+                Long rightHand = LongBitmap.computeTableBitmap(join.right().getInputRelations());
+                join.setBitmap(LongBitmap.or(leftHand, rightHand));
+                collectJoinConstraintList(leading, leftHand, rightHand, join, totalFilterBitMap, nonNullableSlotBitMap);
+
+                return ctx.root;
+            }).toRule(RuleType.COLLECT_JOIN_CONSTRAINT),
+
+            logicalFilter().thenApply(ctx -> {
+                LeadingHint leading = (LeadingHint) ctx.cascadesContext
+                        .getStatementContext().getHintMap().get("Leading");
+                if (leading == null) {
+                    return ctx.root;
+                }
+                LogicalFilter filter = ctx.root;
+                Set<Expression> expressions = filter.getConjuncts();
+                for (Expression expression : expressions) {
+                    Long filterBitMap = calSlotsTableBitMap(leading, expression.getInputSlots(), false);
+                    leading.getFilters().add(Pair.of(filterBitMap, expression));
+                }
+                return ctx.root;
+            }).toRule(RuleType.COLLECT_JOIN_CONSTRAINT),
+
+            logicalProject(logicalOlapScan()).thenApply(
+                ctx -> {
+                    LeadingHint leading = (LeadingHint) ctx.cascadesContext
+                            .getStatementContext().getHintMap().get("Leading");
+                    if (leading == null) {
+                        return ctx.root;
+                    }
+                    LogicalProject<LogicalOlapScan> project = ctx.root;
+                    LogicalOlapScan scan = project.child();
+                    leading.getRelationIdToScanMap().put(scan.getRelationId(), project);
+                    return ctx.root;
+                }
+            ).toRule(RuleType.COLLECT_JOIN_CONSTRAINT)
+        );
+    }
+
+    private void collectJoinConstraintList(LeadingHint leading, Long leftHand, Long rightHand, LogicalJoin join,
+                                            Long filterTableBitMap, Long nonNullableSlotBitMap) {
+        Long totalTables = LongBitmap.or(leftHand, rightHand);
+        if (join.getJoinType().isInnerJoin()) {
+            leading.setInnerJoinBitmap(LongBitmap.or(leading.getInnerJoinBitmap(), totalTables));
+            return;
+        }
+        if (join.getJoinType().isFullOuterJoin()) {
+            JoinConstraint newJoinConstraint = new JoinConstraint(leftHand, rightHand, leftHand, rightHand,
+                    JoinType.FULL_OUTER_JOIN, false);
+            leading.getJoinConstraintList().add(newJoinConstraint);
+            return;
+        }
+        boolean isStrict = LongBitmap.isOverlap(nonNullableSlotBitMap, leftHand);
+        Long minLeftHand = LongBitmap.newBitmapIntersect(filterTableBitMap, leftHand);
+        Long innerJoinTableBitmap = LongBitmap.and(totalTables, leading.getInnerJoinBitmap());
+        Long filterAndInnerBelow = LongBitmap.newBitmapUnion(filterTableBitMap, innerJoinTableBitmap);
+        Long minRightHand = LongBitmap.newBitmapIntersect(filterAndInnerBelow, rightHand);
+        for (JoinConstraint other : leading.getJoinConstraintList()) {
+            if (other.getJoinType() == JoinType.FULL_OUTER_JOIN) {
+                if (LongBitmap.isOverlap(leftHand, other.getLeftHand())
+                        || LongBitmap.isOverlap(leftHand, other.getRightHand())) {
+                    minLeftHand = LongBitmap.or(minLeftHand,
+                        other.getLeftHand());
+                    minLeftHand = LongBitmap.or(minLeftHand,
+                        other.getRightHand());
+                }
+                if (LongBitmap.isOverlap(rightHand, other.getLeftHand())
+                        || LongBitmap.isOverlap(rightHand, other.getRightHand())) {
+                    minRightHand = LongBitmap.or(minRightHand,
+                        other.getLeftHand());
+                    minRightHand = LongBitmap.or(minRightHand,
+                        other.getRightHand());
+                }
+                /* Needn't do anything else with the full join */
+                continue;
+            }
+
+            if (LongBitmap.isOverlap(leftHand, other.getRightHand())) {
+                if (LongBitmap.isOverlap(filterTableBitMap, other.getRightHand())
+                        && (join.getJoinType().isSemiOrAntiJoin()
+                        || !LongBitmap.isOverlap(nonNullableSlotBitMap, other.getMinRightHand()))) {
+                    minLeftHand = LongBitmap.or(minLeftHand,
+                        other.getLeftHand());
+                    minLeftHand = LongBitmap.or(minLeftHand,
+                        other.getRightHand());
+                }
+            }
+
+            if (LongBitmap.isOverlap(rightHand, other.getRightHand())) {
+                if (LongBitmap.isOverlap(filterTableBitMap, other.getRightHand())
+                        || !LongBitmap.isOverlap(filterTableBitMap, other.getMinLeftHand())
+                        || join.getJoinType().isSemiOrAntiJoin()
+                        || other.getJoinType().isSemiOrAntiJoin()
+                        || !other.isLhsStrict()) {
+                    minRightHand = LongBitmap.or(minRightHand, other.getLeftHand());
+                    minRightHand = LongBitmap.or(minRightHand, other.getRightHand());
+                }
+            }
+        }
+
+        JoinConstraint newJoinConstraint = new JoinConstraint(minLeftHand, minRightHand, leftHand, rightHand,
+                join.getJoinType(), isStrict);
+        leading.getJoinConstraintList().add(newJoinConstraint);
+    }
+
+    private long calSlotsTableBitMap(LeadingHint leading, Set<Slot> slots, boolean getNotNullable) {
+        Preconditions.checkArgument(slots.size() != 0);
+        long bitmap = LongBitmap.newBitmap();
+        for (Slot slot : slots) {
+            if (getNotNullable && slot.nullable()) {
+                continue;
+            }
+            if (!slot.isColumnFromTable()) {
+                // we can not get info from column not from table
+                continue;
+            }
+            String tableName = leading.getExprIdToTableNameMap().get(slot.getExprId());
+            if (tableName == null) {
+                tableName = slot.getQualifier().get(slot.getQualifier().size() - 1);
+                leading.getExprIdToTableNameMap().put(slot.getExprId(), tableName);
+            }
+            RelationId id = leading.findRelationIdAndTableName(tableName);
+            if (id == null) {
+                leading.setStatus(Hint.HintStatus.SYNTAX_ERROR);
+                leading.setErrorMessage("can not find table: " + tableName);
+                return bitmap;
+            }
+            long currBitmap = LongBitmap.set(bitmap, id.asInt());
+            bitmap = LongBitmap.or(bitmap, currBitmap);
+        }
+        return bitmap;
+    }
+}

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

@@ -0,0 +1,78 @@
+// 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.hint.Hint;
+import org.apache.doris.nereids.hint.LeadingHint;
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.jobs.joinorder.hypergraph.bitmap.LongBitmap;
+import org.apache.doris.nereids.rules.rewrite.LeadingJoin.LeadingContext;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+
+/**
+ *  Leading join is used to generate leading join and replace original logical join
+*/
+public class LeadingJoin extends DefaultPlanRewriter<LeadingContext> implements CustomRewriter {
+
+    @Override
+    public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+        if (jobContext.getCascadesContext().getStatementContext().isLeadingJoin()) {
+            Hint leadingHint = jobContext.getCascadesContext().getStatementContext().getHintMap().get("Leading");
+            Plan leadingPlan = plan.accept(this, new LeadingContext(
+                    (LeadingHint) leadingHint, ((LeadingHint) leadingHint)
+                        .getLeadingTableBitmap(jobContext.getCascadesContext().getTables())));
+            if (leadingHint.isSuccess()) {
+                jobContext.getCascadesContext().getConnectContext().getSessionVariable().setDisableJoinReorder(true);
+            } else {
+                return plan;
+            }
+            return leadingPlan;
+        }
+        return plan;
+    }
+
+    @Override
+    public Plan visit(Plan plan, LeadingContext context) {
+        Long currentBitMap = LongBitmap.computeTableBitmap(plan.getInputRelations());
+        if (LongBitmap.isSubset(currentBitMap, context.totalBitmap)
+                && plan instanceof LogicalJoin && !context.leading.isSyntaxError()) {
+            Plan leadingJoin = context.leading.generateLeadingJoinPlan();
+            if (context.leading.isSuccess() && leadingJoin != null) {
+                return leadingJoin;
+            }
+        } else {
+            return (LogicalPlan) super.visit(plan, context);
+        }
+        return plan;
+    }
+
+    /** LeadingContext */
+    public static class LeadingContext {
+        public LeadingHint leading;
+        public Long totalBitmap;
+
+        public LeadingContext(LeadingHint leading, Long totalBitmap) {
+            this.leading = leading;
+            this.totalBitmap = totalBitmap;
+        }
+    }
+}

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

@@ -29,7 +29,7 @@ public class SemiJoinCommute extends OneRewriteRuleFactory {
     @Override
     public Rule build() {
         return logicalJoin()
-                .when(join -> join.getJoinType().isRightSemiOrAntiJoin())
+                .when(join -> join.getJoinType().isRightSemiOrAntiJoin() || join.getJoinType().isRightOuterJoin())
                 .whenNot(join -> ConnectContext.get().getSessionVariable().isDisableJoinReorder())
                 .whenNot(LogicalJoin::hasJoinHint)
                 .whenNot(LogicalJoin::isMarkJoin)

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/copier/LogicalPlanDeepCopier.java

@@ -17,7 +17,10 @@
 
 package org.apache.doris.nereids.trees.copier;
 
+import org.apache.doris.common.Pair;
 import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.hint.Hint;
+import org.apache.doris.nereids.hint.LeadingHint;
 import org.apache.doris.nereids.properties.OrderKey;
 import org.apache.doris.nereids.trees.expressions.ExprId;
 import org.apache.doris.nereids.trees.expressions.Expression;
@@ -30,6 +33,7 @@ import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
 import org.apache.doris.nereids.trees.expressions.SubqueryExpr;
 import org.apache.doris.nereids.trees.expressions.functions.Function;
 import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
 import org.apache.doris.nereids.trees.plans.logical.LogicalApply;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAssertNumRows;
@@ -63,6 +67,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
 import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
 import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
 import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -168,6 +173,7 @@ public class LogicalPlanDeepCopier extends DefaultPlanRewriter<DeepCopierContext
                     olapScan.getManuallySpecifiedPartitions(), olapScan.getSelectedTabletIds(),
                     olapScan.getHints());
         }
+        updateLeadingRelationIdMap(newOlapScan.getRelationId(), newOlapScan.getTable().getName(), newOlapScan);
         newOlapScan.getOutput();
         context.putRelation(olapScan.getRelationId(), newOlapScan);
         updateReplaceMapWithOutput(olapScan, newOlapScan, context.exprIdReplaceMap);
@@ -194,6 +200,7 @@ public class LogicalPlanDeepCopier extends DefaultPlanRewriter<DeepCopierContext
         }
         LogicalSchemaScan newSchemaScan = new LogicalSchemaScan(StatementScopeIdGenerator.newRelationId(),
                 schemaScan.getTable(), schemaScan.getQualifier());
+        updateLeadingRelationIdMap(newSchemaScan.getRelationId(), newSchemaScan.getTable().getName(), newSchemaScan);
         updateReplaceMapWithOutput(schemaScan, newSchemaScan, context.exprIdReplaceMap);
         context.putRelation(schemaScan.getRelationId(), newSchemaScan);
         return newSchemaScan;
@@ -206,6 +213,7 @@ public class LogicalPlanDeepCopier extends DefaultPlanRewriter<DeepCopierContext
         }
         LogicalFileScan newFileScan = new LogicalFileScan(StatementScopeIdGenerator.newRelationId(),
                 fileScan.getTable(), fileScan.getQualifier());
+        updateLeadingRelationIdMap(newFileScan.getRelationId(), fileScan.getTable().getName(), newFileScan);
         updateReplaceMapWithOutput(fileScan, newFileScan, context.exprIdReplaceMap);
         context.putRelation(fileScan.getRelationId(), newFileScan);
         Set<Expression> conjuncts = fileScan.getConjuncts().stream()
@@ -233,6 +241,7 @@ public class LogicalPlanDeepCopier extends DefaultPlanRewriter<DeepCopierContext
         }
         LogicalJdbcScan newJdbcScan = new LogicalJdbcScan(StatementScopeIdGenerator.newRelationId(),
                 jdbcScan.getTable(), jdbcScan.getQualifier());
+        updateLeadingRelationIdMap(newJdbcScan.getRelationId(), jdbcScan.getTable().getName(), newJdbcScan);
         updateReplaceMapWithOutput(jdbcScan, newJdbcScan, context.exprIdReplaceMap);
         context.putRelation(jdbcScan.getRelationId(), newJdbcScan);
         return newJdbcScan;
@@ -246,6 +255,7 @@ public class LogicalPlanDeepCopier extends DefaultPlanRewriter<DeepCopierContext
         LogicalEsScan newEsScan = new LogicalEsScan(StatementScopeIdGenerator.newRelationId(),
                 esScan.getTable(), esScan.getQualifier());
         updateReplaceMapWithOutput(esScan, newEsScan, context.exprIdReplaceMap);
+        updateLeadingRelationIdMap(newEsScan.getRelationId(), esScan.getTable().getName(), newEsScan);
         context.putRelation(esScan.getRelationId(), newEsScan);
         return newEsScan;
     }
@@ -429,6 +439,7 @@ public class LogicalPlanDeepCopier extends DefaultPlanRewriter<DeepCopierContext
                 StatementScopeIdGenerator.newRelationId(),
                 cteConsumer.getCteId(), cteConsumer.getName(),
                 consumerToProducerOutputMap, producerToConsumerOutputMap);
+        updateLeadingRelationIdMap(newCTEConsumer.getRelationId(), cteConsumer.getName(), newCTEConsumer);
         context.putRelation(cteConsumer.getRelationId(), newCTEConsumer);
         return newCTEConsumer;
     }
@@ -446,4 +457,13 @@ public class LogicalPlanDeepCopier extends DefaultPlanRewriter<DeepCopierContext
             replaceMap.put(oldOutput.get(i).getExprId(), newOutput.get(i).getExprId());
         }
     }
+
+    private void updateLeadingRelationIdMap(RelationId id, String tableName, LogicalPlan plan) {
+        if (!ConnectContext.get().getStatementContext().isLeadingJoin()) {
+            return;
+        }
+        Hint leading = ConnectContext.get().getStatementContext().getHintMap().get("Leading");
+        ((LeadingHint) leading).updateRelationIdByTableName(Pair.of(id, tableName));
+        ((LeadingHint) leading).getRelationIdToScanMap().put(id, plan);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/Plan.java

@@ -30,6 +30,7 @@ import org.apache.doris.nereids.util.MutableState;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
 
 import java.util.List;
 import java.util.Optional;
@@ -114,6 +115,18 @@ public interface Plan extends TreeNode<Plan> {
         throw new IllegalStateException("Not support compute output for " + getClass().getName());
     }
 
+    /**
+     * Get the input relation ids set of the plan.
+     * @return The result is collected from all inputs relations
+     */
+    default Set<RelationId> getInputRelations() {
+        Set<RelationId> relationIdSet = Sets.newHashSet();
+        children().forEach(
+                plan -> relationIdSet.addAll(plan.getInputRelations())
+        );
+        return relationIdSet;
+    }
+
     String treeString();
 
     Plan withGroupExpression(Optional<GroupExpression> groupExpression);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalJoin.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.nereids.trees.plans.logical;
 
+import org.apache.doris.nereids.jobs.joinorder.hypergraph.bitmap.LongBitmap;
 import org.apache.doris.nereids.memo.GroupExpression;
 import org.apache.doris.nereids.properties.LogicalProperties;
 import org.apache.doris.nereids.rules.exploration.join.JoinReorderContext;
@@ -64,6 +65,8 @@ public class LogicalJoin<LEFT_CHILD_TYPE extends Plan, RIGHT_CHILD_TYPE extends
 
     // Use for top-to-down join reorder
     private final JoinReorderContext joinReorderContext = new JoinReorderContext();
+    // Table bitmap for tables below this join
+    private long bitmap = LongBitmap.newBitmap();
 
     public LogicalJoin(JoinType joinType, LEFT_CHILD_TYPE leftChild, RIGHT_CHILD_TYPE rightChild) {
         this(joinType, ExpressionUtils.EMPTY_CONDITION, ExpressionUtils.EMPTY_CONDITION, JoinHint.NONE,
@@ -94,6 +97,20 @@ public class LogicalJoin<LEFT_CHILD_TYPE extends Plan, RIGHT_CHILD_TYPE extends
                 Optional.empty(), Optional.empty(), leftChild, rightChild);
     }
 
+    public LogicalJoin(
+            long bitmap,
+            JoinType joinType,
+            List<Expression> hashJoinConjuncts,
+            List<Expression> otherJoinConjuncts,
+            JoinHint hint,
+            Optional<MarkJoinSlotReference> markJoinSlotReference,
+            LEFT_CHILD_TYPE leftChild, RIGHT_CHILD_TYPE rightChild) {
+        this(joinType, hashJoinConjuncts,
+                otherJoinConjuncts, hint, markJoinSlotReference,
+                Optional.empty(), Optional.empty(), leftChild, rightChild);
+        this.bitmap = LongBitmap.or(this.bitmap, bitmap);
+    }
+
     public LogicalJoin(
             JoinType joinType,
             List<Expression> hashJoinConjuncts,
@@ -264,6 +281,14 @@ public class LogicalJoin<LEFT_CHILD_TYPE extends Plan, RIGHT_CHILD_TYPE extends
         return markJoinSlotReference;
     }
 
+    public long getBitmap() {
+        return bitmap;
+    }
+
+    public void setBitmap(long bitmap) {
+        this.bitmap = bitmap;
+    }
+
     @Override
     public LEFT_CHILD_TYPE left() {
         return (LEFT_CHILD_TYPE) child(0);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapScan.java

@@ -37,11 +37,13 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 /**
@@ -298,6 +300,13 @@ public class LogicalOlapScan extends LogicalCatalogRelation implements OlapScan
                 .collect(ImmutableList.toImmutableList());
     }
 
+    @Override
+    public Set<RelationId> getInputRelations() {
+        Set<RelationId> relationIdSet = Sets.newHashSet();
+        relationIdSet.add(relationId);
+        return relationIdSet;
+    }
+
     /**
      * Get the slot under the index,
      * and create a new slotReference for the slot that has not appeared in the materialized view.

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalRelation.java

@@ -26,11 +26,13 @@ import org.apache.doris.nereids.trees.plans.algebra.Relation;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
 import org.json.JSONObject;
 
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 
 /**
  * Logical relation plan.
@@ -89,4 +91,11 @@ public abstract class LogicalRelation extends LogicalLeaf implements Relation {
         logicalRelation.put("Properties", properties);
         return logicalRelation;
     }
+
+    @Override
+    public Set<RelationId> getInputRelations() {
+        Set<RelationId> relationIdSet = Sets.newHashSet();
+        relationIdSet.add(relationId);
+        return relationIdSet;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/planner/OriginalPlanner.java

@@ -143,6 +143,13 @@ public class OriginalPlanner extends Planner {
         str.append(plannerContext.getRootAnalyzer().getDescTbl().getExplainString());
     }
 
+    /**
+     * Return hint information.
+     */
+    @Override
+    public void appendHintInfo(StringBuilder str) {
+    }
+
     /**
      * Create plan fragments for an analyzed statement, given a set of execution options. The fragments are returned in
      * a list such that element i of that list can only consume output of the following fragments j > i.

fe/fe-core/src/main/java/org/apache/doris/planner/Planner.java

@@ -85,6 +85,7 @@ public abstract class Planner {
         if (explainLevel == org.apache.doris.thrift.TExplainLevel.VERBOSE) {
             appendTupleInfo(str);
         }
+        appendHintInfo(str);
         return str.toString();
     }
 
@@ -104,6 +105,8 @@ public abstract class Planner {
 
     public void appendTupleInfo(StringBuilder stringBuilder) {}
 
+    public void appendHintInfo(StringBuilder stringBuilder) {}
+
     public List<PlanFragment> getFragments() {
         return fragments;
     }

regression-test/data/nereids_p0/hint/test_leading.out

@@ -0,0 +1,430 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select1 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+--------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t1]
+
+Used: leading(t2 t1)
+UnUsed:
+SyntaxError:
+
+-- !select2 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used: leading(t1 t2)
+UnUsed:
+SyntaxError:
+
+-- !select3 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+--------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t3]
+
+Used: leading(t1 t2 t3)
+UnUsed:
+SyntaxError:
+
+-- !select4 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+------------PhysicalOlapScan[t2]
+------------PhysicalDistribute
+--------------PhysicalOlapScan[t3]
+
+Used: leading(t1 { t2 t3 })
+UnUsed:
+SyntaxError:
+
+-- !select5 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t3.c3 = t4.c4)
+--------PhysicalDistribute
+----------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+------------PhysicalOlapScan[t1]
+------------PhysicalDistribute
+--------------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+----------------PhysicalOlapScan[t2]
+----------------PhysicalDistribute
+------------------PhysicalOlapScan[t3]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t4]
+
+Used: leading(t1 { t2 t3 } t4)
+UnUsed:
+SyntaxError:
+
+-- !select6 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+--------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------hashJoin[INNER_JOIN](t3.c3 = t4.c4)
+------------PhysicalOlapScan[t3]
+------------PhysicalDistribute
+--------------PhysicalOlapScan[t4]
+
+Used: leading({ t1 t2 } { t3 t4 })
+UnUsed:
+SyntaxError:
+
+-- !select7 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+--------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------hashJoin[INNER_JOIN](t3.c3 = t4.c4)
+------------PhysicalOlapScan[t3]
+------------PhysicalDistribute
+--------------PhysicalOlapScan[t4]
+
+Used: leading({ t1 t2 } { t3 t4 })
+UnUsed:
+SyntaxError:
+
+-- !select8 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t3.c3)
+--------hashJoin[LEFT_OUTER_JOIN](t1.c1 = t2.c2)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t3]
+
+-- !select9 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[LEFT_OUTER_JOIN](t1.c1 = t2.c2)
+--------hashJoin[INNER_JOIN](t1.c1 = t3.c3)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t3]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used: leading(t1 t3 t2)
+UnUsed:
+SyntaxError:
+
+-- !select10 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[LEFT_OUTER_JOIN](t1.c1 = t3.c3)
+--------hashJoin[LEFT_OUTER_JOIN](t1.c1 = t2.c2)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t3]
+
+-- !select11 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[LEFT_OUTER_JOIN](t1.c1 = t2.c2)
+--------hashJoin[LEFT_OUTER_JOIN](t1.c1 = t3.c3)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t3]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used: leading(t1 t3 t2)
+UnUsed:
+SyntaxError:
+
+-- !select12 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[LEFT_OUTER_JOIN](t2.c2 = t3.c3)
+--------PhysicalDistribute
+----------hashJoin[LEFT_OUTER_JOIN](t1.c1 = t2.c2)
+------------PhysicalOlapScan[t1]
+------------PhysicalDistribute
+--------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t3]
+
+Used:
+UnUsed: leading(t1 { t2 t3 })
+SyntaxError:
+
+-- !select13 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+------------PhysicalOlapScan[t2]
+------------PhysicalDistribute
+--------------PhysicalOlapScan[t3]
+
+Used: leading(t1 { t2 t3 })
+UnUsed:
+SyntaxError:
+
+-- !select14 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[LEFT_OUTER_JOIN](t1.c1 = tmp.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+------------PhysicalOlapScan[t2]
+------------PhysicalDistribute
+--------------PhysicalOlapScan[t3]
+
+-- !select15 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[LEFT_OUTER_JOIN](t1.c1 = tmp.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+------------PhysicalOlapScan[t2]
+------------PhysicalDistribute
+--------------PhysicalOlapScan[t3]
+
+Used:
+UnUsed: leading(t1 t2 t3)
+SyntaxError:
+
+-- !select16 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[LEFT_SEMI_JOIN](t1.c1 = t2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalProject
+------------PhysicalOlapScan[t2]
+
+-- !select17 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[RIGHT_SEMI_JOIN](t1.c1 = t2.c2)
+--------PhysicalProject
+----------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t1]
+
+Used: leading(t2 t1)
+UnUsed:
+SyntaxError:
+
+-- !select18 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------NestedLoopJoin[CROSS_JOIN]
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalLimit
+------------PhysicalDistribute
+--------------PhysicalLimit
+----------------PhysicalProject
+------------------PhysicalOlapScan[t2]
+
+-- !select19 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------NestedLoopJoin[CROSS_JOIN]
+--------PhysicalProject
+----------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t1]
+
+Used: leading(t2 t1)
+UnUsed:
+SyntaxError:
+
+-- !select20 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------NestedLoopJoin[CROSS_JOIN]
+--------hashJoin[INNER_JOIN](cte.c1 = cte.c2)
+----------PhysicalProject
+------------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalProject
+--------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+-- !select21 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------NestedLoopJoin[CROSS_JOIN]
+--------NestedLoopJoin[CROSS_JOIN](cte.c1 = cte.c2)
+----------PhysicalProject
+------------PhysicalOlapScan[t2]
+----------PhysicalDistribute
+------------PhysicalProject
+--------------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t3]
+
+Used: leading(t2 t1 t3)
+UnUsed:
+SyntaxError:
+
+-- !select22 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used:
+UnUsed:
+SyntaxError: leading(t66 t1) Msg:can not find table: t66
+
+-- !select23 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used:
+UnUsed:
+SyntaxError: leading(t3 t1) Msg:can not find table: t3
+
+-- !select24 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------NestedLoopJoin[CROSS_JOIN]
+--------hashJoin[INNER_JOIN](cte.c1 = cte.c2)
+----------PhysicalProject
+------------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalProject
+--------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used:
+UnUsed:
+SyntaxError: leading(t2 cte t1) Msg:Leading alias can only be table name alias
+
+-- !select25 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+--------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t3]
+
+Used:
+UnUsed:
+SyntaxError: leading(t1 t2) Msg:tables should be same as join tables
+
+-- !select26 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t2.c2 = t3.c3)
+--------hashJoin[INNER_JOIN](t1.c1 = t2.c2)
+----------PhysicalOlapScan[t1]
+----------PhysicalDistribute
+------------PhysicalOlapScan[t2]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t3]
+
+Used:
+UnUsed:
+SyntaxError: leading(t1 t1 t2 t3) Msg:duplicated table
+
+-- !select27 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t_2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used: leading(t1 t_2)
+UnUsed:
+SyntaxError:
+
+-- !select28 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t_2.c2)
+--------PhysicalOlapScan[t1]
+--------PhysicalDistribute
+----------PhysicalOlapScan[t2]
+
+Used:
+UnUsed:
+SyntaxError: leading(t1 t2) Msg:can not find table: t2
+
+-- !select29 --
+PhysicalResultSink
+--PhysicalDistribute
+----PhysicalProject
+------hashJoin[INNER_JOIN](t1.c1 = t_1.c1)
+--------PhysicalOlapScan[t1]
+--------PhysicalOlapScan[t1]
+
+Used: leading(t1 t_1)
+UnUsed:
+SyntaxError:

regression-test/suites/nereids_p0/hint/test_leading.groovy

@@ -0,0 +1,108 @@
+/*
+ * 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.
+ */
+
+suite("test_leading") {
+    // create database and tables
+    sql 'DROP DATABASE IF EXISTS test_leading'
+    sql 'CREATE DATABASE IF NOT EXISTS test_leading'
+    sql 'use test_leading'
+
+    // setting planner to nereids
+    sql 'set enable_nereids_planner=true'
+    sql 'set enable_fallback_to_original_planner=false'
+
+    // create tables
+    sql """drop table if exists t1;"""
+    sql """drop table if exists t2;"""
+    sql """drop table if exists t3;"""
+    sql """drop table if exists t4;"""
+
+    sql """create table t1 (c1 int, c11 int) distributed by hash(c1) buckets 3 properties('replication_num' = '1');"""
+    sql """create table t2 (c2 int, c22 int) distributed by hash(c2) buckets 3 properties('replication_num' = '1');"""
+    sql """create table t3 (c3 int, c33 int) distributed by hash(c3) buckets 3 properties('replication_num' = '1');"""
+    sql """create table t4 (c4 int, c44 int) distributed by hash(c4) buckets 3 properties('replication_num' = '1');"""
+
+//// test inner join with all edge and vertax is complete and equal predicates
+    qt_select1 """explain shape plan select /*+ leading(t2 t1) */ * from t1 join t2 on c1 = c2;"""
+    qt_select2 """explain shape plan select /*+ leading(t1 t2) */ * from t1 join t2 on c1 = c2;"""
+    qt_select3 """explain shape plan select /*+ leading(t1 t2 t3) */ * from t1 join t2 on c1 = c2 join t3 on c2 = c3;"""
+    qt_select4 """explain shape plan select /*+ leading(t1 {t2 t3}) */ * from t1 join t2 on c1 = c2 join t3 on c2 = c3;"""
+    qt_select5 """explain shape plan select /*+ leading(t1 {t2 t3} t4) */ * from t1 join t2 on c1 = c2 join t3 on c2 = c3 join t4 on c3 = c4;"""
+    qt_select6 """explain shape plan select /*+ leading({t1 t2} {t3 t4}) */ * from t1 join t2 on c1 = c2 join t3 on c2 = c3 join t4 on c3 = c4;"""
+
+    // test inner join with part of edge and need cross join
+    qt_select7 """explain shape plan select /*+ leading({t1 t2} {t3 t4}) */ * from t1 join t2 on c1 = c2 join t3 on c2 = c3 join t4 on c3 = c4;"""
+
+//// test outer join which can swap
+    // (A leftjoin B on (Pab)) innerjoin C on (Pac) = (A innerjoin C on (Pac)) leftjoin B on (Pab)
+    qt_select8 """explain shape plan select * from t1 left join t2 on c1 = c2 join t3 on c1 = c3;"""
+    qt_select9 """explain shape plan select /*+ leading(t1 t3 t2) */ * from t1 left join t2 on c1 = c2 join t3 on c1 = c3;"""
+    
+    // (A leftjoin B on (Pab)) leftjoin C on (Pac) = (A leftjoin C on (Pac)) leftjoin B on (Pab)
+    qt_select10 """explain shape plan select * from t1 left join t2 on c1 = c2 left join t3 on c1 = c3;"""
+    qt_select11 """explain shape plan select /*+ leading(t1 t3 t2) */ * from t1 left join t2 on c1 = c2 left join t3 on c1 = c3;"""
+
+    // (A leftjoin B on (Pab)) leftjoin C on (Pbc) = A leftjoin (B leftjoin C on (Pbc)) on (Pab)
+    qt_select12 """explain shape plan select /*+ leading(t1 {t2 t3}) */ * from t1 left join t2 on c1 = c2 left join t3 on c2 = c3;"""
+    
+//// test outer join which can not swap
+    // A leftjoin (B join C on (Pbc)) on (Pab) != (A leftjoin B on (Pab)) join C on (Pbc) output should be unused when explain
+    // this can be done because left join can be eliminated to inner join
+    qt_select13 """explain shape plan select /*+ leading(t1 {t2 t3}) */ * from t1 left join t2 on c1 = c2 join t3 on c2 = c3;"""
+
+    // this can not be done, expect not success but return right deep tree
+    qt_select14 """explain shape plan select * from t1 left join (select * from t2 join t3 on c2 = c3) as tmp on c1 = c2;"""
+    qt_select15 """explain shape plan select /*+ leading(t1 t2 t3) */ * from t1 left join (select * from t2 join t3 on c2 = c3) as tmp on c1 = c2;"""
+
+//// test semi join
+    qt_select16 """explain shape plan select * from t1 where c1 in (select c2 from t2);"""
+    qt_select17 """explain shape plan select /*+ leading(t2 t1) */ * from t1 where c1 in (select c2 from t2);"""
+
+//// test anti join
+    qt_select18 """explain shape plan select * from t1 where exists (select 1 from t2);"""
+    qt_select19 """explain shape plan select /*+ leading (t2 t1) */ * from t1 where exists (select 1 from t2);"""
+
+//// test cte
+    // inline cte, change join order of tables inside cte
+    qt_select20 """explain shape plan with cte as (select * from t1 join t2 on c1 = c2) select * from cte, t2;"""
+    qt_select21 """explain shape plan with cte as (select * from t1 join t2 on c1 = c2) select /*+ leading(t2 t1 t3) */ * from cte, t3;"""
+    // outside cte
+    // inside and outside together (after unnest subquery)
+
+//// test syntax error and unsupported feature
+    // not exist tables in leading: syntax error
+    qt_select22 """explain shape plan select /*+ leading(t66 t1) */ * from t1 join t2 on c1 = c2;"""
+    qt_select23 """explain shape plan select /*+ leading(t3 t1) */ * from t1 join t2 on c1 = c2;"""
+    // subquery alias as leading table
+    qt_select24 """explain shape plan with cte as (select * from t1 join t2 on c1 = c2) select /*+ leading(t2 cte t1) */ * from cte, t2;"""
+    // do not have all tables inside hint
+    qt_select25 """explain shape plan select /*+ leading(t1 t2) */ * from t1 join t2 on c1 = c2 join t3 on c2 = c3;"""
+    // duplicated table
+    qt_select26 """explain shape plan select /*+ leading(t1 t1 t2 t3) */ * from t1 join t2 on c1 = c2 join t3 on c2 = c3;"""
+
+//// test table alias
+    qt_select27 """explain shape plan select /*+ leading(t1 t_2) */ * from t1 join t2 t_2 on c1 = c2;"""
+    qt_select28 """explain shape plan select /*+ leading(t1 t2) */ * from t1 join t2 t_2 on c1 = c2;"""
+    qt_select29 """explain shape plan select /*+ leading(t1 t_1) */ * from t1 join t1 t_1 on t1.c1 = t_1.c1;"""
+
+    sql """drop table if exists t1;"""
+    sql """drop table if exists t2;"""
+    sql """drop table if exists t3;"""
+    sql """drop table if exists t4;"""
+}