File-based CDK: allow to merge schemas with nullable object values (#37773)

Author: Anton Karpets

airbyte-cdk/python/airbyte_cdk/sources/file_based/schema_helpers.py

@@ -104,9 +104,7 @@ def merge_schemas(schema1: SchemaType, schema2: SchemaType) -> SchemaType:
         t1 = merged_schema.get(k2)
         if t1 is None:
             merged_schema[k2] = t2
-        # do not pass fields with null values to the _choose_wider_type method if their keys are already
-        # in the merged schema due to issue observed in https://github.com/airbytehq/oncall/issues/4948
-        elif t1 == t2 or t2["type"] == "null":
+        elif t1 == t2:
             continue
         else:
             merged_schema[k2] = _choose_wider_type(k2, t1, t2)
@@ -119,23 +117,27 @@ def _is_valid_type(t: JsonSchemaSupportedType) -> bool:
 
 
 def _choose_wider_type(key: str, t1: Mapping[str, Any], t2: Mapping[str, Any]) -> Mapping[str, Any]:
-    if (t1["type"] == "array" or t2["type"] == "array") and t1 != t2:
+    t1_type = t1["type"]
+    t2_type = t2["type"]
+
+    if (t1_type == "array" or t2_type == "array") and t1 != t2:
         raise SchemaInferenceError(
             FileBasedSourceError.SCHEMA_INFERENCE_ERROR,
             details="Cannot merge schema for unequal array types.",
             key=key,
             detected_types=f"{t1},{t2}",
         )
-    elif (t1["type"] == "object" or t2["type"] == "object") and t1 != t2:
+    # Schemas can still be merged if a key contains a null value in either t1 or t2, but it is still an object
+    elif (t1_type == "object" or t2_type == "object") and t1_type != "null" and t2_type != "null" and t1 != t2:
         raise SchemaInferenceError(
             FileBasedSourceError.SCHEMA_INFERENCE_ERROR,
             details="Cannot merge schema for unequal object types.",
             key=key,
             detected_types=f"{t1},{t2}",
         )
     else:
-        comparable_t1 = get_comparable_type(TYPE_PYTHON_MAPPING[t1["type"]][0])  # accessing the type_mapping value
-        comparable_t2 = get_comparable_type(TYPE_PYTHON_MAPPING[t2["type"]][0])  # accessing the type_mapping value
+        comparable_t1 = get_comparable_type(TYPE_PYTHON_MAPPING[t1_type][0])  # accessing the type_mapping value
+        comparable_t2 = get_comparable_type(TYPE_PYTHON_MAPPING[t2_type][0])  # accessing the type_mapping value
         if not comparable_t1 and comparable_t2:
             raise SchemaInferenceError(FileBasedSourceError.UNRECOGNIZED_TYPE, key=key, detected_types=f"{t1},{t2}")
         return max(

airbyte-cdk/python/unit_tests/sources/file_based/test_schema_helpers.py

@@ -264,13 +264,20 @@ def test_comparable_types() -> None:
             {"a": {"type": "object"}},
             {"a": {"type": "null"}},
             {"a": {"type": "object"}},
-            id="single-key-with-null-object-schema2"),
+            id="single-key-with-null-object-schema2",
+        ),
         pytest.param(
             {"a": {"type": "object"}},
             {"b": {"type": "null"}},
             {"a": {"type": "object"}, "b": {"type": "null"}},
             id="new-key-with-null-type",
         ),
+        pytest.param(
+            {"a": {"type": "null"}},
+            {"a": {"type": "object"}},
+            {"a": {"type": "object"}},
+            id="single-key-with-null-object-schema1",
+        ),
     ],
 )
 def test_merge_schemas(schema1: SchemaType, schema2: SchemaType, expected_result: Optional[SchemaType]) -> None: