diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java
index b17437b1a2ff..91e10fbb3833 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java
@@ -30,8 +30,10 @@
 import it.unimi.dsi.fastutil.ints.Int2LongOpenHashMap;
 import it.unimi.dsi.fastutil.longs.Long2LongMap;
 import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap;
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
 import it.unimi.dsi.fastutil.objects.ObjectIterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import javax.annotation.Nullable;
 import org.apache.pinot.common.CustomObject;
@@ -54,29 +56,63 @@
 ///
 /// Following arguments are supported:
 ///
-/// - Expression: expression that contains the column to be calculated mode on, can be any Numeric column
+/// - Expression: expression that contains the column to be calculated mode on
 /// - MultiModeReducerType (optional): the reducer to use in case of multiple modes present in data
+///
+/// Numeric calls retain a DOUBLE result and support MIN, MAX and AVG tie reducers. The planner supplies an internal
+/// third STRING or TIMESTAMP literal for non-numeric inputs, which retain their type and support MIN and MAX.
+/// The result type is immutable so a function can be shared across segments and reconstructed on reducing stages.
 @SuppressWarnings({"rawtypes", "unchecked"})
-public class ModeAggregationFunction extends BaseSingleInputAggregationFunction<Map<? extends Number, Long>, Double> {
+public class ModeAggregationFunction extends BaseSingleInputAggregationFunction<Map<?, Long>, Comparable<?>> {
 
   private static final double DEFAULT_FINAL_RESULT = Double.NEGATIVE_INFINITY;
 
   private final MultiModeReducerType _multiModeReducerType;
+  private final ColumnDataType _resultType;
+  private final String _resultColumnName;
 
   public ModeAggregationFunction(List<ExpressionContext> arguments, boolean nullHandlingEnabled) {
-    super(arguments.get(0), nullHandlingEnabled);
+    super(checkArguments(arguments), nullHandlingEnabled);
 
     int numArguments = arguments.size();
-    Preconditions.checkArgument(numArguments <= 2, "Mode expects at most 2 arguments, got: %s", numArguments);
+    if (numArguments == 3) {
+      ExpressionContext typeArgument = arguments.get(2);
+      Preconditions.checkArgument(typeArgument.getType() == ExpressionContext.Type.LITERAL
+              && typeArgument.getLiteral().getType() == DataType.STRING,
+          "MODE result type must be a STRING or TIMESTAMP string literal");
+      String resultType = typeArgument.getLiteral().getStringValue();
+      Preconditions.checkArgument(resultType != null, "MODE result type must be STRING or TIMESTAMP, got: null");
+      resultType = resultType.toUpperCase(Locale.ROOT);
+      Preconditions.checkArgument("STRING".equals(resultType) || "TIMESTAMP".equals(resultType),
+          "MODE result type must be STRING or TIMESTAMP, got: %s", resultType);
+      _resultType = ColumnDataType.valueOf(resultType);
+    } else {
+      _resultType = ColumnDataType.DOUBLE;
+    }
     if (numArguments > 1) {
+      Preconditions.checkArgument(arguments.get(1).getType() == ExpressionContext.Type.LITERAL,
+          "MODE tie reducer must be a literal");
       _multiModeReducerType = MultiModeReducerType.valueOf(arguments.get(1).getLiteral().getStringValue());
     } else {
       _multiModeReducerType = MultiModeReducerType.MIN;
     }
+    Preconditions.checkArgument(
+        _resultType == ColumnDataType.DOUBLE || _multiModeReducerType != MultiModeReducerType.AVG,
+        "MODE for %s supports only MIN or MAX tie reducers, got: %s", _resultType, _multiModeReducerType);
+    // Include the inferred type in the result identity while retaining legacy numeric names.
+    _resultColumnName = numArguments == 3
+        ? "mode(" + _expression + "," + arguments.get(1) + "," + arguments.get(2) + ")"
+        : super.getResultColumnName();
+  }
+
+  private static ExpressionContext checkArguments(List<ExpressionContext> arguments) {
+    Preconditions.checkArgument(!arguments.isEmpty() && arguments.size() <= 3,
+        "MODE expects one to three arguments, got: %s", arguments.size());
+    return arguments.get(0);
   }
 
   /// Helper method to create a value map for the given value type.
-  private static Map<? extends Number, Long> getValueMap(DataType valueType) {
+  private static Map<?, Long> getValueMap(DataType valueType) {
     switch (valueType) {
       case INT:
         return new Int2LongOpenHashMap();
@@ -86,15 +122,17 @@ private static Map<? extends Number, Long> getValueMap(DataType valueType) {
         return new Float2LongOpenHashMap();
       case DOUBLE:
         return new Double2LongOpenHashMap();
+      case STRING:
+        return new Object2LongOpenHashMap<String>();
       default:
         throw new IllegalStateException("Illegal data type for MODE aggregation function: " + valueType);
     }
   }
 
   /// Returns the value map from the result holder or creates a new one if it does not exist.
-  private static Map<? extends Number, Long> getValueMap(AggregationResultHolder aggregationResultHolder,
+  private static Map<?, Long> getValueMap(AggregationResultHolder aggregationResultHolder,
       DataType valueType) {
-    Map<? extends Number, Long> valueMap = aggregationResultHolder.getResult();
+    Map<?, Long> valueMap = aggregationResultHolder.getResult();
     if (valueMap == null) {
       valueMap = getValueMap(valueType);
       aggregationResultHolder.setValue(valueMap);
@@ -142,6 +180,15 @@ private static void setValueForGroupKeys(GroupByResultHolder groupByResultHolder
     valueMap.merge(value, 1, Long::sum);
   }
 
+  private static void setValueForGroupKeys(GroupByResultHolder holder, int groupKey, String value) {
+    Object2LongOpenHashMap<String> counts = holder.getResult(groupKey);
+    if (counts == null) {
+      counts = new Object2LongOpenHashMap<>();
+      holder.setValueForKey(groupKey, counts);
+    }
+    counts.addTo(value, 1L);
+  }
+
   /// Returns the dictionary id count map from the result holder or creates a new one if it does not exist.
   protected static Int2IntOpenHashMap getDictIdCountMap(AggregationResultHolder aggregationResultHolder,
       Dictionary dictionary) {
@@ -165,7 +212,7 @@ protected static Int2IntOpenHashMap getDictIdCountMap(GroupByResultHolder groupB
   }
 
   /// Helper method to read dictionary and convert dictionary ids to values for dictionary-encoded expression.
-  private static Map<? extends Number, Long> convertToValueMap(DictIdsWrapper dictIdsWrapper) {
+  private static Map<?, Long> convertToValueMap(DictIdsWrapper dictIdsWrapper) {
     Dictionary dictionary = dictIdsWrapper._dictionary;
     Int2IntOpenHashMap dictIdCountMap = dictIdsWrapper._dictIdCountMap;
     int numValues = dictIdCountMap.size();
@@ -200,16 +247,24 @@ private static Map<? extends Number, Long> convertToValueMap(DictIdsWrapper dict
           doubleValueMap.put(dictionary.getDoubleValue(next.getIntKey()), next.getIntValue());
         }
         return doubleValueMap;
+      case STRING:
+        Object2LongOpenHashMap<String> stringValueMap = new Object2LongOpenHashMap<>(numValues);
+        while (iterator.hasNext()) {
+          Int2IntMap.Entry next = iterator.next();
+          stringValueMap.put(dictionary.getStringValue(next.getIntKey()), next.getIntValue());
+        }
+        return stringValueMap;
       default:
         throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType);
     }
   }
 
   /// Helper method to extract segment level intermediate result from the inner segment result.
-  private static Map<? extends Number, Long> extractIntermediateResult(@Nullable Object result) {
+  @Nullable
+  private Map<?, Long> extractIntermediateResult(@Nullable Object result) {
     if (result == null) {
-      // NOTE: Return an empty Int2LongOpenHashMap for empty result.
-      return new Int2LongOpenHashMap();
+      // Preserve the legacy numeric empty-result sentinel.
+      return _resultType == ColumnDataType.DOUBLE ? new Int2LongOpenHashMap() : null;
     }
 
     if (result instanceof DictIdsWrapper) {
@@ -226,6 +281,11 @@ public AggregationFunctionType getType() {
     return AggregationFunctionType.MODE;
   }
 
+  @Override
+  public String getResultColumnName() {
+    return _resultColumnName;
+  }
+
   @Override
   public AggregationResultHolder createAggregationResultHolder() {
     return new ObjectAggregationResultHolder();
@@ -257,7 +317,7 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde
 
     // For non-dictionary-encoded expression, store values into the value map
     DataType storedType = blockValSet.getValueType().getStoredType();
-    Map<? extends Number, Long> valueMap = getValueMap(aggregationResultHolder, storedType);
+    Map<?, Long> valueMap = getValueMap(aggregationResultHolder, storedType);
     switch (storedType) {
       case INT:
         Int2LongOpenHashMap intMap = (Int2LongOpenHashMap) valueMap;
@@ -295,6 +355,15 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde
           }
         });
         break;
+      case STRING:
+        Object2LongOpenHashMap<String> stringMap = (Object2LongOpenHashMap<String>) valueMap;
+        String[] stringValues = blockValSet.getStringValuesSV();
+        forEachNotNull(length, blockValSet, (from, to) -> {
+          for (int i = from; i < to; i++) {
+            stringMap.addTo(stringValues[i], 1L);
+          }
+        });
+        break;
       default:
         throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType);
     }
@@ -353,6 +422,14 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol
           }
         });
         break;
+      case STRING:
+        String[] stringValues = blockValSet.getStringValuesSV();
+        forEachNotNull(length, blockValSet, (from, to) -> {
+          for (int i = from; i < to; i++) {
+            setValueForGroupKeys(groupByResultHolder, groupKeyArray[i], stringValues[i]);
+          }
+        });
+        break;
       default:
         throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType);
     }
@@ -420,30 +497,46 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult
           }
         });
         break;
