fix(destination-motherduck): serialize JSON array columns to avoid empty STRUCT failure

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

@@ -16,12 +16,14 @@
 from pydantic import Field
 from sqlalchemy import Executable, TextClause, create_engine, text
 from sqlalchemy.exc import ProgrammingError, SQLAlchemyError
+from sqlalchemy.types import JSON
 
 from airbyte_cdk import DestinationSyncMode
 from airbyte_cdk.sql import exceptions as exc
 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
+from airbyte_cdk.sql.types import SQLTypeConverter
 
 
 if TYPE_CHECKING:
@@ -38,12 +40,16 @@ def _serialize_object_columns(
     json_schema: dict,
 ) -> Dict[str, List[Any]]:
     """
-    Convert object-fields columns into JSON strings. This prevents PyArrow from
-    inferring struct types, which can cause issues with empty structs when the
-    data contains empty dicts {}. PyArrow will then infer a string type for
-    these columns that will convert to JSON again once it is imported into
-    DuckDB because in the destination schema, object-fields columns have JSON
-    as their type.
+    Convert columns stored as JSON in the destination into JSON strings. This
+    prevents PyArrow from inferring struct or list-of-struct types, which break
+    when the data contains empty dicts {} or lists of empty dicts [{}]: PyArrow
+    infers a fieldless `struct<>` (or `list<struct<>>`) type that DuckDB rejects
+    with "Attempted to convert a STRUCT with no fields to DuckDB".
+
+    Both `object` and `array` JSON-schema types map to the SQL JSON type in the
+    CDK type mapping, so both are pre-serialized here. PyArrow then infers a
+    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", {})
     result = {}
@@ -54,17 +60,17 @@ def _serialize_object_columns(
         if col_name not in properties:  # Probably an "Airbyte column"
             continue
 
-        schema_type = properties[col_name].get("type")
-        if isinstance(schema_type, list):  # For nullable types this is a list. We want the first type that is not null
-            schema_type = next((t for t in schema_type if t != "null"), None)
-
-        if schema_type != "object":  # This only applies to properties of type "object"
+        # Serialize the column only if its destination SQL type is JSON. This covers both
+        # `object` and `array` (of objects or scalars) properties while leaving e.g. vector
+        # arrays, which map to a native SQL ARRAY type, untouched.
+        sql_type = SQLTypeConverter().to_sql_type(properties[col_name])
+        if not isinstance(sql_type, JSON):
             continue
 
-        # Convert dicts to JSON strings. Note that `values` is a list of column values. Since Airbyte works with JSON
-        # schemas, we do not have the issue of e.g. having to convert datetimes objects - these will just be passed as
-        # formatted date-time strings.
-        result[col_name] = [orjson.dumps(v).decode() if isinstance(v, dict) else v for v in values]
+        # Convert dicts and lists to JSON strings. Note that `values` is a list of column values. Since Airbyte works
+        # with JSON schemas, we do not have the issue of e.g. having to convert datetime objects - these will just be
+        # passed as formatted date-time strings. `None` and already-serialized values are left as-is.
+        result[col_name] = [orjson.dumps(v).decode() if isinstance(v, (dict, list)) else v for v in values]
 
     return result
 

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

@@ -130,6 +130,7 @@ def table_schema() -> str:
                 },
             },
             "empty_object_key": {"type": ["object"]},
+            "array_of_objects_key": {"type": ["null", "array"], "items": {"type": "object"}},
         },
     }
     return schema
@@ -254,6 +255,7 @@ def airbyte_message1(test_table_name: str):
                 "keyUpperCase": str(fake.ssn()),
                 "object_key": {},
                 "empty_object_key": {},
+                "array_of_objects_key": [{}],
             },
             emitted_at=int(datetime.now().timestamp()) * 1000,
         ),
@@ -273,6 +275,7 @@ def airbyte_message2(test_table_name: str):
                 "keyUpperCase": str(fake.ssn()),
                 "object_key": {},
                 "empty_object_key": {"a": {}},
+                "array_of_objects_key": [{"a": 1}, {}],
             },
             emitted_at=int(datetime.now().timestamp()) * 1000,
         ),
@@ -292,6 +295,7 @@ def airbyte_message2_update(airbyte_message2: AirbyteMessage, test_table_name: s
                 "keyUpperCase": str(fake.ssn()),
                 "object_key": {},
                 "empty_object_key": {},
+                "array_of_objects_key": [],
             },
             emitted_at=int(datetime.now().timestamp()) * 1000,
         ),
@@ -410,7 +414,8 @@ def test_write(
     assert len(result) == 1
 
     sql_result = sql_processor._execute_sql(
-        "SELECT key1, keyuppercase, object_key, empty_object_key, _airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta "
+        "SELECT key1, keyuppercase, object_key, empty_object_key, array_of_objects_key, "
+        "_airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta "
         f"FROM {test_schema_name}.{test_table_name} ORDER BY key1"
     )
 
@@ -423,6 +428,8 @@ def test_write(
     assert sql_result[1][2] == {}
     assert sql_result[0][3] == {"a": {}}
     assert sql_result[1][3] == {}
+    assert sql_result[0][4] == [{"a": 1}, {}]
+    assert sql_result[1][4] == [{}]
 
 
 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.4
+  dockerImageTag: 0.2.5-rc.1
   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.4"
+version = "0.2.5-rc.1"
 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

@@ -0,0 +1,67 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+from __future__ import annotations
+
+import duckdb
+import pyarrow as pa
+import pytest
+from destination_motherduck.processors.duckdb import _serialize_object_columns
+
+
+JSON_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "id": {"type": ["null", "string"]},
+        "count": {"type": ["null", "integer"]},
+        "obj": {"type": ["null", "object"]},
+        "array_of_objects": {"type": ["null", "array"], "items": {"type": "object"}},
+        "array_of_scalars": {"type": ["null", "array"], "items": {"type": "string"}},
+    },
+}
+
+
+@pytest.mark.parametrize(
+    "col_name, values, expected",
+    [
+        pytest.param("id", ["a", None], ["a", None], id="scalar_string_untouched"),
+        pytest.param("count", [1, None], [1, None], id="scalar_integer_untouched"),
+        pytest.param("obj", [{}, {"x": 1}, None], ["{}", '{"x":1}', None], id="object_column_serialized"),
+        pytest.param(
+            "array_of_objects",
+            [[{}], [{"a": 1}], None],
+            ["[{}]", '[{"a":1}]', None],
+            id="array_of_empty_objects_serialized",
+        ),
+        pytest.param(
+            "array_of_scalars",
+            [["a", "b"], [], None],
+            ['["a","b"]', "[]", None],
+            id="array_of_scalars_serialized",
+        ),
+        pytest.param("not_in_schema", [{"x": 1}], [{"x": 1}], id="airbyte_column_untouched"),
+    ],
+)
+def test_serialize_object_columns(col_name, values, expected) -> None:
+    result = _serialize_object_columns({col_name: values}, JSON_SCHEMA)
+    assert result[col_name] == expected
+
+
+def test_serialize_object_columns_prevents_empty_struct_error() -> None:
+    """Regression test for empty STRUCT failure (oncall #13118).
+
+    A `type: array` column whose items are objects, containing an empty object `[{}]`, previously
+    reached PyArrow un-serialized and was inferred as `list<struct<>>`, which DuckDB rejects with
+    "Attempted to convert a STRUCT with no fields to DuckDB". Serializing it to a JSON string first
+    avoids the fieldless struct.
+    """
+    buffer_data = {"id": ["1"], "array_of_objects": [[{}]]}
+
+    serialized = _serialize_object_columns(buffer_data, JSON_SCHEMA)
+    pa_table = pa.Table.from_pydict(serialized)
+
+    # The column must be a string, not a list-of-struct type.
+    assert pa.types.is_string(pa_table.schema.field("array_of_objects").type)
+
+    # Registering the table in DuckDB must not raise the empty-struct error.
+    con = duckdb.connect()
+    con.register("buf", pa_table)
+    assert con.execute("SELECT array_of_objects FROM buf").fetchall() == [("[{}]",)]

docs/integrations/destinations/motherduck.md

@@ -75,6 +75,7 @@ This destination supports [namespaces](https://docs.airbyte.com/platform/using-a
 
 | Version | Date | Pull Request | Subject |
 | :------ | :--- | :----------- | :------ |
+| 0.2.5-rc.1 | 2026-07-16 | [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-08 | [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 |
 | 0.2.2 | 2025-02-02 | [70438](https://github.com/airbytehq/airbyte/pull/70438) | Fix for camelCase columns being `NULL` |