[fix](sync mv) fix rewrite wrongly with sync mv (#44866)

Cherry-picked from #39284

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

@@ -144,6 +144,21 @@ protected static List<Expression> getPrunedPredicates(List<Expression> aggExpres
         return prunedExpr;
     }
 
+    protected static boolean containAllKeyColumns(OlapTable table, MaterializedIndex index) {
+        if (table.getKeysType() == KeysType.UNIQUE_KEYS) {
+            return true;
+        }
+        Set<String> mvColNames = table.getKeyColumnsByIndexId(index.getId()).stream()
+                .map(c -> normalizeName(parseMvColumnToSql(c.getNameWithoutMvPrefix())))
+                .collect(Collectors.toCollection(() -> new TreeSet<String>(String.CASE_INSENSITIVE_ORDER)));
+
+        Set<String> keyColNames = table.getBaseSchemaKeyColumns().stream()
+                .map(c -> normalizeName(parseMvColumnToSql(c.getNameWithoutMvPrefix())))
+                .collect(Collectors.toCollection(() -> new TreeSet<String>(String.CASE_INSENSITIVE_ORDER)));
+
+        return keyColNames.containsAll(mvColNames);
+    }
+
     protected static boolean containAllRequiredColumns(MaterializedIndex index, LogicalOlapScan scan,
             Set<Slot> requiredScanOutput, Set<? extends Expression> requiredExpr, Set<Expression> predicateExpr) {
         OlapTable table = scan.getTable();

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

@@ -219,6 +219,7 @@ public static LogicalOlapScan select(
             // So only base index and indexes that have all the keys could be used.
             List<MaterializedIndex> candidates = table.getVisibleIndex().stream()
                     .filter(index -> table.getKeyColumnsByIndexId(index.getId()).size() == baseIndexKeySize)
+                    .filter(index -> containAllKeyColumns(table, index))
                     .filter(index -> containAllRequiredColumns(index, scan, requiredScanOutputSupplier.get(),
                             requiredExpr.get(), predicatesSupplier.get()))
                     .collect(Collectors.toList());

regression-test/data/nereids_rules_p0/mv/join/inner/inner_join_x.out

@@ -0,0 +1,25 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !query_before --
+1
+2
+2
+2
+
+-- !query_after --
+1
+2
+2
+2
+
+-- !query_before --
+1
+2
+2
+2
+
+-- !query_after --
+1
+2
+2
+2
+

regression-test/suites/nereids_rules_p0/mv/join/inner/inner_join_x.groovy

@@ -0,0 +1,127 @@
+// 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("inner_join_x") {
+    String db = context.config.getDbNameByFile(context.file)
+    sql "use ${db}"
+    sql "set runtime_filter_mode=OFF"
+
+
+    // =======================  test table with aggregate key ============================
+    sql """
+    drop table if  exists t1;
+    """
+
+    sql """
+    CREATE TABLE IF NOT EXISTS t1 (
+                k int,
+                a int,
+                int_value int sum,
+                char_value char(10) max,
+                date_value date max
+            )
+            ENGINE=OLAP
+            aggregate KEY(k,a)
+            DISTRIBUTED BY HASH(k) BUCKETS 2 properties("replication_num" = "1")
+
+    """
+
+    def mv_name="v_t1"
+    createMV ( """
+     create materialized view ${mv_name} as select k%2 as kk,a, sum(int_value), max(date_value) from t1 group by kk, a;
+    """)
+
+    sql """
+    insert into t1 values
+    (1,1,1,'a', '2020-12-01'),
+    (2,2,2,'b', '2021-12-01'),
+    (3,2,2,'c', '2022-12-01'),
+    (4,2,4,'c', '2023-12-01');
+    """
+
+    def query =  """
+    select a from t1
+    """
+
+    explain {
+        sql("${query}")
+        notContains("${mv_name}(${mv_name})")
+    }
+
+    order_qt_query_before "${query}"
+
+
+    sql """ DROP MATERIALIZED VIEW IF EXISTS  ${mv_name} on t1"""
+
+    order_qt_query_after "${query}"
+
+    sql """
+    drop table if exists t1 
+    """
+
+    // =======================  test table with duplicate key ============================
+    sql """
+    drop table if  exists t1;
+    """
+
+    sql """
+    CREATE TABLE IF NOT EXISTS t1 (
+                k int,
+                a int,
+                int_value int,
+                char_value char(10),
+                date_value date
+            )
+            ENGINE=OLAP
+            duplicate KEY(k,a)
+            DISTRIBUTED BY HASH(k) BUCKETS 2 properties("replication_num" = "1")
+
+    """
+
+    mv_name="v_t1"
+    createMV ( """
+     create materialized view ${mv_name} as select k%2 as kk,a, sum(int_value), max(date_value) from t1 group by kk, a;
+    """)
+
+    sql """
+    insert into t1 values
+    (1,1,1,'a', '2020-12-01'),
+    (2,2,2,'b', '2021-12-01'),
+    (3,2,2,'c', '2022-12-01'),
+    (4,2,4,'c', '2023-12-01');
+    """
+
+    query =  """
+    select a from t1
+    """
+
+    explain {
+        sql("${query}")
+        notContains("t1(${mv_name})")
+    }
+
+    order_qt_query_before "${query}"
+
+
+    sql """ DROP MATERIALIZED VIEW IF EXISTS  ${mv_name} on t1"""
+
+    order_qt_query_after "${query}"
+
+    sql """
+    drop table if exists t1 
+    """
+}