[fix](nereids)fix bug of select mv in nereids

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

@@ -77,7 +77,28 @@
  * Base class for selecting materialized index rules.
  */
 public abstract class AbstractSelectMaterializedIndexRule {
-    protected boolean shouldSelectIndex(LogicalOlapScan scan) {
+    protected boolean shouldSelectIndexWithAgg(LogicalOlapScan scan) {
+        switch (scan.getTable().getKeysType()) {
+            case AGG_KEYS:
+            case UNIQUE_KEYS:
+            case DUP_KEYS:
+                // SelectMaterializedIndexWithAggregate(R1) run before SelectMaterializedIndexWithoutAggregate(R2)
+                // if R1 selects baseIndex and preAggStatus is off
+                // we should give a chance to R2 to check if some prefix-index can be selected
+                // so if R1 selects baseIndex and preAggStatus is off, we keep scan's index unselected in order to
+                // let R2 to get a chance to do its work
+                // at last, after R1, the scan may be the 4 status
+                // 1. preAggStatus is ON and baseIndex is selected, it means select baseIndex is correct.
+                // 2. preAggStatus is ON and some other Index is selected, this is correct, too.
+                // 3. preAggStatus is OFF, no index is selected, it means R2 could get a chance to run
+                // so we check the preAggStatus and if some index is selected to make sure R1 can be run only once
+                return scan.getPreAggStatus().isOn() && !scan.isIndexSelected();
+            default:
+                return false;
+        }
+    }
+
+    protected boolean shouldSelectIndexWithoutAgg(LogicalOlapScan scan) {
         switch (scan.getTable().getKeysType()) {
             case AGG_KEYS:
             case UNIQUE_KEYS:

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

@@ -24,6 +24,8 @@
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.RuleType;
 import org.apache.doris.nereids.rules.rewrite.RewriteRuleFactory;
+import org.apache.doris.nereids.rules.rewrite.mv.AbstractSelectMaterializedIndexRule.ReplaceExpressions;
+import org.apache.doris.nereids.rules.rewrite.mv.AbstractSelectMaterializedIndexRule.SlotContext;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.plans.PreAggStatus;
@@ -59,7 +61,7 @@ public List<Rule> buildRules() {
         return ImmutableList.of(
                 // project with pushdown filter.
                 // Project(Filter(Scan))
-                logicalProject(logicalFilter(logicalOlapScan().when(this::shouldSelectIndex)))
+                logicalProject(logicalFilter(logicalOlapScan().when(this::shouldSelectIndexWithoutAgg)))
                         .thenApply(ctx -> {
                             LogicalProject<LogicalFilter<LogicalOlapScan>> project = ctx.root;
                             LogicalFilter<LogicalOlapScan> filter = project.child();
@@ -79,7 +81,7 @@ public List<Rule> buildRules() {
 
                 // project with filter that cannot be pushdown.
                 // Filter(Project(Scan))
-                logicalFilter(logicalProject(logicalOlapScan().when(this::shouldSelectIndex)))
+                logicalFilter(logicalProject(logicalOlapScan().when(this::shouldSelectIndexWithoutAgg)))
                         .thenApply(ctx -> {
                             LogicalFilter<LogicalProject<LogicalOlapScan>> filter = ctx.root;
                             LogicalProject<LogicalOlapScan> project = filter.child();
@@ -98,13 +100,14 @@ public List<Rule> buildRules() {
 
                 // scan with filters could be pushdown.
                 // Filter(Scan)
-                logicalFilter(logicalOlapScan().when(this::shouldSelectIndex))
+                logicalFilter(logicalOlapScan().when(this::shouldSelectIndexWithoutAgg))
                         .thenApply(ctx -> {
                             LogicalFilter<LogicalOlapScan> filter = ctx.root;
                             LogicalOlapScan scan = filter.child();
                             LogicalOlapScan mvPlan = select(
                                     scan, filter::getOutputSet, filter::getConjuncts,
-                                    new HashSet<>(filter.getExpressions()));
+                                    Stream.concat(filter.getExpressions().stream(),
+                                            filter.getOutputSet().stream()).collect(ImmutableSet.toImmutableSet()));
                             SlotContext slotContext = generateBaseScanExprToMvExpr(mvPlan);
 
                             return new LogicalProject(
@@ -116,7 +119,7 @@ public List<Rule> buildRules() {
 
                 // project and scan.
                 // Project(Scan)
-                logicalProject(logicalOlapScan().when(this::shouldSelectIndex))
+                logicalProject(logicalOlapScan().when(this::shouldSelectIndexWithoutAgg))
                         .thenApply(ctx -> {
                             LogicalProject<LogicalOlapScan> project = ctx.root;
                             LogicalOlapScan scan = project.child();
@@ -135,7 +138,7 @@ public List<Rule> buildRules() {
 
                 // only scan.
                 logicalOlapScan()
-                        .when(this::shouldSelectIndex)
+                        .when(this::shouldSelectIndexWithoutAgg)
                         .thenApply(ctx -> {
                             LogicalOlapScan scan = ctx.root;
 
@@ -196,7 +199,9 @@ private LogicalOlapScan select(
                 // PreAggStatus could be enabled by pre-aggregation hint for agg-keys and unique-keys.
                 preAggStatus = PreAggStatus.on();
             } else {
-                preAggStatus = PreAggStatus.off("No aggregate on scan.");
+                // if PreAggStatus is OFF, we use the message from SelectMaterializedIndexWithAggregate
+                preAggStatus = scan.getPreAggStatus().isOff() ? scan.getPreAggStatus()
+                        : PreAggStatus.off("No aggregate on scan.");
             }
             if (table.getIndexIdToMeta().size() == 1) {
                 return scan.withMaterializedIndexSelected(preAggStatus, baseIndexId);

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

@@ -178,8 +178,12 @@ public OlapTable getTable() {
 
     @Override
     public String toString() {
-        return Utils.toSqlString("LogicalOlapScan", "qualified", qualifiedName(), "indexName",
-                getSelectedMaterializedIndexName(), "selectedIndexId", selectedIndexId, "preAgg", preAggStatus);
+        return Utils.toSqlString("LogicalOlapScan",
+                "qualified", qualifiedName(),
+                "indexName", getSelectedMaterializedIndexName().orElse("<index_not_selected>"),
+                "selectedIndexId", selectedIndexId,
+                "preAgg", preAggStatus
+        );
     }
 
     @Override
@@ -287,8 +291,9 @@ public PreAggStatus getPreAggStatus() {
     }
 
     @VisibleForTesting
-    public String getSelectedMaterializedIndexName() {
-        return ((OlapTable) table).getIndexNameById(selectedIndexId);
+    public Optional<String> getSelectedMaterializedIndexName() {
+        return indexSelected ? Optional.ofNullable(((OlapTable) table).getIndexNameById(selectedIndexId))
+                : Optional.empty();
     }
 
     @Override

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

@@ -875,11 +875,18 @@ void testTwoTupleInQuery() throws Exception {
         String query = "select * from (select user_id, bitmap_union_count(to_bitmap(tag_id)) x from "
                 + USER_TAG_TABLE_NAME + " group by user_id) a, (select user_name, bitmap_union_count(to_bitmap(tag_id))"
                 + "" + " y from " + USER_TAG_TABLE_NAME + " group by user_name) b where a.x=b.y;";
-        PlanChecker.from(connectContext).analyze(query).rewrite().matches(logicalJoin(
-                        logicalProject(logicalAggregate(logicalOlapScan()
-                                        .when(scan -> "user_tags_mv".equals(scan.getSelectedMaterializedIndexName())))),
-                        logicalAggregate(logicalProject(logicalOlapScan()
-                                        .when(scan -> "user_tags".equals(scan.getSelectedMaterializedIndexName()))))));
+        PlanChecker.from(connectContext)
+                .analyze(query)
+                .rewrite()
+                .matches(logicalJoin(
+                        logicalProject(
+                                logicalAggregate(
+                                        logicalOlapScan().when(scan -> "user_tags_mv".equals(
+                                                scan.getSelectedMaterializedIndexName().get())))),
+                        logicalAggregate(
+                                logicalProject(
+                                        logicalOlapScan().when(scan -> "user_tags".equals(
+                                                scan.getSelectedMaterializedIndexName().get()))))));
 
     }
 

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

@@ -112,7 +112,7 @@ public void testMatchingBase() {
                 .applyTopDown(new SelectMaterializedIndexWithAggregate())
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertTrue(scan.getPreAggStatus().isOn());
-                    Assertions.assertEquals("t", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("t", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -124,7 +124,7 @@ void testAggFilterScan() {
                 .applyTopDown(new SelectMaterializedIndexWithAggregate())
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertTrue(scan.getPreAggStatus().isOn());
-                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -139,6 +139,9 @@ void testTranslate() {
     public void testTranslateWhenPreAggIsOff() {
         singleTableTest("select k2, min(v1) from t group by k2", scan -> {
             Assertions.assertFalse(scan.isPreAggregation());
+            Assertions.assertEquals("Aggregate operator don't match, "
+                            + "aggregate function: min(v1), column aggregate type: SUM",
+                    scan.getReasonOfPreAggregation());
         });
     }
 
@@ -149,7 +152,7 @@ public void testWithEqualFilter() {
                 .applyTopDown(new SelectMaterializedIndexWithAggregate())
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertTrue(scan.getPreAggStatus().isOn());
-                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -161,7 +164,7 @@ public void testWithNonEqualFilter() {
                 .applyTopDown(new SelectMaterializedIndexWithAggregate())
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertTrue(scan.getPreAggStatus().isOn());
-                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -173,7 +176,7 @@ public void testWithFilter() {
                 .applyTopDown(new SelectMaterializedIndexWithAggregate())
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertTrue(scan.getPreAggStatus().isOn());
-                    Assertions.assertEquals("r1", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r1", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -192,7 +195,7 @@ public void testWithFilterAndProject() {
                 .applyTopDown(new SelectMaterializedIndexWithoutAggregate())
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertTrue(scan.getPreAggStatus().isOn());
-                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r2", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -224,6 +227,8 @@ public void testAggregateTypeNotMatch() {
                 .matches(logicalOlapScan().when(scan -> {
                     PreAggStatus preAgg = scan.getPreAggStatus();
                     Assertions.assertTrue(preAgg.isOff());
+                    Assertions.assertEquals("Aggregate operator don't match, "
+                            + "aggregate function: min(v1), column aggregate type: SUM", preAgg.getOffReason());
                     return true;
                 }));
     }
@@ -237,6 +242,8 @@ public void testInvalidSlotInAggFunction() {
                 .matches(logicalOlapScan().when(scan -> {
                     PreAggStatus preAgg = scan.getPreAggStatus();
                     Assertions.assertTrue(preAgg.isOff());
+                    Assertions.assertEquals("Slot((v1 + 1)) in sum((v1 + 1)) is neither key column nor value column.",
+                            preAgg.getOffReason());
                     return true;
                 }));
     }
@@ -250,6 +257,8 @@ public void testKeyColumnInAggFunction() {
                 .matches(logicalOlapScan().when(scan -> {
                     PreAggStatus preAgg = scan.getPreAggStatus();
                     Assertions.assertTrue(preAgg.isOff());
+                    Assertions.assertEquals("Aggregate function sum(k2) contains key column k2.",
+                            preAgg.getOffReason());
                     return true;
                 }));
     }
@@ -264,7 +273,7 @@ public void testMaxCanUseKeyColumn() {
                 .matches(logicalOlapScan().when(scan -> {
                     PreAggStatus preAgg = scan.getPreAggStatus();
                     Assertions.assertTrue(preAgg.isOn());
-                    Assertions.assertEquals("r4", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r4", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -279,7 +288,7 @@ public void testMinCanUseKeyColumn() {
                 .matches(logicalOlapScan().when(scan -> {
                     PreAggStatus preAgg = scan.getPreAggStatus();
                     Assertions.assertTrue(preAgg.isOn());
-                    Assertions.assertEquals("r4", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r4", scan.getSelectedMaterializedIndexName().get());
                     return true;
                 }));
     }
@@ -367,6 +376,8 @@ public void testCountDistinctKeyColumn() {
     public void testCountDistinctValueColumn() {
         singleTableTest("select k1, count(distinct v1) from t group by k1", scan -> {
             Assertions.assertFalse(scan.isPreAggregation());
+            Assertions.assertEquals("Count distinct is only valid for key columns, but meet count(DISTINCT v1).",
+                    scan.getReasonOfPreAggregation());
             Assertions.assertEquals("t", scan.getSelectedIndexName());
         });
     }
@@ -414,7 +425,7 @@ public void testPreAggHint() throws Exception {
                 .rewrite()
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertTrue(scan.getHints().isEmpty());
-                    Assertions.assertEquals("r1", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r1", scan.getSelectedMaterializedIndexName().get());
                     PreAggStatus preAggStatus = scan.getPreAggStatus();
                     Assertions.assertTrue(preAggStatus.isOff());
                     Assertions.assertEquals("No aggregate on scan.", preAggStatus.getOffReason());
@@ -433,7 +444,7 @@ public void testPreAggHint() throws Exception {
                 .matches(logicalOlapScan().when(scan -> {
                     Assertions.assertEquals(1, scan.getHints().size());
                     Assertions.assertEquals("PREAGGOPEN", scan.getHints().get(0));
-                    Assertions.assertEquals("r1", scan.getSelectedMaterializedIndexName());
+                    Assertions.assertEquals("r1", scan.getSelectedMaterializedIndexName().get());
                     PreAggStatus preAggStatus = scan.getPreAggStatus();
                     Assertions.assertTrue(preAggStatus.isOn());
                     return true;

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/PlanToStringTest.java

@@ -80,8 +80,10 @@ public void testLogicalJoin(@Mocked Plan left, @Mocked Plan right) {
     @Test
     public void testLogicalOlapScan() {
         LogicalOlapScan plan = PlanConstructor.newLogicalOlapScan(0, "table", 0);
-        Assertions.assertTrue(plan.toString().matches("LogicalOlapScan \\( qualified=db\\.table, " + "indexName=table, "
-                + "selectedIndexId=-1, preAgg=ON \\)"), plan.toString());
+        Assertions.assertTrue(
+                plan.toString().matches("LogicalOlapScan \\( qualified=db\\.table, "
+                        + "indexName=<index_not_selected>, "
+                        + "selectedIndexId=-1, preAgg=ON \\)"));
     }
 
     @Test

regression-test/suites/mv_p0/ut/testProjectionMV1/testProjectionMV1.groovy

@@ -34,6 +34,7 @@ suite ("testProjectionMV1") {
     sql """insert into emps values("2020-01-01",1,"a",1,1,1);"""
     sql """insert into emps values("2020-01-02",2,"b",2,2,2);"""
 
+    sql """set enable_nereids_planner=false"""
     test {
         sql "create materialized view emps_mv as select deptno, empid from emps t order by deptno;"
         exception "errCode = 2,"

regression-test/suites/nereids_p0/test_mv_select.groovy

@@ -0,0 +1,45 @@
+// 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_mv_select") {
+    sql "SET enable_nereids_planner=true"
+    sql "SET enable_fallback_to_original_planner=false"
+
+    sql "DROP TABLE IF EXISTS mv_test_table_t"
+    sql """
+	    CREATE TABLE `mv_test_table_t` (
+        `Uid` bigint(20) NOT NULL,
+        `DateCode` int(11) NOT NULL,
+        `ProductId` bigint(20) NOT NULL,
+        `LiveSales` int(11) REPLACE NULL
+        ) ENGINE=OLAP
+        AGGREGATE KEY(`Uid`, `DateCode`, `ProductId`)
+        DISTRIBUTED BY HASH(`Uid`, `ProductId`) BUCKETS 8
+        PROPERTIES (
+        "replication_allocation" = "tag.location.default: 1"
+        );
+    """
+    sql "ALTER TABLE mv_test_table_t ADD ROLLUP rollup_mv_test_table_t(ProductId,DateCode,Uid);"
+
+    explain {
+        sql ("""select Uid
+                        from mv_test_table_t  
+                where ProductId = 3570093298674738221  and DateCode >=20230919 and DateCode <=20231018
+                        group by Uid;""")
+        contains "mv_test_table_t"
+    }
+}