With compatible name (#63724)

airbyte-integrations/connectors/destination-clickhouse/metadata.yaml

@@ -2,7 +2,7 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: ce0d828e-1dc4-496c-b122-2da42e637e48
-  dockerImageTag: 2.0.5
+  dockerImageTag: 2.0.6
   dockerRepository: airbyte/destination-clickhouse
   githubIssueLabel: destination-clickhouse
   icon: clickhouse.svg

airbyte-integrations/connectors/destination-clickhouse/src/main/kotlin/io/airbyte/integrations/destination/clickhouse/client/ClickhouseAirbyteClient.kt

@@ -22,6 +22,7 @@ import io.airbyte.cdk.load.orchestration.db.direct_load_table.DirectLoadTableNat
 import io.airbyte.cdk.load.orchestration.db.direct_load_table.DirectLoadTableSqlOperations
 import io.airbyte.integrations.destination.clickhouse.client.ClickhouseSqlGenerator.Companion.DATETIME_WITH_PRECISION
 import io.airbyte.integrations.destination.clickhouse.config.ClickhouseFinalTableNameGenerator
+import io.airbyte.integrations.destination.clickhouse.config.toClickHouseCompatibleName
 import io.airbyte.integrations.destination.clickhouse.model.AlterationSummary
 import io.airbyte.integrations.destination.clickhouse.model.hasApplicableAlterations
 import io.airbyte.integrations.destination.clickhouse.spec.ClickhouseConfiguration
@@ -124,16 +125,7 @@ class ClickhouseAirbyteClient(
         }
 
         val airbyteSchemaWithClickhouseType: Map<String, String> =
-            stream.schema
-                .asColumns()
-                .map { (fieldName, fieldType) ->
-                    // We don't need to nullable information here because we are setting all fields
-                    // as
-                    // nullable in the destination
-                    // Add map key
-                    fieldName to fieldType.type.toDialectType(clickhouseConfiguration.enableJson)
-                }
-                .toMap()
+            getAirbyteSchemaWithClickhouseType(stream)
 
         val clickhousePks: List<String> =
             tableSchemaWithoutAirbyteColumns.filterNot { it.isNullable }.map { it.columnName }
@@ -190,6 +182,21 @@ class ClickhouseAirbyteClient(
         }
     }
 
+    internal fun getAirbyteSchemaWithClickhouseType(
+        stream: DestinationStream
+    ): Map<String, String> =
+        stream.schema
+            .asColumns()
+            .map { (fieldName, fieldType) ->
+                // We don't need to nullable information here because we are setting all fields
+                // as
+                // nullable in the destination
+                // Add map key
+                fieldName.toClickHouseCompatibleName() to
+                    fieldType.type.toDialectType(clickhouseConfiguration.enableJson)
+            }
+            .toMap()
+
     internal fun getChangedColumns(
         tableColumns: List<ClickHouseColumn>,
         catalogColumns: Map<String, String>,

airbyte-integrations/connectors/destination-clickhouse/src/test/kotlin/io/airbyte/integrations/destination/clickhouse/client/ClickhouseAirbyteClientTest.kt

@@ -11,6 +11,8 @@ import com.clickhouse.data.ClickHouseColumn
 import com.clickhouse.data.ClickHouseDataType
 import io.airbyte.cdk.load.command.Append
 import io.airbyte.cdk.load.command.DestinationStream
+import io.airbyte.cdk.load.data.FieldType
+import io.airbyte.cdk.load.data.StringType
 import io.airbyte.cdk.load.orchestration.db.ColumnNameMapping
 import io.airbyte.cdk.load.orchestration.db.TableName
 import io.airbyte.cdk.load.orchestration.db.TempTableNameGenerator
@@ -451,6 +453,34 @@ class ClickhouseAirbyteClientTest {
         }
     }
 
+    @Test
+    fun `test getAirbyteSchemaWithClickhouseType with simple schema`() {
+        val columns = LinkedHashMap.newLinkedHashMap<String, FieldType>(1)
+        columns["field 1"] = FieldType(StringType, true)
+
+        val stream =
+            mockk<DestinationStream>() {
+                every { mappedDescriptor } returns
+                    mockk(relaxed = true) {
+                        every { name } returns "my_table"
+                        every { namespace } returns "my_namespace"
+                    }
+                every { schema } returns
+                    mockk(relaxed = true) {
+                        every { isObject } returns true
+                        every { asColumns() } returns columns
+                    }
+                every { importType } returns Append
+            }
+
+        val expected =
+            mapOf(
+                "field_1" to "String",
+            )
+        val actual = clickhouseAirbyteClient.getAirbyteSchemaWithClickhouseType(stream)
+        Assertions.assertEquals(expected, actual)
+    }
+
     companion object {
         // Constants
         private const val COL1 = "col1"

docs/integrations/destinations/clickhouse.md

@@ -95,6 +95,7 @@ You can also use a pre-existing user but we highly recommend creating a dedicate
 
 | Version | Date       | Pull Request                                               | Subject                                                                        |
 |:--------|:-----------|:-----------------------------------------------------------|:-------------------------------------------------------------------------------|
+| 2.0.6   | 2025-07-22 | [\#63724](https://github.com/airbytehq/airbyte/pull/63724) | Apply clickhouse column name transformation for columns.                       |
 | 2.0.5   | 2025-07-22 | [\#63721](https://github.com/airbytehq/airbyte/pull/63721) | Fix schema change with PKs.                                                    |
 | 2.0.4   | 2025-07-21 | [\#62948](https://github.com/airbytehq/airbyte/pull/62948) | SSH support BETA.                                                              |
 | 2.0.3   | 2025-07-11 | [\#62946](https://github.com/airbytehq/airbyte/pull/62946) | Publish metadata changes.                                                      |