[Fix](count on index) fix count on index opt when count project expr #…

…41772 (#42229)

cherry pick from #41772

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

@@ -501,26 +501,53 @@ private LogicalAggregate<? extends Plan> pushdownCountOnIndex(
             LogicalFilter<? extends Plan> filter,
             LogicalOlapScan olapScan,
             CascadesContext cascadesContext) {
-        PhysicalOlapScan physicalOlapScan
-                = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
+
+        PhysicalOlapScan physicalOlapScan = (PhysicalOlapScan) new LogicalOlapScanToPhysicalOlapScan()
                 .build()
                 .transform(olapScan, cascadesContext)
                 .get(0);
+
+        List<Expression> argumentsOfAggregateFunction = normalizeArguments(agg.getAggregateFunctions(), project);
+
+        if (!onlyContainsSlot(argumentsOfAggregateFunction)) {
+            return agg;
+        }
+
+        return agg.withChildren(ImmutableList.of(
+                project != null
+                        ? project.withChildren(ImmutableList.of(
+                        filter.withChildren(ImmutableList.of(
+                                new PhysicalStorageLayerAggregate(
+                                        physicalOlapScan, PushDownAggOp.COUNT_ON_MATCH)))))
+                        : filter.withChildren(ImmutableList.of(
+                                new PhysicalStorageLayerAggregate(
+                                        physicalOlapScan, PushDownAggOp.COUNT_ON_MATCH)))
+        ));
+    }
+
+    private List<Expression> normalizeArguments(Set<AggregateFunction> aggregateFunctions,
+            @Nullable LogicalProject<? extends Plan> project) {
+        List<Expression> arguments = aggregateFunctions.stream()
+                .flatMap(aggregateFunction -> aggregateFunction.getArguments().stream())
+                .collect(ImmutableList.toImmutableList());
+
         if (project != null) {
-            return agg.withChildren(ImmutableList.of(
-                    project.withChildren(ImmutableList.of(
-                            filter.withChildren(ImmutableList.of(
-                                    new PhysicalStorageLayerAggregate(
-                                            physicalOlapScan,
-                                            PushDownAggOp.COUNT_ON_MATCH)))))
-            ));
-        } else {
-            return agg.withChildren(ImmutableList.of(
-                            filter.withChildren(ImmutableList.of(
-                                    new PhysicalStorageLayerAggregate(
-                                            physicalOlapScan,
-                                            PushDownAggOp.COUNT_ON_MATCH)))));
+            arguments = Project.findProject(arguments, project.getProjects())
+                    .stream()
+                    .map(p -> p instanceof Alias ? p.child(0) : p)
+                    .collect(ImmutableList.toImmutableList());
         }
+
+        return arguments;
+    }
+
+    private boolean onlyContainsSlot(List<Expression> arguments) {
+        return arguments.stream().allMatch(argument -> {
+            if (argument instanceof SlotReference) {
+                return true;
+            }
+            return false;
+        });
     }
 
     //select /*+SET_VAR(enable_pushdown_minmax_on_unique=true) */min(user_id) from table_unique;

regression-test/data/inverted_index_p0/test_count_on_index.out

@@ -77,3 +77,6 @@
 -- !sql_bad --
 0	1
 
+-- !sql_bad2 --
+0	1
+

regression-test/suites/inverted_index_p0/test_count_on_index.groovy

@@ -308,6 +308,30 @@ suite("test_count_on_index_httplogs", "p0") {
                 contains "pushAggOp=NONE"
         }
         qt_sql_bad "${bad_sql}"
+        def bad_sql2 = """
+        SELECT
+            COUNT(cond1) AS num1,
+            COUNT(cond2) AS num2
+        FROM (
+            SELECT
+                CASE
+                    WHEN c IN ('c1', 'c2', 'c3') AND d = 'd1' THEN b
+                END AS cond1,
+                CASE
+                    WHEN e = 'e1' AND c IN ('c1', 'c2', 'c3') THEN b
+                END AS cond2
+            FROM
+                ${tableName5}
+            WHERE
+                a = '2024-07-26'
+                AND e = 'e1'
+        ) AS project;
+        """
+        explain {
+            sql("${bad_sql2}")
+                contains "pushAggOp=NONE"
+        }
+        qt_sql_bad2 "${bad_sql2}"
     } finally {
         //try_sql("DROP TABLE IF EXISTS ${testTable}")
     }