diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index bcfb300ba1..d9beff52b1 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import functools
+import json
 import logging
 import re
 import sys
@@ -1649,7 +1650,11 @@ def _transform_value(value: t.Any, dtype: exp.DataType) -> t.Any:
             return exp.Struct(expressions=expressions)
 
         if dtype.is_type(exp.DataType.Type.JSON):
-            return exp.func("PARSE_JSON", f"'{value}'")
+            if value is None or isinstance(value, exp.Null):
+                return exp.null()
+            if isinstance(value, (dict, list, tuple, bool, int, float)):
+                value = json.dumps(value)
+            return exp.func("PARSE_JSON", exp.Literal.string(str(value)))
 
         return exp.convert(value)
 
diff --git a/tests/core/test_dialect.py b/tests/core/test_dialect.py
index 8d19ffee36..5ce35ff51d 100644
--- a/tests/core/test_dialect.py
+++ b/tests/core/test_dialect.py
@@ -1116,6 +1116,26 @@ def test_select_from_values_for_batch_range_json():
     )
 
 
+def test_select_from_values_for_batch_range_json_values():
+    values = [
+        (1, None),
+        (2, {"name": "O'Brien", "tags": ["a"]}),
+        (3, '{"name": "O\'Brien"}'),
+    ]
+    columns_to_types = {
+        "id": exp.DataType.build("int"),
+        "json_col": exp.DataType.build("json"),
+    }
+
+    assert select_from_values_for_batch_range(values, columns_to_types, 0, len(values)).sql() == (
+        "SELECT CAST(id AS INT) AS id, CAST(json_col AS JSON) AS json_col "
+        "FROM (VALUES (1, CAST(NULL AS JSON)), "
+        """(2, PARSE_JSON('{"name": "O''Brien", "tags": ["a"]}')), """
+        """(3, PARSE_JSON('{"name": "O''Brien"}'))) """
+        "AS t(id, json_col)"
+    )
+
+
 def test_select_from_values_that_include_null():
     values = [(1, exp.null())]
     columns_to_types = {
diff --git a/tests/core/test_test.py b/tests/core/test_test.py
index 0fbe8e926c..b6a3fa8d21 100644
--- a/tests/core/test_test.py
+++ b/tests/core/test_test.py
@@ -1254,6 +1254,38 @@ def test_empty_rows(sushi_context: Context) -> None:
     )
 
 
+def test_json_column_missing_from_input_rows(sushi_context: Context) -> None:
+    _check_successful_or_raise(
+        _create_test(
+            body=load_yaml(
+                """
+test_a:
+  model: a
+  inputs:
+    b:
+      columns:
+        id: int
+        payload: json
+      rows:
+        - id: 1
+  outputs:
+    query:
+      - id: 1
+        payload_is_null: true
+                """
+            ),
+            test_name="test_a",
+            model=sushi_context.upsert_model(
+                _create_model(
+                    "SELECT id, payload IS NULL AS payload_is_null FROM b",
+                    default_catalog="memory",
+                )
+            ),
+            context=Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))),
+        ).run()
+    )
+
+
 @pytest.mark.parametrize("full_model_without_ctes", ["snowflake"], indirect=True)
 def test_normalization(full_model_without_ctes: SqlModel) -> None:
     normalized_body = _create_test(