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 03cf1af1f3ce..8d65d5cb2e47 100644 --- a/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/processors/duckdb.py +++ b/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: 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 7cd6f3e70470..b2a499813540 100644 --- a/airbyte-integrations/connectors/destination-motherduck/integration_tests/integration_test.py +++ b/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( diff --git a/airbyte-integrations/connectors/destination-motherduck/metadata.yaml b/airbyte-integrations/connectors/destination-motherduck/metadata.yaml index 2b5e22e706a9..935ee499154f 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.5 + dockerImageTag: 0.2.6 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 908c7c6d8c5f..e37143001150 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.5" +version = "0.2.6" 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 index a6e18f1cf5ae..ff6c8711b7eb 100644 --- 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 @@ -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>>`, + 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":{}}]',)] diff --git a/docs/integrations/destinations/motherduck.md b/docs/integrations/destinations/motherduck.md index e61a5abaff30..c06cd432e435 100644 --- a/docs/integrations/destinations/motherduck.md +++ b/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 |