fix(destination-motherduck): handle camel cased object/array columns

airbyte-integrations/connectors/destination-motherduck/destination_motherduck/processors/duckdb.py

@@ -20,6 +20,7 @@
 
 from airbyte_cdk import DestinationSyncMode
 from airbyte_cdk.sql import exceptions as exc
+from airbyte_cdk.sql._util.name_normalizers import NameNormalizerBase
 from airbyte_cdk.sql.constants import AB_EXTRACTED_AT_COLUMN, DEBUG_MODE
 from airbyte_cdk.sql.secrets import SecretString
 from airbyte_cdk.sql.shared.sql_processor import SqlConfig, SqlProcessorBase, SQLRuntimeError
@@ -38,6 +39,7 @@
 def _serialize_object_columns(
     buffer_data: Dict[str, List[Any]],
     json_schema: dict,
+    normalizer: type[NameNormalizerBase],
 ) -> Dict[str, List[Any]]:
     """
     Convert columns stored as JSON in the destination into JSON strings. This
@@ -51,7 +53,9 @@ def _serialize_object_columns(
     string type for these columns, which DuckDB converts back to JSON on import
     because the destination column type is JSON.
     """
-    properties = json_schema.get("properties", {})
+    # Buffer keys are normalized column names (see _get_sql_column_definitions), while schema
+    # properties carry the source's original names
+    properties = {normalizer.normalize(name): prop for name, prop in json_schema.get("properties", {}).items()}
     result = {}
 
     for col_name, values in buffer_data.items():
@@ -398,7 +402,9 @@ def write_stream_data_from_buffer(
         temp_table_name = self._create_table_for_loading(stream_name, batch_id=None)
         try:
             serialized_buffer = _serialize_object_columns(
-                buffer[stream_name], self.catalog_provider.get_configured_stream_info(stream_name).stream.json_schema
+                buffer[stream_name],
+                self.catalog_provider.get_configured_stream_info(stream_name).stream.json_schema,
+                self.normalizer,
             )
             pa_table = pa.Table.from_pydict(serialized_buffer)
         except Exception:

airbyte-integrations/connectors/destination-motherduck/integration_tests/integration_test.py

@@ -131,6 +131,7 @@ def table_schema() -> str:
             },
             "empty_object_key": {"type": ["object"]},
             "array_of_objects_key": {"type": ["null", "array"], "items": {"type": "object"}},
+            "ArrayOfObjectsUpperCase": {"type": ["null", "array"], "items": {"type": "object"}},
         },
     }
     return schema
@@ -256,6 +257,7 @@ def airbyte_message1(test_table_name: str):
                 "object_key": {},
                 "empty_object_key": {},
                 "array_of_objects_key": [{}],
+                "ArrayOfObjectsUpperCase": [{"Amount": 100.0, "SubTotalLineDetail": {}}],
             },
             emitted_at=int(datetime.now().timestamp()) * 1000,
         ),
@@ -276,6 +278,7 @@ def airbyte_message2(test_table_name: str):
                 "object_key": {},
                 "empty_object_key": {"a": {}},
                 "array_of_objects_key": [{"a": 1}, {}],
+                "ArrayOfObjectsUpperCase": [{}],
             },
             emitted_at=int(datetime.now().timestamp()) * 1000,
         ),
