fix(sql): skip unsafe eager rewrite for avg-style aggregates (#19740)

* fix(sql): correct eager aggregate decimal metadata (#19738)

* fix(sql): sync eager aggregate output types (#19738)

* fix(sql): repair eager count aggregate rewrite (#19738)

* fix(sql): skip unsafe eager rewrite for avg-style aggregates (#19738)

* fix(sql): narrow eager avg fix to rule guard (#19738)

Author: KKould

src/query/sql/src/planner/optimizer/optimizers/rule/agg_rules/rule_eager_aggregation.rs

@@ -258,6 +258,10 @@ impl<'a> EagerInput<'a> {
             .flat_map(|item| item.scalar.used_columns())
             .collect::<ColumnSet>();
 
+        if self.has_mixed_sum_and_count_aggregates(&eval_scalar_used_columns) {
+            return Ok(vec![]);
+        }
+
         let function_factory = AggregateFunctionFactory::instance();
         let eager_candidates = EagerCandidates::collect(
             &self.final_agg,
@@ -355,6 +359,36 @@ impl<'a> EagerInput<'a> {
             .collect())
     }
 
+    fn has_mixed_sum_and_count_aggregates(&self, eval_scalar_used_columns: &ColumnSet) -> bool {
+        let mut has_sum = false;
+        let mut has_count = false;
+
+        for aggregate in &self.final_agg.aggregate_functions {
+            if !eval_scalar_used_columns.contains(&aggregate.index) {
+                continue;
+            }
+
+            let ScalarExpr::AggregateFunction(aggregate_function) = &aggregate.scalar else {
+                continue;
+            };
+
+            match aggregate_function.func_name.as_str() {
+                "sum" => has_sum = true,
+                "count" => has_count = true,
+                _ => {}
+            }
+
+            if has_sum && has_count {
+                // Mixed sum/count outputs are used by rewrites such as AVG = SUM / COUNT.
+                // The current eager-count rewrite does not preserve this shape, so keep the
+                // original aggregate plan instead of applying eager aggregation.
+                return true;
+            }
+        }
+
+        false
+    }
+
     fn expand_analyses(
         &self,
         final_agg: &Aggregate,

tests/sqllogictests/suites/query/aggregate.test

@@ -105,6 +105,24 @@ SELECT AVG(1 + a), sum(1 + a), count(1 + a) FROM tc;
 ----
 3.0 9 3
 
+statement ok
+create or replace table t0(c0 int);
+
+statement ok
+create or replace table t1(c0 int);
+
+statement ok
+insert into t0 values (1), (2);
+
+statement ok
+insert into t1 values (1), (2);
+
+# https://github.com/databendlabs/databend/issues/19738
+query R
+select avg(0.5) from t0, t1;
+----
+0.5000000
+
 ## Aggregate with alias and window function
 statement ok
 select avg(1+a) score, a, percent_rank() over (partition by a % 3 order by score) d, d + 3 from tc group by a;