[improvement](mtmv) Support rewrite by materialized view when join has other join conjuncts

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/joinorder/hypergraph/edge/Edge.java

@@ -225,10 +225,6 @@ public String getTypeName() {
 
     @Override
     public String toString() {
-        if (!leftRejectEdges.isEmpty() || !rightRejectEdges.isEmpty()) {
-            return String.format("<%s --%s-- %s>[%s , %s]", LongBitmap.toString(leftExtendedNodes),
-                    this.getTypeName(), LongBitmap.toString(rightExtendedNodes), leftRejectEdges, rightRejectEdges);
-        }
         return String.format("<%s --%s-- %s>", LongBitmap.toString(leftExtendedNodes),
                 this.getTypeName(), LongBitmap.toString(rightExtendedNodes));
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/HyperGraphComparator.java

@@ -455,7 +455,9 @@ private boolean compareEdgeWithNode(Edge query, Edge view) {
         if (query instanceof FilterEdge && view instanceof FilterEdge) {
             return compareFilterEdgeWithNode((FilterEdge) query, viewFilterEdgesAfterInferring.get(view.getIndex()));
         } else if (query instanceof JoinEdge && view instanceof JoinEdge) {
-            return compareJoinEdgeWithNode((JoinEdge) query, viewJoinEdgesAfterInferring.get(view.getIndex()));
+            // compare original or inferred join edge
+            return compareJoinEdgeWithNode((JoinEdge) query, viewJoinEdgesAfterInferring.get(view.getIndex()))
+                    || compareJoinEdgeWithNode((JoinEdge) query, (JoinEdge) view);
         }
         return false;
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/StructInfo.java

@@ -210,28 +210,25 @@ private static boolean collectStructInfoFromGraph(HyperGraph hyperGraph,
         });
         // Collect expression from join condition in hyper graph
         for (JoinEdge edge : hyperGraph.getJoinEdges()) {
-            List<Expression> hashJoinConjuncts = edge.getHashJoinConjuncts();
+            List<? extends Expression> joinConjunctExpressions = edge.getExpressions();
             // shuttle expression in edge for the build of LogicalCompatibilityContext later.
             // Record the exprId to expr map in the processing to strut info
             // TODO get exprId to expr map when complex project is ready in join dege
-            hashJoinConjuncts.forEach(conjunctExpr -> {
-                ExpressionLineageReplacer.ExpressionReplaceContext replaceContext =
-                        new ExpressionLineageReplacer.ExpressionReplaceContext(
-                                Lists.newArrayList(conjunctExpr), ImmutableSet.of(),
-                                ImmutableSet.of(), new BitSet());
-                topPlan.accept(ExpressionLineageReplacer.INSTANCE, replaceContext);
-                // Replace expressions by expression map
-                List<Expression> replacedExpressions = replaceContext.getReplacedExpressions();
+            ExpressionLineageReplacer.ExpressionReplaceContext replaceContext =
+                    new ExpressionLineageReplacer.ExpressionReplaceContext(
+                            joinConjunctExpressions.stream().map(expr -> (Expression) expr)
+                                    .collect(Collectors.toList()),
+                            ImmutableSet.of(), ImmutableSet.of(), new BitSet());
+            topPlan.accept(ExpressionLineageReplacer.INSTANCE, replaceContext);
+            // Replace expressions by expression map
+            List<Expression> replacedExpressions = replaceContext.getReplacedExpressions();
+            for (int i = 0; i < replacedExpressions.size(); i++) {
                 putShuttledExpressionsToExpressionsMap(shuttledExpressionsToExpressionsMap,
-                        ExpressionPosition.JOIN_EDGE, replacedExpressions.get(0), conjunctExpr);
-                // Record this, will be used in top level expression shuttle later, see the method
-                // ExpressionLineageReplacer#visitGroupPlan
-                namedExprIdAndExprMapping.putAll(replaceContext.getExprIdExpressionMap());
-            });
-            List<Expression> otherJoinConjuncts = edge.getOtherJoinConjuncts();
-            if (!otherJoinConjuncts.isEmpty()) {
-                return false;
+                        ExpressionPosition.JOIN_EDGE, replacedExpressions.get(i), joinConjunctExpressions.get(i));
             }
+            // Record this, will be used in top level expression shuttle later, see the method
+            // ExpressionLineageReplacer#visitGroupPlan
+            namedExprIdAndExprMapping.putAll(replaceContext.getExprIdExpressionMap());
         }
         // Collect expression from where in hyper graph
         hyperGraph.getFilterEdges().forEach(filterEdge -> {
@@ -621,9 +618,6 @@ public Boolean visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join
             if (!checkContext.getSupportJoinTypes().contains(join.getJoinType())) {
                 return false;
             }
-            if (!join.getOtherJoinConjuncts().isEmpty()) {
-                return false;
-            }
             return visit(join, checkContext);
         }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/mapping/RelationMapping.java

@@ -29,9 +29,9 @@
 import com.google.common.collect.ImmutableBiMap.Builder;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -92,39 +92,25 @@ public static List<RelationMapping> generate(List<CatalogRelation> sources, List
             }
             // relation appear more than once, should cartesian them and power set to correct combination
             // if query is select * from tableA0, tableA1, materialized view is select * from tableA2, tableA3,
-            // tableA is the same table used by both query and materialized view
-            // relationMapping will be
-            // tableA0 tableA2
-            // tableA0 tableA3
-            // tableA1 tableA2
-            // tableA1 tableA3
-            ImmutableList<Pair<MappedRelation, MappedRelation>> relationMapping = Sets.cartesianProduct(
-                            sourceMappedRelations, targetMappedRelations)
-                    .stream()
-                    .map(listPair -> Pair.of(listPair.get(0), listPair.get(1)))
-                    .collect(ImmutableList.toImmutableList());
-
-            // the mapping in relationMappingPowerList should be bi-direction
+            // the relationMappingPowerList in relationMappingPowerList should be bi-direction
             // [
-            //    {tableA0 tableA2, tableA1 tableA3}
-            //    {tableA0 tableA3, tableA1 tableA2}
+            //    {tableA0 -> tableA2, tableA1 -> tableA3}
+            //    {tableA0 -> tableA3, tableA1 -> tableA2}
             // ]
+            // query is select * from tableA0, tableA1, tableA4
             List<BiMap<MappedRelation, MappedRelation>> relationMappingPowerList = new ArrayList<>();
-            int relationMappingSize = relationMapping.size();
-            int relationMappingMinSize = Math.min(sourceMappedRelations.size(), targetMappedRelations.size());
-            for (int i = 0; i < relationMappingSize; i++) {
-                HashBiMap<MappedRelation, MappedRelation> relationBiMap = HashBiMap.create();
-                relationBiMap.put(relationMapping.get(i).key(), relationMapping.get(i).value());
-                for (int j = i + 1; j < relationMappingSize; j++) {
-                    if (!relationBiMap.containsKey(relationMapping.get(j).key())
-                            && !relationBiMap.containsValue(relationMapping.get(j).value())) {
-                        relationBiMap.put(relationMapping.get(j).key(), relationMapping.get(j).value());
-                    }
-                }
-                // mapping should contain min num of relation in source or target at least
-                if (relationBiMap.size() >= relationMappingMinSize) {
-                    relationMappingPowerList.add(relationBiMap);
+            List<Pair<MappedRelation[], MappedRelation[]>> combinations = getUniquePermutation(
+                    sourceMappedRelations.toArray(sourceMappedRelations.toArray(new MappedRelation[0])),
+                    targetMappedRelations.toArray(new MappedRelation[0]));
+            for (Pair<MappedRelation[], MappedRelation[]> combination : combinations) {
+                BiMap<MappedRelation, MappedRelation> combinationBiMap = HashBiMap.create();
+                MappedRelation[] key = combination.key();
+                MappedRelation[] value = combination.value();
+                int length = Math.min(key.length, value.length);
+                for (int i = 0; i < length; i++) {
+                    combinationBiMap.put(key[i], value[i]);
                 }
+                relationMappingPowerList.add(combinationBiMap);
             }
             mappedRelations.add(relationMappingPowerList);
         }
@@ -167,4 +153,58 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(mappedRelationMap);
     }
+
+    /**
+     * Permutation and remove duplicated element
+     * For example:
+     * Given [1, 4, 5] and [191, 194, 195]
+     * This would return
+     * [
+     * [(1, 191) (4, 194) (5, 195)],
+     * [(1, 191) (4, 195) (5, 194)],
+     * [(1, 194) (4, 191) (5, 195)],
+     * [(1, 194) (4, 195) (5, 191)],
+     * [(1, 195) (4, 191) (5, 194)],
+     * [(1, 195) (4, 194) (5, 191)]
+     * ]
+     * */
+    private static List<Pair<MappedRelation[], MappedRelation[]>> getUniquePermutation(
+            MappedRelation[] left, MappedRelation[] right) {
+        boolean needSwap = left.length > right.length;
+        if (needSwap) {
+            MappedRelation[] temp = left;
+            left = right;
+            right = temp;
+        }
+
+        boolean[] used = new boolean[right.length];
+        MappedRelation[] current = new MappedRelation[left.length];
+        List<Pair<MappedRelation[], MappedRelation[]>> results = new ArrayList<>();
+        backtrack(left, right, 0, used, current, results);
+        if (needSwap) {
+            List<Pair<MappedRelation[], MappedRelation[]>> tmpResults = results;
+            results = new ArrayList<>();
+            for (Pair<MappedRelation[], MappedRelation[]> relation : tmpResults) {
+                results.add(Pair.of(relation.value(), relation.key()));
+            }
+        }
+        return results;
+    }
+
+    private static void backtrack(MappedRelation[] left, MappedRelation[] right, int index,
+            boolean[] used, MappedRelation[] current, List<Pair<MappedRelation[], MappedRelation[]>> results) {
+        if (index == left.length) {
+            results.add(Pair.of(Arrays.copyOf(left, left.length), Arrays.copyOf(current, current.length)));
+            return;
+        }
+
+        for (int i = 0; i < right.length; i++) {
+            if (!used[i]) {
+                used[i] = true;
+                current[index] = right[i];
+                backtrack(left, right, index + 1, used, current, results);
+                used[i] = false;
+            }
+        }
+    }
 }