start implementing bigquery stuff

Author: edgao

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/client/BigqueryClientFactory.kt

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.bigquery.client
+
+import com.google.auth.oauth2.GoogleCredentials
+import com.google.cloud.bigquery.BigQuery
+import com.google.cloud.bigquery.BigQueryOptions
+import io.airbyte.integrations.destination.bigquery.BigQueryUtils
+import io.airbyte.integrations.destination.bigquery.spec.BigqueryConfiguration
+import io.github.oshai.kotlinlogging.KotlinLogging
+import io.micronaut.context.annotation.Factory
+import jakarta.inject.Singleton
+import java.io.ByteArrayInputStream
+import java.nio.charset.StandardCharsets
+
+private val logger = KotlinLogging.logger {}
+
+@Factory
+class BigqueryClientFactory(private val config: BigqueryConfiguration) {
+    @Singleton
+    fun make(): BigQuery {
+        // Follows this order of resolution:
+        // https://cloud.google.com/java/docs/reference/google-auth-library/latest/com.google.auth.oauth2.GoogleCredentials#com_google_auth_oauth2_GoogleCredentials_getApplicationDefault
+        val credentials =
+            if (config.credentialsJson == null) {
+                logger.info {
+                    "No service account key json is provided. It is required if you are using Airbyte cloud."
+                }
+                logger.info { "Using the default service account credential from environment." }
+                GoogleCredentials.getApplicationDefault()
+            } else {
+                // The JSON credential can either be a raw JSON object, or a serialized JSON object.
+                GoogleCredentials.fromStream(
+                    ByteArrayInputStream(
+                        config.credentialsJson.toByteArray(StandardCharsets.UTF_8)
+                    ),
+                )
+            }
+        return BigQueryOptions.newBuilder()
+            .setProjectId(config.projectId)
+            .setCredentials(credentials)
+            .setHeaderProvider(BigQueryUtils.headerProvider)
+            .build()
+            .service
+    }
+}

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/operation/BigQueryDirectLoadingStorageOperation.kt

@@ -43,10 +43,10 @@ class BigQueryDirectLoadingStorageOperation(
     ) {
     private val rateLimiter: RateLimiter = RateLimiter.create(0.07)
     companion object {
-        private const val HTTP_STATUS_CODE_FORBIDDEN = 403
-        private const val HTTP_STATUS_CODE_NOT_FOUND = 404
+        const val HTTP_STATUS_CODE_FORBIDDEN = 403
+        const val HTTP_STATUS_CODE_NOT_FOUND = 404
 
-        private val CONFIG_ERROR_MSG =
+        val CONFIG_ERROR_MSG =
             """
             |Failed to write to destination schema.
             |   1. Make sure you have all required permissions for writing to the schema.

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryDirectLoader.kt

@@ -0,0 +1,96 @@
+/*
+ * Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.bigquery.write.standard_insert
+
+import com.google.cloud.bigquery.BigQuery
+import com.google.cloud.bigquery.BigQueryException
+import com.google.cloud.bigquery.FormatOptions
+import com.google.cloud.bigquery.JobId
+import com.google.cloud.bigquery.JobInfo
+import com.google.cloud.bigquery.TableDataWriteChannel
+import com.google.cloud.bigquery.TableId
+import com.google.cloud.bigquery.WriteChannelConfiguration
+import io.airbyte.cdk.ConfigErrorException
+import io.airbyte.cdk.load.command.DestinationCatalog
+import io.airbyte.cdk.load.command.DestinationStream
+import io.airbyte.cdk.load.message.DestinationRecordRaw
+import io.airbyte.cdk.load.write.DirectLoader
+import io.airbyte.cdk.load.write.DirectLoaderFactory
+import io.airbyte.integrations.destination.bigquery.BigQueryUtils
+import io.airbyte.integrations.destination.bigquery.formatter.BigQueryRecordFormatter
+import io.airbyte.integrations.destination.bigquery.operation.BigQueryDirectLoadingStorageOperation.Companion.CONFIG_ERROR_MSG
+import io.airbyte.integrations.destination.bigquery.operation.BigQueryDirectLoadingStorageOperation.Companion.HTTP_STATUS_CODE_FORBIDDEN
+import io.airbyte.integrations.destination.bigquery.operation.BigQueryDirectLoadingStorageOperation.Companion.HTTP_STATUS_CODE_NOT_FOUND
+import io.airbyte.integrations.destination.bigquery.spec.BigqueryConfiguration
+import jakarta.inject.Singleton
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets
+
+class BigqueryDirectLoader(
+    private val stream: DestinationStream,
+    private val writer: TableDataWriteChannel,
+) : DirectLoader {
+    override fun accept(record: DestinationRecordRaw): DirectLoader.DirectLoadResult {
+        // TODO wtf is a bigquery record formatter
+        val byteArray =
+            "${bigQueryRecordFormatter.formatRecord(record, stream.generationId)} ${System.lineSeparator()}".toByteArray(
+                StandardCharsets.UTF_8,
+            )
+        writer.write(ByteBuffer.wrap(byteArray))
+        return DirectLoader.Incomplete
+    }
+
+    override fun finish() {
+        writer.close()
+        BigQueryUtils.waitForJobFinish(writer.job)
+    }
+
+    override fun close() {
+        // do nothing
+    }
+}
+
+@Singleton
+class BigqueryDirectLoaderFactory(
+    private val catalog: DestinationCatalog,
+    private val bigquery: BigQuery,
+    private val config: BigqueryConfiguration,
+) : DirectLoaderFactory<BigqueryDirectLoader> {
+    override fun create(
+        streamDescriptor: DestinationStream.Descriptor,
+        part: Int
+    ): BigqueryDirectLoader {
+        // TODO there was a RateLimiter here for some reason...?
+        // TODO we need to handle special chars in stream name/namespace
+        val writeChannelConfiguration =
+            WriteChannelConfiguration.newBuilder(
+                    TableId.of(streamDescriptor.namespace, streamDescriptor.name)
+                )
+                .setCreateDisposition(JobInfo.CreateDisposition.CREATE_IF_NEEDED)
+                .setSchema(BigQueryRecordFormatter.SCHEMA_V2)
+                .setFormatOptions(FormatOptions.json())
+                .build() // new-line delimited json.
+
+        val job =
+            JobId.newBuilder()
+                .setRandomJob()
+                .setLocation(config.datasetLocation.region)
+                .setProject(bigquery.options.projectId)
+                .build()
+
+        val writer =
+            try {
+                bigquery.writer(job, writeChannelConfiguration)
+            } catch (e: BigQueryException) {
+                if (e.code == HTTP_STATUS_CODE_FORBIDDEN || e.code == HTTP_STATUS_CODE_NOT_FOUND) {
+                    throw ConfigErrorException(CONFIG_ERROR_MSG + e)
+                } else {
+                    throw BigQueryException(e.code, e.message)
+                }
+            }
+
+        return BigqueryDirectLoader(catalog.getStream(streamDescriptor), writer)
+    }
+}