@@ -414,7 +417,7 @@ def test_write(
     assert len(result) == 1
 
     sql_result = sql_processor._execute_sql(
-        "SELECT key1, keyuppercase, object_key, empty_object_key, array_of_objects_key, "
+        "SELECT key1, keyuppercase, object_key, empty_object_key, array_of_objects_key, arrayofobjectsuppercase, "
         "_airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta "
         f"FROM {test_schema_name}.{test_table_name} ORDER BY key1"
     )
@@ -430,6 +433,8 @@ def test_write(
     assert sql_result[1][3] == {}
     assert sql_result[0][4] == [{"a": 1}, {}]
     assert sql_result[1][4] == [{}]
+    assert sql_result[0][5] == [{}]
+    assert sql_result[1][5] == [{"Amount": 100.0, "SubTotalLineDetail": {}}]
 
 
 def test_write_dupe(

airbyte-integrations/connectors/destination-motherduck/metadata.yaml

@@ -4,7 +4,7 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: 042ee9b5-eb98-4e99-a4e5-3f0d573bee66
-  dockerImageTag: 0.2.5
+  dockerImageTag: 0.2.6
   dockerRepository: airbyte/destination-motherduck
   githubIssueLabel: destination-motherduck
   icon: duckdb.svg

airbyte-integrations/connectors/destination-motherduck/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "airbyte-destination-motherduck"
-version = "0.2.5"
+version = "0.2.6"
 description = "Destination implementation for MotherDuck."
 authors = ["Guen Prawiroatmodjo, Simon Späti, Airbyte"]
 license = "ELv2"

airbyte-integrations/connectors/destination-motherduck/unit_tests/test_serialize_object_columns.py

@@ -6,6 +6,8 @@
 import pytest
 from destination_motherduck.processors.duckdb import _serialize_object_columns
 
+from airbyte_cdk.sql._util.name_normalizers import LowerCaseNormalizer
+
 
 JSON_SCHEMA = {
     "type": "object",
@@ -15,6 +17,7 @@
         "obj": {"type": ["null", "object"]},
         "array_of_objects": {"type": ["null", "array"], "items": {"type": "object"}},
         "array_of_scalars": {"type": ["null", "array"], "items": {"type": "string"}},
+        "Line": {"type": ["null", "array"], "items": {"type": "object"}},
     },
 }
 
@@ -38,10 +41,11 @@
             id="array_of_scalars_serialized",
         ),
         pytest.param("not_in_schema", [{"x": 1}], [{"x": 1}], id="airbyte_column_untouched"),
+        pytest.param("line", [[{}], None], ["[{}]", None], id="pascal_case_property_matched_via_normalization"),
     ],
 )
 def test_serialize_object_columns(col_name, values, expected) -> None:
-    result = _serialize_object_columns({col_name: values}, JSON_SCHEMA)
+    result = _serialize_object_columns({col_name: values}, JSON_SCHEMA, LowerCaseNormalizer)
     assert result[col_name] == expected
 
 
@@ -55,7 +59,7 @@ def test_serialize_object_columns_prevents_empty_struct_error() -> None:
     """
     buffer_data = {"id": ["1"], "array_of_objects": [[{}]]}
 
-    serialized = _serialize_object_columns(buffer_data, JSON_SCHEMA)
+    serialized = _serialize_object_columns(buffer_data, JSON_SCHEMA, LowerCaseNormalizer)
     pa_table = pa.Table.from_pydict(serialized)
 
     # The column must be a string, not a list-of-struct type.
@@ -65,3 +69,23 @@ def test_serialize_object_columns_prevents_empty_struct_error() -> None:
     con = duckdb.connect()
     con.register("buf", pa_table)
     assert con.execute("SELECT array_of_objects FROM buf").fetchall() == [("[{}]",)]
+
+
+def test_serialize_object_columns_normalized_column_names() -> None:
+    """Regression test for the empty STRUCT failure.
+
+    Buffer keys are normalized column names ("line"), but the schema property is the source's
+    original name ("Line"). Matching on the raw property names skips serialization entirely, so
+    `[{"SubTotalLineDetail": {}}]` reached PyArrow as `list<struct<SubTotalLineDetail: struct<>>>`,
+    which DuckDB rejects with "Attempted to convert a STRUCT with no fields to DuckDB".
+    """
+    buffer_data = {"id": ["1"], "line": [[{"Amount": 100.0, "SubTotalLineDetail": {}}]]}
+
+    serialized = _serialize_object_columns(buffer_data, JSON_SCHEMA, LowerCaseNormalizer)
+    pa_table = pa.Table.from_pydict(serialized)
+
+    assert pa.types.is_string(pa_table.schema.field("line").type)
+
+    con = duckdb.connect()
+    con.register("buf", pa_table)
+    assert con.execute("SELECT line FROM buf").fetchall() == [('[{"Amount":100.0,"SubTotalLineDetail":{}}]',)]

docs/integrations/destinations/motherduck.md

@@ -112,6 +112,7 @@ If you only need local DuckDB files, consider using the [DuckDB destination](/in
 
 | Version | Date | Pull Request | Subject |
 | :------ | :--- | :----------- | :------ |
+| 0.2.6 | 2026-08-04 | [83694](https://github.com/airbytehq/airbyte/pull/83694) | Fix JSON column serialization being skipped for non-lowercase source property names, causing empty STRUCT failures |
 | 0.2.5 | 2026-07-22 | [82244](https://github.com/airbytehq/airbyte/pull/82244) | Fix sync failures on array fields containing empty objects by serializing JSON array columns before load |
 | 0.2.4 | 2026-07-14 | [81511](https://github.com/airbytehq/airbyte/pull/81511) | Fix silent data loss on multi-stream syncs by no longer discarding other streams' buffered records when one stream is flushed |
 | 0.2.3 | 2026-03-31 | [75645](https://github.com/airbytehq/airbyte/pull/75645) | Bump version to force registry update for `supportLevel` change to certified |