+      case STRING:
+        String[] stringValues = blockValSet.getStringValuesSV();
+        forEachNotNull(length, blockValSet, (from, to) -> {
+          for (int i = from; i < to; i++) {
+            for (int groupKey : groupKeysArray[i]) {
+              setValueForGroupKeys(groupByResultHolder, groupKey, stringValues[i]);
+            }
+          }
+        });
+        break;
       default:
         throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType);
     }
   }
 
+  @Nullable
   @Override
-  public Map<? extends Number, Long> extractAggregationResult(AggregationResultHolder aggregationResultHolder) {
+  public Map<?, Long> extractAggregationResult(AggregationResultHolder aggregationResultHolder) {
     return extractIntermediateResult(aggregationResultHolder.getResult());
   }
 
+  @Nullable
   @Override
-  public Map<? extends Number, Long> extractGroupByResult(GroupByResultHolder groupByResultHolder, int groupKey) {
+  public Map<?, Long> extractGroupByResult(GroupByResultHolder groupByResultHolder, int groupKey) {
     return extractIntermediateResult(groupByResultHolder.getResult(groupKey));
   }
 
   @Override
-  public Map<? extends Number, Long> merge(Map<? extends Number, Long> intermediateResult1,
-      Map<? extends Number, Long> intermediateResult2) {
+  public Map<?, Long> merge(Map<?, Long> intermediateResult1, Map<?, Long> intermediateResult2) {
     if (intermediateResult1.isEmpty()) {
       return intermediateResult2;
     }
     if (intermediateResult2.isEmpty()) {
       return intermediateResult1;
     }
+    if (_resultType != ColumnDataType.DOUBLE) {
+      Map<Object, Long> counts = (Map<Object, Long>) intermediateResult1;
+      intermediateResult2.forEach((value, count) -> counts.merge(value, count, Long::sum));
+      return counts;
+    }
     if (intermediateResult1 instanceof Int2LongOpenHashMap && intermediateResult2 instanceof Int2LongOpenHashMap) {
       ((Int2LongOpenHashMap) intermediateResult2).int2LongEntrySet().fastForEach(
           e -> ((Int2LongOpenHashMap) intermediateResult1).merge(e.getIntKey(), e.getLongValue(), Long::sum));
@@ -473,7 +566,16 @@ public ColumnDataType getIntermediateResultColumnType() {
   }
 
   @Override
-  public SerializedIntermediateResult serializeIntermediateResult(Map<? extends Number, Long> longMap) {
+  @SuppressWarnings("deprecation")
+  public SerializedIntermediateResult serializeIntermediateResult(Map<?, Long> longMap) {
+    if (_resultType == ColumnDataType.STRING) {
+      // Reuse the existing map wire encoding for generic aggregation bridges.
+      return new SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.Map.getValue(),
+          ObjectSerDeUtils.MAP_SER_DE.serialize((Map) longMap));
+    }
+    if (_resultType == ColumnDataType.TIMESTAMP && !(longMap instanceof Long2LongMap)) {
+      longMap = new Long2LongOpenHashMap((Map<Long, Long>) longMap);
+    }
     if (longMap instanceof Int2LongMap) {
       return new SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.Int2LongMap.getValue(),
           ObjectSerDeUtils.INT_2_LONG_MAP_SER_DE.serialize((Int2LongMap) longMap));
@@ -494,18 +596,21 @@ public SerializedIntermediateResult serializeIntermediateResult(Map<? extends Nu
   }
 
   @Override
-  public Map<? extends Number, Long> deserializeIntermediateResult(CustomObject customObject) {
+  public Map<?, Long> deserializeIntermediateResult(CustomObject customObject) {
     return ObjectSerDeUtils.deserialize(customObject);
   }
 
   @Override
   public ColumnDataType getFinalResultColumnType() {
-    return ColumnDataType.DOUBLE;
+    return _resultType;
   }
 
   @Nullable
   @Override
-  public Double extractFinalResult(@Nullable Map<? extends Number, Long> intermediateResult) {
+  public Comparable<?> extractFinalResult(@Nullable Map<?, Long> intermediateResult) {
+    if (_resultType != ColumnDataType.DOUBLE) {
+      return extractComparableFinalResult(intermediateResult);
+    }
     // A null intermediate result means nothing was aggregated, and the mode of nothing is NULL. An empty map is a
     // different thing: it is what an untouched single-stage result holder produces, and it keeps its historical
     // sentinel below so that path is not silently changed.
@@ -519,13 +624,13 @@ public Double extractFinalResult(@Nullable Map<? extends Number, Long> intermedi
         return DEFAULT_FINAL_RESULT;
       }
     } else if (intermediateResult instanceof Int2LongOpenHashMap) {
-      return extractFinalResult((Int2LongOpenHashMap) intermediateResult);
+      return extractNumericFinalResult((Int2LongOpenHashMap) intermediateResult);
     } else if (intermediateResult instanceof Long2LongOpenHashMap) {
-      return extractFinalResult((Long2LongOpenHashMap) intermediateResult);
+      return extractNumericFinalResult((Long2LongOpenHashMap) intermediateResult);
     } else if (intermediateResult instanceof Float2LongOpenHashMap) {
-      return extractFinalResult((Float2LongOpenHashMap) intermediateResult);
+      return extractNumericFinalResult((Float2LongOpenHashMap) intermediateResult);
     } else if (intermediateResult instanceof Double2LongOpenHashMap) {
-      return extractFinalResult((Double2LongOpenHashMap) intermediateResult);
+      return extractNumericFinalResult((Double2LongOpenHashMap) intermediateResult);
     } else {
       throw new IllegalStateException(
           "Illegal data type for Intermediate Result of MODE aggregation function: " + intermediateResult.getClass()
@@ -533,7 +638,7 @@ public Double extractFinalResult(@Nullable Map<? extends Number, Long> intermedi
     }
   }
 
-  public double extractFinalResult(Int2LongOpenHashMap intermediateResult) {
+  private double extractNumericFinalResult(Int2LongOpenHashMap intermediateResult) {
     ObjectIterator<Int2LongMap.Entry> iterator = intermediateResult.int2LongEntrySet().fastIterator();
     Int2LongMap.Entry first = iterator.next();
     long maxFrequency = first.getLongValue();
@@ -578,7 +683,7 @@ public double extractFinalResult(Int2LongOpenHashMap intermediateResult) {
     }
   }
 
-  public double extractFinalResult(Long2LongOpenHashMap intermediateResult) {
+  private double extractNumericFinalResult(Long2LongOpenHashMap intermediateResult) {
     ObjectIterator<Long2LongMap.Entry> iterator = intermediateResult.long2LongEntrySet().fastIterator();
     Long2LongMap.Entry first = iterator.next();
     long maxFrequency = first.getLongValue();
@@ -625,7 +730,7 @@ public double extractFinalResult(Long2LongOpenHashMap intermediateResult) {
     }
   }
 
-  public double extractFinalResult(Float2LongOpenHashMap intermediateResult) {
+  private double extractNumericFinalResult(Float2LongOpenHashMap intermediateResult) {
     ObjectIterator<Float2LongMap.Entry> iterator = intermediateResult.float2LongEntrySet().fastIterator();
     Float2LongMap.Entry first = iterator.next();
     long maxFrequency = first.getLongValue();
@@ -672,7 +777,7 @@ public double extractFinalResult(Float2LongOpenHashMap intermediateResult) {
     }
   }
 
-  public Double extractFinalResult(Double2LongOpenHashMap intermediateResult) {
+  private Double extractNumericFinalResult(Double2LongOpenHashMap intermediateResult) {
     ObjectIterator<Double2LongMap.Entry> iterator = intermediateResult.double2LongEntrySet().fastIterator();
     Double2LongMap.Entry first = iterator.next();
     long maxFrequency = first.getLongValue();
@@ -719,6 +824,25 @@ public Double extractFinalResult(Double2LongOpenHashMap intermediateResult) {
     }
   }
 
+  @Nullable
+  private Comparable<?> extractComparableFinalResult(@Nullable Map<?, Long> counts) {
+    Comparable mode = null;
+    long maxCount = 0;
+    if (counts != null) {
+      for (Map.Entry<?, Long> entry : counts.entrySet()) {
+        Comparable value = (Comparable) entry.getKey();
+        long count = entry.getValue();
+        if (mode == null || count > maxCount || (count == maxCount
+            && (_multiModeReducerType == MultiModeReducerType.MIN
+                ? value.compareTo(mode) < 0 : value.compareTo(mode) > 0))) {
+          mode = value;
+          maxCount = count;
+        }
+      }
+    }
+    return mode;
+  }
+
   private enum MultiModeReducerType {
     MIN, MAX, AVG
   }
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java
index f88c1cf372de..0451cefc1498 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java
@@ -18,18 +18,28 @@
  */
 package org.apache.pinot.core.query.optimizer.statement;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 import javax.annotation.Nullable;
+import org.apache.pinot.common.function.FunctionInfo;
+import org.apache.pinot.common.function.FunctionRegistry;
+import org.apache.pinot.common.function.FunctionUtils;
 import org.apache.pinot.common.request.Expression;
 import org.apache.pinot.common.request.Function;
 import org.apache.pinot.common.request.PinotQuery;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.common.utils.request.RequestUtils;
 import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
 import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
 import org.apache.pinot.spi.data.Schema;
 import org.apache.pinot.spi.utils.CommonConstants;
 
 
-/// Rewrites certain aggregation functions based on operand types to support polymorphic aggregations.
+/// Binds inferred arguments declared by aggregation functions and optionally rewrites type-specific variants.
 ///
 /// Currently supported rewrites:
 /// - MIN(stringType) -> MINSTRING
@@ -46,48 +56,69 @@ public void optimize(PinotQuery pinotQuery, @Nullable Schema schema) {
       return;
     }
 
-    // Only perform auto rewrite when enabled through query option.
-    if (pinotQuery.getQueryOptions() == null || !Boolean.parseBoolean(pinotQuery.getQueryOptions().get(
-        CommonConstants.Broker.Request.QueryOptionKey.AUTO_REWRITE_AGGREGATION_TYPE))) {
-      return;
-    }
+    boolean autoRewrite = pinotQuery.getQueryOptions() != null && Boolean.parseBoolean(pinotQuery.getQueryOptions().get(
+        CommonConstants.Broker.Request.QueryOptionKey.AUTO_REWRITE_AGGREGATION_TYPE));
 
     List<Expression> selectList = pinotQuery.getSelectList();
     if (selectList != null) {
       for (Expression expression : selectList) {
-        maybeRewriteAggregateFunction(expression, schema);
+        maybeRewriteAggregateFunction(expression, schema, autoRewrite);
       }
     }
 
     List<Expression> groupByList = pinotQuery.getGroupByList();
     if (groupByList != null) {
       for (Expression expression : groupByList) {
-        maybeRewriteAggregateFunction(expression, schema);
+        maybeRewriteAggregateFunction(expression, schema, autoRewrite);
       }
     }
 
     List<Expression> orderByList = pinotQuery.getOrderByList();
     if (orderByList != null) {
       for (Expression expression : orderByList) {
-        maybeRewriteAggregateFunction(expression, schema);
+        maybeRewriteAggregateFunction(expression, schema, autoRewrite);
       }
     }
 
-    maybeRewriteAggregateFunction(pinotQuery.getFilterExpression(), schema);
-    maybeRewriteAggregateFunction(pinotQuery.getHavingExpression(), schema);
+    maybeRewriteAggregateFunction(pinotQuery.getFilterExpression(), schema, autoRewrite);
+    maybeRewriteAggregateFunction(pinotQuery.getHavingExpression(), schema, autoRewrite);
   }
 
-  private void maybeRewriteAggregateFunction(@Nullable Expression expression, Schema schema) {
+  private void maybeRewriteAggregateFunction(@Nullable Expression expression, Schema schema, boolean autoRewrite) {
     if (expression == null || !expression.isSetFunctionCall()) {
       return;
     }
 
     Function function = expression.getFunctionCall();
+    List<Expression> operands = function.getOperands();
+    for (Expression operand : operands) {
+      // Infer arguments within scalar expressions while preserving the existing top-level variant rewrites.
+      maybeRewriteAggregateFunction(operand, schema, false);
+    }
     String functionName = function.getOperator();
     if (!AggregationFunctionType.isAggregationFunction(functionName)) {
       return;
     }
 
+    List<String> inferredArguments = AggregationFunctionType.getAggregationFunctionType(functionName)
+        .inferArguments(operands.size(), i -> {
+          ColumnDataType type = getOperandType(operands.get(i), schema);
+          if (type == null || type.isArray() || type == ColumnDataType.OBJECT) {
+            return DataType.UNKNOWN;
+          }
+          return type == ColumnDataType.MAP ? DataType.MAP : type.toDataType();
+        });
+    if (!inferredArguments.isEmpty()) {
+      List<Expression> typedOperands = new ArrayList<>(operands);
+      for (String argument : inferredArguments) {
+        typedOperands.add(RequestUtils.getLiteralExpression(argument));
+      }
+      function.setOperands(typedOperands);
+    }
+    if (!autoRewrite || operands.isEmpty()) {
+      return;
+    }
+
     Expression operand = function.getOperands().get(0);
     FieldSpec.DataType operandType;
     // TODO: Handle more complex expressions (e.g. MIN(trim(stringCol)) )
@@ -133,4 +164,96 @@ private void maybeRewriteAggregateFunction(@Nullable Expression expression, Sche
       }
     }
   }
+
+  @Nullable
+  private static ColumnDataType getOperandType(Expression operand, Schema schema) {
+    if (operand.isSetIdentifier()) {
+      FieldSpec fieldSpec = schema.getFieldSpecFor(operand.getIdentifier().getName());
+      return fieldSpec != null
+          ? ColumnDataType.fromDataType(fieldSpec.getDataType(), fieldSpec.isSingleValueField())
+          : null;
+    }
+    if (operand.isSetLiteral()) {
+      return RequestUtils.getLiteralTypeAndValue(operand.getLiteral()).getLeft();
+    }
+    if (!operand.isSetFunctionCall()) {
+      return null;
+    }
+    Function function = operand.getFunctionCall();
+    List<Expression> arguments = function.getOperands();
+    String name = FunctionRegistry.canonicalize(function.getOperator());
+    switch (name) {
+      case "cast":
+        return literalType(arguments, 1);
+      case "jsonextractscalar":
+      case "jsonextractscalarfast":
+      case "jsonextractscalarfirstmatch":
+      case "jsonextractscalarfory":
+        return literalType(arguments, 2);
+      case "case":
+        // CASE stores alternating condition/result pairs followed by an optional ELSE result.
+        ColumnDataType resultType = ColumnDataType.UNKNOWN;
+        for (int i = 1; i < arguments.size(); i += 2) {
+          resultType = commonType(resultType, getOperandType(arguments.get(i), schema));
+        }
+        if (arguments.size() % 2 == 1) {
+          resultType = commonType(resultType, getOperandType(arguments.get(arguments.size() - 1), schema));
+        }
+        return resultType;
+      case "datetimeconvert":
+        if (arguments.size() < 3 || !arguments.get(2).isSetLiteral()
+            || !arguments.get(2).getLiteral().isSetStringValue()) {
+          return null;
+        }
+        DateTimeFieldSpec.TimeFormat format =
+            new DateTimeFormatSpec(arguments.get(2).getLiteral().getStringValue()).getTimeFormat();
+        return format == DateTimeFieldSpec.TimeFormat.EPOCH || format == DateTimeFieldSpec.TimeFormat.TIMESTAMP
+            ? ColumnDataType.LONG
+            : ColumnDataType.STRING;
+      default:
+        ColumnDataType[] argumentTypes = new ColumnDataType[arguments.size()];
+        for (int i = 0; i < arguments.size(); i++) {
+          argumentTypes[i] = getOperandType(arguments.get(i), schema);
+          if (argumentTypes[i] == null) {
+            return null;
+          }
+        }
+        FunctionInfo functionInfo = FunctionRegistry.lookupFunctionInfo(name, argumentTypes);
+        return functionInfo != null ? FunctionUtils.getColumnDataType(functionInfo.getMethod().getReturnType()) : null;
+    }
+  }
+
+  @Nullable
+  private static ColumnDataType commonType(@Nullable ColumnDataType left, @Nullable ColumnDataType right) {
+    if (left == null || right == null) {
+      return null;
+    }
+    if (left == ColumnDataType.UNKNOWN) {
+      return right;
+    }
+    return right == ColumnDataType.UNKNOWN || left == right ? left : null;
+  }
+
+  @Nullable
+  private static ColumnDataType literalType(List<Expression> arguments, int position) {
+    if (arguments.size() <= position || !arguments.get(position).isSetLiteral()
+        || !arguments.get(position).getLiteral().isSetStringValue()) {
+      return null;
+    }
+    String type = arguments.get(position).getLiteral().getStringValue().toUpperCase(Locale.ROOT);
+    return switch (type) {
+      case "VARCHAR", "CHAR", "JSON" -> ColumnDataType.STRING;
+      case "BIGINT" -> ColumnDataType.LONG;
+      case "INTEGER" -> ColumnDataType.INT;
+      case "REAL" -> ColumnDataType.FLOAT;
+      case "DECIMAL" -> ColumnDataType.BIG_DECIMAL;
+      default -> {
+        try {
+          yield ColumnDataType.valueOf(type);
+        } catch (IllegalArgumentException e) {
+          yield null;
+        }
+      }
+    };
+  }
 }
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
index 96f53690f5d6..66d84f0fe83e 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java
@@ -222,7 +222,9 @@ private ResultTable reduceToResultTable(DataSchema dataSchema, Object[] finalRes
     int numColumns = columnDataTypes.length;
     for (Object[] rewrittenRow : rows) {
       for (int j = 0; j < numColumns; j++) {
-        rewrittenRow[j] = columnDataTypes[j].format(rewrittenRow[j]);
+        if (rewrittenRow[j] != null) {
+          rewrittenRow[j] = columnDataTypes[j].format(rewrittenRow[j]);
+        }
       }
     }
 
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java
index 78a7cdcada99..01f781930ca2 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java
@@ -19,6 +19,13 @@
 
 package org.apache.pinot.core.query.aggregation.function;
 
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.CustomObject;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
 import org.apache.pinot.queries.FluentQueryTest;
 import org.apache.pinot.spi.config.table.FieldConfig;
 import org.apache.pinot.spi.data.FieldSpec.DataType;
@@ -26,6 +33,10 @@
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.expectThrows;
+
 
 public class ModeAggregationFunctionTest extends AbstractAggregationFunctionTest {
 
@@ -40,6 +51,58 @@ Object[] scenarios() {
     };
   }
 
+  @DataProvider
+  Object[] typedScenarios() {
+    return new Object[]{new Scenario(DataType.STRING, true), new Scenario(DataType.STRING, false),
+        new Scenario(DataType.TIMESTAMP, true), new Scenario(DataType.TIMESTAMP, false)};
+  }
+
+  @Test(dataProvider = "typedScenarios")
+  void typedModeMergesCountsAndHandlesNulls(Scenario scenario) {
+    String type = scenario._dataType.name();
+    String first = scenario._dataType == DataType.STRING ? "apple" : "2026-09-03 10:11:12.123";
+    String shared = scenario._dataType == DataType.STRING ? "banana" : "2026-09-04 10:11:12.456";
+    String last = scenario._dataType == DataType.STRING ? "cherry" : "2026-09-05 10:11:12.789";
+    // Each instance has a different local mode; the shared value wins only after merging full counts.
+    scenario.getDeclaringTable(true)
+        .onFirstInstance("myField", first, first, first, shared, shared, "null")
+        .andOnSecondInstance("myField", last, last, last, shared, shared, "null")
+        .whenQuery("select mode(myField, 'MIN', '" + type + "') as mode from testTable")
+        .thenResultTextIs("mode[" + type + "]\n" + shared)
+        .whenQuery("select mode(myField, 'MIN', '" + type + "') from testTable where myField is null")
+        .thenResultIs(new Object[]{null})
+        .whenQuery("select myField, mode(myField, 'MIN', '" + type + "') from testTable "
+            + "group by myField order by myField")
+        .thenResultIs(new Object[]{first, first}, new Object[]{shared, shared}, new Object[]{last, last},
+            new Object[]{null, null});
+  }
+
+  @DataProvider
+  Object[][] typedStates() {
+    return new Object[][]{{"STRING", "", "zebra"}, {"TIMESTAMP", 9_007_199_254_740_992L, 9_007_199_254_740_993L}};
+  }
+
+  @Test(dataProvider = "typedStates")
+  void typedModePreservesSerializedValuesAndTieReducers(String type, Object smaller, Object larger) {
+    ModeAggregationFunction min = typedMode(type, "MIN");
+    AggregationFunction.SerializedIntermediateResult serialized =
+        min.serializeIntermediateResult(Map.of(smaller, 2L, larger, 1L));
+    Map<?, Long> counts = min.deserializeIntermediateResult(
+        new CustomObject(serialized.getType(), ByteBuffer.wrap(serialized.getBytes())));
+    counts = min.merge(counts, Map.of(larger, 1L));
+    assertEquals(min.getFinalResultColumnType(), ColumnDataType.valueOf(type));
+    assertEquals(min.extractFinalResult(counts), smaller);
+    assertEquals(typedMode(type, "MAX").extractFinalResult(counts), larger);
+    assertNull(min.extractFinalResult(null));
+    expectThrows(IllegalArgumentException.class, () -> typedMode(type, "AVG"));
+  }
+
+  private static ModeAggregationFunction typedMode(String type, String reducer) {
+    return new ModeAggregationFunction(List.of(ExpressionContext.forIdentifier("value"),
+        ExpressionContext.forLiteral(Literal.stringValue(reducer)),
+        ExpressionContext.forLiteral(Literal.stringValue(type))), true);
+  }
+
   public class Scenario {
     private final DataType _dataType;
     private final boolean _dictionary;
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java
new file mode 100644
index 000000000000..6941f16c6f38
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java
@@ -0,0 +1,86 @@
+/**
+ * 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.pinot.core.query.optimizer.statement;
+
+import org.apache.pinot.common.request.PinotQuery;
+import org.apache.pinot.core.query.optimizer.QueryOptimizer;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// Covers inferred aggregate arguments and unchanged legacy calls.
+public class AggregateFunctionRewriteOptimizerTest {
+  private static final QueryOptimizer OPTIMIZER = new QueryOptimizer();
+  private static final Schema SCHEMA = new Schema.SchemaBuilder().setSchemaName("testTable")
+      .addSingleValueDimension("stringCol", DataType.STRING)
+      .addSingleValueDimension("longCol", DataType.LONG)
+      .addMultiValueDimension("mvStringCol", DataType.STRING)
+      .addDateTime("timestampCol", DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS")
+      .build();
+
+  @DataProvider
+  public Object[][] modeExpressions() {
+    return new Object[][]{
+        {"MODE(stringCol)", "MODE(stringCol, 'MIN', 'STRING')"},
+        {"MODE(timestampCol, 'MAX')", "MODE(timestampCol, 'MAX', 'TIMESTAMP')"},
+        {"MODE(CONCAT(stringCol, 'suffix'))", "MODE(CONCAT(stringCol, 'suffix'), 'MIN', 'STRING')"},
+        {"MODE(CASE WHEN stringCol = '' THEN NULL ELSE JSONEXTRACTSCALAR(stringCol, '$.user', 'STRING', '') END)",
+            "MODE(CASE WHEN stringCol = '' THEN NULL ELSE JSONEXTRACTSCALAR(stringCol, '$.user', 'STRING', '') END, "
+                + "'MIN', 'STRING')"},
+        {"MODE(CAST(stringCol AS TIMESTAMP))", "MODE(CAST(stringCol AS TIMESTAMP), 'MIN', 'TIMESTAMP')"},
+        {"fromTimestamp(MODE(timestampCol))", "fromTimestamp(MODE(timestampCol, 'MIN', 'TIMESTAMP'))"}
+    };
+  }
+
+  @Test(dataProvider = "modeExpressions")
+  public void testModeExpressions(String original, String rewritten) {
+    String prefix = "SELECT ";
+    TestHelper.assertEqualsQuery(prefix + original + " FROM testTable", prefix + rewritten + " FROM testTable", SCHEMA);
+  }
+
+  @Test
+  public void testLegacyCallsAndOtherRewritesAreUnchanged() {
+    assertUnchanged("SELECT MODE(longCol, 'AVG'), MIN(stringCol), SUM(longCol), "
+        + "MODE(stringCol, 'MAX', 'STRING'), MODE(timestampCol, 'MIN', 'TIMESTAMP') FROM testTable", SCHEMA);
+    TestHelper.assertEqualsQuery(
+        "SET autoRewriteAggregationType=true; SELECT MODE(stringCol), MIN(stringCol), MAX(longCol), SUM(longCol), "
+            + "COUNT(*) FROM testTable",
+        "SET autoRewriteAggregationType=true; SELECT MODE(stringCol, 'MIN', 'STRING'), MINSTRING(stringCol), "
+            + "MAXLONG(longCol), SUMLONG(longCol), COUNT(*) FROM testTable", SCHEMA);
+  }
+
+  @Test
+  public void testUnknownTypesDoNotInitializeServerTransforms() {
+    assertUnchanged("SELECT MODE(unknownCol), MODE(mvStringCol), MODE(NULLIF(longCol, 0)), "
+        + "MODE(LOOKUP('baseballTeams', 'teamInteger', 'teamID', stringCol)) FROM testTable", SCHEMA);
+    assertUnchanged("SELECT MODE(stringCol) FROM testTable", null);
+  }
+
+  private static void assertUnchanged(String sql, Schema schema) {
+    PinotQuery original = CalciteSqlParser.compileToPinotQuery(sql);
+    PinotQuery optimized = original.deepCopy();
+    OPTIMIZER.optimize(optimized, schema);
+    assertEquals(optimized, original);
+  }
+}
diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateExchangeNodeInsertRule.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateExchangeNodeInsertRule.java
index 61a0b33cfebb..fc52168c8d18 100644
--- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateExchangeNodeInsertRule.java
+++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateExchangeNodeInsertRule.java
@@ -511,6 +511,7 @@ private static PinotLogicalAggregate convertAggFromIntermediateInput(Aggregate a
           }
         }
       }
+      rexList = PinotRuleUtils.inferAggregateArguments(orgAggCall, input, rexList);
       aggCalls.add(buildAggCall(exchange, orgAggCall, rexList, aggColumnOffset, aggType, leafReturnFinalResult));
     }
 
@@ -550,6 +551,7 @@ private static List<AggregateCall> buildAggCalls(Aggregate aggRel, List<Aggregat
           }
         }
       }
+      rexList = PinotRuleUtils.inferAggregateArguments(orgAggCall, input, rexList);
       aggCalls.add(buildAggCall(input, orgAggCall, rexList, aggRel.getGroupCount(), aggType, leafReturnFinalResult));
     }
     return aggCalls;
diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java
index b7ad7a4edb2c..aba6b4dc3a06 100644
--- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java
+++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java
@@ -27,6 +27,7 @@
 import org.apache.calcite.plan.hep.HepRelVertex;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.AggregateCall;
 import org.apache.calcite.rel.core.Exchange;
 import org.apache.calcite.rel.core.Filter;
 import org.apache.calcite.rel.core.Join;
@@ -34,6 +35,7 @@
 import org.apache.calcite.rel.core.RelFactories;
 import org.apache.calcite.rel.core.TableScan;
 import org.apache.calcite.rel.core.Window;
+import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLiteral;
@@ -51,6 +53,10 @@
 import org.apache.calcite.util.Util;
 import org.apache.pinot.calcite.rel.hint.PinotHintStrategyTable;
 import org.apache.pinot.common.function.sql.PinotSqlFunction;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.query.planner.logical.RelToPlanNodeConverter;
+import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
 
 
 public class PinotRuleUtils {
@@ -97,6 +103,36 @@ public static boolean isAggregate(RelNode rel) {
     return unboxRel(rel) instanceof Aggregate;
   }
 
+  /// Appends arguments inferred by the aggregation's metadata using its original input types. Final stages must
+  /// resolve against the original input rather than the intermediate accumulator referenced by `arguments`.
+  public static List<RexNode> inferAggregateArguments(AggregateCall aggregateCall, RelNode originalInput,
+      List<RexNode> arguments) {
+    String functionName = aggregateCall.getAggregation().getName();
+    if (!AggregationFunctionType.isAggregationFunction(functionName)) {
+      return arguments;
+    }
+    List<Integer> argList = aggregateCall.getArgList();
+    List<String> inferredArguments = AggregationFunctionType.getAggregationFunctionType(functionName)
+        .inferArguments(argList.size(), ordinal -> {
+          ColumnDataType type = RelToPlanNodeConverter.convertToColumnDataType(
+              originalInput.getRowType().getFieldList().get(argList.get(ordinal)).getType());
+          if (type.isArray() || type == ColumnDataType.OBJECT) {
+            return DataType.UNKNOWN;
+          }
+          return type == ColumnDataType.MAP ? DataType.MAP : type.toDataType();
+        });
+    if (inferredArguments.isEmpty()) {
+      return arguments;
+    }
+    List<RexNode> resolvedArguments = new ArrayList<>(arguments.size() + inferredArguments.size());
+    resolvedArguments.addAll(arguments);
+    RexBuilder rexBuilder = originalInput.getCluster().getRexBuilder();
+    for (String argument : inferredArguments) {
+      resolvedArguments.add(rexBuilder.makeLiteral(argument));
+    }
+    return resolvedArguments;
+  }
+
   /// utility logic to determine if a JOIN can be pushed down to the leaf-stage execution and leverage the
   /// segment-local info (indexing and others) to speed up the execution.
   ///
diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java
index e05ee827547d..eec0041d47b2 100644
--- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java
+++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java
@@ -192,7 +192,6 @@ public static PinotOperatorTable instance(boolean nullHandlingEnabled) {
       PinotMinMaxFunction.MIN,
       PinotMinMaxFunction.MAX,
       PinotAvgFunction.INSTANCE,
-      SqlStdOperatorTable.MODE,
       SqlStdOperatorTable.STDDEV_POP,
       SqlStdOperatorTable.COVAR_POP,
       SqlStdOperatorTable.COVAR_SAMP,
diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/AggregatePushdownRule.java b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/AggregatePushdownRule.java
index ca30c851a013..cd0ca78fbc1e 100644
--- a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/AggregatePushdownRule.java
+++ b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/AggregatePushdownRule.java
@@ -236,6 +236,7 @@ private static PhysicalAggregate convertAggForGroupingSets(PhysicalAggregate phy
           }
         }
       }
+      rexList = PinotRuleUtils.inferAggregateArguments(orgAggCall, aggRel.getInput(), rexList);
       aggCalls.add(buildAggCall(exchange, orgAggCall, rexList, finalGroupCount, AggType.FINAL, false));
     }
     ImmutableBitSet groupSet = ImmutableBitSet.range(finalGroupCount);
@@ -322,6 +323,7 @@ private static PhysicalAggregate convertAggFromIntermediateInput(PhysicalAggrega
           }
         }
       }
+      rexList = PinotRuleUtils.inferAggregateArguments(orgAggCall, input, rexList);
       aggCalls.add(buildAggCall(exchange, orgAggCall, rexList, groupCount, aggType, leafReturnFinalResult));
     }
     ImmutableBitSet.Builder groupSetBuilder = ImmutableBitSet.builder();
@@ -370,6 +372,7 @@ public static List<AggregateCall> buildAggCalls(Aggregate aggRel, List<Aggregate
           }
         }
       }
+      rexList = PinotRuleUtils.inferAggregateArguments(orgAggCall, input, rexList);
       aggCalls.add(buildAggCall(input, orgAggCall, rexList, aggRel.getGroupCount(), aggType, leafReturnFinalResult));
     }
     return aggCalls;
diff --git a/pinot-query-planner/src/test/java/org/apache/pinot/query/queries/ModeSqlPlannerTest.java b/pinot-query-planner/src/test/java/org/apache/pinot/query/queries/ModeSqlPlannerTest.java
new file mode 100644
index 000000000000..52daa164811e
--- /dev/null
+++ b/pinot-query-planner/src/test/java/org/apache/pinot/query/queries/ModeSqlPlannerTest.java
@@ -0,0 +1,151 @@
+/**
+ * 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.pinot.query.queries;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.query.QueryEnvironmentTestBase;
+import org.apache.pinot.query.planner.logical.RexExpression;
+import org.apache.pinot.query.planner.physical.DispatchablePlanFragment;
+import org.apache.pinot.query.planner.physical.DispatchableSubPlan;
+import org.apache.pinot.query.planner.plannode.AggregateNode;
+import org.apache.pinot.query.planner.plannode.AggregateNode.AggType;
+import org.apache.pinot.query.planner.plannode.PlanNode;
+import org.apache.pinot.spi.exception.QueryException;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Verifies automatic distributed MODE type inference with both physical planners.
+public class ModeSqlPlannerTest extends QueryEnvironmentTestBase {
+  @DataProvider
+  public Object[][] physicalOptimizers() {
+    return new Object[][]{{false}, {true}};
+  }
+
+  @Test(dataProvider = "physicalOptimizers")
+  public void testDistributedModeTypes(boolean usePhysicalOptimizer) {
+    DispatchableSubPlan plan = _queryEnvironment.planQuery("SET usePhysicalOptimizer=" + usePhysicalOptimizer + "; "
+        + "SELECT MODE(NULLIF(col1, '')), MODE(ts_timestamp, 'MAX'), MODE(col3, 'AVG'), "
+        + "MODE(col1, 'MAX', 'STRING') FROM a");
+    PlanNode root = plan.getQueryStageMap().get(0).getPlanFragment().getFragmentRoot();
+    assertEquals(root.getDataSchema().getColumnDataTypes(),
+        new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.TIMESTAMP, ColumnDataType.DOUBLE,
+            ColumnDataType.STRING});
+
+    List<AggregateNode> aggregates = findAggregates(plan);
+    assertFalse(aggregates.isEmpty());
+    boolean sawIntermediate = false;
+    boolean sawFinal = false;
+    for (AggregateNode aggregate : aggregates) {
+      List<RexExpression.FunctionCall> calls = aggregate.getAggCalls();
+      assertEquals(calls.stream().map(RexExpression.FunctionCall::getFunctionName).toList(),
+          List.of("MODE", "MODE", "MODE", "MODE"));
+      assertTypedCall(calls.get(0), "MIN", "STRING");
+      assertTypedCall(calls.get(1), "MAX", "TIMESTAMP");
+      assertEquals(calls.get(2).getFunctionOperands().size(), 2);
+      assertTypedCall(calls.get(3), "MAX", "STRING");
+      if (aggregate.getAggType().isOutputIntermediateFormat() && !aggregate.isLeafReturnFinalResult()) {
+        sawIntermediate = true;
+        assertEquals(aggregate.getDataSchema().getColumnDataTypes(),
+            new ColumnDataType[]{ColumnDataType.OBJECT, ColumnDataType.OBJECT, ColumnDataType.OBJECT,
+                ColumnDataType.OBJECT});
+      } else {
+        sawFinal = true;
+        assertEquals(aggregate.getDataSchema().getColumnDataTypes(),
+            new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.TIMESTAMP, ColumnDataType.DOUBLE,
+                ColumnDataType.STRING});
+      }
+    }
+    assertTrue(sawIntermediate, "Distributed MODE must exchange frequency counts");
+    assertTrue(sawFinal, "Distributed MODE must retain its final result types");
+  }
+
+  @Test(dataProvider = "physicalOptimizers")
+  public void testGroupedModeTypes(boolean usePhysicalOptimizer) {
+    List<String> queries = List.of(
+        "SELECT /*+ aggOptions(is_skip_leaf_stage_group_by='true') */ col2, MODE(col1), MODE(ts_timestamp, 'MAX') "
+            + "FROM a GROUP BY col2",
+        "SELECT col3, GROUPING(col3), MODE(col1), MODE(ts_timestamp, 'MAX') "
+            + "FROM a GROUP BY GROUPING SETS ((col3), ())");
+    for (int i = 0; i < queries.size(); i++) {
+      DispatchableSubPlan plan = _queryEnvironment.planQuery(
+          "SET usePhysicalOptimizer=" + usePhysicalOptimizer + "; " + queries.get(i));
+      List<AggregateNode> aggregates = findAggregates(plan);
+      assertFalse(aggregates.isEmpty());
+      for (AggregateNode aggregate : aggregates) {
+        List<RexExpression.FunctionCall> calls = aggregate.getAggCalls();
+        assertEquals(calls.size(), 2);
+        assertTypedCall(calls.get(0), "MIN", "STRING");
+        assertTypedCall(calls.get(1), "MAX", "TIMESTAMP");
+        if (aggregate.getAggType() == AggType.FINAL) {
+          for (int j = 0; j < calls.size(); j++) {
+            RexExpression.InputRef input = (RexExpression.InputRef) calls.get(j).getFunctionOperands().get(0);
+            assertEquals(input.getIndex(), aggregate.getGroupKeys().size() + j);
+          }
+        }
+      }
+      AggType expectedType = i == 0 ? AggType.DIRECT : AggType.FINAL;
+      assertTrue(aggregates.stream().anyMatch(aggregate -> aggregate.getAggType() == expectedType));
+    }
+  }
+
+  @Test
+  public void testInvalidTypeAnnotations() {
+    for (String expression : List.of("MODE(col1, 'MIN', 'TIMESTAMP')", "MODE(ts_timestamp, 'MIN', 'STRING')",
+        "MODE(col1, 'MIN', 'INVALID')", "MODE(col1, 'MIN', col1)")) {
+      QueryException error = expectThrows(QueryException.class,
+          () -> _queryEnvironment.compile("SELECT " + expression + " FROM a"));
+      assertTrue(error.getMessage().contains("MODE type argument"), error.getMessage());
+    }
+  }
+
+  private static void assertTypedCall(RexExpression.FunctionCall call, String reducer, String type) {
+    assertEquals(call.getFunctionName(), "MODE");
+    List<RexExpression> arguments = call.getFunctionOperands();
+    assertEquals(arguments.size(), 3);
+    assertTrue(arguments.get(1) instanceof RexExpression.Literal);
+    assertTrue(arguments.get(2) instanceof RexExpression.Literal);
+    assertEquals(((RexExpression.Literal) arguments.get(1)).getValue(), reducer);
+    assertEquals(((RexExpression.Literal) arguments.get(2)).getValue(), type);
+  }
+
+  private static List<AggregateNode> findAggregates(DispatchableSubPlan plan) {
+    List<AggregateNode> aggregates = new ArrayList<>();
+    for (DispatchablePlanFragment fragment : plan.getQueryStageMap().values()) {
+      findAggregates(fragment.getPlanFragment().getFragmentRoot(), aggregates);
+    }
+    return aggregates;
+  }
+
+  private static void findAggregates(PlanNode node, List<AggregateNode> aggregates) {
+    if (node instanceof AggregateNode) {
+      aggregates.add((AggregateNode) node);
+    }
+    for (PlanNode input : node.getInputs()) {
+      findAggregates(input, aggregates);
+    }
+  }
+}
diff --git a/pinot-query-runtime/src/test/resources/queries/ModeAggregates.json b/pinot-query-runtime/src/test/resources/queries/ModeAggregates.json
new file mode 100644
index 000000000000..de72a8941234
--- /dev/null
+++ b/pinot-query-runtime/src/test/resources/queries/ModeAggregates.json
@@ -0,0 +1,54 @@
+{
+  "mode_string_timestamp_distributed": {
+    "comments": "Merge STRING and TIMESTAMP frequency states across four partitions on two servers.",
+    "extraProps": {"enableColumnBasedNullHandling": true},
+    "tables": {
+      "items": {
+        "schema": [
+          {"name": "lpn_id", "type": "STRING"},
+          {"name": "pallet_id", "type": "STRING"},
+          {"name": "created_on", "type": "TIMESTAMP"}
+        ],
+        "inputs": [
+          ["a", "p2", "2026-09-03 08:00:00.001"],
+          ["a", "p2", "2026-09-03 08:00:00.001"],
+          ["a", "p2", "2026-09-03 09:00:00.123"],
+          ["a", "p1", "2026-09-03 09:00:00.123"],
+          ["a", "p1", "2026-09-03 09:00:00.123"],
+          ["a", "p3", "2026-09-03 10:00:00.999"],
+          ["nulls", null, null],
+          ["nulls", null, null],
+          ["ties", "z", "1970-01-01 00:00:00.001"],
+          ["ties", "a", "1969-12-31 23:59:59.999"],
+          ["ties", "z", "1970-01-01 00:00:00.001"],
+          ["ties", "a", "1969-12-31 23:59:59.999"],
+          ["empty", "", "2026-09-03 00:00:00.001"],
+          ["empty", "z", null],
+          ["empty", "", null]
+        ]
+      }
+    },
+    "queries": [
+      {
+        "description": "Grouped modes ignore nulls, retain empty strings, and break ties by the smallest value",
+        "sql": "SELECT lpn_id, MODE(pallet_id), MODE(created_on) FROM {items} GROUP BY lpn_id",
+        "outputs": [
+          ["a", "p2", "2026-09-03 09:00:00.123"],
+          ["nulls", null, null],
+          ["ties", "a", "1969-12-31 23:59:59.999"],
+          ["empty", "", "2026-09-03 00:00:00.001"]
+        ]
+      },
+      {
+        "description": "Typed MAX tie reducers select the largest original value",
+        "sql": "SELECT MODE(pallet_id, 'MAX'), MODE(created_on, 'MAX') FROM {items} WHERE lpn_id = 'ties'",
+        "outputs": [["z", "1970-01-01 00:00:00.001"]]
+      },
+      {
+        "description": "Empty aggregate inputs return null for both types",
+        "sql": "SELECT MODE(pallet_id), MODE(created_on) FROM {items} WHERE lpn_id = 'missing'",
+        "outputs": [[null, null]]
+      }
+    ]
+  }
+}
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java
index 080138b3a327..83865ef1cd27 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java
@@ -20,12 +20,15 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import java.util.Set;
+import java.util.function.IntFunction;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.SqlCallBinding;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperatorBinding;
 import org.apache.calcite.sql.type.OperandTypes;
@@ -36,17 +39,19 @@
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.Strings;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
 import org.apache.pinot.spi.utils.CommonConstants;
 
+import static com.google.common.base.Preconditions.checkArgument;
+
 
 /// NOTES:
 /// - No underscore is allowed in the enum name.
 /// - '$' is allowed in the name field but not in the enum name.
 ///
 /// This enum is used both in the v1 engine and multistage engine to define the allowed Pinot aggregation functions.
-/// The v1 engine only relies on the 'name' field, whereas all the other fields are used in the multistage engine
-/// to register the aggregation function with Calcite. This allows using a unified approach to aggregations across both
-/// the v1 and multistage engines.
+/// Both engines use the function name and argument inference. The SQL metadata also registers the aggregation
+/// function with Calcite in the multistage engine.
 public enum AggregationFunctionType {
   // Aggregation functions for single-valued columns
   COUNT("count"),
@@ -63,7 +68,25 @@ public enum AggregationFunctionType {
   SUMLONG("sumLong", ReturnTypes.AGG_SUM, OperandTypes.or(OperandTypes.INTEGER, OperandTypes.ARRAY_OF_INTEGER)),
   SUMPRECISION("sumPrecision", ReturnTypes.explicit(SqlTypeName.DECIMAL), OperandTypes.ANY, SqlTypeName.OTHER),
   AVG("avg", SqlTypeName.OTHER, SqlTypeName.DOUBLE),
-  MODE("mode", SqlTypeName.OTHER, SqlTypeName.DOUBLE),
+  MODE("mode", new ModeReturnTypeInference(), OperandTypes.or(
+      OperandTypes.NUMERIC, OperandTypes.CHARACTER, OperandTypes.TIMESTAMP,
+      OperandTypes.family(List.of(SqlTypeFamily.NUMERIC, SqlTypeFamily.CHARACTER), i -> i == 1),
+      OperandTypes.family(List.of(SqlTypeFamily.CHARACTER, SqlTypeFamily.CHARACTER), i -> i == 1),
+      OperandTypes.family(List.of(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.CHARACTER), i -> i == 1),
+      OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER, SqlTypeFamily.CHARACTER)),
+      ReturnTypes.explicit(SqlTypeName.OTHER), null, SqlKind.MODE) {
+    @Override
+    public List<String> inferArguments(int argumentCount, IntFunction<DataType> argumentTypes) {
+      if (argumentCount == 0 || argumentCount >= 3) {
+        return List.of();
+      }
+      DataType inputType = argumentTypes.apply(0);
+      if (inputType != DataType.STRING && inputType != DataType.TIMESTAMP) {
+        return List.of();
+      }
+      return argumentCount == 1 ? List.of("MIN", inputType.name()) : List.of(inputType.name());
+    }
+  },
   ANYVALUE("anyValue", ReturnTypes.ARG0, OperandTypes.ANY, SqlTypeName.OTHER),
   FIRSTWITHTIME("firstWithTime", ReturnTypes.ARG0,
       OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER), SqlTypeName.OTHER),
@@ -365,6 +388,13 @@ public static String getNormalizedAggregationFunctionName(String functionName) {
     return Strings.CS.remove(StringUtils.remove(functionName, '_').toUpperCase(), "$");
   }
 
+  /// Infers internal string literal arguments to append before execution.
+  /// Argument types are resolved lazily so functions without inferred arguments do not require expression typing.
+  /// Unknown or multi-value argument types are represented by [DataType#UNKNOWN].
+  public List<String> inferArguments(int argumentCount, IntFunction<DataType> argumentTypes) {
+    return List.of();
+  }
+
   /// Returns the corresponding aggregation function type for the given function name.
   ///
   /// NOTE: Underscores in the function name are ignored.
@@ -413,6 +443,31 @@ public static AggregationFunctionType getAggregationFunctionType(String function
     }
   }
 
+  /// Preserves the legacy DOUBLE result for numeric MODE while retaining the type of string and timestamp inputs.
+  private static class ModeReturnTypeInference implements SqlReturnTypeInference {
+    @Override
+    public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
+      RelDataType operandType = opBinding.getOperandType(0);
+      // Aggregate bindings contain operand types only. Validate the annotation while SQL literals are available;
+      // later planning derives the same result type from the original input and carries it through merge stages.
+      if (opBinding.getOperandCount() == 3 && opBinding instanceof SqlCallBinding) {
+        checkArgument(opBinding.isOperandLiteral(2, false),
+            "MODE type argument must be a STRING or TIMESTAMP literal");
+        String type = opBinding.getOperandLiteralValue(2, String.class);
+        type = type != null ? type.toUpperCase(Locale.ROOT) : null;
+        boolean matches = "STRING".equals(type) && SqlTypeName.STRING_TYPES.contains(operandType.getSqlTypeName())
+            || "TIMESTAMP".equals(type) && operandType.getSqlTypeName() == SqlTypeName.TIMESTAMP;
+        checkArgument(matches, "MODE type argument must match the STRING or TIMESTAMP input, got: %s", type);
+      }
+      if (SqlTypeName.STRING_TYPES.contains(operandType.getSqlTypeName())
+          || operandType.getSqlTypeName() == SqlTypeName.TIMESTAMP) {
+        return ReturnTypes.ARG0_NULLABLE_IF_EMPTY.inferReturnType(opBinding);
+      }
+      return opBinding.getTypeFactory().createTypeWithNullability(
+          opBinding.getTypeFactory().createSqlType(SqlTypeName.DOUBLE), true);
+    }
+  }
+
   private static class ArrayReturnTypeInference implements SqlReturnTypeInference {
     final SqlTypeName _sqlTypeName;