BigQuery Destination: Proposal to impersonate account on bigquery

airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryConsts.java

@@ -11,6 +11,7 @@ public class BigQueryConsts {
   public static final String CONFIG_PROJECT_ID = "project_id";
   public static final String CONFIG_DATASET_LOCATION = "dataset_location";
   public static final String CONFIG_CREDS = "credentials_json";
+  public static final String CONFIG_IMPERSONATE_ACCOUNT = "impersonate_account";
   public static final String BIG_QUERY_CLIENT_CHUNK_SIZE = "big_query_client_buffer_size_mb";
 
   public static final String LOADING_METHOD = "loading_method";

airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDestination.java

@@ -7,6 +7,7 @@
 import com.codepoetics.protonpack.StreamUtils;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.ImpersonatedCredentials;
 import com.google.cloud.bigquery.BigQuery;
 import com.google.cloud.bigquery.BigQueryOptions;
 import com.google.cloud.bigquery.Dataset;
@@ -49,6 +50,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -178,18 +180,34 @@ protected BigQuery getBigQuery(final JsonNode config) {
   }
 
   private static GoogleCredentials getServiceAccountCredentials(final JsonNode config) throws IOException {
+    GoogleCredentials credentials = null;
     if (!BigQueryUtils.isUsingJsonCredentials(config)) {
       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.");
-      return GoogleCredentials.getApplicationDefault();
+      credentials = GoogleCredentials.getApplicationDefault();
+    } else {
+      // The JSON credential can either be a raw JSON object, or a serialized JSON object.
+      final String credentialsString = config.get(BigQueryConsts.CONFIG_CREDS).isObject()
+          ? Jsons.serialize(config.get(BigQueryConsts.CONFIG_CREDS))
+          : config.get(BigQueryConsts.CONFIG_CREDS).asText();
+      credentials = GoogleCredentials.fromStream(
+          new ByteArrayInputStream(credentialsString.getBytes(Charsets.UTF_8)));
     }
 
-    // The JSON credential can either be a raw JSON object, or a serialized JSON object.
-    final String credentialsString = config.get(BigQueryConsts.CONFIG_CREDS).isObject()
-        ? Jsons.serialize(config.get(BigQueryConsts.CONFIG_CREDS))
-        : config.get(BigQueryConsts.CONFIG_CREDS).asText();
-    return GoogleCredentials.fromStream(
-        new ByteArrayInputStream(credentialsString.getBytes(Charsets.UTF_8)));
+    if (config.has(BigQueryConsts.CONFIG_IMPERSONATE_ACCOUNT)){
+      return ImpersonatedCredentials.create(
+        credentials, 
+        config.get(BigQueryConsts.CONFIG_IMPERSONATE_ACCOUNT).asText(), 
+        null,
+        Arrays.asList(
+          "https://www.googleapis.com/auth/bigquery",
+          "https://www.googleapis.com/auth/bigquery.insertdata",
+          "https://www.googleapis.com/auth/devstorage.read_write"
+        ), 
+        0); //0 = max = 3600 seconds
+    } else {
+      return credentials;
+    }
   }
 
   @Override

airbyte-integrations/connectors/destination-bigquery/src/main/resources/spec.json

@@ -169,13 +169,19 @@
         "airbyte_secret": true,
         "order": 4
       },
+      "impersonate_account": {
+        "type": "string",
+        "description": "The name of the service account you want to impersonate, if there is one.",
+        "title": "Service account name to impersonate (Optional)",
+        "order": 5
+      },
       "transformation_priority": {
         "type": "string",
         "description": "Interactive run type means that the query is executed as soon as possible, and these queries count towards concurrent rate limit and daily limit. Read more about interactive run type <a href=\"https://cloud.google.com/bigquery/docs/running-queries#queries\">here</a>. Batch queries are queued and started as soon as idle resources are available in the BigQuery shared resource pool, which usually occurs within a few minutes. Batch queries don’t count towards your concurrent rate limit. Read more about batch queries <a href=\"https://cloud.google.com/bigquery/docs/running-queries#batch\">here</a>. The default \"interactive\" value is used if not set explicitly.",
         "title": "Transformation Query Run Type",
         "default": "interactive",
         "enum": ["interactive", "batch"],
-        "order": 5
+        "order": 6
       },
       "big_query_client_buffer_size_mb": {
         "title": "Google BigQuery Client Chunk Size",
@@ -185,7 +191,7 @@
         "maximum": 15,
         "default": 15,
         "examples": ["15"],
-        "order": 6
+        "order": 7
       }
     }
   }