Rbroughan/productionalize direct loader (#61528)

tryangul · octavia-squidington-iii · web-flow · commit ee4139732423 · 2025-06-17T13:35:44.000-07:00
Co-authored-by: Octavia Squidington III &lt;octavia-squidington-iii@users.noreply.github.com&gt;
diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt
@@ -120,6 +120,9 @@ data class Meta(
         const val CHECKPOINT_ID_NAME: String = "partition_id"
         const val CHECKPOINT_INDEX_NAME: String = "id"
 
+        const val AIRBYTE_META_SYNC_ID = "sync_id"
+        const val AIRBYTE_META_CHANGES = "changes"
+
         const val COLUMN_NAME_AB_RAW_ID: String = "_airbyte_raw_id"
         const val COLUMN_NAME_AB_EXTRACTED_AT: String = "_airbyte_extracted_at"
         const val COLUMN_NAME_AB_META: String = "_airbyte_meta"
@@ -289,8 +292,8 @@ data class EnrichedDestinationRecordAirbyteValue(
             EnrichedAirbyteValue(
                 ObjectValue(
                     linkedMapOf(
-                        "sync_id" to IntegerValue(stream.syncId),
-                        "changes" to
+                        Meta.AIRBYTE_META_SYNC_ID to IntegerValue(stream.syncId),
+                        Meta.AIRBYTE_META_CHANGES to
                             ArrayValue(
                                 (sourceMeta.changes.toAirbyteValues()) +
                                     declaredFields
diff --git a/airbyte-cdk/bulk/toolkits/load-db/src/main/kotlin/io/airbyte/cdk/load/orchestration/db/legacy_typing_deduping/TableCatalogFactory.kt b/airbyte-cdk/bulk/toolkits/load-db/src/main/kotlin/io/airbyte/cdk/load/orchestration/db/legacy_typing_deduping/TableCatalogFactory.kt
@@ -23,11 +23,17 @@ const val DEFAULT_AIRBYTE_INTERNAL_NAMESPACE = "airbyte_internal"
 data class TableNameInfo(val tableNames: TableNames, val columnNameMapping: ColumnNameMapping)
 
 data class TableCatalog(private val catalog: Map<DestinationStream, TableNameInfo>) :
-    Map<DestinationStream, TableNameInfo> by catalog
+    Map<DestinationStream, TableNameInfo> by catalog {
+    fun getMappedColumnName(stream: DestinationStream, colName: String): String? =
+        this[stream]?.columnNameMapping?.get(colName)
+}
 
 data class TableCatalogByDescriptor(
     private val catalog: Map<DestinationStream.Descriptor, TableNameInfo>
-) : Map<DestinationStream.Descriptor, TableNameInfo> by catalog
+) : Map<DestinationStream.Descriptor, TableNameInfo> by catalog {
+    fun getFinalTableName(desc: DestinationStream.Descriptor): TableName? =
+        this[desc]?.tableNames?.finalTableName
+}
 
 @Factory
 class TableCatalogFactory {
diff --git a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/formatter/BigQueryRecordFormatter.kt b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/formatter/BigQueryRecordFormatter.kt
@@ -36,7 +36,8 @@ class BigQueryRecordFormatter {
                     //   includes changes in-connector type coercion
                     //   and for raw tables, we only want changes that originated from the source
                     val protocolMeta = enrichedRecord.sourceMeta.asProtocolObject()
-                    protocolMeta.additionalProperties["sync_id"] = record.stream.syncId
+                    protocolMeta.additionalProperties[Meta.AIRBYTE_META_SYNC_ID] =
+                        record.stream.syncId
                     outputRecord[key] = protocolMeta.serializeToString()
                     // TODO we should do this for direct-load tables
                     // val serializedAirbyteMeta = (value.abValue as
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/build.gradle b/airbyte-integrations/connectors/destination-clickhouse-v2/build.gradle
@@ -34,7 +34,6 @@ application {
 }
 
 dependencies {
-
     implementation 'com.clickhouse:client-v2:0.8.6'
 
     testImplementation("io.mockk:mockk:1.14.2")
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/client/ClickhouseSqlGenerator.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/client/ClickhouseSqlGenerator.kt
@@ -63,7 +63,7 @@ class ClickhouseSqlGenerator {
               $COLUMN_NAME_AB_RAW_ID String NOT NULL,
               $COLUMN_NAME_AB_EXTRACTED_AT DateTime64(3) NOT NULL,
               $COLUMN_NAME_AB_META String NOT NULL,
-              $COLUMN_NAME_AB_GENERATION_ID UInt32,
+              $COLUMN_NAME_AB_GENERATION_ID UInt32 NOT NULL,
               $columnDeclarations
             )
             ENGINE = ${engine}
@@ -286,7 +286,7 @@ class ClickhouseSqlGenerator {
             .map { (fieldName, type) ->
                 val columnName = columnNameMapping[fieldName]!!
                 val typeName = type.type.toDialectType()
-                "`$columnName` $typeName"
+                "`$columnName` Nullable($typeName)"
             }
             .joinToString(",\n")
 
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/config/ClickhouseBeanFactory.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/config/ClickhouseBeanFactory.kt
@@ -28,6 +28,7 @@ class ClickhouseBeanFactory {
             .setUsername(config.username)
             .setPassword(config.password)
             .setDefaultDatabase(config.resolvedDatabase)
+            .compressClientRequest(true)
             .build()
     }
 
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/config/ClickhouseDirectLoadDatabaseInitialStatusGatherer.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/config/ClickhouseDirectLoadDatabaseInitialStatusGatherer.kt
@@ -2,7 +2,7 @@
  * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  */
 
-package io.airbyte.integrations.destination.clickhouse_v2.write.direct
+package io.airbyte.integrations.destination.clickhouse_v2.config
 
 import io.airbyte.cdk.load.client.AirbyteClient
 import io.airbyte.cdk.load.orchestration.db.BaseDirectLoadInitialStatusGatherer
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/config/ClickhouseNameGenerators.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/config/ClickhouseNameGenerators.kt
@@ -7,23 +7,11 @@ package io.airbyte.integrations.destination.clickhouse_v2.config
 import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.orchestration.db.ColumnNameGenerator
 import io.airbyte.cdk.load.orchestration.db.FinalTableNameGenerator
-import io.airbyte.cdk.load.orchestration.db.RawTableNameGenerator
 import io.airbyte.cdk.load.orchestration.db.TableName
 import io.airbyte.integrations.destination.clickhouse_v2.spec.ClickhouseConfiguration
 import jakarta.inject.Singleton
 import java.util.Locale
 
-// Unused but needed by another bean
-@Singleton
-class ClickhouseRawTableNameGenerators(val config: ClickhouseConfiguration) :
-    RawTableNameGenerator {
-    override fun getTableName(streamDescriptor: DestinationStream.Descriptor): TableName =
-        TableName(
-            config.resolvedDatabase,
-            "test_${streamDescriptor.name}",
-        )
-}
-
 @Singleton
 class ClickhouseFinalTableNameGenerator(private val config: ClickhouseConfiguration) :
     FinalTableNameGenerator {
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/RecordMunger.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/RecordMunger.kt
@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.clickhouse_v2.write
+
+import io.airbyte.cdk.load.data.AirbyteValue
+import io.airbyte.cdk.load.message.DestinationRecordRaw
+import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.TableCatalog
+import jakarta.inject.Singleton
+
+/*
+ * Munges values and keys into a simple map form. Encapsulates
+ * EnrichedAirybteValue logic from other classes, so it can be replaced if
+ * necessary, as we know it's slow.
+ *
+ * The HashMap construction is deliberate for speed considerations.
+ */
+@Singleton
+class RecordMunger(
+    private val catalogInfo: TableCatalog,
+) {
+    fun transformForDest(record: DestinationRecordRaw): Map<String, AirbyteValue> {
+        // this actually munges and coerces data
+        val enriched =
+            record.asEnrichedDestinationRecordAirbyteValue(
+                extractedAtAsTimestampWithTimezone = true
+            )
+
+        val munged = HashMap<String, AirbyteValue>()
+        enriched.declaredFields.forEach {
+            val mappedKey = catalogInfo.getMappedColumnName(record.stream, it.key)!!
+            munged[mappedKey] = it.value.abValue
+        }
+        enriched.airbyteMetaFields.forEach { munged[it.key] = it.value.abValue }
+
+        return munged
+    }
+}
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/SizedWindow.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/SizedWindow.kt
@@ -0,0 +1,17 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.clickhouse_v2.write
+
+/*
+ * Encapsulates basic sized windowing logic. As we implement other windowing,
+ * we should look to break out a shared interface.
+ */
+class SizedWindow(private val size: Long) {
+    private var accumulated = 0L
+
+    fun increment(quantity: Long): SizedWindow = this.apply { accumulated += quantity }
+
+    fun isComplete(): Boolean = accumulated >= size
+}
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/BinaryRowInsertBuffer.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/BinaryRowInsertBuffer.kt
@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.clickhouse_v2.write.direct
+
+import com.clickhouse.client.api.Client
+import com.clickhouse.client.api.data_formats.RowBinaryFormatWriter
+import com.clickhouse.data.ClickHouseFormat
+import com.google.common.annotations.VisibleForTesting
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings
+import io.airbyte.cdk.load.data.AirbyteValue
+import io.airbyte.cdk.load.data.ArrayValue
+import io.airbyte.cdk.load.data.BooleanValue
+import io.airbyte.cdk.load.data.DateValue
+import io.airbyte.cdk.load.data.IntegerValue
+import io.airbyte.cdk.load.data.NullValue
+import io.airbyte.cdk.load.data.NumberValue
+import io.airbyte.cdk.load.data.ObjectValue
+import io.airbyte.cdk.load.data.StringValue
+import io.airbyte.cdk.load.data.TimeWithTimezoneValue
+import io.airbyte.cdk.load.data.TimeWithoutTimezoneValue
+import io.airbyte.cdk.load.data.TimestampWithTimezoneValue
+import io.airbyte.cdk.load.data.TimestampWithoutTimezoneValue
+import io.airbyte.cdk.load.orchestration.db.TableName
+import io.airbyte.cdk.load.util.serializeToString
+import io.github.oshai.kotlinlogging.KotlinLogging
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import kotlinx.coroutines.future.await
+
+private val log = KotlinLogging.logger {}
+
+/*
+ * Encapsulates ClickHouse specific buffering and writing logic separate from the loader
+ * state machine.
+ */
+@SuppressFBWarnings(
+    value = ["NP_NONNULL_PARAM_VIOLATION"],
+    justification = "suspend and fb's non-null analysis don't play well"
+)
+class BinaryRowInsertBuffer(
+    val tableName: TableName,
+    private val clickhouseClient: Client,
+) {
+    // Initialize the inner buffer
+    private val schema = clickhouseClient.getTableSchema(tableName.name, tableName.namespace)
+    @VisibleForTesting internal var inner = InputOutputBuffer()
+    @VisibleForTesting
+    internal var writer = RowBinaryFormatWriter(inner, schema, ClickHouseFormat.RowBinary)
+
+    fun accumulate(recordFields: Map<String, AirbyteValue>) {
+        recordFields.forEach { writeAirbyteValue(it.key, it.value) }
+
+        writer.commitRow()
+    }
+
+    suspend fun flush() {
+        log.info { "Beginning insert into ${tableName.name}" }
+
+        val insertResult =
+            clickhouseClient
+                .insert(
+                    "`${tableName.namespace}`.`${tableName.name}`",
+                    inner.toInputStream(),
+                    ClickHouseFormat.RowBinary
+                )
+                .await()
+
+        log.info { "Finished insert of ${insertResult.writtenRows} rows into ${tableName.name}" }
+    }
+
+    private fun writeAirbyteValue(columnName: String, abValue: AirbyteValue) {
+        when (abValue) {
+            // TODO: let's consider refactoring AirbyteValue so we don't have to do this
+            is NullValue -> writer.setValue(columnName, null)
+            is ObjectValue -> writer.setValue(columnName, abValue.values.serializeToString())
+            is ArrayValue -> writer.setValue(columnName, abValue.values.serializeToString())
+            is BooleanValue -> writer.setValue(columnName, abValue.value)
+            is IntegerValue -> writer.setValue(columnName, abValue.value)
+            is NumberValue -> writer.setValue(columnName, abValue.value)
+            is StringValue -> writer.setValue(columnName, abValue.value)
+            is DateValue -> writer.setValue(columnName, abValue.value)
+            is TimeWithTimezoneValue -> writer.setValue(columnName, abValue.value)
+            is TimeWithoutTimezoneValue -> writer.setValue(columnName, abValue.value)
+            is TimestampWithTimezoneValue -> writer.setValue(columnName, abValue.value)
+            is TimestampWithoutTimezoneValue -> writer.setValue(columnName, abValue.value)
+        }
+    }
+
+    /**
+     * The CH writer wants an output stream and the client an input stream. This is a naive wrapper
+     * class to avoid having to copy the buffer contents around.
+     */
+    internal class InputOutputBuffer : ByteArrayOutputStream() {
+        /**
+         * Get an input stream based on the contents of this output stream. Do not use the output
+         * stream after calling this method.
+         * @return an {@link InputStream}
+         */
+        fun toInputStream(): ByteArrayInputStream {
+            return ByteArrayInputStream(this.buf, 0, this.count)
+        }
+    }
+}
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoader.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoader.kt
@@ -4,92 +4,50 @@
 
 package io.airbyte.integrations.destination.clickhouse_v2.write.direct
 
-import com.clickhouse.client.api.Client
-import com.clickhouse.data.ClickHouseFormat
-import com.fasterxml.jackson.databind.node.ObjectNode
+import com.google.common.annotations.VisibleForTesting
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings
-import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.message.DestinationRecordRaw
-import io.airbyte.cdk.load.util.write
 import io.airbyte.cdk.load.write.DirectLoader
-import io.airbyte.integrations.destination.clickhouse_v2.write.direct.ClickhouseDirectLoader.Constants.DELIMITER
-import io.airbyte.protocol.models.Jsons
-import io.github.oshai.kotlinlogging.KotlinLogging
-import java.io.ByteArrayInputStream
-import java.io.ByteArrayOutputStream
-import java.util.UUID
-import kotlinx.coroutines.future.await
-
-private val log = KotlinLogging.logger {}
+import io.airbyte.integrations.destination.clickhouse_v2.write.RecordMunger
+import io.airbyte.integrations.destination.clickhouse_v2.write.SizedWindow
 
 @SuppressFBWarnings(
     value = ["NP_NONNULL_PARAM_VIOLATION"],
     justification = "suspend and fb's non-null analysis don't play well"
 )
 class ClickhouseDirectLoader(
-    private val descriptor: DestinationStream.Descriptor,
-    private val clickhouseClient: Client,
+    @VisibleForTesting val munger: RecordMunger,
+    @VisibleForTesting val buffer: BinaryRowInsertBuffer,
 ) : DirectLoader {
-    private var buffer: ByteArrayOutputStream = ByteArrayOutputStream()
-    private var recordCount = 0
-
-    object Constants {
-        const val UUID = "bf7d3df8-8a91-4fd4-bd4c-89c293ba1d6b"
-        const val BATCH_SIZE_RECORDS = 500000
-        const val DELIMITER = "\n"
-
-        const val FIELD_RAW_ID = "_airbyte_raw_id"
-        const val FIELD_EXTRACTED_AT = "_airbyte_extracted_at"
-        const val FIELD_META = "_airbyte_meta"
-        const val FIELD_GEN_ID = "_airbyte_generation_id"
-    }
+    private var recordCountWindow = SizedWindow(Constants.MAX_BATCH_SIZE_RECORDS)
+    // the sum of serialized json bytes we've accumulated
+    private var bytesWindow = SizedWindow(Constants.MAX_BATCH_SIZE_BYTES)
 
     override suspend fun accept(record: DestinationRecordRaw): DirectLoader.DirectLoadResult {
-        val protocolRecord = record.asJsonRecord() as ObjectNode
-
-        protocolRecord.put(Constants.FIELD_EXTRACTED_AT, record.rawData.emittedAtMs)
-        protocolRecord.put(Constants.FIELD_GEN_ID, record.stream.generationId)
-        protocolRecord.put(Constants.FIELD_RAW_ID, UUID.randomUUID().toString())
-
-        val meta = Jsons.jsonNode(record.rawData.sourceMeta) as ObjectNode
-        meta.put("sync_id", record.stream.syncId)
-        protocolRecord.set<ObjectNode>(Constants.FIELD_META, meta)
+        val munged = munger.transformForDest(record)
+        buffer.accumulate(munged)
 
-        buffer.write(protocolRecord.toString())
-        buffer.write(DELIMITER)
+        recordCountWindow.increment(1)
+        bytesWindow.increment(record.serializedSizeBytes)
 
-        recordCount++
-
-        if (recordCount >= Constants.BATCH_SIZE_RECORDS) {
-            flush()
+        if (bytesWindow.isComplete() || recordCountWindow.isComplete()) {
+            buffer.flush()
             return DirectLoader.Complete
         }
 
         return DirectLoader.Incomplete
     }
 
-    private suspend fun flush() {
-        val jsonBytes = ByteArrayInputStream(buffer.toByteArray())
-        buffer = ByteArrayOutputStream()
-
-        val insertResult =
-            clickhouseClient
-                .insert(
-                    "`${descriptor.namespace ?: "default"}`.`${descriptor.name}`",
-                    jsonBytes,
-                    ClickHouseFormat.JSONEachRow,
-                )
-                .await()
+    override suspend fun finish() = buffer.flush()
 
-        log.info { "Finished insert of ${insertResult.writtenRows} rows into ${descriptor.name}" }
-        recordCount = 0
-    }
+    override fun close() {}
 
-    // only calls this on force complete
-    override suspend fun finish() {
-        flush()
+    object Constants {
+        // CH recommends 10k-100k batch sizes — since we block on IO we aim on the high side
+        // to amortize the overheads.
+        const val MAX_BATCH_SIZE_RECORDS = 100000L
+        // To prevent undue backpressure, we try to cap the buffer size at something that will
+        // safely fit in the "reserved" memory, which in practice is ~180MB.
+        const val MAX_BATCH_SIZE_BYTES = 70000000L
     }
-
-    // this is always called on complete
-    override fun close() {}
 }
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoaderFactory.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/main/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoaderFactory.kt
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/RecordMungerTest.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/RecordMungerTest.kt
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/SizedWindowTest.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/SizedWindowTest.kt
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/BinaryRowInsertBufferTest.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/BinaryRowInsertBufferTest.kt
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoaderFactoryTest.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoaderFactoryTest.kt
diff --git a/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoaderTest.kt b/airbyte-integrations/connectors/destination-clickhouse-v2/src/test/kotlin/io/airbyte/integrations/destination/clickhouse_v2/write/direct/ClickhouseDirectLoaderTest.kt

Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,6 @@ application {`
`34`	`34`	`}`
`35`	`35`
`36`	`36`	`dependencies {`
`37`		`-`
`38`	`37`	`implementation 'com.clickhouse:client-v2:0.8.6'`
`39`	`38`
`40`	`39`	`testImplementation("io.mockk:mockk:1.14.2")`
Original file line number	Diff line number	Diff line change
`@@ -28,6 +28,7 @@ class ClickhouseBeanFactory {`
`28`	`28`	`.setUsername(config.username)`
`29`	`29`	`.setPassword(config.password)`
`30`	`30`	`.setDefaultDatabase(config.resolvedDatabase)`
	`31`	`+ .compressClientRequest(true)`
`31`	`32`	`.build()`
`32`	`33`	`}`
`33`	`34`