fix parent project in BigQueryToParquet - this should be transparent change as BQIO is following same logic, it was strange that for schema pulling, pipeline is using dataset's project, while for reading it is following options. now it will be aligned. (#2530)

Author: stankiewicz

v2/bigquery-to-parquet/src/main/java/com/google/cloud/teleport/v2/templates/BigQueryToParquet.java

@@ -42,6 +42,7 @@
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.TypedRead.Method;
+import org.apache.beam.sdk.io.gcp.bigquery.BigQueryOptions;
 import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
 import org.apache.beam.sdk.io.parquet.ParquetIO;
 import org.apache.beam.sdk.options.Default;
@@ -108,14 +109,22 @@ static class ReadSessionFactory {
      * Creates ReadSession for schema extraction.
      *
      * @param client BigQueryStorage client used to create ReadSession.
-     * @param tableString String that represents table to export from.
+     * @param options BigQueryToParquetOptions options.
      * @param tableReadOptions TableReadOptions that specify any fields in the table to filter on.
      * @return session ReadSession object that contains the schema for the export.
      */
     static ReadSession create(
-        BigQueryStorageClient client, String tableString, TableReadOptions tableReadOptions) {
+        BigQueryStorageClient client,
+        BigQueryToParquetOptions options,
+        TableReadOptions tableReadOptions) {
+      String tableString = options.getTableRef();
       TableReference tableReference = BigQueryHelpers.parseTableSpec(tableString);
-      String parentProjectId = "projects/" + tableReference.getProjectId();
+      BigQueryOptions bigQueryOptions = options.as(BigQueryOptions.class);
+      String parentProjectId =
+          bigQueryOptions.getBigQueryProject() == null
+              ? bigQueryOptions.getProject()
+              : bigQueryOptions.getBigQueryProject();
+      String parentProjectIdResource = "projects/" + parentProjectId;
 
       TableReferenceProto.TableReference storageTableRef =
           TableReferenceProto.TableReference.newBuilder()
@@ -126,7 +135,7 @@ static ReadSession create(
 
       CreateReadSessionRequest.Builder builder =
           CreateReadSessionRequest.newBuilder()
-              .setParent(parentProjectId)
+              .setParent(parentProjectIdResource)
               .setReadOptions(tableReadOptions)
               .setTableReference(storageTableRef);
       try {
@@ -254,8 +263,7 @@ private static PipelineResult run(BigQueryToParquetOptions options) {
 
     TableReadOptions tableReadOptions = builder.build();
     BigQueryStorageClient client = BigQueryStorageClientFactory.create();
-    ReadSession session =
-        ReadSessionFactory.create(client, options.getTableRef(), tableReadOptions);
+    ReadSession session = ReadSessionFactory.create(client, options, tableReadOptions);
 
     // Extract schema from ReadSession
     Schema schema = getTableSchema(session);

v2/bigquery-to-parquet/src/test/java/com/google/cloud/teleport/v2/templates/BigQueryToParquetTest.java

@@ -23,6 +23,7 @@
 import com.google.cloud.bigquery.storage.v1beta1.Storage.ReadSession;
 import com.google.cloud.teleport.v2.templates.BigQueryToParquet.ReadSessionFactory;
 import org.apache.avro.Schema;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -48,8 +49,11 @@ public void testReadSessionFactoryBadTable() {
     // Test input
     final String badTableRef = "fantasmic-999999;great_data.table";
     final TableReadOptions tableReadOptions = TableReadOptions.newBuilder().build();
+    BigQueryToParquet.BigQueryToParquetOptions options =
+        PipelineOptionsFactory.create().as(BigQueryToParquet.BigQueryToParquetOptions.class);
+    options.setTableRef(badTableRef);
     ReadSessionFactory trsf = new ReadSessionFactory();
-    ReadSession trs = trsf.create(client, badTableRef, tableReadOptions);
+    ReadSession trs = trsf.create(client, options, tableReadOptions);
   }
 
   /** Test Schema Parser is working as expected. */