diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunction.java
index 97b34e05e2..2386750ae8 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunction.java
@@ -24,14 +24,17 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+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.segment.spi.index.IndexService;
 import org.apache.pinot.segment.spi.index.IndexType;
 import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
 import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.CommonConstants.NullValuePlaceHolder;
 import org.apache.pinot.spi.utils.JsonUtils;
 import org.roaringbitmap.RoaringBitmap;
 
@@ -41,6 +44,12 @@
  * implementation changed to read values from the JSON index. For large JSON blobs this can be faster than parsing
  * GBs of JSON at query time. For small JSON blobs/highly filtered input this is generally slower than the *scalar
  * implementation. The inflection point is highly dependent on the number of docs remaining post filter.
+ *
+ * <p><b>Null handling:</b> when query-level null handling is enabled (e.g. {@code SET enableNullHandling = true})
+ * and no default literal is supplied, an unresolved JSON path on a row surfaces as SQL {@code NULL} via the
+ * {@link #getNullBitmap(ValueBlock)} bitmap rather than throwing. With null handling disabled, the legacy throw
+ * is preserved. A user-supplied non-null default literal takes precedence and is written for unresolved rows
+ * regardless of null-handling state.
  */
 public class JsonExtractIndexTransformFunction extends BaseTransformFunction {
   public static final String FUNCTION_NAME = "jsonExtractIndex";
@@ -54,14 +63,22 @@ public class JsonExtractIndexTransformFunction extends BaseTransformFunction {
   private boolean _isSingleValue;
   private String _filterJsonPath;
 
+  // Cache for the per-ValueBlock SV scan result. Both the SV transforms and getNullBitmap need
+  // the same `String[]` for the same ValueBlock; without caching, we'd hit the JSON index twice
+  // per block on the null-handling path. Identity comparison is safe — the server constructs a
+  // fresh ValueBlock per batch and never mutates a previously returned one.
+  private ValueBlock _cachedValueBlock;
+  private String[] _cachedValuesSV;
+
   @Override
   public String getName() {
     return FUNCTION_NAME;
   }
 
   @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 or 5 arguments
     if (arguments.size() < 3 || arguments.size() > 5) {
       throw new IllegalArgumentException(
@@ -118,22 +135,27 @@ public void init(List<TransformFunction> arguments, Map<String, ColumnContext> c
       if (!(fourthArgument instanceof LiteralTransformFunction)) {
         throw new IllegalArgumentException("Default value must be a literal");
       }
-
-      if (_isSingleValue) {
-        _defaultValue = dataType.convert(((LiteralTransformFunction) fourthArgument).getStringLiteral());
-      } else {
-        try {
-          JsonNode mvArray = JsonUtils.stringToJsonNode(((LiteralTransformFunction) fourthArgument).getStringLiteral());
-          if (!mvArray.isArray()) {
+      LiteralTransformFunction literalFn = (LiteralTransformFunction) fourthArgument;
+      // SQL NULL literal as the default ⇒ behave like the 3-arg form: leave `_defaultValue` null
+      // so unresolved rows surface as SQL NULL (NH on) or throw (NH off). This mirrors the scalar
+      // function's `_defaultIsNull` handling without needing a separate flag.
+      if (!literalFn.isNull()) {
+        if (_isSingleValue) {
+          _defaultValue = dataType.convert(literalFn.getStringLiteral());
+        } else {
+          try {
+            JsonNode mvArray = JsonUtils.stringToJsonNode(literalFn.getStringLiteral());
+            if (!mvArray.isArray()) {
+              throw new IllegalArgumentException("Default value must be a valid JSON array");
+            }
+            Object[] defaultValues = new Object[mvArray.size()];
+            for (int i = 0; i < mvArray.size(); i++) {
+              defaultValues[i] = dataType.convert(mvArray.get(i).asText());
+            }
+            _defaultValue = defaultValues;
+          } catch (IOException e) {
             throw new IllegalArgumentException("Default value must be a valid JSON array");
           }
-          Object[] defaultValues = new Object[mvArray.size()];
-          for (int i = 0; i < mvArray.size(); i++) {
-            defaultValues[i] = dataType.convert(mvArray.get(i).asText());
-          }
-          _defaultValue = defaultValues;
-        } catch (IOException e) {
-          throw new IllegalArgumentException("Default value must be a valid JSON array");
         }
       }
     }
@@ -159,8 +181,7 @@ public int[] transformToIntValuesSV(ValueBlock valueBlock) {
     int numDocs = valueBlock.getNumDocs();
     int[] inputDocIds = valueBlock.getDocIds();
     initIntValuesSV(numDocs);
-    String[] valuesFromIndex = _jsonIndexReader.getValuesSV(valueBlock.getDocIds(), valueBlock.getNumDocs(),
-        getValueToMatchingDocsMap(), false);
+    String[] valuesFromIndex = getCachedValuesFromIndex(valueBlock);
     for (int i = 0; i < numDocs; i++) {
       String value = valuesFromIndex[i];
       if (value == null) {
@@ -168,6 +189,10 @@ public int[] transformToIntValuesSV(ValueBlock valueBlock) {
           _intValuesSV[i] = (int) _defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _intValuesSV[i] = NullValuePlaceHolder.INT;
+          continue;
+        }
         throw new RuntimeException(
             String.format("Illegal Json Path: [%s], for docId [%s]", _jsonPathString, inputDocIds[i]));
       }
@@ -181,8 +206,7 @@ public long[] transformToLongValuesSV(ValueBlock valueBlock) {
     int numDocs = valueBlock.getNumDocs();
     int[] inputDocIds = valueBlock.getDocIds();
     initLongValuesSV(numDocs);
-    String[] valuesFromIndex = _jsonIndexReader.getValuesSV(valueBlock.getDocIds(), valueBlock.getNumDocs(),
-        getValueToMatchingDocsMap(), false);
+    String[] valuesFromIndex = getCachedValuesFromIndex(valueBlock);
     for (int i = 0; i < numDocs; i++) {
       String value = valuesFromIndex[i];
       if (value == null) {
@@ -190,6 +214,10 @@ public long[] transformToLongValuesSV(ValueBlock valueBlock) {
           _longValuesSV[i] = (long) _defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _longValuesSV[i] = NullValuePlaceHolder.LONG;
+          continue;
+        }
         throw new RuntimeException(
             String.format("Illegal Json Path: [%s], for docId [%s]", _jsonPathString, inputDocIds[i]));
       }
@@ -203,8 +231,7 @@ public float[] transformToFloatValuesSV(ValueBlock valueBlock) {
     int numDocs = valueBlock.getNumDocs();
     int[] inputDocIds = valueBlock.getDocIds();
     initFloatValuesSV(numDocs);
-    String[] valuesFromIndex = _jsonIndexReader.getValuesSV(valueBlock.getDocIds(), valueBlock.getNumDocs(),
-        getValueToMatchingDocsMap(), false);
+    String[] valuesFromIndex = getCachedValuesFromIndex(valueBlock);
     for (int i = 0; i < numDocs; i++) {
       String value = valuesFromIndex[i];
       if (value == null) {
@@ -212,6 +239,10 @@ public float[] transformToFloatValuesSV(ValueBlock valueBlock) {
           _floatValuesSV[i] = (float) _defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _floatValuesSV[i] = NullValuePlaceHolder.FLOAT;
+          continue;
+        }
         throw new RuntimeException(
             String.format("Illegal Json Path: [%s], for docId [%s]", _jsonPathString, inputDocIds[i]));
       }
@@ -225,8 +256,7 @@ public double[] transformToDoubleValuesSV(ValueBlock valueBlock) {
     int numDocs = valueBlock.getNumDocs();
     int[] inputDocIds = valueBlock.getDocIds();
     initDoubleValuesSV(numDocs);
-    String[] valuesFromIndex = _jsonIndexReader.getValuesSV(valueBlock.getDocIds(), valueBlock.getNumDocs(),
-        getValueToMatchingDocsMap(), false);
+    String[] valuesFromIndex = getCachedValuesFromIndex(valueBlock);
     for (int i = 0; i < numDocs; i++) {
       String value = valuesFromIndex[i];
       if (value == null) {
@@ -234,6 +264,10 @@ public double[] transformToDoubleValuesSV(ValueBlock valueBlock) {
           _doubleValuesSV[i] = (double) _defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _doubleValuesSV[i] = NullValuePlaceHolder.DOUBLE;
+          continue;
+        }
         throw new RuntimeException(
             String.format("Illegal Json Path: [%s], for docId [%s]", _jsonPathString, inputDocIds[i]));
       }
@@ -247,8 +281,7 @@ public BigDecimal[] transformToBigDecimalValuesSV(ValueBlock valueBlock) {
     int numDocs = valueBlock.getNumDocs();
     int[] inputDocIds = valueBlock.getDocIds();
     initBigDecimalValuesSV(numDocs);
-    String[] valuesFromIndex = _jsonIndexReader.getValuesSV(valueBlock.getDocIds(), valueBlock.getNumDocs(),
-        getValueToMatchingDocsMap(), false);
+    String[] valuesFromIndex = getCachedValuesFromIndex(valueBlock);
     for (int i = 0; i < numDocs; i++) {
       String value = valuesFromIndex[i];
       if (value == null) {
@@ -256,6 +289,10 @@ public BigDecimal[] transformToBigDecimalValuesSV(ValueBlock valueBlock) {
           _bigDecimalValuesSV[i] = (BigDecimal) _defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _bigDecimalValuesSV[i] = NullValuePlaceHolder.BIG_DECIMAL;
+          continue;
+        }
         throw new RuntimeException(
             String.format("Illegal Json Path: [%s], for docId [%s]", _jsonPathString, inputDocIds[i]));
       }
@@ -269,8 +306,7 @@ public String[] transformToStringValuesSV(ValueBlock valueBlock) {
     int numDocs = valueBlock.getNumDocs();
     int[] inputDocIds = valueBlock.getDocIds();
     initStringValuesSV(numDocs);
-    String[] valuesFromIndex = _jsonIndexReader.getValuesSV(valueBlock.getDocIds(), valueBlock.getNumDocs(),
-        getValueToMatchingDocsMap(), false);
+    String[] valuesFromIndex = getCachedValuesFromIndex(valueBlock);
     for (int i = 0; i < numDocs; i++) {
       String value = valuesFromIndex[i];
       if (value == null) {
@@ -278,6 +314,10 @@ public String[] transformToStringValuesSV(ValueBlock valueBlock) {
           _stringValuesSV[i] = (String) _defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _stringValuesSV[i] = NullValuePlaceHolder.STRING;
+          continue;
+        }
         throw new RuntimeException(
             String.format("Illegal Json Path: [%s], for docId [%s]", _jsonPathString, inputDocIds[i]));
       }
@@ -286,6 +326,28 @@ public String[] transformToStringValuesSV(ValueBlock valueBlock) {
     return _stringValuesSV;
   }
 
+  @Nullable
+  @Override
+  public RoaringBitmap getNullBitmap(ValueBlock valueBlock) {
+    // Short-circuit to argument-bitmap propagation when this function isn't introducing nulls of
+    // its own: non-SV output, null handling disabled, or a non-null default literal supplied (the
+    // SV transform writes it for unresolved rows). The SQL NULL literal case is handled in `init`
+    // by leaving `_defaultValue` null so it falls through to the scan below — same as the 3-arg
+    // form.
+    if (!_isSingleValue || !_nullHandlingEnabled || _defaultValue != null) {
+      return super.getNullBitmap(valueBlock);
+    }
+    String[] valuesFromIndex = getCachedValuesFromIndex(valueBlock);
+    int numDocs = valueBlock.getNumDocs();
+    RoaringBitmap nullBitmap = new RoaringBitmap();
+    for (int i = 0; i < numDocs; i++) {
+      if (valuesFromIndex[i] == null) {
+        nullBitmap.add(i);
+      }
+    }
+    return nullBitmap.isEmpty() ? null : nullBitmap;
+  }
+
   @Override
   public int[][] transformToIntValuesMV(ValueBlock valueBlock) {
     int numDocs = valueBlock.getNumDocs();
@@ -433,4 +495,17 @@ private Map<String, RoaringBitmap> getValueToMatchingDocsMap() {
     }
     return _valueToMatchingDocsMap;
   }
+
+  /**
+   * Returns the JSON-index SV scan result for the given block, caching it so that the SV
+   * transform and {@link #getNullBitmap} don't each pay the cost on the null-handling path.
+   */
+  private String[] getCachedValuesFromIndex(ValueBlock valueBlock) {
+    if (valueBlock != _cachedValueBlock) {
+      _cachedValuesSV = _jsonIndexReader.getValuesSV(valueBlock.getDocIds(), valueBlock.getNumDocs(),
+          getValueToMatchingDocsMap(), false);
+      _cachedValueBlock = valueBlock;
+    }
+    return _cachedValuesSV;
+  }
 }
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 94da2934f3..54fd5cbee1 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
@@ -22,6 +22,7 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.jayway.jsonpath.Configuration;
+import com.jayway.jsonpath.InvalidJsonException;
 import com.jayway.jsonpath.JsonPath;
 import com.jayway.jsonpath.Option;
 import com.jayway.jsonpath.ParseContext;
@@ -41,6 +42,7 @@
 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.CommonConstants.NullValuePlaceHolder;
 import org.apache.pinot.spi.utils.JsonUtils;
 import org.apache.pinot.spi.utils.TimestampUtils;
 import org.roaringbitmap.RoaringBitmap;
@@ -55,9 +57,11 @@
 /// - `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.
+/// - `defaultValue` (optional) — used when the path resolves to `null` or fails. Without it, the
+///   behavior depends on query-level null handling: when enabled, unresolved SV rows surface as SQL
+///   `NULL` (typed zero/empty sentinel + a bit in [#getNullBitmap]); when disabled, unresolved SV
+///   rows throw `IllegalArgumentException`. MV rows always surface as empty arrays when the path
+///   doesn't resolve; null elements within a resolved array still throw if no default is set.
 ///
 /// **Supported `resultsType`:** `INT`, `LONG`, `FLOAT`, `DOUBLE`, `BIG_DECIMAL`, `BOOLEAN`, `TIMESTAMP`,
 /// `STRING`, `JSON`, `BYTES`, plus `_ARRAY` variants of `INT` / `LONG` / `FLOAT` / `DOUBLE` /
@@ -189,7 +193,14 @@ public TransformResultMetadata getResultMetadata() {
   @Override
   @Nullable
   public RoaringBitmap getNullBitmap(ValueBlock valueBlock) {
-    if (!_defaultIsNull) {
+    // Short-circuit to argument-bitmap propagation when this function isn't introducing nulls of
+    // its own. Two cases qualify:
+    //  - null handling is disabled (the SV transform either fills a default or throws), or
+    //  - a real (non-SQL-NULL) default was supplied (the SV transform writes the default for
+    //    unresolved rows).
+    // The 4-arg SQL-NULL case has _defaultIsNull=true and falls through to the per-row scan
+    // below, so unresolved rows surface as SQL NULL via the bitmap.
+    if (!_nullHandlingEnabled || (_defaultValue != null && !_defaultIsNull)) {
       return super.getNullBitmap(valueBlock);
     }
     RoaringBitmap bitmap = new RoaringBitmap();
@@ -206,7 +217,16 @@ public RoaringBitmap getNullBitmap(ValueBlock valueBlock) {
       Object result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         nullBitmap.add(i);
@@ -235,13 +255,28 @@ public int[] transformToIntValuesSV(ValueBlock valueBlock) {
       Object result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         if (_defaultValue != null) {
           _intValuesSV[i] = defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          // Write the null placeholder so stale data from a reused buffer can't surface.
+          // getNullBitmap marks the row null; consumers should read the bitmap, not this value.
+          _intValuesSV[i] = NullValuePlaceHolder.INT;
+          continue;
+        }
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
@@ -264,13 +299,26 @@ public long[] transformToLongValuesSV(ValueBlock valueBlock) {
       Object result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         if (_defaultValue != null) {
           _longValuesSV[i] = defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _longValuesSV[i] = NullValuePlaceHolder.LONG;
+          continue;
+        }
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
@@ -292,13 +340,26 @@ public float[] transformToFloatValuesSV(ValueBlock valueBlock) {
       Object result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         if (_defaultValue != null) {
           _floatValuesSV[i] = defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _floatValuesSV[i] = NullValuePlaceHolder.FLOAT;
+          continue;
+        }
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
@@ -320,13 +381,26 @@ public double[] transformToDoubleValuesSV(ValueBlock valueBlock) {
       Object result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         if (_defaultValue != null) {
           _doubleValuesSV[i] = defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _doubleValuesSV[i] = NullValuePlaceHolder.DOUBLE;
+          continue;
+        }
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
@@ -348,13 +422,26 @@ public BigDecimal[] transformToBigDecimalValuesSV(ValueBlock valueBlock) {
       Object result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         if (_defaultValue != null) {
           _bigDecimalValuesSV[i] = defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _bigDecimalValuesSV[i] = NullValuePlaceHolder.BIG_DECIMAL;
+          continue;
+        }
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
@@ -376,13 +463,26 @@ public String[] transformToStringValuesSV(ValueBlock valueBlock) {
       Object result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         if (_defaultValue != null) {
           _stringValuesSV[i] = defaultValue;
           continue;
         }
+        if (_nullHandlingEnabled) {
+          _stringValuesSV[i] = NullValuePlaceHolder.STRING;
+          continue;
+        }
         throw new IllegalArgumentException(
             "Cannot resolve JSON path on some records. Consider setting a default value.");
       }
@@ -405,7 +505,16 @@ public int[][] transformToIntValuesMV(ValueBlock valueBlock) {
       List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         _intValuesMV[i] = new int[0];
@@ -445,7 +554,16 @@ public long[][] transformToLongValuesMV(ValueBlock valueBlock) {
       List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         _longValuesMV[i] = new long[0];
@@ -484,7 +602,16 @@ public float[][] transformToFloatValuesMV(ValueBlock valueBlock) {
       List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         _floatValuesMV[i] = new float[0];
@@ -523,7 +650,16 @@ public double[][] transformToDoubleValuesMV(ValueBlock valueBlock) {
       List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         _doubleValuesMV[i] = new double[0];
@@ -562,7 +698,16 @@ public String[][] transformToStringValuesMV(ValueBlock valueBlock) {
       List<Object> result = null;
       try {
         result = resultExtractor.apply(i);
+      } catch (InvalidJsonException e) {
+        // Input that is not valid JSON at all (as opposed to valid JSON with a missing path) is a
+        // caller / schema modeling error — e.g. jsonExtractScalar invoked on a non-JSON column.
+        // Surface it regardless of _nullHandlingEnabled so the bug isn't silently masked as SQL NULL.
+        throw new IllegalArgumentException(
+            "jsonExtractScalar received input that is not valid JSON. "
+                + "This function requires a JSON-typed column or a JSON-formatted string.", e);
       } catch (Exception ignored) {
+        // Null / empty input or a missing path: result stays null and the cascade below applies
+        // null-handling or the default value (preserves PR #205's behavior).
       }
       if (result == null) {
         _stringValuesMV[i] = new String[0];
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunctionTest.java
index 7fcfb31b53..5167c66b84 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunctionTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractIndexTransformFunctionTest.java
@@ -18,6 +18,8 @@
  */
 package org.apache.pinot.core.operator.transform.function;
 
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.jayway.jsonpath.Configuration;
 import com.jayway.jsonpath.JsonPath;
 import com.jayway.jsonpath.Option;
@@ -25,15 +27,29 @@
 import com.jayway.jsonpath.TypeRef;
 import com.jayway.jsonpath.spi.json.JacksonJsonProvider;
 import com.jayway.jsonpath.spi.mapper.JacksonMappingProvider;
+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.HashMap;
 import java.util.List;
+import org.apache.commons.io.FileUtils;
 import org.apache.pinot.common.function.JsonPathCache;
 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.FieldConfig;
+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.utils.builder.TableConfigBuilder;
+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;
 
@@ -423,4 +439,462 @@ private String getValueForKey(String blob, JsonPath path) {
   private <T> T getValueForKey(String blob, JsonPath path, TypeRef<T> typeRef) {
     return JSON_PARSER_CONTEXT.parse(blob).read(path, typeRef);
   }
+
+  // ============================================================================================
+  // Country/click null-handling tests for jsonExtractIndex (issue #18568).
+  //
+  // The SV null-handling fix: with `enableNullHandling = true` and no default literal, an
+  // unresolved JSON path must surface as SQL NULL instead of throwing.
+  //
+  // All tests below share a single `clicks` table fixture with two JSON-indexed columns and four
+  // rows. The fixture mirrors the one in JsonExtractScalarTransformFunctionTest so that, modulo
+  // the function name, expected results are identical — `jsonExtractIndex` and
+  // `jsonExtractScalar` must agree on null semantics after the fix.
+  //
+  //   `flatJson`   — 1-level scalar shapes. Each row exercises one case: resolved value, explicit
+  //                  JSON null, missing key, empty document.
+  //   `nestedJson` — multi-level shapes (nested object, array, array of objects) with mixed
+  //                  resolved / null / missing data at depth.
+  //
+  // Tests are grouped by query shape in this order: projection → DISTINCT → GROUP BY. Each section
+  // covers the same four sub-cases, in order:
+  //
+  //   .1  NH on, 3-arg                          — unresolved rows surface as SQL NULL (E4 fix)
+  //   .2  NH on, 4-arg with SQL NULL literal     — same result as .1; `init` detects the null
+  //                                                literal and leaves `_defaultValue` null so the
+  //                                                3-arg semantics apply
+  //   .3  NH on, 4-arg with a non-null default   — default wins over NH placeholder
+  //   .4  NH off                                  — legacy throw preserved
+  //
+  // Result rows in every assertion are ordered by the projected expression ASC NULLS LAST so the
+  // comparison is deterministic across the framework's segment-duplication behavior (one segment
+  // becomes two, so per-row counts ×2).
+  // ============================================================================================
+
+  protected File _baseDir;
+
+  @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);
+    }
+  }
+
+  // ---------------- Fixture ----------------
+  //
+  // Pretty-printed contents of the 4-row × 2-column fixture:
+  //
+  //   row 0 — fully populated:
+  //     flatJson:   { "country": "US", "clicks": 5 }
+  //     nestedJson: { "location": {"city": "SF", "country": "US"},
+  //                   "tags":     ["red", "blue", "green"],
+  //                   "events":   [{"country": "US"}, {"country": "CA"}] }
+  //
+  //   row 1 — partial / explicit-null at depth:
+  //     flatJson:   { "country": "CA", "clicks": 3 }
+  //     nestedJson: { "location": {"city": "Tor"},          // <-- no country key
+  //                   "tags":     ["green"],                 // <-- only one element
+  //                   "events":   [{"country": null}] }     // <-- inner explicit null
+  //
+  //   row 2 — every field explicit JSON null:
+  //     flatJson:   { "country": null, "clicks": null }
+  //     nestedJson: { "location": null, "tags": null, "events": null }
+  //
+  //   row 3 — empty documents:
+  //     flatJson:   {}
+  //     nestedJson: {}
+
+  private static final Object[][] COUNTRY_CLICK_FIXTURE = {
+      // {flatJson, nestedJson}
+      {
+          "{\"country\":\"US\",\"clicks\":5}",
+          "{\"location\":{\"city\":\"SF\",\"country\":\"US\"},"
+              + "\"tags\":[\"red\",\"blue\",\"green\"],"
+              + "\"events\":[{\"country\":\"US\"},{\"country\":\"CA\"}]}"
+      },
+      {
+          "{\"country\":\"CA\",\"clicks\":3}",
+          "{\"location\":{\"city\":\"Tor\"},"
+              + "\"tags\":[\"green\"],"
+              + "\"events\":[{\"country\":null}]}"
+      },
+      {
+          "{\"country\":null,\"clicks\":null}",
+          "{\"location\":null,\"tags\":null,\"events\":null}"
+      },
+      {"{}", "{}"}
+  };
+
+  // ---------------- Helper ----------------
+  //
+  // `jsonExtractIndex` requires the column to have a JSON index. The helper wires a JSON index
+  // onto both columns via FieldConfig — same shape as the JSON index in
+  // BaseTransformFunctionTest.getTableConfig().
+
+  private FluentQueryTest.OnFirstInstance givenCountryClickTable(boolean nullHandling) {
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("clicks")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("flatJson", DataType.JSON)
+        .addDimensionField("nestedJson", DataType.JSON)
+        .build();
+    ObjectNode jsonIndexNode = JsonNodeFactory.instance.objectNode();
+    jsonIndexNode.set("json", JsonNodeFactory.instance.objectNode());
+    List<FieldConfig> fieldConfigs = List.of(
+        new FieldConfig("flatJson", FieldConfig.EncodingType.DICTIONARY, null, null, null, null, jsonIndexNode, null,
+            null),
+        new FieldConfig("nestedJson", FieldConfig.EncodingType.DICTIONARY, null, null, null, null, jsonIndexNode, null,
+            null));
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+        .setTableName("clicks")
+        .setFieldConfigList(fieldConfigs)
+        .build();
+    return FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(nullHandling)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(COUNTRY_CLICK_FIXTURE);
+  }
+
+  // ============================================================================================
+  // 1. Projection
+  // ============================================================================================
+
+  /**
+   * 1.1 — NH on, 3-arg. The SV transform must surface SQL NULL for unresolved rows and pass
+   * resolved values through.
+   * <pre>
+   *   Example — `flatJson.country` (STRING):
+   *     per-row: row 0 -> "US", row 1 -> "CA", row 2 -> NULL (explicit), row 3 -> NULL (missing)
+   *     Query:  SET enableNullHandling = true;
+   *             SELECT jsonExtractIndex(flatJson, '$.country', 'STRING') AS c
+   *             FROM clicks ORDER BY c ASC NULLS LAST
+   *     Result (×2 dup): "CA", "CA", "US", "US", NULL, NULL, NULL, NULL
+   * </pre>
+   */
+  @Test(dataProvider = "projectionCases")
+  public void testProjectionNullHandlingOn(String column, String jsonPath, String resultsType,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractIndex(%s, '%s', '%s')", column, jsonPath, resultsType);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT %s FROM clicks ORDER BY %s ASC NULLS LAST", expr, expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /**
+   * 1.2 — NH on, 4-arg with `NULL` literal. Same result as 1.1: `init` detects the null literal
+   * and leaves `_defaultValue` null, so the 3-arg code path applies.
+   */
+  @Test(dataProvider = "sqlNullDefaultProjectionCases")
+  public void testProjectionWithSqlNullDefaultEmitsNull(String column, String jsonPath,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractIndex(%s, '%s', 'STRING', NULL)", column, jsonPath);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT %s AS c FROM clicks ORDER BY c ASC NULLS LAST", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /**
+   * 1.3 — NH on, 4-arg with non-null default. The SV transform's priority is: real default >
+   * null-handling placeholder > throw. The user-supplied default surfaces for unresolved rows;
+   * no null bit is set in the bitmap.
+   */
+  @Test(dataProvider = "defaultPrecedenceProjectionCases")
+  public void testProjectionDefaultBeatsNullHandlingPlaceholder(String column, String jsonPath,
+      String defaultLiteral, Object[][] expectedRows, String label) {
+    String expr = String.format(
+        "jsonExtractIndex(%s, '%s', 'STRING', %s)", column, jsonPath, defaultLiteral);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT %s AS c FROM clicks ORDER BY c ASC", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /**
+   * 1.4 — NH off: any unresolved row throws. One representative path; throw is path-independent.
+   * The index function's error message differs from the scalar's ("Illegal Json Path" vs "Cannot
+   * resolve JSON path").
+   */
+  @Test
+  public void testProjectionNullHandlingOffThrows() {
+    try {
+      givenCountryClickTable(false)
+          .whenQuery("SELECT jsonExtractIndex(flatJson, '$.country', 'STRING') FROM clicks")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected projection to fail when null handling is off and rows are unresolved");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("Illegal Json Path");
+    }
+  }
+
+  /** 1.1 cases — `(column, jsonPath, resultsType, expected 8-row result, label)`, ordered ASC NULLS LAST. */
+  @DataProvider(name = "projectionCases")
+  public static Object[][] projectionCases() {
+    return new Object[][]{
+        // ----- flatJson: 1-level scalar shape -----
+        // $.country: row0="US", row1="CA", row2=null (explicit), row3=null (missing).
+        {"flatJson", "$.country", "STRING", new Object[][]{
+            {"CA"}, {"CA"}, {"US"}, {"US"}, {null}, {null}, {null}, {null}
+        }, "flatJson.country STRING"},
+        // $.clicks across all 6 numeric SV types: row0=5, row1=3, row2=null, row3=null.
+        {"flatJson", "$.clicks", "INT", new Object[][]{
+            {3}, {3}, {5}, {5}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks INT"},
+        {"flatJson", "$.clicks", "LONG", new Object[][]{
+            {3L}, {3L}, {5L}, {5L}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks LONG"},
+        {"flatJson", "$.clicks", "FLOAT", new Object[][]{
+            {3f}, {3f}, {5f}, {5f}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks FLOAT"},
+        {"flatJson", "$.clicks", "DOUBLE", new Object[][]{
+            {3d}, {3d}, {5d}, {5d}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks DOUBLE"},
+        // BIG_DECIMAL is formatted as String by the broker.
+        {"flatJson", "$.clicks", "BIG_DECIMAL", new Object[][]{
+            {"3"}, {"3"}, {"5"}, {"5"}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks BIG_DECIMAL"},
+
+        // ----- nestedJson: multi-level shapes -----
+        {"nestedJson", "$.location.country", "STRING", new Object[][]{
+            {"US"}, {"US"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.location.country STRING"},
+        {"nestedJson", "$.tags[0]", "STRING", new Object[][]{
+            {"green"}, {"green"}, {"red"}, {"red"}, {null}, {null}, {null}, {null}
+        }, "nestedJson.tags[0] STRING"},
+        {"nestedJson", "$.tags[1]", "STRING", new Object[][]{
+            {"blue"}, {"blue"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.tags[1] STRING (OOB on row 1)"},
+        {"nestedJson", "$.events[0].country", "STRING", new Object[][]{
+            {"US"}, {"US"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.events[0].country STRING"}
+    };
+  }
+
+  /** 1.2 cases — `(column, jsonPath, expected 8-row result ordered ASC NULLS LAST, label)`. */
+  @DataProvider(name = "sqlNullDefaultProjectionCases")
+  public static Object[][] sqlNullDefaultProjectionCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", new Object[][]{
+            {"CA"}, {"CA"}, {"US"}, {"US"}, {null}, {null}, {null}, {null}
+        }, "flatJson.country NULL default"},
+        {"nestedJson", "$.location.country", new Object[][]{
+            {"US"}, {"US"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.location.country NULL default"}
+    };
+  }
+
+  /** 1.3 cases — `(column, jsonPath, default SQL literal, expected 8-row result ASC, label)`. */
+  @DataProvider(name = "defaultPrecedenceProjectionCases")
+  public static Object[][] defaultPrecedenceProjectionCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", "'foobar'", new Object[][]{
+            {"CA"}, {"CA"}, {"US"}, {"US"},
+            {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}
+        }, "flatJson.country default 'foobar'"},
+        {"nestedJson", "$.location.country", "'foobar'", new Object[][]{
+            {"US"}, {"US"},
+            {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}
+        }, "nestedJson.location.country default 'foobar'"}
+    };
+  }
+
+  // ============================================================================================
+  // 2. DISTINCT
+  //
+  // Pinot rejects positional `ORDER BY 1` for DISTINCT, so each DISTINCT test repeats the
+  // projected expression in ORDER BY.
+  // ============================================================================================
+
+  /**
+   * 2.1 — NH on, 3-arg. The distinct set must include exactly one null entry alongside the
+   * resolved values (deduped across segments).
+   */
+  @Test(dataProvider = "distinctCases")
+  public void testDistinctNullHandlingOn(String column, String jsonPath, String resultsType,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractIndex(%s, '%s', '%s')", column, jsonPath, resultsType);
+    String query = String.format("SELECT DISTINCT %s FROM clicks ORDER BY %s ASC NULLS LAST", expr, expr);
+    givenCountryClickTable(true).whenQuery(query).thenResultIs(expectedRows);
+  }
+
+  /** 2.2 — NH on, 4-arg with `NULL` literal. Same distinct set as 2.1 (includes a null entry). */
+  @Test(dataProvider = "sqlNullDefaultDistinctCases")
+  public void testDistinctWithSqlNullDefaultIncludesNull(String column, String jsonPath,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractIndex(%s, '%s', 'STRING', NULL)", column, jsonPath);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT DISTINCT %s FROM clicks ORDER BY %s ASC NULLS LAST", expr, expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 2.3 — NH on, 4-arg with non-null default. Default appears as a regular distinct entry (no null). */
+  @Test(dataProvider = "defaultPrecedenceDistinctCases")
+  public void testDistinctDefaultBeatsNullHandlingPlaceholder(String column, String jsonPath,
+      String defaultLiteral, Object[][] expectedRows, String label) {
+    String expr = String.format(
+        "jsonExtractIndex(%s, '%s', 'STRING', %s)", column, jsonPath, defaultLiteral);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT DISTINCT %s FROM clicks ORDER BY %s ASC", expr, expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 2.4 — NH off: DISTINCT throws on unresolved rows. */
+  @Test
+  public void testDistinctNullHandlingOffThrows() {
+    try {
+      givenCountryClickTable(false)
+          .whenQuery("SELECT DISTINCT jsonExtractIndex(flatJson, '$.country', 'STRING') FROM clicks")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected DISTINCT to fail when null handling is off and rows are unresolved");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("Illegal Json Path");
+    }
+  }
+
+  /** 2.1 cases — `(column, jsonPath, resultsType, expected distinct rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "distinctCases")
+  public static Object[][] distinctCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", "STRING", new Object[][]{{"CA"}, {"US"}, {null}}, "flatJson.country"},
+        {"flatJson", "$.clicks", "INT", new Object[][]{{3}, {5}, {null}}, "flatJson.clicks INT"},
+        {"nestedJson", "$.location.country", "STRING", new Object[][]{{"US"}, {null}},
+            "nestedJson.location.country"},
+        {"nestedJson", "$.tags[0]", "STRING", new Object[][]{{"green"}, {"red"}, {null}},
+            "nestedJson.tags[0]"},
+        {"nestedJson", "$.events[0].country", "STRING", new Object[][]{{"US"}, {null}},
+            "nestedJson.events[0].country"}
+    };
+  }
+
+  /** 2.2 cases — `(column, jsonPath, expected distinct rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "sqlNullDefaultDistinctCases")
+  public static Object[][] sqlNullDefaultDistinctCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", new Object[][]{{"CA"}, {"US"}, {null}}, "flatJson.country NULL default"},
+        {"nestedJson", "$.location.country", new Object[][]{{"US"}, {null}},
+            "nestedJson.location.country NULL default"}
+    };
+  }
+
+  /** 2.3 cases — `(column, jsonPath, default literal, expected distinct rows ASC, label)`. */
+  @DataProvider(name = "defaultPrecedenceDistinctCases")
+  public static Object[][] defaultPrecedenceDistinctCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", "'foobar'",
+            new Object[][]{{"CA"}, {"US"}, {"foobar"}}, "flatJson.country default 'foobar'"},
+        {"nestedJson", "$.location.country", "'foobar'",
+            new Object[][]{{"US"}, {"foobar"}}, "nestedJson.location.country default 'foobar'"}
+    };
+  }
+
+  // ============================================================================================
+  // 3. GROUP BY
+  // ============================================================================================
+
+  /**
+   * 3.1 — NH on, 3-arg. Unresolved rows collapse into a single null group with the correct count,
+   * alongside the resolved value groups.
+   */
+  @Test(dataProvider = "groupByCases")
+  public void testGroupByNullHandlingOn(String column, String jsonPath, String resultsType,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractIndex(%s, '%s', '%s')", column, jsonPath, resultsType);
+    givenCountryClickTable(true)
+        .whenQuery(String.format(
+            "SELECT %s AS v, COUNT(*) FROM clicks GROUP BY v ORDER BY v ASC NULLS LAST", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 3.2 — NH on, 4-arg with `NULL` literal. Same groups as 3.1 (null group with correct count). */
+  @Test(dataProvider = "sqlNullDefaultGroupByCases")
+  public void testGroupByWithSqlNullDefaultProducesNullGroup(String column, String jsonPath,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractIndex(%s, '%s', 'STRING', NULL)", column, jsonPath);
+    givenCountryClickTable(true)
+        .whenQuery(String.format(
+            "SELECT %s AS v, COUNT(*) FROM clicks GROUP BY v ORDER BY v ASC NULLS LAST", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /**
+   * 3.3 — NH on, 4-arg with non-null default. Unresolved rows count toward the default's group;
+   * no NULL group surfaces.
+   */
+  @Test(dataProvider = "defaultPrecedenceGroupByCases")
+  public void testGroupByDefaultBeatsNullHandlingPlaceholder(String column, String jsonPath,
+      String defaultLiteral, Object[][] expectedRows, String label) {
+    String expr = String.format(
+        "jsonExtractIndex(%s, '%s', 'STRING', %s)", column, jsonPath, defaultLiteral);
+    givenCountryClickTable(true)
+        .whenQuery(String.format(
+            "SELECT %s AS v, COUNT(*) FROM clicks GROUP BY v ORDER BY v ASC", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 3.4 — NH off: GROUP BY throws on unresolved rows. */
+  @Test
+  public void testGroupByNullHandlingOffThrows() {
+    try {
+      givenCountryClickTable(false)
+          .whenQuery("SELECT jsonExtractIndex(flatJson, '$.country', 'STRING'), COUNT(*) FROM clicks GROUP BY 1")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected GROUP BY to fail when null handling is off and rows are unresolved");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("Illegal Json Path");
+    }
+  }
+
+  /** 3.1 cases — `(column, jsonPath, resultsType, expected (value, count) rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "groupByCases")
+  public static Object[][] groupByCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", "STRING", new Object[][]{
+            {"CA", 2L}, {"US", 2L}, {null, 4L}
+        }, "flatJson.country"},
+        {"flatJson", "$.clicks", "INT", new Object[][]{
+            {3, 2L}, {5, 2L}, {null, 4L}
+        }, "flatJson.clicks INT"},
+        {"nestedJson", "$.location.country", "STRING", new Object[][]{
+            {"US", 2L}, {null, 6L}
+        }, "nestedJson.location.country"},
+        {"nestedJson", "$.tags[0]", "STRING", new Object[][]{
+            {"green", 2L}, {"red", 2L}, {null, 4L}
+        }, "nestedJson.tags[0]"},
+        {"nestedJson", "$.events[0].country", "STRING", new Object[][]{
+            {"US", 2L}, {null, 6L}
+        }, "nestedJson.events[0].country"}
+    };
+  }
+
+  /** 3.2 cases — `(column, jsonPath, expected (value, count) rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "sqlNullDefaultGroupByCases")
+  public static Object[][] sqlNullDefaultGroupByCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", new Object[][]{{"CA", 2L}, {"US", 2L}, {null, 4L}},
+            "flatJson.country NULL default"},
+        {"nestedJson", "$.location.country", new Object[][]{{"US", 2L}, {null, 6L}},
+            "nestedJson.location.country NULL default"}
+    };
+  }
+
+  /** 3.3 cases — `(column, jsonPath, default literal, expected (value, count) rows ASC, label)`. */
+  @DataProvider(name = "defaultPrecedenceGroupByCases")
+  public static Object[][] defaultPrecedenceGroupByCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", "'foobar'", new Object[][]{
+            {"CA", 2L}, {"US", 2L}, {"foobar", 4L}
+        }, "flatJson.country default 'foobar'"},
+        {"nestedJson", "$.location.country", "'foobar'", new Object[][]{
+            {"US", 2L}, {"foobar", 6L}
+        }, "nestedJson.location.country default 'foobar'"}
+    };
+  }
 }
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 586880ae1f..0b47dcd2ed 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
@@ -447,6 +447,7 @@ public void testIllegalArguments(String expressionStr) {
     TransformFunctionFactory.get(expression, _dataSourceMap);
   }
 
+
   @Test(dataProvider = "testParsingIllegalQueries", expectedExceptions = {SqlCompilationException.class})
   public void testParsingIllegalQueries(String expressionStr) {
     RequestContextUtils.getExpression(expressionStr);
@@ -903,4 +904,524 @@ public void testCrossTypeConversionFromStringResult() {
         .whenQuery("SELECT CAST(jsonExtractScalar(json, '$.v', 'STRING') AS LONG) FROM testTable")
         .thenResultIs(new Object[]{42L}, new Object[]{42L});
   }
+
+  // ============================================================================================
+  // Country/click null-handling tests for jsonExtractScalar (issue #18568).
+  //
+  // The SV null-handling fix: with `enableNullHandling = true` and no default literal, an
+  // unresolved JSON path must surface as SQL NULL instead of throwing.
+  //
+  // All tests below share a single `clicks` table fixture with two JSON columns and four rows:
+  //
+  //   `flatJson`   — 1-level scalar shapes. Each row exercises one case: resolved value, explicit
+  //                  JSON null, missing key, empty document.
+  //   `nestedJson` — multi-level shapes (nested object, array, array of objects) with mixed
+  //                  resolved / null / missing data at depth.
+  //
+  // Tests are grouped by query shape in this order: projection → DISTINCT → GROUP BY. Each section
+  // covers the same four sub-cases, in order:
+  //
+  //   .1  NH on, 3-arg                          — unresolved rows surface as SQL NULL (E4 fix)
+  //   .2  NH on, 4-arg with SQL NULL literal     — same result as .1 via the `_defaultIsNull`
+  //                                                code path in `getNullBitmap`
+  //   .3  NH on, 4-arg with a non-null default   — default wins over NH placeholder
+  //   .4  NH off                                  — legacy throw preserved
+  //
+  // Result rows in every assertion are ordered by the projected expression ASC NULLS LAST so the
+  // comparison is deterministic across the framework's segment-duplication behavior (a single
+  // segment shows up twice, so per-row counts are ×2).
+  // ============================================================================================
+
+  // ---------------- Fixture ----------------
+  //
+  // Pretty-printed contents of the 4-row × 2-column fixture:
+  //
+  //   row 0 — fully populated:
+  //     flatJson:   { "country": "US", "clicks": 5 }
+  //     nestedJson: { "location": {"city": "SF", "country": "US"},
+  //                   "tags":     ["red", "blue", "green"],
+  //                   "events":   [{"country": "US"}, {"country": "CA"}] }
+  //
+  //   row 1 — partial / explicit-null at depth:
+  //     flatJson:   { "country": "CA", "clicks": 3 }
+  //     nestedJson: { "location": {"city": "Tor"},          // <-- no country key
+  //                   "tags":     ["green"],                 // <-- only one element
+  //                   "events":   [{"country": null}] }     // <-- inner explicit null
+  //
+  //   row 2 — every field explicit JSON null:
+  //     flatJson:   { "country": null, "clicks": null }
+  //     nestedJson: { "location": null, "tags": null, "events": null }
+  //
+  //   row 3 — empty documents:
+  //     flatJson:   {}
+  //     nestedJson: {}
+  //
+  // Across the four rows, every path covered by the tests has at least one resolved value AND
+  // at least one unresolved occurrence (missing key, explicit JSON null, null parent, or
+  // out-of-bounds array index), so both the value groups and the null group must show up.
+
+  private static final Object[][] COUNTRY_CLICK_FIXTURE = {
+      // {flatJson, nestedJson}
+      {
+          "{\"country\":\"US\",\"clicks\":5}",
+          "{\"location\":{\"city\":\"SF\",\"country\":\"US\"},"
+              + "\"tags\":[\"red\",\"blue\",\"green\"],"
+              + "\"events\":[{\"country\":\"US\"},{\"country\":\"CA\"}]}"
+      },
+      {
+          "{\"country\":\"CA\",\"clicks\":3}",
+          "{\"location\":{\"city\":\"Tor\"},"
+              + "\"tags\":[\"green\"],"
+              + "\"events\":[{\"country\":null}]}"
+      },
+      {
+          "{\"country\":null,\"clicks\":null}",
+          "{\"location\":null,\"tags\":null,\"events\":null}"
+      },
+      {"{}", "{}"}
+  };
+
+  // ---------------- Helper ----------------
+
+  private FluentQueryTest.OnFirstInstance givenCountryClickTable(boolean nullHandling) {
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("clicks")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("flatJson", DataType.JSON)
+        .addDimensionField("nestedJson", DataType.JSON)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("clicks").build();
+    return FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(nullHandling)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(COUNTRY_CLICK_FIXTURE);
+  }
+
+  // ============================================================================================
+  // 1. Projection
+  // ============================================================================================
+
+  /**
+   * 1.1 — NH on, 3-arg. Each row's resolution depends on the column + path; the SV transform must
+   * surface SQL NULL for unresolved rows and pass resolved values through.
+   * <pre>
+   *   Example — `flatJson.country` (STRING):
+   *     per-row: row 0 -> "US", row 1 -> "CA", row 2 -> NULL (explicit), row 3 -> NULL (missing)
+   *
+   *     Query:  SET enableNullHandling = true;
+   *             SELECT jsonExtractScalar(flatJson, '$.country', 'STRING') AS c
+   *             FROM clicks ORDER BY c ASC NULLS LAST
+   *
+   *     Result rows (×2 for segment duplication):
+   *             "CA", "CA", "US", "US", NULL, NULL, NULL, NULL
+   * </pre>
+   */
+  @Test(dataProvider = "projectionCases")
+  public void testProjectionNullHandlingOn(String column, String jsonPath, String resultsType,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractScalar(%s, '%s', '%s')", column, jsonPath, resultsType);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT %s FROM clicks ORDER BY %s ASC NULLS LAST", expr, expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /**
+   * 1.2 — NH on, 4-arg with `NULL` literal. Same result as 1.1, but exercises the `_defaultIsNull`
+   * code path in `getNullBitmap` (the SV transform writes the typed zero via `_defaultValue != null`,
+   * then `getNullBitmap` scans and marks unresolved rows null via the `!_defaultIsNull` carve-out).
+   */
+  @Test(dataProvider = "sqlNullDefaultProjectionCases")
+  public void testProjectionWithSqlNullDefaultEmitsNull(String column, String jsonPath,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractScalar(%s, '%s', 'STRING', NULL)", column, jsonPath);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT %s AS c FROM clicks ORDER BY c ASC NULLS LAST", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /**
+   * 1.3 — NH on, 4-arg with non-null default. The SV transform's priority is: real default >
+   * null-handling placeholder > throw. The user-supplied default surfaces for unresolved rows;
+   * no null bit is set in the bitmap.
+   * <pre>
+   *   Example — flatJson.country with default 'foobar':
+   *     per-row: row 0 -> "US", row 1 -> "CA", row 2 -> "foobar", row 3 -> "foobar"
+   *
+   *     Query: SET enableNullHandling = true;
+   *            SELECT jsonExtractScalar(flatJson, '$.country', 'STRING', 'foobar') AS c
+   *            FROM clicks ORDER BY c ASC
+   *
+   *     Result (×2 dup; ASCII order): "CA", "CA", "US", "US", "foobar"×4
+   * </pre>
+   */
+  @Test(dataProvider = "defaultPrecedenceProjectionCases")
+  public void testProjectionDefaultBeatsNullHandlingPlaceholder(String column, String jsonPath,
+      String defaultLiteral, Object[][] expectedRows, String label) {
+    String expr = String.format(
+        "jsonExtractScalar(%s, '%s', 'STRING', %s)", column, jsonPath, defaultLiteral);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT %s AS c FROM clicks ORDER BY c ASC", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 1.4 — NH off: any unresolved row throws. One representative path; throw is path-independent. */
+  @Test
+  public void testProjectionNullHandlingOffThrows() {
+    try {
+      givenCountryClickTable(false)
+          .whenQuery("SELECT jsonExtractScalar(flatJson, '$.country', 'STRING') FROM clicks")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected projection to fail when null handling is off and rows are unresolved");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("Cannot resolve JSON path on some records");
+    }
+  }
+
+  /** 1.1 cases — `(column, jsonPath, resultsType, expected 8-row result, label)`, ordered ASC NULLS LAST. */
+  @DataProvider(name = "projectionCases")
+  public static Object[][] projectionCases() {
+    return new Object[][]{
+        // ----- flatJson: 1-level scalar shape -----
+        // $.country: row0="US", row1="CA", row2=null (explicit), row3=null (missing).
+        {"flatJson", "$.country", "STRING", new Object[][]{
+            {"CA"}, {"CA"}, {"US"}, {"US"}, {null}, {null}, {null}, {null}
+        }, "flatJson.country STRING"},
+        // $.clicks across all 6 numeric SV types: row0=5, row1=3, row2=null, row3=null.
+        {"flatJson", "$.clicks", "INT", new Object[][]{
+            {3}, {3}, {5}, {5}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks INT"},
+        {"flatJson", "$.clicks", "LONG", new Object[][]{
+            {3L}, {3L}, {5L}, {5L}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks LONG"},
+        {"flatJson", "$.clicks", "FLOAT", new Object[][]{
+            {3f}, {3f}, {5f}, {5f}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks FLOAT"},
+        {"flatJson", "$.clicks", "DOUBLE", new Object[][]{
+            {3d}, {3d}, {5d}, {5d}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks DOUBLE"},
+        // BIG_DECIMAL is formatted as String by the broker (BigDecimal.toPlainString).
+        {"flatJson", "$.clicks", "BIG_DECIMAL", new Object[][]{
+            {"3"}, {"3"}, {"5"}, {"5"}, {null}, {null}, {null}, {null}
+        }, "flatJson.clicks BIG_DECIMAL"},
+
+        // ----- nestedJson: multi-level shapes -----
+        // $.location.country: row0="US", row1=missing (no country key), row2=parent null, row3=missing.
+        {"nestedJson", "$.location.country", "STRING", new Object[][]{
+            {"US"}, {"US"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.location.country STRING"},
+        // $.tags[0]: row0="red", row1="green", row2=null parent, row3=missing.
+        {"nestedJson", "$.tags[0]", "STRING", new Object[][]{
+            {"green"}, {"green"}, {"red"}, {"red"}, {null}, {null}, {null}, {null}
+        }, "nestedJson.tags[0] STRING"},
+        // $.tags[1]: row0="blue", row1=out-of-bounds (only one element), row2=null parent, row3=missing.
+        {"nestedJson", "$.tags[1]", "STRING", new Object[][]{
+            {"blue"}, {"blue"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.tags[1] STRING (OOB on row 1)"},
+        // $.events[0].country: row0="US", row1=explicit inner null, row2=null array, row3=missing.
+        {"nestedJson", "$.events[0].country", "STRING", new Object[][]{
+            {"US"}, {"US"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.events[0].country STRING"}
+    };
+  }
+
+  /** 1.2 cases — `(column, jsonPath, expected 8-row result ordered ASC NULLS LAST, label)`. */
+  @DataProvider(name = "sqlNullDefaultProjectionCases")
+  public static Object[][] sqlNullDefaultProjectionCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", new Object[][]{
+            {"CA"}, {"CA"}, {"US"}, {"US"}, {null}, {null}, {null}, {null}
+        }, "flatJson.country NULL default"},
+        {"nestedJson", "$.location.country", new Object[][]{
+            {"US"}, {"US"}, {null}, {null}, {null}, {null}, {null}, {null}
+        }, "nestedJson.location.country NULL default"}
+    };
+  }
+
+  /** 1.3 cases — `(column, jsonPath, default SQL literal, expected 8-row result ASC, label)`. */
+  @DataProvider(name = "defaultPrecedenceProjectionCases")
+  public static Object[][] defaultPrecedenceProjectionCases() {
+    return new Object[][]{
+        // flatJson.country: row0="US", row1="CA", row2=null, row3=missing -> 2 each, 4 'foobar'.
+        {"flatJson", "$.country", "'foobar'", new Object[][]{
+            {"CA"}, {"CA"}, {"US"}, {"US"},
+            {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}
+        }, "flatJson.country default 'foobar'"},
+        // nestedJson.location.country: row0="US", rows 1/2/3 unresolved -> 2 "US" + 6 'foobar'.
+        {"nestedJson", "$.location.country", "'foobar'", new Object[][]{
+            {"US"}, {"US"},
+            {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}, {"foobar"}
+        }, "nestedJson.location.country default 'foobar'"}
+    };
+  }
+
+  // ============================================================================================
+  // 2. DISTINCT
+  //
+  // Pinot rejects positional `ORDER BY 1` for DISTINCT ("ORDER-BY columns should be included in
+  // the DISTINCT columns"), so each DISTINCT test repeats the projected expression in ORDER BY.
+  // ============================================================================================
+
+  /**
+   * 2.1 — NH on, 3-arg. The distinct set must include exactly one null entry alongside the
+   * resolved values (deduped across segments).
+   * <pre>
+   *   Example — `flatJson.country`: Result: "CA", "US", NULL
+   * </pre>
+   */
+  @Test(dataProvider = "distinctCases")
+  public void testDistinctNullHandlingOn(String column, String jsonPath, String resultsType,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractScalar(%s, '%s', '%s')", column, jsonPath, resultsType);
+    String query = String.format("SELECT DISTINCT %s FROM clicks ORDER BY %s ASC NULLS LAST", expr, expr);
+    givenCountryClickTable(true).whenQuery(query).thenResultIs(expectedRows);
+  }
+
+  /** 2.2 — NH on, 4-arg with `NULL` literal. Same distinct set as 2.1 (includes a null entry). */
+  @Test(dataProvider = "sqlNullDefaultDistinctCases")
+  public void testDistinctWithSqlNullDefaultIncludesNull(String column, String jsonPath,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractScalar(%s, '%s', 'STRING', NULL)", column, jsonPath);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT DISTINCT %s FROM clicks ORDER BY %s ASC NULLS LAST", expr, expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 2.3 — NH on, 4-arg with non-null default. Default value appears as a regular distinct entry (no null). */
+  @Test(dataProvider = "defaultPrecedenceDistinctCases")
+  public void testDistinctDefaultBeatsNullHandlingPlaceholder(String column, String jsonPath,
+      String defaultLiteral, Object[][] expectedRows, String label) {
+    String expr = String.format(
+        "jsonExtractScalar(%s, '%s', 'STRING', %s)", column, jsonPath, defaultLiteral);
+    givenCountryClickTable(true)
+        .whenQuery(String.format("SELECT DISTINCT %s FROM clicks ORDER BY %s ASC", expr, expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 2.4 — NH off: DISTINCT throws on unresolved rows. */
+  @Test
+  public void testDistinctNullHandlingOffThrows() {
+    try {
+      givenCountryClickTable(false)
+          .whenQuery("SELECT DISTINCT jsonExtractScalar(flatJson, '$.country', 'STRING') FROM clicks")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected DISTINCT to fail when null handling is off and rows are unresolved");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("Cannot resolve JSON path on some records");
+    }
+  }
+
+  /** 2.1 cases — `(column, jsonPath, resultsType, expected distinct rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "distinctCases")
+  public static Object[][] distinctCases() {
+    return new Object[][]{
+        // flatJson: 1-level scalar.
+        {"flatJson", "$.country", "STRING", new Object[][]{{"CA"}, {"US"}, {null}}, "flatJson.country"},
+        {"flatJson", "$.clicks", "INT", new Object[][]{{3}, {5}, {null}}, "flatJson.clicks INT"},
+        // nestedJson: multi-level paths.
+        {"nestedJson", "$.location.country", "STRING", new Object[][]{{"US"}, {null}},
+            "nestedJson.location.country"},
+        {"nestedJson", "$.tags[0]", "STRING", new Object[][]{{"green"}, {"red"}, {null}},
+            "nestedJson.tags[0]"},
+        {"nestedJson", "$.events[0].country", "STRING", new Object[][]{{"US"}, {null}},
+            "nestedJson.events[0].country"}
+    };
+  }
+
+  /** 2.2 cases — `(column, jsonPath, expected distinct rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "sqlNullDefaultDistinctCases")
+  public static Object[][] sqlNullDefaultDistinctCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", new Object[][]{{"CA"}, {"US"}, {null}}, "flatJson.country NULL default"},
+        {"nestedJson", "$.location.country", new Object[][]{{"US"}, {null}},
+            "nestedJson.location.country NULL default"}
+    };
+  }
+
+  /** 2.3 cases — `(column, jsonPath, default literal, expected distinct rows ASC, label)`. */
+  @DataProvider(name = "defaultPrecedenceDistinctCases")
+  public static Object[][] defaultPrecedenceDistinctCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", "'foobar'",
+            new Object[][]{{"CA"}, {"US"}, {"foobar"}}, "flatJson.country default 'foobar'"},
+        {"nestedJson", "$.location.country", "'foobar'",
+            new Object[][]{{"US"}, {"foobar"}}, "nestedJson.location.country default 'foobar'"}
+    };
+  }
+
+  // ============================================================================================
+  // 3. GROUP BY
+  // ============================================================================================
+
+  /**
+   * 3.1 — NH on, 3-arg. Unresolved rows collapse into a single null group with the correct count,
+   * alongside the resolved value groups.
+   * <pre>
+   *   Example — `flatJson.country`: Result (counts ×2 for segment dup): ("CA", 2), ("US", 2), (NULL, 4)
+   * </pre>
+   */
+  @Test(dataProvider = "groupByCases")
+  public void testGroupByNullHandlingOn(String column, String jsonPath, String resultsType,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractScalar(%s, '%s', '%s')", column, jsonPath, resultsType);
+    givenCountryClickTable(true)
+        .whenQuery(String.format(
+            "SELECT %s AS v, COUNT(*) FROM clicks GROUP BY v ORDER BY v ASC NULLS LAST", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 3.2 — NH on, 4-arg with `NULL` literal. Same groups as 3.1 (null group with correct count). */
+  @Test(dataProvider = "sqlNullDefaultGroupByCases")
+  public void testGroupByWithSqlNullDefaultProducesNullGroup(String column, String jsonPath,
+      Object[][] expectedRows, String label) {
+    String expr = String.format("jsonExtractScalar(%s, '%s', 'STRING', NULL)", column, jsonPath);
+    givenCountryClickTable(true)
+        .whenQuery(String.format(
+            "SELECT %s AS v, COUNT(*) FROM clicks GROUP BY v ORDER BY v ASC NULLS LAST", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /**
+   * 3.3 — NH on, 4-arg with non-null default. Unresolved rows count toward the default's group,
+   * not a null group — so the result has no NULL group at all.
+   */
+  @Test(dataProvider = "defaultPrecedenceGroupByCases")
+  public void testGroupByDefaultBeatsNullHandlingPlaceholder(String column, String jsonPath,
+      String defaultLiteral, Object[][] expectedRows, String label) {
+    String expr = String.format(
+        "jsonExtractScalar(%s, '%s', 'STRING', %s)", column, jsonPath, defaultLiteral);
+    givenCountryClickTable(true)
+        .whenQuery(String.format(
+            "SELECT %s AS v, COUNT(*) FROM clicks GROUP BY v ORDER BY v ASC", expr))
+        .thenResultIs(expectedRows);
+  }
+
+  /** 3.4 — NH off: GROUP BY throws on unresolved rows. */
+  @Test
+  public void testGroupByNullHandlingOffThrows() {
+    try {
+      givenCountryClickTable(false)
+          .whenQuery("SELECT jsonExtractScalar(flatJson, '$.country', 'STRING'), COUNT(*) FROM clicks GROUP BY 1")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected GROUP BY to fail when null handling is off and rows are unresolved");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("Cannot resolve JSON path on some records");
+    }
+  }
+
+  /** 3.1 cases — `(column, jsonPath, resultsType, expected (value, count) rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "groupByCases")
+  public static Object[][] groupByCases() {
+    return new Object[][]{
+        // flatJson.country: row0="US", row1="CA", row2=null, row3=null -> 2 each + 4 null.
+        {"flatJson", "$.country", "STRING", new Object[][]{
+            {"CA", 2L}, {"US", 2L}, {null, 4L}
+        }, "flatJson.country"},
+        // flatJson.clicks INT: row0=5, row1=3, row2=null, row3=null -> 2 each + 4 null.
+        {"flatJson", "$.clicks", "INT", new Object[][]{
+            {3, 2L}, {5, 2L}, {null, 4L}
+        }, "flatJson.clicks INT"},
+        // nestedJson.location.country: row0="US", rows 1/2/3 unresolved -> 2 "US" + 6 null.
+        {"nestedJson", "$.location.country", "STRING", new Object[][]{
+            {"US", 2L}, {null, 6L}
+        }, "nestedJson.location.country"},
+        // nestedJson.tags[0]: row0="red", row1="green", rows 2/3 unresolved -> 2 each + 4 null.
+        {"nestedJson", "$.tags[0]", "STRING", new Object[][]{
+            {"green", 2L}, {"red", 2L}, {null, 4L}
+        }, "nestedJson.tags[0]"},
+        // nestedJson.events[0].country: row0="US", rows 1/2/3 unresolved -> 2 "US" + 6 null.
+        {"nestedJson", "$.events[0].country", "STRING", new Object[][]{
+            {"US", 2L}, {null, 6L}
+        }, "nestedJson.events[0].country"}
+    };
+  }
+
+  /** 3.2 cases — `(column, jsonPath, expected (value, count) rows ASC NULLS LAST, label)`. */
+  @DataProvider(name = "sqlNullDefaultGroupByCases")
+  public static Object[][] sqlNullDefaultGroupByCases() {
+    return new Object[][]{
+        {"flatJson", "$.country", new Object[][]{{"CA", 2L}, {"US", 2L}, {null, 4L}},
+            "flatJson.country NULL default"},
+        {"nestedJson", "$.location.country", new Object[][]{{"US", 2L}, {null, 6L}},
+            "nestedJson.location.country NULL default"}
+    };
+  }
+
+  /** 3.3 cases — `(column, jsonPath, default literal, expected (value, count) rows ASC, label)`. */
+  @DataProvider(name = "defaultPrecedenceGroupByCases")
+  public static Object[][] defaultPrecedenceGroupByCases() {
+    return new Object[][]{
+        // flatJson.country: 2 "CA" + 2 "US" + 4 'foobar'.
+        {"flatJson", "$.country", "'foobar'", new Object[][]{
+            {"CA", 2L}, {"US", 2L}, {"foobar", 4L}
+        }, "flatJson.country default 'foobar'"},
+        // nestedJson.location.country: 2 "US" + 6 'foobar'.
+        {"nestedJson", "$.location.country", "'foobar'", new Object[][]{
+            {"US", 2L}, {"foobar", 6L}
+        }, "nestedJson.location.country default 'foobar'"}
+    };
+  }
+
+  // ============================================================================================
+  // 4. Non-JSON input rejection
+  // ============================================================================================
+  // jsonExtractScalar invoked on a column whose values are not parseable as JSON is a caller /
+  // schema modeling error, not a missing value. The function must throw regardless of the null
+  // handling setting — coercing parse failures to SQL NULL would silently mask the bug. Null
+  // handling applies only to valid-JSON-with-missing-path / explicit-JSON-null cases (covered
+  // above), not to inputs that aren't JSON at all.
+
+  private static final Object[][] NON_JSON_FIXTURE = {{"value1"}, {"value2"}, {"value3"}, {"value4"}};
+
+  private FluentQueryTest.OnFirstInstance givenNonJsonStringTable(boolean nullHandling) {
+    Schema schema = new Schema.SchemaBuilder()
+        .setSchemaName("notJson")
+        .setEnableColumnBasedNullHandling(true)
+        .addDimensionField("plainString", DataType.STRING)
+        .build();
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("notJson").build();
+    return FluentQueryTest.withBaseDir(_baseDir)
+        .withNullHandling(nullHandling)
+        .givenTable(schema, tableConfig)
+        .onFirstInstance(NON_JSON_FIXTURE);
+  }
+
+  @Test
+  public void testNonJsonInputThrowsWithNullHandlingOn() {
+    try {
+      givenNonJsonStringTable(true)
+          .whenQuery("SELECT jsonExtractScalar(plainString, '$.path', 'STRING') FROM notJson")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected jsonExtractScalar on non-JSON input to throw even under null handling");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("not valid JSON");
+    }
+  }
+
+  @Test
+  public void testNonJsonInputThrowsWithNullHandlingOff() {
+    try {
+      givenNonJsonStringTable(false)
+          .whenQuery("SELECT jsonExtractScalar(plainString, '$.path', 'STRING') FROM notJson")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected jsonExtractScalar on non-JSON input to throw with null handling off");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("not valid JSON");
+    }
+  }
+
+  // A configured default value substitutes for a *missing* value (valid JSON, absent path, or null
+  // cell — see testDefaultValue). It must NOT mask non-JSON input: parse failure is thrown before the
+  // default cascade is reached, so the default never gets a chance to swallow a schema modeling error.
+  // Null handling on + a default is the most masking-prone config; if it throws here, it throws anywhere.
+  @Test
+  public void testNonJsonInputThrowsEvenWithDefaultValue() {
+    try {
+      givenNonJsonStringTable(true)
+          .whenQuery("SELECT jsonExtractScalar(plainString, '$.path', 'STRING', 'fallback') FROM notJson")
+          .thenResultIs(new Object[]{"unused"});
+      Assert.fail("Expected jsonExtractScalar on non-JSON input to throw even when a default value is set");
+    } catch (AssertionError e) {
+      Assertions.assertThat(e.getMessage()).contains("not valid JSON");
+    }
+  }
 }
diff --git a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
index 557a6a65fe..68080184c2 100644
--- a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
+++ b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
@@ -330,10 +330,13 @@ protected Iterator<Object[]> provideTestSqlWithExecutionException() {
     testCases.add(new Object[]{"SELECT jsonExtractKey(col1, 'path') FROM a", "was expecting (JSON String"});
     testCases.add(new Object[]{"SELECT json_extract_key(col1, 'path') FROM a", "was expecting (JSON String"});
 
-    //  - PlaceholderScalarFunction registered will throw on intermediate stage, but works on leaf stage.
-    //    - checked "Illegal Json Path" as col1 is not actually a json string, but the call is correctly triggered.
-    testCases.add(
-        new Object[]{"SELECT CAST(jsonExtractScalar(col1, 'path', 'INT') AS INT) FROM a", "Cannot resolve JSON path"});
+    //  - col1 is a non-JSON STRING column. jsonExtractScalar must reject the input with a clear
+    //    caller-error message rather than silently coercing it to NULL — null handling applies only
+    //    to valid-JSON-with-missing-path, not to input that isn't JSON at all.
+    testCases.add(new Object[]{
+        "SELECT CAST(jsonExtractScalar(col1, 'path', 'INT') AS INT) FROM a",
+        "jsonExtractScalar received input that is not valid JSON"
+    });
     //    - checked function cannot be found b/c there's no intermediate stage impl for json_extract_scalar
     testCases.add(new Object[]{
         "SELECT CAST(json_extract_scalar(a.col1, b.col2, 'INT') AS INT) FROM a JOIN b ON a.col1 = b.col1",