fix(destination-motherduck): fix for camelCase columns being NULL

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:

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

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

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"

docs/integrations/destinations/motherduck.md

@@ -73,8 +73,9 @@ This destination supports [namespaces](https://docs.airbyte.com/platform/using-a
 <details>
   <summary>Expand to review</summary>
 
-| 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 |