diff --git a/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/destination.py b/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/destination.py index d67d8dbe18a8..fbb5c2ec09b3 100644 --- a/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/destination.py +++ b/airbyte-integrations/connectors/destination-motherduck/destination_motherduck/destination.py @@ -229,6 +229,7 @@ def write( db_path=path, motherduck_token=motherduck_api_key, ) + normalizer = self.normalizer() for configured_stream in configured_catalog.streams: processor.prepare_stream_table(stream_name=configured_stream.stream.name, sync_mode=configured_stream.destination_sync_mode) @@ -273,11 +274,36 @@ def write( if stream_name not in streams: logger.debug(f"Stream {stream_name} was not present in configured streams, skipping") continue + + # The data here has the original column names from the source, but _get_sql_column_definitions() below + # returns the normalized schema. So to match the right fields in data with the normalized schema, we + # need to map the normalized keys back to the keys in the data dictionary here. + normalized_keys = {normalizer.normalize(key): key for key in data.keys()} + + if len(normalized_keys) < len(data): + # Because we find the key in the data dictionary through the normalized_key mapping, + # only the values in the normalized_keys dict will get pulled from the data. + # So all the keys in the data that are NOT in the values of the normalized_keys would get skipped, + # hence we log those fields. + logger.warning( + "Data contained duplicate keys after normalization: keys %s were dropped. Make sure " + "the column names in the source data stay unique after applying these operations: \n" + " - Converts ASCII letters to lowercase\n" + " - Replaces whitespace with underscores\n" + " - Preserves Unicode letters and numbers\n" + " - Adds underscore prefix if name starts with ASCII digit\n" + " - Replaces other special characters with underscores\n" + "skipping", + set(data) - set(normalized_keys.values()), + ) + continue + # add to buffer record_meta: dict[str, str] = {} for column_name in processor._get_sql_column_definitions(stream_name): - if column_name in data: - buffer[stream_name][column_name].append(data[column_name]) + if column_name in normalized_keys.keys(): + # Find the key in the data dictionary through the mapping for this (normalized) column name. + buffer[stream_name][column_name].append(data[normalized_keys[column_name]]) elif column_name not in AB_INTERNAL_COLUMNS: buffer[stream_name][column_name].append(None) @@ -334,6 +360,7 @@ def _flush_buffer( processor = self._get_sql_processor( configured_catalog=configured_catalog, schema_name=schema_name, db_path=db_path, motherduck_token=motherduck_api_key ) + processor.write_stream_data_from_buffer(buffer, configured_stream.stream.name, configured_stream.destination_sync_mode) def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus: 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 37e87786e619..0a356d67d5f8 100644 --- a/airbyte-integrations/connectors/destination-motherduck/integration_tests/integration_test.py +++ b/airbyte-integrations/connectors/destination-motherduck/integration_tests/integration_test.py @@ -102,6 +102,11 @@ def other_test_table_name(test_table_name) -> str: return test_table_name + "_other" +@pytest.fixture(scope="module") +def duplicate_column_name_test_table_name(test_table_name) -> str: + return test_table_name + "_broken" + + @pytest.fixture def test_large_table_name() -> str: letters = string.ascii_lowercase @@ -115,7 +120,7 @@ def table_schema() -> str: "type": "object", "properties": { "key1": {"type": ["null", "string"]}, - "key2": {"type": ["null", "string"]}, + "keyUpperCase": {"type": ["null", "string"]}, "object_key": { "type": ["object"], "properties": { @@ -142,13 +147,27 @@ def other_table_schema() -> str: return schema +@pytest.fixture +def duplicate_column_name_table_schema() -> str: + schema = { + "type": "object", + "properties": { + "uppercase": {"type": ["null", "string"]}, + "upperCase": {"type": ["null", "string"]}, + }, + } + return schema + + @pytest.fixture def configured_catalogue( test_table_name: str, other_test_table_name: str, + duplicate_column_name_test_table_name: str, test_large_table_name: str, table_schema: str, other_table_schema: str, + duplicate_column_name_table_schema: str, ) -> ConfiguredAirbyteCatalog: append_stream = ConfiguredAirbyteStream( stream=AirbyteStream( @@ -170,6 +189,16 @@ def configured_catalogue( destination_sync_mode=DestinationSyncMode.append, primary_key=[["key3"]], ) + duplicate_column_name_append_stream = ConfiguredAirbyteStream( + stream=AirbyteStream( + name=duplicate_column_name_test_table_name, + json_schema=duplicate_column_name_table_schema, + supported_sync_modes=[SyncMode.full_refresh, SyncMode.incremental], + ), + sync_mode=SyncMode.incremental, + destination_sync_mode=DestinationSyncMode.append, + primary_key=[["uppercase"]], + ) append_stream_large = ConfiguredAirbyteStream( stream=AirbyteStream( name=test_large_table_name, @@ -222,7 +251,7 @@ def airbyte_message1(test_table_name: str): stream=test_table_name, data={ "key1": fake.unique.first_name(), - "key2": str(fake.ssn()), + "keyUpperCase": str(fake.ssn()), "object_key": {}, "empty_object_key": {}, }, @@ -241,7 +270,7 @@ def airbyte_message2(test_table_name: str): stream=test_table_name, data={ "key1": fake.unique.first_name(), - "key2": str(fake.ssn()), + "keyUpperCase": str(fake.ssn()), "object_key": {}, "empty_object_key": {"a": {}}, }, @@ -260,7 +289,7 @@ def airbyte_message2_update(airbyte_message2: AirbyteMessage, test_table_name: s stream=test_table_name, data={ "key1": airbyte_message2.record.data["key1"], - "key2": str(fake.ssn()), + "keyUpperCase": str(fake.ssn()), "object_key": {}, "empty_object_key": {}, }, @@ -302,6 +331,20 @@ def airbyte_message5(other_test_table_name: str): ) +@pytest.fixture +def duplicate_column_name_airbyte_message(test_table_name: str): + fake = Faker() + Faker.seed(0) + return AirbyteMessage( + type=Type.RECORD, + record=AirbyteRecordMessage( + stream=test_table_name, + data={"uppercase": fake.unique.first_name(), "upperCase": str(fake.ssn())}, + emitted_at=int(datetime.now().timestamp()) * 1000, + ), + ) + + @pytest.mark.disable_autouse def test_check_fails(invalid_config, request): destination = DestinationMotherDuck() @@ -367,7 +410,7 @@ def test_write( assert len(result) == 1 sql_result = sql_processor._execute_sql( - "SELECT key1, key2, object_key, empty_object_key, _airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta " + "SELECT key1, keyuppercase, object_key, empty_object_key, _airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta " f"FROM {test_schema_name}.{test_table_name} ORDER BY key1" ) @@ -405,7 +448,7 @@ def test_write_dupe( assert len(result) == 1 sql_result = sql_processor._execute_sql( - "SELECT key1, key2, _airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta " + "SELECT key1, keyuppercase, _airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta " f"FROM {test_schema_name}.{test_table_name} ORDER BY key1" ) @@ -416,6 +459,34 @@ def test_write_dupe( assert sql_result[1][1] == "777-54-0664" +def test_writing_to_schema_with_duplicate_column_names_after_normalization_doesnt_work( + config: Dict[str, str], + request, + configured_catalogue: ConfiguredAirbyteCatalog, + duplicate_column_name_airbyte_message: AirbyteMessage, + test_table_name: str, + test_schema_name: str, + sql_processor, +): + destination = DestinationMotherDuck() + generator = destination.write( + config, + configured_catalogue, + [duplicate_column_name_airbyte_message], + ) + + assert len(list(generator)) == 0 + assert ( + len( + sql_processor._execute_sql( + "SELECT key1, keyuppercase, _airbyte_raw_id, _airbyte_extracted_at, _airbyte_meta " + f"FROM {test_schema_name}.{test_table_name} ORDER BY key1" + ) + ) + == 0 + ) + + def _airbyte_messages(n: int, batch_size: int, table_name: str) -> Generator[AirbyteMessage, None, None]: fake = Faker() Faker.seed(0) @@ -431,7 +502,7 @@ def _airbyte_messages(n: int, batch_size: int, table_name: str) -> Generator[Air type=Type.RECORD, record=AirbyteRecordMessage( stream=table_name, - data={"key1": fake.unique.name(), "key2": str(fake.ssn())}, + data={"key1": fake.unique.name(), "keyUpperCase": str(fake.ssn())}, emitted_at=int(datetime.now().timestamp()) * 1000, ), ) @@ -458,7 +529,7 @@ def _airbyte_messages_with_inconsistent_json_fields(n: int, batch_size: int, tab data=( { "key1": fake.unique.name(), - "key2": str(fake.ssn()) if random.random() < 0.5 else str(random.randrange(1000, 9999999999999)), + "keyUpperCase": str(fake.ssn()) if random.random() < 0.5 else str(random.randrange(1000, 9999999999999)), "nested1": ( {} if random.random() < 0.1 diff --git a/airbyte-integrations/connectors/destination-motherduck/metadata.yaml b/airbyte-integrations/connectors/destination-motherduck/metadata.yaml index f978e9f6aaf7..34ce42295ab6 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.1 + dockerImageTag: 0.2.2 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 b38b5e37d390..2b5f39cb8e64 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.1" +version = "0.2.2" description = "Destination implementation for MotherDuck." authors = ["Guen Prawiroatmodjo, Simon Späti, Airbyte"] license = "ELv2" diff --git a/docs/integrations/destinations/motherduck.md b/docs/integrations/destinations/motherduck.md index 0bd6d0c3aa9b..e7b8de9e6d21 100644 --- a/docs/integrations/destinations/motherduck.md +++ b/docs/integrations/destinations/motherduck.md @@ -73,8 +73,9 @@ This destination supports [namespaces](https://docs.airbyte.com/platform/using-a
Expand to review -| Version | Date | Pull Request | Subject | -| :------ | :--------- | :------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------- | +| Version | Date | Pull Request | Subject | +| :------ | :--------- | :------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------ | +| 0.2.2 | 2025-02-02 | [70438](https://github.com/airbytehq/airbyte/pull/70438) | Fix for camelCase columns being `NULL` | | 0.2.1 | 2025-12-19 | [70999](https://github.com/airbytehq/airbyte/pull/70999) | Fix for empty STRUCTs | | 0.2.0 | 2025-12-01 | [70221](https://github.com/airbytehq/airbyte/pull/70221) | Upgrade DuckDB to v1.4.2 and duckdb-engine to v0.17.0 | | 0.1.26 | 2025-10-21 | [68338](https://github.com/airbytehq/airbyte/pull/68338) | Update dependencies |