diff --git a/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/processors/duckdb.py b/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/processors/duckdb.py index 16163a22bb1d..03cf1af1f3ce 100644 --- a/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/processors/duckdb.py +++ b/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>`) 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 diff --git a/airbyte-integrations/connectors/destination-motherduck/integration_tests/integration_test.py b/airbyte-integrations/connectors/destination-motherduck/integration_tests/integration_test.py index 0a356d67d5f8..7cd6f3e70470 100644 --- a/airbyte-integrations/connectors/destination-motherduck/integration_tests/integration_test.py +++ b/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( diff --git a/airbyte-integrations/connectors/destination-motherduck/metadata.yaml b/airbyte-integrations/connectors/destination-motherduck/metadata.yaml index fccb944a6f83..4f95338c3d76 100644 --- a/airbyte-integrations/connectors/destination-motherduck/metadata.yaml +++ b/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 diff --git a/airbyte-integrations/connectors/destination-motherduck/pyproject.toml b/airbyte-integrations/connectors/destination-motherduck/pyproject.toml index 8780be96ccc7..6d27a2d6d768 100644 --- a/airbyte-integrations/connectors/destination-motherduck/pyproject.toml +++ b/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" diff --git a/airbyte-integrations/connectors/destination-motherduck/unit_tests/test_serialize_object_columns.py b/airbyte-integrations/connectors/destination-motherduck/unit_tests/test_serialize_object_columns.py new file mode 100644 index 000000000000..a6e18f1cf5ae --- /dev/null +++ b/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>`, 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() == [("[{}]",)] diff --git a/docs/integrations/destinations/motherduck.md b/docs/integrations/destinations/motherduck.md index 333432607a55..6bd7e3fa3b0b 100644 --- a/docs/integrations/destinations/motherduck.md +++ b/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` |