diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java
index 0241202d99..94da2934f3 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pinot.core.operator.transform.function;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.jayway.jsonpath.Configuration;
@@ -30,34 +31,50 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.IntFunction;
+import javax.annotation.Nullable;
 import org.apache.pinot.common.function.JsonPathCache;
 import org.apache.pinot.core.operator.ColumnContext;
 import org.apache.pinot.core.operator.blocks.ValueBlock;
 import org.apache.pinot.core.operator.transform.TransformResultMetadata;
+import org.apache.pinot.core.query.request.context.QueryContext;
 import org.apache.pinot.core.util.NumberUtils;
 import org.apache.pinot.core.util.NumericException;
 import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.BooleanUtils;
 import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.TimestampUtils;
+import org.roaringbitmap.RoaringBitmap;
 
-/**
- * The <code>JsonExtractScalarTransformFunction</code> class implements the json
- * path transformation based on
- * <a href="https://goessner.net/articles/JsonPath/">Stefan Goessner JsonPath
- * implementation.</a>.
- *
- * Please note, currently this method only works with String field. The values
- * in this field should be Json String.
- *
- * Usage:
- * jsonExtractScalar(jsonFieldName, 'jsonPath', 'resultsType')
- * <code>jsonFieldName</code> is the Json String field/expression.
- * <code>jsonPath</code> is a JsonPath expression which used to read from JSON
- * document
- * <code>results_type</code> refers to the results data type, could be INT,
- * LONG, FLOAT, DOUBLE, BIG_DECIMAL, STRING,
- * INT_ARRAY, LONG_ARRAY, FLOAT_ARRAY, DOUBLE_ARRAY, STRING_ARRAY.
- *
- */
+
+/// Implements the `jsonExtractScalar(jsonField, jsonPath, resultsType[, defaultValue])` transform.
+/// Reads a JSON document from `jsonField` for each row, resolves the
+/// [Stefan Goessner JsonPath](https://goessner.net/articles/JsonPath/) expression against it, and
+/// converts the resolved value to `resultsType`.
+///
+/// **Arguments:**
+/// - `jsonField` — single-value `STRING` or `BYTES` column / transform expression containing JSON.
+/// - `jsonPath` — JsonPath expression used to read the value.
+/// - `resultsType` — Pinot data type for the output. Append `_ARRAY` for multi-value results.
+/// - `defaultValue` (optional) — used when the path resolves to `null` or fails. Without it, unresolved
+///   SV rows throw `IllegalArgumentException`; MV rows surface as empty arrays, but null elements within
+///   a resolved array still throw.
+///
+/// **Supported `resultsType`:** `INT`, `LONG`, `FLOAT`, `DOUBLE`, `BIG_DECIMAL`, `BOOLEAN`, `TIMESTAMP`,
+/// `STRING`, `JSON`, `BYTES`, plus `_ARRAY` variants of `INT` / `LONG` / `FLOAT` / `DOUBLE` /
+/// `STRING`.
+///
+/// **Per-row coercion** of the JsonPath result to `resultsType`:
+/// - `BOOLEAN` (stored as `INT`) follows Pinot's numeric convention — any non-zero `Number` is true;
+///   `Boolean` and `"true"` / `"TRUE"` / `"1"` strings (via [BooleanUtils#toInt(String)]) are also true.
+/// - `TIMESTAMP` (stored as `LONG`) accepts numeric epoch millis directly; strings go through
+///   [TimestampUtils#toMillisSinceEpoch] (ISO-8601 and numeric millis strings).
+/// - `STRING` returns `String` values as-is; other JSON values are serialized via
+///   [JsonUtils#objectToString].
+/// - `BIG_DECIMAL` and `STRING` paths use a BigDecimal-preserving JSON parser
+///   (`JSON_PARSER_CONTEXT_WITH_BIG_DECIMAL`) to avoid precision loss on numeric values; other paths use
+///   the default parser.
+/// - Other types coerce via `Number` cast (preserved as the canonical primitive form) or
+///   `parse*(toString())`.
 public class JsonExtractScalarTransformFunction extends BaseTransformFunction {
   public static final String FUNCTION_NAME = "jsonExtractScalar";
 
@@ -77,7 +94,10 @@ public class JsonExtractScalarTransformFunction extends BaseTransformFunction {
 
   private TransformFunction _jsonFieldTransformFunction;
   private JsonPath _jsonPath;
+  private DataType _dataType;
+  private DataType _storedType;
   private Object _defaultValue;
+  private boolean _defaultIsNull;
   private TransformResultMetadata _resultMetadata;
 
   @Override
@@ -86,8 +106,9 @@ public String getName() {
   }
 
   @Override
-  public void init(List<TransformFunction> arguments, Map<String, ColumnContext> columnContextMap) {
-    super.init(arguments, columnContextMap);
+  public void init(List<TransformFunction> arguments, Map<String, ColumnContext> columnContextMap,
+      QueryContext queryContext) {
+    super.init(arguments, columnContextMap, queryContext);
     // Check that there are exactly 3 or 4 arguments
     if (arguments.size() < 3 || arguments.size() > 4) {
       throw new IllegalArgumentException(
@@ -106,20 +127,58 @@ public void init(List<TransformFunction> arguments, Map<String, ColumnContext> c
     _jsonPath = JsonPathCache.INSTANCE.getOrCompute(jsonPathString);
     String resultsType = ((LiteralTransformFunction) arguments.get(2)).getStringLiteral().toUpperCase();
     boolean isSingleValue = !resultsType.endsWith("_ARRAY");
-    DataType dataType;
     try {
-      dataType = DataType.valueOf(isSingleValue ? resultsType : resultsType.substring(0, resultsType.length() - 6));
+      _dataType = DataType.valueOf(isSingleValue ? resultsType : resultsType.substring(0, resultsType.length() - 6));
     } catch (Exception e) {
       throw new IllegalArgumentException(String.format(
           "Unsupported results type: %s for jsonExtractScalar function. Supported types are: "
-              + "INT/LONG/FLOAT/DOUBLE/BOOLEAN/BIG_DECIMAL/TIMESTAMP/STRING/INT_ARRAY/LONG_ARRAY/FLOAT_ARRAY"
-              + "/DOUBLE_ARRAY/STRING_ARRAY",
-          resultsType));
+              + "INT/LONG/FLOAT/DOUBLE/BIG_DECIMAL/BOOLEAN/TIMESTAMP/STRING/JSON/BYTES/"
+              + "INT_ARRAY/LONG_ARRAY/FLOAT_ARRAY/DOUBLE_ARRAY/STRING_ARRAY", resultsType));
     }
+    _storedType = _dataType.getStoredType();
     if (arguments.size() == 4) {
-      _defaultValue = dataType.convert(((LiteralTransformFunction) arguments.get(3)).getStringLiteral());
+      LiteralTransformFunction literalTransformFun = (LiteralTransformFunction) arguments.get(3);
+      _defaultIsNull = literalTransformFun.isNull() && _nullHandlingEnabled;
+      switch (_dataType) {
+        case INT:
+          _defaultValue = literalTransformFun.getIntLiteral();
+          break;
+        case LONG:
+          _defaultValue = literalTransformFun.getLongLiteral();
+          break;
+        case FLOAT:
+          _defaultValue = literalTransformFun.getFloatLiteral();
+          break;
+        case DOUBLE:
+          _defaultValue = literalTransformFun.getDoubleLiteral();
+          break;
+        case BIG_DECIMAL:
+          _defaultValue = literalTransformFun.getBigDecimalLiteral();
+          break;
+        case BOOLEAN:
+          // Stored as Integer 0 / 1 to match BOOLEAN's storedType (INT) so per-row consumers can
+          // unbox directly without a Boolean → Integer conversion.
+          _defaultValue = literalTransformFun.getBooleanLiteral() ? 1 : 0;
+          break;
+        case TIMESTAMP:
+          // Use long literal so numeric millis stay exact and string timestamps use LiteralContext parsing.
+          _defaultValue = literalTransformFun.getLongLiteral();
+          break;
+        case STRING:
+        case JSON:
+          _defaultValue = literalTransformFun.getStringLiteral();
+          break;
+        case BYTES:
+          _defaultValue = literalTransformFun.getBytesLiteral();
+          break;
+        default:
+          throw new IllegalArgumentException(
+              "Unsupported results type: " + _dataType + " for jsonExtractScalar function. Supported types are: "
+                  + "INT/LONG/FLOAT/DOUBLE/BIG_DECIMAL/BOOLEAN/TIMESTAMP/STRING/JSON/BYTES"
+          );
+      }
     }
-    _resultMetadata = new TransformResultMetadata(dataType, isSingleValue, false);
+    _resultMetadata = new TransformResultMetadata(_dataType, isSingleValue, false);
   }
 
   @Override
@@ -127,22 +186,50 @@ public TransformResultMetadata getResultMetadata() {
     return _resultMetadata;
   }
 
+  @Override
+  @Nullable
+  public RoaringBitmap getNullBitmap(ValueBlock valueBlock) {
+    if (!_defaultIsNull) {
+      return super.getNullBitmap(valueBlock);
+    }
+    RoaringBitmap bitmap = new RoaringBitmap();
+    for (TransformFunction arg : _arguments.subList(1, _arguments.size() - 1)) {
+      RoaringBitmap argBitmap = arg.getNullBitmap(valueBlock);
+      if (argBitmap != null) {
+        bitmap.or(argBitmap);
+      }
+    }
+    int numDocs = valueBlock.getNumDocs();
+    RoaringBitmap nullBitmap = new RoaringBitmap();
+    IntFunction<Object> resultExtractor = getResultExtractor(valueBlock);
+    for (int i = 0; i < numDocs; i++) {
+      Object result = null;
+      try {
+        result = resultExtractor.apply(i);
+      } catch (Exception ignored) {
+      }
+      if (result == null) {
+        nullBitmap.add(i);
+      }
+    }
+    if (!nullBitmap.isEmpty()) {
+      bitmap.or(nullBitmap);
+    }
+    if (bitmap.isEmpty()) {
+      return null;
+    }
+    return bitmap;
+  }
+
   @Override
   public int[] transformToIntValuesSV(ValueBlock valueBlock) {
-    if (_resultMetadata.getDataType().getStoredType() != DataType.INT) {
+    if (_storedType != DataType.INT) {
       return super.transformToIntValuesSV(valueBlock);
     }
-
     initIntValuesSV(valueBlock.getNumDocs());
     IntFunction<Object> resultExtractor = getResultExtractor(valueBlock);
-    int defaultValue = 0;
-    if (_defaultValue != null) {
-      if (_defaultValue instanceof Number) {
-        defaultValue = ((Number) _defaultValue).intValue();
-      } else {
-        defaultValue = Integer.parseInt(_defaultValue.toString());
-      }
-    }
+    int defaultValue = _defaultValue != null ? (Integer) _defaultValue : 0;
+    boolean isBoolean = _dataType == DataType.BOOLEAN;
     int numDocs = valueBlock.getNumDocs();
     for (int i = 0; i < numDocs; i++) {
       Object result = null;
@@ -158,30 +245,20 @@ public int[] transformToIntValuesSV(ValueBlock valueBlock) {
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
-      if (result instanceof Number) {
-        _intValuesSV[i] = ((Number) result).intValue();
-      } else {
-        _intValuesSV[i] = Integer.parseInt(result.toString());
-      }
+      _intValuesSV[i] = toInt(result, isBoolean);
     }
     return _intValuesSV;
   }
 
   @Override
   public long[] transformToLongValuesSV(ValueBlock valueBlock) {
-    if (_resultMetadata.getDataType().getStoredType() != DataType.LONG) {
+    if (_storedType != DataType.LONG) {
       return super.transformToLongValuesSV(valueBlock);
     }
     initLongValuesSV(valueBlock.getNumDocs());
     IntFunction<Object> resultExtractor = getResultExtractor(valueBlock);
-    long defaultValue = 0;
-    if (_defaultValue != null) {
-      if (_defaultValue instanceof Number) {
-        defaultValue = ((Number) _defaultValue).longValue();
-      } else {
-        defaultValue = Long.parseLong(_defaultValue.toString());
-      }
-    }
+    long defaultValue = _defaultValue != null ? (Long) _defaultValue : 0L;
+    boolean isTimestamp = _dataType == DataType.TIMESTAMP;
     int numDocs = valueBlock.getNumDocs();
     for (int i = 0; i < numDocs; i++) {
       Object result = null;
@@ -197,31 +274,19 @@ public long[] transformToLongValuesSV(ValueBlock valueBlock) {
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
-      if (result instanceof Number) {
-        _longValuesSV[i] = ((Number) result).longValue();
-      } else {
-        try {
-          _longValuesSV[i] = NumberUtils.parseJsonLong(result.toString());
-        } catch (NumericException nfe) {
-          throw new NumberFormatException("For input string: \"" + result + "\"");
-        }
-      }
+      _longValuesSV[i] = toLong(result, isTimestamp);
     }
     return _longValuesSV;
   }
 
   @Override
   public float[] transformToFloatValuesSV(ValueBlock valueBlock) {
+    if (_storedType != DataType.FLOAT) {
+      return super.transformToFloatValuesSV(valueBlock);
+    }
     initFloatValuesSV(valueBlock.getNumDocs());
     IntFunction<Object> resultExtractor = getResultExtractor(valueBlock);
-    float defaultValue = 0;
-    if (_defaultValue != null) {
-      if (_defaultValue instanceof Number) {
-        defaultValue = ((Number) _defaultValue).floatValue();
-      } else {
-        defaultValue = Float.parseFloat(_defaultValue.toString());
-      }
-    }
+    float defaultValue = _defaultValue != null ? (Float) _defaultValue : 0f;
     int numDocs = valueBlock.getNumDocs();
     for (int i = 0; i < numDocs; i++) {
       Object result = null;
@@ -237,27 +302,19 @@ public float[] transformToFloatValuesSV(ValueBlock valueBlock) {
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
-      if (result instanceof Number) {
-        _floatValuesSV[i] = ((Number) result).floatValue();
-      } else {
-        _floatValuesSV[i] = Float.parseFloat(result.toString());
-      }
+      _floatValuesSV[i] = toFloat(result);
     }
     return _floatValuesSV;
   }
 
   @Override
   public double[] transformToDoubleValuesSV(ValueBlock valueBlock) {
+    if (_storedType != DataType.DOUBLE) {
+      return super.transformToDoubleValuesSV(valueBlock);
+    }
     initDoubleValuesSV(valueBlock.getNumDocs());
     IntFunction<Object> resultExtractor = getResultExtractor(valueBlock);
-    double defaultValue = 0;
-    if (_defaultValue != null) {
-      if (_defaultValue instanceof Number) {
-        defaultValue = ((Number) _defaultValue).doubleValue();
-      } else {
-        defaultValue = Double.parseDouble(_defaultValue.toString());
-      }
-    }
+    double defaultValue = _defaultValue != null ? (Double) _defaultValue : 0d;
     int numDocs = valueBlock.getNumDocs();
     for (int i = 0; i < numDocs; i++) {
       Object result = null;
@@ -273,27 +330,19 @@ public double[] transformToDoubleValuesSV(ValueBlock valueBlock) {
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
-      if (result instanceof Number) {
-        _doubleValuesSV[i] = ((Number) result).doubleValue();
-      } else {
-        _doubleValuesSV[i] = Double.parseDouble(result.toString());
-      }
+      _doubleValuesSV[i] = toDouble(result);
     }
     return _doubleValuesSV;
   }
 
   @Override
   public BigDecimal[] transformToBigDecimalValuesSV(ValueBlock valueBlock) {
-    initBigDecimalValuesSV(valueBlock.getNumDocs());
-    IntFunction<Object> resultExtractor = getResultExtractor(valueBlock, JSON_PARSER_CONTEXT_WITH_BIG_DECIMAL);
-    BigDecimal defaultValue = null;
-    if (_defaultValue != null) {
-      if (_defaultValue instanceof BigDecimal) {
-        defaultValue = (BigDecimal) _defaultValue;
-      } else {
-        defaultValue = new BigDecimal(_defaultValue.toString());
-      }
+    if (_storedType != DataType.BIG_DECIMAL) {
+      return super.transformToBigDecimalValuesSV(valueBlock);
     }
+    initBigDecimalValuesSV(valueBlock.getNumDocs());
+    IntFunction<Object> resultExtractor = getResultExtractorWithBigDecimal(valueBlock);
+    BigDecimal defaultValue = (BigDecimal) _defaultValue;
     int numDocs = valueBlock.getNumDocs();
     for (int i = 0; i < numDocs; i++) {
       Object result = null;
@@ -309,23 +358,19 @@ public BigDecimal[] transformToBigDecimalValuesSV(ValueBlock valueBlock) {
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
-      if (result instanceof BigDecimal) {
-        _bigDecimalValuesSV[i] = (BigDecimal) result;
-      } else {
-        _bigDecimalValuesSV[i] = new BigDecimal(result.toString());
-      }
+      _bigDecimalValuesSV[i] = toBigDecimal(result);
     }
     return _bigDecimalValuesSV;
   }
 
   @Override
   public String[] transformToStringValuesSV(ValueBlock valueBlock) {
-    initStringValuesSV(valueBlock.getNumDocs());
-    IntFunction<Object> resultExtractor = getResultExtractor(valueBlock, JSON_PARSER_CONTEXT_WITH_BIG_DECIMAL);
-    String defaultValue = null;
-    if (_defaultValue != null) {
-      defaultValue = _defaultValue.toString();
+    if (_storedType != DataType.STRING) {
+      return super.transformToStringValuesSV(valueBlock);
     }
+    initStringValuesSV(valueBlock.getNumDocs());
+    IntFunction<Object> resultExtractor = getResultExtractorWithBigDecimal(valueBlock);
+    String defaultValue = (String) _defaultValue;
     int numDocs = valueBlock.getNumDocs();
     for (int i = 0; i < numDocs; i++) {
       Object result = null;
@@ -341,22 +386,23 @@ public String[] transformToStringValuesSV(ValueBlock valueBlock) {
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
-      if (result instanceof String) {
-        _stringValuesSV[i] = (String) result;
-      } else {
-        _stringValuesSV[i] = JsonUtils.objectToJsonNode(result).toString();
-      }
+      _stringValuesSV[i] = toString(result);
     }
     return _stringValuesSV;
   }
 
   @Override
   public int[][] transformToIntValuesMV(ValueBlock valueBlock) {
+    if (_storedType != DataType.INT) {
+      return super.transformToIntValuesMV(valueBlock);
+    }
     initIntValuesMV(valueBlock.getNumDocs());
-    IntFunction<List<Integer>> resultExtractor = getResultExtractor(valueBlock);
+    IntFunction<List<Object>> resultExtractor = getResultExtractor(valueBlock);
+    int defaultValue = _defaultValue != null ? (Integer) _defaultValue : 0;
+    boolean isBoolean = _dataType == DataType.BOOLEAN;
     int numDocs = valueBlock.getNumDocs();
     for (int i = 0; i < numDocs; i++) {
-      List<Integer> result = null;
+      List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
       } catch (Exception ignored) {
@@ -368,7 +414,17 @@ public int[][] transformToIntValuesMV(ValueBlock valueBlock) {
       int numValues = result.size();
       int[] values = new int[numValues];
       for (int j = 0; j < numValues; j++) {
-        values[j] = result.get(j);
+        Object element = result.get(j);
+        if (element == null) {
+          if (_defaultValue != null) {
+            values[j] = defaultValue;
+            continue;
+          }
+          throw new IllegalArgumentException(
+              "At least one of the resolved JSON arrays include nulls, which is not supported in Pinot. "
+                  + "Consider setting a default value as the fourth argument of json_extract_scalar.");
+        }
+        values[j] = toInt(element, isBoolean);
       }
       _intValuesMV[i] = values;
     }
@@ -377,11 +433,16 @@ public int[][] transformToIntValuesMV(ValueBlock valueBlock) {
 
   @Override
   public long[][] transformToLongValuesMV(ValueBlock valueBlock) {
+    if (_storedType != DataType.LONG) {
+      return super.transformToLongValuesMV(valueBlock);
+    }
     initLongValuesMV(valueBlock.getNumDocs());
-    IntFunction<List<Long>> resultExtractor = getResultExtractor(valueBlock);
-    int length = valueBlock.getNumDocs();
-    for (int i = 0; i < length; i++) {
-      List<Long> result = null;
+    IntFunction<List<Object>> resultExtractor = getResultExtractor(valueBlock);
+    long defaultValue = _defaultValue != null ? (Long) _defaultValue : 0L;
+    boolean isTimestamp = _dataType == DataType.TIMESTAMP;
+    int numDocs = valueBlock.getNumDocs();
+    for (int i = 0; i < numDocs; i++) {
+      List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
       } catch (Exception ignored) {
@@ -393,7 +454,17 @@ public long[][] transformToLongValuesMV(ValueBlock valueBlock) {
       int numValues = result.size();
       long[] values = new long[numValues];
       for (int j = 0; j < numValues; j++) {
-        values[j] = result.get(j);
+        Object element = result.get(j);
+        if (element == null) {
+          if (_defaultValue != null) {
+            values[j] = defaultValue;
+            continue;
+          }
+          throw new IllegalArgumentException(
+              "At least one of the resolved JSON arrays include nulls, which is not supported in Pinot. "
+                  + "Consider setting a default value as the fourth argument of json_extract_scalar.");
+        }
+        values[j] = toLong(element, isTimestamp);
       }
       _longValuesMV[i] = values;
     }
@@ -402,11 +473,15 @@ public long[][] transformToLongValuesMV(ValueBlock valueBlock) {
 
   @Override
   public float[][] transformToFloatValuesMV(ValueBlock valueBlock) {
+    if (_storedType != DataType.FLOAT) {
+      return super.transformToFloatValuesMV(valueBlock);
+    }
     initFloatValuesMV(valueBlock.getNumDocs());
-    IntFunction<List<Float>> resultExtractor = getResultExtractor(valueBlock);
-    int length = valueBlock.getNumDocs();
-    for (int i = 0; i < length; i++) {
-      List<Float> result = null;
+    IntFunction<List<Object>> resultExtractor = getResultExtractor(valueBlock);
+    float defaultValue = _defaultValue != null ? (Float) _defaultValue : 0f;
+    int numDocs = valueBlock.getNumDocs();
+    for (int i = 0; i < numDocs; i++) {
+      List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
       } catch (Exception ignored) {
@@ -418,7 +493,17 @@ public float[][] transformToFloatValuesMV(ValueBlock valueBlock) {
       int numValues = result.size();
       float[] values = new float[numValues];
       for (int j = 0; j < numValues; j++) {
-        values[j] = result.get(j);
+        Object element = result.get(j);
+        if (element == null) {
+          if (_defaultValue != null) {
+            values[j] = defaultValue;
+            continue;
+          }
+          throw new IllegalArgumentException(
+              "At least one of the resolved JSON arrays include nulls, which is not supported in Pinot. "
+                  + "Consider setting a default value as the fourth argument of json_extract_scalar.");
+        }
+        values[j] = toFloat(element);
       }
       _floatValuesMV[i] = values;
     }
@@ -427,11 +512,15 @@ public float[][] transformToFloatValuesMV(ValueBlock valueBlock) {
 
   @Override
   public double[][] transformToDoubleValuesMV(ValueBlock valueBlock) {
+    if (_storedType != DataType.DOUBLE) {
+      return super.transformToDoubleValuesMV(valueBlock);
+    }
     initDoubleValuesMV(valueBlock.getNumDocs());
-    IntFunction<List<Double>> resultExtractor = getResultExtractor(valueBlock);
-    int length = valueBlock.getNumDocs();
-    for (int i = 0; i < length; i++) {
-      List<Double> result = null;
+    IntFunction<List<Object>> resultExtractor = getResultExtractor(valueBlock);
+    double defaultValue = _defaultValue != null ? (Double) _defaultValue : 0d;
+    int numDocs = valueBlock.getNumDocs();
+    for (int i = 0; i < numDocs; i++) {
+      List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
       } catch (Exception ignored) {
@@ -443,7 +532,17 @@ public double[][] transformToDoubleValuesMV(ValueBlock valueBlock) {
       int numValues = result.size();
       double[] values = new double[numValues];
       for (int j = 0; j < numValues; j++) {
-        values[j] = result.get(j);
+        Object element = result.get(j);
+        if (element == null) {
+          if (_defaultValue != null) {
+            values[j] = defaultValue;
+            continue;
+          }
+          throw new IllegalArgumentException(
+              "At least one of the resolved JSON arrays include nulls, which is not supported in Pinot. "
+                  + "Consider setting a default value as the fourth argument of json_extract_scalar.");
+        }
+        values[j] = toDouble(element);
       }
       _doubleValuesMV[i] = values;
     }
@@ -452,11 +551,15 @@ public double[][] transformToDoubleValuesMV(ValueBlock valueBlock) {
 
   @Override
   public String[][] transformToStringValuesMV(ValueBlock valueBlock) {
+    if (_storedType != DataType.STRING) {
+      return super.transformToStringValuesMV(valueBlock);
+    }
     initStringValuesMV(valueBlock.getNumDocs());
-    IntFunction<List<String>> resultExtractor = getResultExtractor(valueBlock);
-    int length = valueBlock.getNumDocs();
-    for (int i = 0; i < length; i++) {
-      List<String> result = null;
+    IntFunction<List<Object>> resultExtractor = getResultExtractorWithBigDecimal(valueBlock);
+    String defaultValue = (String) _defaultValue;
+    int numDocs = valueBlock.getNumDocs();
+    for (int i = 0; i < numDocs; i++) {
+      List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
       } catch (Exception ignored) {
@@ -468,13 +571,102 @@ public String[][] transformToStringValuesMV(ValueBlock valueBlock) {
       int numValues = result.size();
       String[] values = new String[numValues];
       for (int j = 0; j < numValues; j++) {
-        values[j] = result.get(j);
+        Object element = result.get(j);
+        if (element == null) {
+          if (_defaultValue != null) {
+            values[j] = defaultValue;
+            continue;
+          }
+          throw new IllegalArgumentException(
+              "At least one of the resolved JSON arrays include nulls, which is not supported in Pinot. "
+                  + "Consider setting a default value as the fourth argument of json_extract_scalar.");
+        }
+        values[j] = toString(element);
       }
       _stringValuesMV[i] = values;
     }
     return _stringValuesMV;
   }
 
+  private static int toInt(Object value, boolean isBoolean) {
+    if (isBoolean) {
+      if (value instanceof Boolean) {
+        return (Boolean) value ? 1 : 0;
+      }
+      // For BOOLEAN result, follow PinotDataType numeric convention: non-zero number → true.
+      if (value instanceof Number) {
+        return ((Number) value).doubleValue() != 0 ? 1 : 0;
+      }
+      // String fallback: BooleanUtils.toInt accepts "true" / "TRUE" / "1".
+      return BooleanUtils.toInt(value.toString());
+    }
+    if (value instanceof Number) {
+      return ((Number) value).intValue();
+    }
+    if (value instanceof Boolean) {
+      return (Boolean) value ? 1 : 0;
+    }
+    return Integer.parseInt(value.toString());
+  }
+
+  private static long toLong(Object value, boolean isTimestamp) {
+    if (value instanceof Number) {
+      return ((Number) value).longValue();
+    }
+    if (isTimestamp) {
+      return TimestampUtils.toMillisSinceEpoch(value.toString());
+    }
+    if (value instanceof Boolean) {
+      return (Boolean) value ? 1L : 0L;
+    }
+    try {
+      return NumberUtils.parseJsonLong(value.toString());
+    } catch (NumericException nfe) {
+      throw new NumberFormatException("For input string: \"" + value + "\"");
+    }
+  }
+
+  private static float toFloat(Object value) {
+    if (value instanceof Number) {
+      return ((Number) value).floatValue();
+    }
+    if (value instanceof Boolean) {
+      return (Boolean) value ? 1f : 0f;
+    }
+    return Float.parseFloat(value.toString());
+  }
+
+  private static double toDouble(Object value) {
+    if (value instanceof Number) {
+      return ((Number) value).doubleValue();
+    }
+    if (value instanceof Boolean) {
+      return (Boolean) value ? 1d : 0d;
+    }
+    return Double.parseDouble(value.toString());
+  }
+
+  private static BigDecimal toBigDecimal(Object value) {
+    if (value instanceof BigDecimal) {
+      return (BigDecimal) value;
+    }
+    if (value instanceof Boolean) {
+      return (Boolean) value ? BigDecimal.ONE : BigDecimal.ZERO;
+    }
+    return new BigDecimal(value.toString());
+  }
+
+  private static String toString(Object value) {
+    if (value instanceof String) {
+      return (String) value;
+    }
+    try {
+      return JsonUtils.objectToString(value);
+    } catch (JsonProcessingException e) {
+      throw new RuntimeException("Caught exception while serializing JSON value: " + value, e);
+    }
+  }
+
   private <T> IntFunction<T> getResultExtractor(ValueBlock valueBlock, ParseContext parseContext) {
     if (_jsonFieldTransformFunction.getResultMetadata().getDataType() == DataType.BYTES) {
       byte[][] jsonBytes = _jsonFieldTransformFunction.transformToBytesValuesSV(valueBlock);
@@ -488,4 +680,8 @@ private <T> IntFunction<T> getResultExtractor(ValueBlock valueBlock, ParseContex
   private <T> IntFunction<T> getResultExtractor(ValueBlock valueBlock) {
     return getResultExtractor(valueBlock, JSON_PARSER_CONTEXT);
   }
+
+  private <T> IntFunction<T> getResultExtractorWithBigDecimal(ValueBlock valueBlock) {
+    return getResultExtractor(valueBlock, JSON_PARSER_CONTEXT_WITH_BIG_DECIMAL);
+  }
 }
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java
index 57a8ae04d9..586880ae1f 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java
@@ -18,26 +18,40 @@
  */
 package org.apache.pinot.core.operator.transform.function;
 
+import java.io.File;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.math.BigDecimal;
+import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Random;
+import org.apache.commons.io.FileUtils;
 import org.apache.pinot.common.request.context.ExpressionContext;
 import org.apache.pinot.common.request.context.RequestContextUtils;
+import org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
 import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
 import org.apache.pinot.spi.exception.BadQueryRequestException;
 import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
 import org.apache.pinot.sql.parsers.SqlCompilationException;
+import org.assertj.core.api.Assertions;
 import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 
 public class JsonExtractScalarTransformFunctionTest extends BaseTransformFunctionTest {
 
+  protected File _baseDir;
+
   @Test(dataProvider = "testJsonPathTransformFunction")
   public void testJsonPathTransformFunction(String expressionStr, DataType resultsDataType, boolean isSingleValue) {
     ExpressionContext expression = RequestContextUtils.getExpression(expressionStr);
@@ -467,4 +481,426 @@ public Object[][] testParsingIllegalQueries() {
         new Object[]{String.format("json_extract_key(%s, \"$.*\")", JSON_COLUMN)}};
     //@formatter:on
   }
+
+  @BeforeClass
+  void createBaseDir() {
+    try {
+      _baseDir = Files.createTempDirectory(getClass().getSimpleName()).toFile();
+    } catch (IOException ex) {
+      throw new UncheckedIOException(ex);
+    }
+  }
+
+  @AfterClass
+  void destroyBaseDir()
+      throws IOException {
+    if (_baseDir != null) {
+      FileUtils.deleteDirectory(_baseDir);
+    }
+  }
+
+  @DataProvider
+  public static Object[][] multiValuesAndDefaults() {
+    return new Object[][]{
+        {"INT_ARRAY", "-123"},
+        {"LONG_ARRAY", "-123"},
+        {"FLOAT_ARRAY", "0.5"},
+        {"DOUBLE_ARRAY", "0.5"},
+        {"STRING_ARRAY", "'default'"}
+    };
+  }
+
+  /**
+   *
+   * @param resultsType The result type used in the query, either 'INT_ARRAY' or 'STRING_ARRAY'
+   * @param notUsed Not used, just here to be able to use the same provider as
+   *                {@link #mvWithNullsWithDefault(String, String)}
+   */
+  @Test(dataProvider = "multiValuesAndDefaults")
+  public void mvWithNullsWithoutDefault(String resultsType, String notUsed) {
+    // schema with a single column called "json" of type JSON
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    // trivial table config
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+    try {
+      FluentQueryTest.withBaseDir(_baseDir)
+          .withNullHandling(false)
+          .givenTable(schema, tableConfig)
+          .onFirstInstance(new Object[]{"{\"name\": [null]}"})
+          .whenQuery("SELECT jsonExtractScalar(json, '$.name', '" + resultsType + "') FROM testTable")
+          // TODO: Change the framework to do not duplicate segments when only one segment is used
+          .thenResultIs(new Object[]{"doesn't matter, it should fail"});
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage())
+          .describedAs("Expected to fail because the JSON array contains nulls")
+          .contains("At least one of the resolved JSON arrays include nulls");
+    }
+  }
+
+  /**
+   *
+   * @param resultsType The result type used in the query, either 'INT_ARRAY' or 'STRING_ARRAY'
+   * @param defaultValSql The default value to use in the query, as passed to jsonExtractScalar forth argument
+   */
+  @Test(dataProvider = "multiValuesAndDefaults")
+  public void mvWithNullsWithDefault(String resultsType, String defaultValSql) {
+    // schema with a single column called "json" of type JSON
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    // trivial table config
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+
+    Object[] expectedRow;
+    switch (resultsType) {
+      case "INT_ARRAY": {
+        int[] rowValue = new int[]{Integer.parseInt(defaultValSql)};
+        expectedRow = new Object[]{rowValue};
+        break;
+      }
+      case "LONG_ARRAY": {
+        long[] rowValue = new long[]{Long.parseLong(defaultValSql)};
+        expectedRow = new Object[]{rowValue};
+        break;
+      }
+      case "FLOAT_ARRAY": {
+        float[] rowValue = new float[]{Float.parseFloat(defaultValSql)};
+        expectedRow = new Object[]{rowValue};
+        break;
+      }
+      case "DOUBLE_ARRAY": {
+        double[] rowValue = new double[]{Double.parseDouble(defaultValSql)};
+        expectedRow = new Object[]{rowValue};
+        break;
+      }
+      case "STRING_ARRAY": {
+        // defaultValSql is quoted, need to remove quotes
+        String str = defaultValSql.substring(1, defaultValSql.length() - 1);
+        String[] rowValue = new String[]{str};
+        expectedRow = new Object[]{rowValue};
+        break;
+      }
+      default:
+        throw new UnsupportedOperationException("Not support data type - " + resultsType);
+    }
+
+    FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(false)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(new Object[]{"{\"name\": [null]}"})
+        .whenQuery("SELECT jsonExtractScalar(json, '$.name', '" + resultsType + "', " + defaultValSql + ") "
+            + "FROM testTable")
+        // TODO: Change the framework to do not duplicate segments when only one segment is used
+        .thenResultIs(expectedRow, expectedRow); // 2 rows because of segment duplication
+  }
+
+  // === Per-stored-type coercion tests for the value-handling helpers. Tests are grouped first by SV
+  //     vs MV result type, then ordered within each group by canonical Pinot type order
+  //     (INT/LONG/FLOAT/DOUBLE → BIG_DECIMAL → BOOLEAN → TIMESTAMP → STRING). ===
+
+  // -- Single-value (SV) tests --
+
+  /// Runs `SELECT jsonExtractScalar(json, '$.v', resultsType) FROM testTable` against a single-row table
+  /// containing the given JSON document, and asserts the result for the (always-duplicated) two
+  /// expected rows.
+  private void assertJsonExtractScalar(String json, String resultsType, Object expectedValue) {
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+    Object[] expectedRow = new Object[]{expectedValue};
+    FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(false)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(new Object[]{json})
+        .whenQuery("SELECT jsonExtractScalar(json, '$.v', '" + resultsType + "') FROM testTable")
+        .thenResultIs(expectedRow, expectedRow);
+  }
+
+  @Test
+  public void testExtractBooleanAsNumeric() {
+    // JSON true / false coerces to 1 / 0 across all numeric result types — matches PinotDataType's
+    // BOOLEAN.toInt / toLong / toFloat / toDouble / toBigDecimal convention.
+    assertJsonExtractScalar("{\"v\": true}", "INT", 1);
+    assertJsonExtractScalar("{\"v\": false}", "INT", 0);
+    assertJsonExtractScalar("{\"v\": true}", "LONG", 1L);
+    assertJsonExtractScalar("{\"v\": false}", "LONG", 0L);
+    assertJsonExtractScalar("{\"v\": true}", "FLOAT", 1f);
+    assertJsonExtractScalar("{\"v\": false}", "FLOAT", 0f);
+    assertJsonExtractScalar("{\"v\": true}", "DOUBLE", 1d);
+    assertJsonExtractScalar("{\"v\": false}", "DOUBLE", 0d);
+    // BIG_DECIMAL formatted as String via toPlainString.
+    assertJsonExtractScalar("{\"v\": true}", "BIG_DECIMAL", "1");
+    assertJsonExtractScalar("{\"v\": false}", "BIG_DECIMAL", "0");
+  }
+
+  @Test
+  public void testExtractBigDecimalPreservesPrecision() {
+    // The BIG_DECIMAL parser preserves full numeric precision — beyond what Double can represent.
+    // Broker formats BIG_DECIMAL as String via BigDecimal.toPlainString().
+    assertJsonExtractScalar("{\"v\": 12345678901234567890.123456789}", "BIG_DECIMAL",
+        "12345678901234567890.123456789");
+  }
+
+  @DataProvider(name = "booleanCoercion")
+  public Object[][] booleanCoercion() {
+    return new Object[][]{
+        // JSON boolean — direct map.
+        {"true", 1},
+        {"false", 0},
+        // JSON number — Pinot's numeric BOOLEAN convention: any non-zero → 1.
+        {"1", 1},
+        {"0", 0},
+        {"5", 1},
+        {"-1", 1},
+        {"0.5", 1},
+        {"0.0", 0},
+        // JSON string — only "true" / "TRUE" / "1" → 1; anything else (including "yes", "TRUE ") → 0.
+        {"\"true\"", 1},
+        {"\"TRUE\"", 1},
+        {"\"True\"", 1},
+        {"\"1\"", 1},
+        {"\"false\"", 0},
+        {"\"0\"", 0},
+        {"\"yes\"", 0},
+        {"\"\"", 0}
+    };
+  }
+
+  @Test(dataProvider = "booleanCoercion")
+  public void testExtractBoolean(String jsonValueLiteral, int expected) {
+    // The SELECT projection for a BOOLEAN result surfaces as Boolean true / false in the broker rows.
+    Object expectedBoolean = expected == 1;
+    assertJsonExtractScalar("{\"v\": " + jsonValueLiteral + "}", "BOOLEAN", expectedBoolean);
+  }
+
+  @DataProvider(name = "timestampCoercion")
+  public Object[][] timestampCoercion() {
+    long epochMillis = 1700000000000L;
+    return new Object[][]{
+        // Numeric epoch millis — straight longValue path.
+        {String.valueOf(epochMillis), epochMillis},
+        // Numeric epoch millis as string — TimestampUtils accepts numeric strings.
+        {"\"" + epochMillis + "\"", epochMillis},
+        // ISO-8601 string — TimestampUtils parses to epoch millis.
+        {"\"2023-11-14T22:13:20Z\"", epochMillis},
+        {"\"2023-11-14 22:13:20\"", epochMillis - java.util.TimeZone.getDefault().getOffset(epochMillis)}
+    };
+  }
+
+  @Test(dataProvider = "timestampCoercion")
+  public void testExtractTimestamp(String jsonValueLiteral, long expectedMillis) {
+    // Broker formats TIMESTAMP via Timestamp.toString() (local-TZ wall-clock representation).
+    assertJsonExtractScalar("{\"v\": " + jsonValueLiteral + "}", "TIMESTAMP",
+        new java.sql.Timestamp(expectedMillis).toString());
+  }
+
+  @Test
+  public void testExtractTimestampRejectsBoolean() {
+    // TIMESTAMP doesn't accept Boolean — Boolean → epoch millis is semantically nonsensical, matching
+    // PinotDataType.TIMESTAMP.toBoolean throwing UnsupportedOperationException. JSON true is routed
+    // through TimestampUtils.toMillisSinceEpoch("true") which throws IllegalArgumentException; the
+    // broker surfaces the failure as a query error.
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+    try {
+      FluentQueryTest.withBaseDir(_baseDir)
+          .withNullHandling(false)
+          .givenTable(schema, tableConfig)
+          .onFirstInstance(new Object[]{"{\"v\": true}"})
+          .whenQuery("SELECT jsonExtractScalar(json, '$.v', 'TIMESTAMP') FROM testTable")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected query to fail when extracting JSON boolean as TIMESTAMP");
+    } catch (AssertionError e) {
+      // Expected — broker surfaces the parse failure as a query error.
+    }
+  }
+
+  @Test
+  public void testExtractStringFromNonStringJson() {
+    // A String JSON value passes through as-is.
+    assertJsonExtractScalar("{\"v\": \"hello\"}", "STRING", "hello");
+    // Numbers, booleans, arrays, and objects are JSON-serialized via JsonUtils.objectToString.
+    assertJsonExtractScalar("{\"v\": 42}", "STRING", "42");
+    assertJsonExtractScalar("{\"v\": 3.14}", "STRING", "3.14");
+    assertJsonExtractScalar("{\"v\": true}", "STRING", "true");
+    assertJsonExtractScalar("{\"v\": [1,2,3]}", "STRING", "[1,2,3]");
+    assertJsonExtractScalar("{\"v\": {\"a\":1}}", "STRING", "{\"a\":1}");
+  }
+
+  @Test
+  public void testExtractStringPreservesNumericPrecision() {
+    // STRING uses JSON_PARSER_CONTEXT_WITH_BIG_DECIMAL so floats that exceed Double precision survive
+    // the JSON → Java → JSON round-trip without truncation. Symmetric with BIG_DECIMAL precision.
+    assertJsonExtractScalar(
+        "{\"v\": 12345678901234567890.123456789}", "STRING",
+        "12345678901234567890.123456789");
+  }
+
+  // -- Multi-value (MV) tests --
+
+  /// Asserts that `SELECT jsonExtractScalar(json, '$.v', resultsType)` over a single-row table with the
+  /// given JSON document produces the given primitive-array result.
+  private void assertJsonExtractMv(String json, String resultsType, Object expectedArray) {
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+    Object[] expectedRow = new Object[]{expectedArray};
+    FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(false)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(new Object[]{json})
+        .whenQuery("SELECT jsonExtractScalar(json, '$.v', '" + resultsType + "') FROM testTable")
+        .thenResultIs(expectedRow, expectedRow);
+  }
+
+  @Test
+  public void testIntArrayFromHeterogeneousJsonElements() {
+    // JSON numbers — direct intValue path.
+    assertJsonExtractMv("{\"v\": [1, 2, 3]}", "INT_ARRAY", new int[]{1, 2, 3});
+    // JSON string-form numbers — Integer.parseInt(toString()) fallback.
+    assertJsonExtractMv("{\"v\": [\"1\", \"2\", \"3\"]}", "INT_ARRAY", new int[]{1, 2, 3});
+    // Mixed numeric and string-numeric — both forms coerce.
+    assertJsonExtractMv("{\"v\": [1, \"2\", 3]}", "INT_ARRAY", new int[]{1, 2, 3});
+  }
+
+  @Test
+  public void testNumericArrayWithBooleanElements() {
+    // JSON true / false elements coerce to 1 / 0 in numeric MV paths.
+    assertJsonExtractMv("{\"v\": [true, false, true]}", "INT_ARRAY", new int[]{1, 0, 1});
+    assertJsonExtractMv("{\"v\": [true, false]}", "LONG_ARRAY", new long[]{1L, 0L});
+    assertJsonExtractMv("{\"v\": [true, false]}", "FLOAT_ARRAY", new float[]{1f, 0f});
+    assertJsonExtractMv("{\"v\": [true, false]}", "DOUBLE_ARRAY", new double[]{1d, 0d});
+    // Mixed numeric / boolean elements — Boolean coerces to 1/0, Number values pass through.
+    assertJsonExtractMv("{\"v\": [1, true, 3, false]}", "INT_ARRAY", new int[]{1, 1, 3, 0});
+  }
+
+  @Test
+  public void testBooleanArrayResultType() {
+    // BOOLEAN_ARRAY uses storedType INT and the BOOLEAN per-element convention (non-zero Number → 1),
+    // distinct from INT_ARRAY which uses intValue() directly. JSON [1, 5, 0, 2.5] therefore yields
+    // [true, true, false, true] for BOOLEAN_ARRAY but [1, 5, 0, 2] for INT_ARRAY.
+    assertJsonExtractMv("{\"v\": [1, 5, 0, 2.5]}", "BOOLEAN_ARRAY",
+        new boolean[]{true, true, false, true});
+    // Mixed Boolean / Number / String — each element coerced via PinotDataType BOOLEAN convention.
+    assertJsonExtractMv("{\"v\": [true, 0, \"true\", false]}", "BOOLEAN_ARRAY",
+        new boolean[]{true, false, true, false});
+  }
+
+  @Test
+  public void testTimestampArrayResultType() {
+    // TIMESTAMP_ARRAY: storedType LONG, transformToLongValuesMV with isTimestamp=true. JSON numeric
+    // millis pass through Number.longValue(); JSON ISO strings parse via TimestampUtils.
+    long epochMillis = 1700000000000L;
+    String tsString = new java.sql.Timestamp(epochMillis).toString();
+    // Broker formats TIMESTAMP_ARRAY as String[] (per-element Timestamp.toString()).
+    assertJsonExtractMv(
+        "{\"v\": [" + epochMillis + ", \"2023-11-14T22:13:20Z\"]}",
+        "TIMESTAMP_ARRAY",
+        new String[]{tsString, tsString});
+  }
+
+  @Test
+  public void testStringArrayFromNonStringJsonElements() {
+    // JSON strings pass through.
+    assertJsonExtractMv("{\"v\": [\"a\", \"b\"]}", "STRING_ARRAY", new String[]{"a", "b"});
+    // JSON numbers, booleans, and nested structures get JSON-serialized.
+    assertJsonExtractMv("{\"v\": [1, 2, 3]}", "STRING_ARRAY", new String[]{"1", "2", "3"});
+    assertJsonExtractMv("{\"v\": [true, false]}", "STRING_ARRAY", new String[]{"true", "false"});
+    assertJsonExtractMv("{\"v\": [[1, 2], [3, 4]]}", "STRING_ARRAY", new String[]{"[1,2]", "[3,4]"});
+  }
+
+  // -- Default values for newly-handled result types --
+
+  @Test
+  public void testDefaultValueForBoolean() {
+    // BOOLEAN default literal stored as Integer 0/1 in init() to match INT storedType. When the JSON
+    // path doesn't resolve, the default surfaces correctly through transformToIntValuesSV.
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+    FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(false)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(new Object[]{"{\"other\": 1}"})
+        .whenQuery("SELECT jsonExtractScalar(json, '$.missing', 'BOOLEAN', true) FROM testTable")
+        .thenResultIs(new Object[]{true}, new Object[]{true});
+  }
+
+  @Test
+  public void testDefaultValueForTimestamp() {
+    // TIMESTAMP default literal stored as Long in init() to match LONG storedType. When the JSON path
+    // doesn't resolve, the default surfaces through transformToLongValuesSV.
+    long defaultMillis = 1234567890000L;
+    String expected = new java.sql.Timestamp(defaultMillis).toString();
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+    FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(false)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(new Object[]{"{\"other\": 1}"})
+        .whenQuery("SELECT jsonExtractScalar(json, '$.missing', 'TIMESTAMP', " + defaultMillis + ") "
+            + "FROM testTable")
+        .thenResultIs(new Object[]{expected}, new Object[]{expected});
+  }
+
+  // -- Cross-type guard: requesting a type other than the function's declared result type should
+  //    route through the base class's cross-type conversion path. --
+
+  @Test
+  public void testCrossTypeConversionFromStringResult() {
+    // The function's declared result type is STRING, but the caller requests INT — base class should
+    // handle the STRING→INT conversion via parsing.
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("testTable")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("json", DataType.JSON)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("testTable")
+        .build();
+    FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(false)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(new Object[]{"{\"v\": \"42\"}"})
+        // Cast STRING-result to LONG triggers the base-class cross-type path: STRING → parseLong.
+        .whenQuery("SELECT CAST(jsonExtractScalar(json, '$.v', 'STRING') AS LONG) FROM testTable")
+        .thenResultIs(new Object[]{42L}, new Object[]{42L});
+  }
 }
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/BaseJsonQueryTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/BaseJsonQueryTest.java
index 6ef6e89d66..d089fe1d59 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/BaseJsonQueryTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/BaseJsonQueryTest.java
@@ -34,6 +34,7 @@
 import org.apache.pinot.spi.config.table.TableConfig;
 import org.apache.pinot.spi.data.Schema;
 import org.apache.pinot.spi.data.readers.GenericRow;
+import org.intellij.lang.annotations.Language;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.DataProvider;
 
@@ -145,6 +146,7 @@ public void setUp()
     records.add(createRecord(16, 16, "john doe", "{\"longVal\": \"-9223372036854775808\" }"));
     records.add(createRecord(17, 17, "john doe", "{\"longVal\": \"-100.12345\" }"));
     records.add(createRecord(18, 18, "john doe", "{\"longVal\": \"10e2\" }"));
+    records.add(createRecord(19, 19, "john doe", "{\"longVal\": null }"));
 
     tableConfig.getIndexingConfig().setJsonIndexColumns(List.of("jsonColumn"));
     SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(tableConfig, schema);
@@ -178,7 +180,7 @@ protected List<IndexSegment> getIndexSegments() {
     return _indexSegments;
   }
 
-  protected void checkResult(String query, Object[][] expectedResults) {
+  protected void checkResult(@Language("sql") String query, Object[][] expectedResults) {
     BrokerResponseNative brokerResponse = getBrokerResponseForOptimizedQuery(query, tableConfig(), schema());
     QueriesTestUtils.testInterSegmentsResult(brokerResponse, Arrays.asList(expectedResults));
   }
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/JsonExtractScalarTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/JsonExtractScalarTest.java
index 8af1349096..ba33ed824c 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/JsonExtractScalarTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/JsonExtractScalarTest.java
@@ -171,4 +171,42 @@ public void testExtractJsonLongValue(String column) {
             + "limit 4",
         new Object[][]{{15, Long.MAX_VALUE}, {16, Long.MIN_VALUE}, {17, -100L}, {18, 1000L}});
   }
+
+  @Test
+  public void testNullAsDefaultValueWithNullHandlingDisabled() {
+    checkResult(
+        "SET enableNullHandling=false;"
+            + "SELECT intColumn, jsonextractscalar(" + JSON_COLUMN + ", '$.longVal', 'long', null) "
+            + "FROM testTable "
+            + "where intColumn >= 15 and intColumn <= 19 "
+            + "group by 1, 2 "
+            + "order by 1, 2 ",
+        new Object[][]{
+            {15, Long.MAX_VALUE},
+            {16, Long.MIN_VALUE},
+            {17, -100L},
+            {18, 1000L},
+            {19, 0L} // when enableNullHandling is false, null is treated as 0 for LONG type
+        }
+    );
+  }
+
+  @Test
+  public void testNullAsDefaultValueWithNullHandlingEnabled() {
+    checkResult(
+        "SET enableNullHandling=true;"
+            + "SELECT intColumn, jsonextractscalar(" + JSON_COLUMN + ", '$.longVal', 'long', null) "
+            + "FROM testTable "
+            + "where intColumn >= 15 and intColumn <= 19 "
+            + "group by 1, 2 "
+            + "order by 1, 2 ",
+        new Object[][]{
+            {15, Long.MAX_VALUE},
+            {16, Long.MIN_VALUE},
+            {17, -100L},
+            {18, 1000L},
+            {19, null} // when enableNullHandling is true, null is treated as null
+        }
+    );
+  }
 }