branch-4.1: [fix](eager-agg) null-to-non-null safety check and remove redundant normalize calls in eager aggregation

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java

@@ -21,14 +21,13 @@
 import org.apache.doris.nereids.rules.rewrite.StatsDerive;
 import org.apache.doris.nereids.stats.ExpressionEstimation;
 import org.apache.doris.nereids.trees.expressions.Alias;
-import org.apache.doris.nereids.trees.expressions.CaseWhen;
 import org.apache.doris.nereids.trees.expressions.Cast;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
-import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
 import org.apache.doris.nereids.trees.plans.JoinType;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
@@ -269,7 +268,9 @@ private PushDownAggContext createContextFromProject(
         boolean newHasCaseWhen = context.hasCaseWhen;
         if (!newHasCaseWhen) {
             for (AggregateFunction aggFunc : aggFunctions) {
-                if (aggFunc.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+                if (aggFunc.children().stream().anyMatch(
+                        arg -> arg.anyMatch(e ->
+                                NullToNonNullFunction.canConvertNullToNonNull((Expression) e)))) {
                     newHasCaseWhen = true;
                     break;
                 }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java

@@ -37,9 +37,9 @@
 import org.apache.doris.nereids.jobs.JobContext;
 import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
 import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
-import org.apache.doris.nereids.trees.expressions.CaseWhen;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
 import org.apache.doris.nereids.trees.expressions.functions.Function;
@@ -164,7 +164,10 @@ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobConte
                     // of an outer join produces wrong results: null-extended rows make "col IS NULL"
                     // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
                     // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
-                    if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+                    if (!hasCaseWhen
+                            && aggFunction.children().stream().anyMatch(
+                                    arg -> arg.anyMatch(e ->
+                                            NullToNonNullFunction.canConvertNullToNonNull((Expression) e)))) {
                         hasCaseWhen = true;
                     }
                     if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/CaseWhen.java

@@ -43,7 +43,7 @@
  *
  * Children layout: [value?, WhenClause+, defaultValue?]
  */
-public class CaseWhen extends Expression implements NeedSessionVarGuard {
+public class CaseWhen extends Expression implements NeedSessionVarGuard, NullToNonNullFunction {
 
     private final Optional<Expression> value;
     private final List<WhenClause> whenClauses;

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/IsNull.java

@@ -33,7 +33,7 @@
 /**
  * expr is null predicate.
  */
-public class IsNull extends Expression implements UnaryExpression, AlwaysNotNullable {
+public class IsNull extends Expression implements UnaryExpression, AlwaysNotNullable, NullToNonNullFunction {
 
     public IsNull(Expression e) {
         super(ImmutableList.of(e));

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/NullSafeEqual.java

@@ -29,7 +29,7 @@
  * Null safe equal expression: a <=> b.
  * Unlike normal equal to expression, null <=> null is true.
  */
-public class NullSafeEqual extends EqualPredicate implements AlwaysNotNullable {
+public class NullSafeEqual extends EqualPredicate implements AlwaysNotNullable, NullToNonNullFunction {
     public NullSafeEqual(Expression left, Expression right) {
         this(left, right, false);
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/NullToNonNullFunction.java

@@ -0,0 +1,57 @@
+// 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.
+
+package org.apache.doris.nereids.trees.expressions;
+
+import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
+
+/**
+ * Marker interface for expressions that can convert NULL input into a non-NULL output.
+ *
+ * For example: Coalesce(NULL, 2) → 2, Nvl(NULL, 0) → 0, NullOrEmpty(NULL) → true.
+ *
+ * This is significant for outer-join push-down safety: when an aggregate function contains
+ * a NullToNonNull expression wrapping a column from the nullable side of an outer join,
+ * the aggregation must NOT be pushed down. Null-extended rows (produced by the join for
+ * unmatched rows) have NULL for all nullable-side columns. The NullToNonNull expression
+ * would convert those NULLs to non-NULL values, and the pre-aggregation would miss those
+ * contributions because null-extended rows do not exist in the base table.
+ *
+ * <p>Note: {@link AlwaysNotNullable} expressions with input slots (e.g. Array, JsonArray,
+ * JsonObject, CreateStruct, CreateMap) are also blocked from being pushed to the nullable
+ * side of outer joins via a separate check in {@link #canConvertNullToNonNull(Expression)}.
+ */
+public interface NullToNonNullFunction {
+
+    /**
+     * Check whether an expression can convert NULL input to non-NULL output.
+     * This covers both {@link NullToNonNullFunction} (e.g. Coalesce, Nvl, If, CaseWhen,
+     * NullOrEmpty, IsNull, IsTrue, IsFalse, NonNullable)
+     * and {@link AlwaysNotNullable} expressions with input slots (e.g. Array, JsonArray,
+     * CreateStruct, CreateMap), which always produce non-NULL output regardless of NULL inputs.
+     *
+     * <p>In outer-join push-down safety checks, any expression matching this predicate
+     * must NOT be pushed to the nullable side, because null-extended rows (produced by the
+     * join for unmatched rows) would produce non-NULL values that get aggregated, but the
+     * pre-aggregation on the base table cannot see those rows — resulting in wrong results.
+     */
+    static boolean canConvertNullToNonNull(Expression e) {
+        return e instanceof NullToNonNullFunction
+                || (e instanceof AlwaysNotNullable
+                        && !e.getInputSlots().isEmpty());
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/Coalesce.java

@@ -20,6 +20,7 @@
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NeedSessionVarGuard;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
 import org.apache.doris.nereids.trees.expressions.functions.ExpressionTrait;
 import org.apache.doris.nereids.trees.expressions.functions.SearchSignature;
@@ -39,7 +40,7 @@
 /**
  * ScalarFunction 'coalesce'. This class is generated by GenerateFunction.
  */
-public class Coalesce extends ScalarFunction implements CustomSignature, NeedSessionVarGuard {
+public class Coalesce extends ScalarFunction implements CustomSignature, NeedSessionVarGuard, NullToNonNullFunction {
 
     /**
      * constructor with 1 or more arguments.

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/If.java

@@ -21,6 +21,7 @@
 import org.apache.doris.nereids.analyzer.Unbound;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NeedSessionVarGuard;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
 import org.apache.doris.nereids.trees.expressions.functions.SearchSignature;
 import org.apache.doris.nereids.trees.expressions.shape.TernaryExpression;
@@ -38,7 +39,7 @@
  * ScalarFunction 'if'. This class is generated by GenerateFunction.
  */
 public class If extends ScalarFunction
-        implements TernaryExpression, CustomSignature, NeedSessionVarGuard {
+        implements TernaryExpression, CustomSignature, NeedSessionVarGuard, NullToNonNullFunction {
 
     /**
      * constructor with 3 arguments.

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/Ipv4StringToNumOrDefault.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
 import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
@@ -36,7 +37,7 @@
  * scalar function ipv4_string_to_num_or_default
  */
 public class Ipv4StringToNumOrDefault extends ScalarFunction
-        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable {
+        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable, NullToNonNullFunction {
 
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(BigIntType.INSTANCE).args(VarcharType.SYSTEM_DEFAULT),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/Ipv6StringToNumOrDefault.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
 import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
@@ -35,7 +36,7 @@
  * scalar function ipv6_string_to_num_or_default
  */
 public class Ipv6StringToNumOrDefault extends ScalarFunction
-        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable {
+        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable, NullToNonNullFunction {
 
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(VarcharType.SYSTEM_DEFAULT).args(VarcharType.SYSTEM_DEFAULT),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/NonNullable.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
 import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
@@ -32,7 +33,8 @@
 /**
  * change nullable input col to non_nullable col
  */
-public class NonNullable extends ScalarFunction implements UnaryExpression, CustomSignature, AlwaysNotNullable {
+public class NonNullable extends ScalarFunction
+        implements UnaryExpression, CustomSignature, AlwaysNotNullable, NullToNonNullFunction {
 
     public NonNullable(Expression expr) {
         super("non_nullable", expr);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/NotNullOrEmpty.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
 import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
@@ -36,7 +37,7 @@
  * ScalarFunction 'not_null_or_empty'. This class is generated by GenerateFunction.
  */
 public class NotNullOrEmpty extends ScalarFunction
-        implements UnaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable {
+        implements UnaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable, NullToNonNullFunction {
 
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(BooleanType.INSTANCE).args(VarcharType.SYSTEM_DEFAULT),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/NullOrEmpty.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
 import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
@@ -36,7 +37,7 @@
  * ScalarFunction 'null_or_empty'. This class is generated by GenerateFunction.
  */
 public class NullOrEmpty extends ScalarFunction
-        implements UnaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable {
+        implements UnaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable, NullToNonNullFunction {
 
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(BooleanType.INSTANCE).args(VarcharType.SYSTEM_DEFAULT),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/Nvl.java

@@ -20,6 +20,7 @@
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.NeedSessionVarGuard;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
 import org.apache.doris.nereids.trees.expressions.functions.SearchSignature;
 import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
@@ -36,7 +37,7 @@
  * ScalarFunction 'nvl'. This class is generated by GenerateFunction.
  */
 public class Nvl extends ScalarFunction
-        implements BinaryExpression, CustomSignature, NeedSessionVarGuard {
+        implements BinaryExpression, CustomSignature, NeedSessionVarGuard, NullToNonNullFunction {
 
     /**
      * constructor with 2 arguments.

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ToIpv4OrDefault.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
 import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
@@ -36,7 +37,7 @@
  * scalar function to_ipv4_or_default
  */
 public class ToIpv4OrDefault extends ScalarFunction
-        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable {
+        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable, NullToNonNullFunction {
 
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(IPv4Type.INSTANCE).args(VarcharType.SYSTEM_DEFAULT),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ToIpv6OrDefault.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.catalog.FunctionSignature;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NullToNonNullFunction;
 import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
 import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
 import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
@@ -36,7 +37,7 @@
  * scalar function to_ipv6_or_default
  */
 public class ToIpv6OrDefault extends ScalarFunction
-        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable {
+        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNotNullable, NullToNonNullFunction {
 
     public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
             FunctionSignature.ret(IPv6Type.INSTANCE).args(VarcharType.SYSTEM_DEFAULT),