Revert "[performance](agg) support count push agg in no null column"

This reverts commit 13cce98.

Author: HappenLee

be/src/olap/rowset/segment_v2/segment_iterator.cpp

@@ -131,10 +131,11 @@ void SegmentIterator::_init_row_bitmap_by_condition_cache() {
             if (_find_condition_cache) {
                 DorisMetrics::instance()->condition_cache_hit_count->increment(1);
                 if (_opts.runtime_state) {
-                    LOG(INFO) << "Condition cache hit, query id: " << print_id(_opts.runtime_state->query_id()) <<
-                             ", segment id: " << _segment->id() <<
-                             ", cache digest: " << _opts.condition_cache_digest <<
-                             ", rowset id: " << _opts.rowset_id.to_string();
+                    LOG(INFO) << "Condition cache hit, query id: "
+                              << print_id(_opts.runtime_state->query_id())
+                              << ", segment id: " << _segment->id()
+                              << ", cache digest: " << _opts.condition_cache_digest
+                              << ", rowset id: " << _opts.rowset_id.to_string();
                 }
             }
 
@@ -2422,10 +2423,11 @@ Status SegmentIterator::next_batch(vectorized::Block* block) {
                     auto* condition_cache = ConditionCache::instance();
                     ConditionCache::CacheKey cache_key(_opts.rowset_id, _segment->id(),
                                                        _opts.condition_cache_digest);
-                    LOG(INFO) << "Condition cache insert, query id: " << print_id(_opts.runtime_state->query_id()) <<
-                            ", rowset id: " << _opts.rowset_id.to_string() <<
-                             ", segment id: " << _segment->id() <<
-                             ", cache digest: " << _opts.condition_cache_digest;
+                    LOG(INFO) << "Condition cache insert, query id: "
+                              << print_id(_opts.runtime_state->query_id())
+                              << ", rowset id: " << _opts.rowset_id.to_string()
+                              << ", segment id: " << _segment->id()
+                              << ", cache digest: " << _opts.condition_cache_digest;
                     condition_cache->insert(cache_key, std::move(_condition_cache));
                 }
                 return res;

be/src/vec/exprs/vruntimefilter_wrapper.h

@@ -62,7 +62,13 @@ class VRuntimeFilterWrapper final : public VExpr {
     const std::string& expr_name() const override;
     const VExprSPtrs& children() const override { return _impl->children(); }
 
-    uint64_t get_digest(uint64_t seed) const override { return _impl->get_digest(seed); }
+    uint64_t get_digest(uint64_t seed) const override {
+        seed = _impl->get_digest(seed);
+        if (seed) {
+            return HashUtil::crc_hash64(&_null_aware, sizeof(_null_aware), seed);
+        }
+        return seed;
+    }
 
     VExprSPtr get_impl() const override { return _impl; }
 

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

@@ -61,7 +61,6 @@
 
 import com.google.common.collect.ImmutableList;
 
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -559,54 +558,28 @@ private LogicalAggregate<? extends Plan> storageLayerAggregate(
         }
 
         Set<AggregateFunction> aggregateFunctions = aggregate.getAggregateFunctions();
-        // Use for loop to replace Stream API
-        Set<Class<? extends AggregateFunction>> functionClasses = new HashSet<>();
-        Map<Class<? extends AggregateFunction>, PushDownAggOp> supportedAgg = PushDownAggOp.supportedFunctions();
-
-        boolean containsCount = false;
-        Set<SlotReference> checkNullSlots = new HashSet<>();
-
-        // Single loop through aggregateFunctions to handle multiple logic
-        for (AggregateFunction function : aggregateFunctions) {
-            Class<? extends AggregateFunction> functionClass = function.getClass();
-            functionClasses.add(functionClass);
-            // Check if any function has arity > 1
-            if (function.arity() > 1) {
-                return canNotPush;
-            }
-
-            // Check if contains Count function
-            if (functionClass.equals(Count.class)) {
-                containsCount = true;
-                if (!function.getArguments().isEmpty()) {
-                    Expression arg0 = function.getArguments().get(0);
-                    if (arg0 instanceof SlotReference) {
-                        checkNullSlots.add((SlotReference) arg0);
-                    } else if (arg0 instanceof Cast) {
-                        Expression child0 = arg0.child(0);
-                        if (child0 instanceof SlotReference) {
-                            checkNullSlots.add((SlotReference) child0);
-                        }
-                    }
-                }
-            }
+        Set<Class<? extends AggregateFunction>> functionClasses = aggregateFunctions
+                .stream()
+                .map(AggregateFunction::getClass)
+                .collect(Collectors.toSet());
 
-            // Check if function is supported by supportedAgg
-            if (!supportedAgg.containsKey(functionClass)) {
-                return canNotPush;
-            }
+        Map<Class<? extends AggregateFunction>, PushDownAggOp> supportedAgg = PushDownAggOp.supportedFunctions();
+        if (!supportedAgg.keySet().containsAll(functionClasses)) {
+            return canNotPush;
         }
-
         if (logicalScan instanceof LogicalOlapScan) {
             LogicalOlapScan logicalOlapScan = (LogicalOlapScan) logicalScan;
             KeysType keysType = logicalOlapScan.getTable().getKeysType();
-            if (containsCount && keysType != KeysType.DUP_KEYS) {
+            if (functionClasses.contains(Count.class) && keysType != KeysType.DUP_KEYS) {
                 return canNotPush;
             }
-            if (containsCount && logicalOlapScan.isDirectMvScan()) {
+            if (functionClasses.contains(Count.class) && logicalOlapScan.isDirectMvScan()) {
                 return canNotPush;
             }
         }
+        if (aggregateFunctions.stream().anyMatch(fun -> fun.arity() > 1)) {
+            return canNotPush;
+        }
 
         // TODO: refactor this to process slot reference or expression together
         boolean onlyContainsSlotOrNumericCastSlot = aggregateFunctions.stream()
@@ -692,7 +665,7 @@ private LogicalAggregate<? extends Plan> storageLayerAggregate(
                 // NULL value behavior in `count` function is zero, so
                 // we should not use row_count to speed up query. the col
                 // must be not null
-                if (column.isAllowNull() && checkNullSlots.contains(slot)) {
+                if (column.isAllowNull()) {
                     return canNotPush;
                 }
             }

regression-test/suites/nereids_p0/explain/test_pushdown_explain.groovy

@@ -74,56 +74,6 @@ suite("test_pushdown_explain") {
         contains "pushAggOp=NONE"
     }
 
-    // Test cases for NULL column handling in count pushdown optimization
-    sql "DROP TABLE IF EXISTS test_null_columns"
-    sql """ CREATE TABLE `test_null_columns` (
-        `id` INT NOT NULL COMMENT 'ID',
-        `nullable_col` VARCHAR(11) NULL COMMENT 'Nullable column',
-        `non_nullable_col` VARCHAR(11) NOT NULL COMMENT 'Non-nullable column'
-    ) ENGINE=OLAP
-    DUPLICATE KEY(`id`)
-    DISTRIBUTED BY HASH(`id`) BUCKETS 48
-    PROPERTIES (
-        "replication_allocation" = "tag.location.default: 1",
-        "min_load_replica_num" = "-1",
-        "is_being_synced" = "false",
-        "colocate_with" = "groupa1",
-        "storage_format" = "V2",
-        "light_schema_change" = "true",
-        "disable_auto_compaction" = "false",
-        "enable_single_replica_compaction" = "false"
-    ); """
-    sql """ insert into test_null_columns values(1, NULL, "value1"); """
-    sql """ insert into test_null_columns values(2, NULL, "value2"); """
-    sql """ insert into test_null_columns values(3, "not_null", "value3"); """
-    
-    // Test count(1) and count(*) with NULL columns - should push Count optimization
-    explain {
-        sql("select count(1) from test_null_columns;")
-        contains "pushAggOp=COUNT"
-    }
-    explain {
-        sql("select count(*) from test_null_columns;")
-        contains "pushAggOp=COUNT"
-    }
-
-    explain {
-        sql("select count(non_nullable_col), min(non_nullable_col), max(non_nullable_col) from test_null_columns;")
-        contains "pushAggOp=MIX"
-    }
-    explain {
-        sql("select count(), min(non_nullable_col), max(non_nullable_col) from test_null_columns;")
-        contains "pushAggOp=MIX"
-    }
-    explain {
-        sql("select count(*), min(non_nullable_col), max(non_nullable_col) from test_null_columns;")
-        contains "pushAggOp=MIX"
-    }
-    explain {
-        sql("select count(nullable_col), min(nullable_col), max(nullable_col) from test_null_columns;")
-        contains "pushAggOp=NONE"
-    }
-
     sql "DROP TABLE IF EXISTS table_unique0"
     sql """ 
         CREATE TABLE `table_unique0` (