diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValue.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValue.kt
index d84f6b6136fb..0d95b18fe11f 100644
--- a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValue.kt
+++ b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValue.kt
@@ -18,6 +18,9 @@ import java.math.BigDecimal
  */
 class JsonToAirbyteValue {
     fun convert(json: JsonNode, schema: AirbyteType): AirbyteValue {
+        if (json.isNull) {
+            return NullValue
+        }
         try {
             return when (schema) {
                 is ArrayType -> toArray(json, schema.items.type)
@@ -103,6 +106,11 @@ class JsonToAirbyteValue {
         return ObjectValue(
             values =
                 schema.properties
+                    // Note that this will create an ObjectValue where properties in the schema
+                    // might not exist in the value.
+                    // This matches JSON behavior (i.e. explicit null != property not set),
+                    // but we maybe would prefer to set an explicit NullValue.
+                    .filter { (name, _) -> json.has(name) }
                     .mapValues { (name, field) -> convert(json.get(name), field.type) }
                     .toMap(LinkedHashMap())
         )
diff --git a/airbyte-cdk/bulk/core/load/src/test/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValueTest.kt b/airbyte-cdk/bulk/core/load/src/test/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValueTest.kt
index 58019e3c2431..9b5fa2cc35a1 100644
--- a/airbyte-cdk/bulk/core/load/src/test/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValueTest.kt
+++ b/airbyte-cdk/bulk/core/load/src/test/kotlin/io/airbyte/cdk/load/data/JsonToAirbyteValueTest.kt
@@ -5,6 +5,7 @@
 package io.airbyte.cdk.load.data
 
 import com.fasterxml.jackson.databind.node.JsonNodeFactory
+import io.airbyte.cdk.util.Jsons
 import java.math.BigDecimal
 import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.Test
@@ -160,4 +161,24 @@ class JsonToAirbyteValueTest {
         Assertions.assertTrue(value is TimeValue)
         Assertions.assertEquals("00:00:00", (value as TimeValue).value)
     }
+
+    @Test
+    fun testMissingObjectField() {
+        val value =
+            JsonToAirbyteValue()
+                .convert(
+                    Jsons.readTree("""{"foo": 1}"""),
+                    ObjectType(
+                        properties =
+                            linkedMapOf(
+                                "foo" to FieldType(IntegerType, nullable = true),
+                                "bar" to FieldType(IntegerType, nullable = true),
+                            )
+                    )
+                )
+        Assertions.assertEquals(
+            ObjectValue(linkedMapOf("foo" to IntegerValue(1))),
+            value,
+        )
+    }
 }