[yaml] - add bigquery destination test

.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "revision": 6
+  "revision": 8
 }

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryStorageSourceBase.java

@@ -156,13 +156,27 @@ public List<BigQueryStorageStreamSource<T>> split(
       streamCount = Math.max(streamCount, MIN_SPLIT_COUNT);
     }
 
+    String project =
+        bqOptions.getBigQueryProject() == null
+            ? bqOptions.getProject()
+            : bqOptions.getBigQueryProject();
+    if (project == null) {
+      if (targetTable != null
+          && targetTable.getTableReference() != null
+          && targetTable.getTableReference().getProjectId() != null) {
+        project = targetTable.getTableReference().getProjectId();
+      } else {
+        @Nullable String tableReferenceId = getTargetTableId(bqOptions);
+        if (tableReferenceId != null) {
+          TableReference tableReference = BigQueryHelpers.parseTableUrn(tableReferenceId);
+          project = tableReference.getProjectId();
+        }
+      }
+    }
+
     CreateReadSessionRequest createReadSessionRequest =
         CreateReadSessionRequest.newBuilder()
-            .setParent(
-                BigQueryHelpers.toProjectResourceName(
-                    bqOptions.getBigQueryProject() == null
-                        ? bqOptions.getProject()
-                        : bqOptions.getBigQueryProject()))
+            .setParent(BigQueryHelpers.toProjectResourceName(project))
             .setReadSession(readSessionBuilder)
             .setMaxStreamCount(streamCount)
             .build();

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOStorageReadTest.java

@@ -1600,6 +1600,66 @@ public void testReadFromBigQueryIO() throws Exception {
     p.run();
   }
 
+  @Test
+  public void testReadFromBigQueryIOWithFallbackProject() throws Exception {
+    fakeDatasetService.createDataset("fallback-project", "dataset", "", "", null);
+    TableReference tableRef = BigQueryHelpers.parseTableSpec("fallback-project:dataset.table");
+    Table table = new Table().setTableReference(tableRef).setNumBytes(10L).setSchema(TABLE_SCHEMA);
+    fakeDatasetService.createTable(table);
+
+    CreateReadSessionRequest expectedCreateReadSessionRequest =
+        CreateReadSessionRequest.newBuilder()
+            .setParent("projects/fallback-project")
+            .setReadSession(
+                ReadSession.newBuilder()
+                    .setTable("projects/fallback-project/datasets/dataset/tables/table")
+                    .setDataFormat(DataFormat.AVRO)
+                    .setReadOptions(ReadSession.TableReadOptions.newBuilder()))
+            .setMaxStreamCount(10)
+            .build();
+
+    ReadSession readSession =
+        ReadSession.newBuilder()
+            .setName("readSessionName")
+            .setAvroSchema(AvroSchema.newBuilder().setSchema(AVRO_SCHEMA_STRING))
+            .addStreams(ReadStream.newBuilder().setName("streamName"))
+            .setDataFormat(DataFormat.AVRO)
+            .build();
+
+    ReadRowsRequest expectedReadRowsRequest =
+        ReadRowsRequest.newBuilder().setReadStream("streamName").build();
+
+    List<GenericRecord> records = Lists.newArrayList(createRecord("A", 1, AVRO_SCHEMA));
+
+    List<ReadRowsResponse> readRowsResponses =
+        Lists.newArrayList(createResponse(AVRO_SCHEMA, records.subList(0, 1), 0.0, 1.0));
+
+    StorageClient fakeStorageClient = mock(StorageClient.class, withSettings().serializable());
+    when(fakeStorageClient.createReadSession(expectedCreateReadSessionRequest))
+        .thenReturn(readSession);
+    when(fakeStorageClient.readRows(expectedReadRowsRequest, ""))
+        .thenReturn(new FakeBigQueryServerStream<>(readRowsResponses));
+
+    // Explicitly set the pipeline's project option to null to simulate missing
+    // cross-language parameters, and verify it uses the project from the TableReference.
+    options.as(BigQueryOptions.class).setProject(null);
+
+    PCollection<KV<String, Long>> output =
+        p.apply(
+            BigQueryIO.read(new ParseKeyValue())
+                .from("fallback-project:dataset.table")
+                .withMethod(Method.DIRECT_READ)
+                .withFormat(DataFormat.AVRO)
+                .withTestServices(
+                    new FakeBigQueryServices()
+                        .withDatasetService(fakeDatasetService)
+                        .withStorageClient(fakeStorageClient)));
+
+    PAssert.that(output).containsInAnyOrder(ImmutableList.of(KV.of("A", 1L)));
+
+    p.run();
+  }
+
   @Test
   public void testReadFromBigQueryIOWithTrimmedSchema() throws Exception {
     fakeDatasetService.createDataset("foo.com:project", "dataset", "", "", null);

sdks/python/apache_beam/yaml/extended_tests/databases/bigquery_dynamic_destinations.yaml

@@ -0,0 +1,97 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+fixtures:
+  - name: BQ_TABLE
+    type: "apache_beam.yaml.integration_tests.temp_bigquery_table"
+    config:
+      project: "apache-beam-testing"
+  - name: TEMP_DIR_0
+    # Need distributed filesystem to be able to read and write from a container.
+    type: "apache_beam.yaml.integration_tests.gcs_temp_dir"
+    config:
+      bucket: "gs://temp-storage-for-end-to-end-tests/temp-it"
+
+pipelines:
+  - pipeline:
+      type: chain
+      transforms:
+        - type: Create
+          name: CreateUsers
+          config:
+            elements:
+              - {id: 1, name: "Alice", country: "US"}
+              - {id: 2, name: "Bob", country: "UK"}
+              - {id: 3, name: "Charlie", country: "CN"}
+              - {id: 4, name: "David", country: "US"}
+        - type: WriteToBigQuery
+          name: WriteWithDynamicDestinations
+          config:
+            # NOTE: This will create 3 tables: BQ_TABLE_US, BQ_TABLE_UK, BQ_TABLE_CN
+            # This is because the table name is dynamically generated using the country field.
+            # The {{country}} syntax is used to dynamically generate the table name.
+            # For this testing example we are using {{country}} to dynamically generate the table name.
+            # In production, we would use {country} to dynamically generate the table name.
+            table: "{BQ_TABLE}_{{country}}"
+            create_disposition: CREATE_IF_NEEDED
+            write_disposition: WRITE_APPEND
+    options:
+      project: "apache-beam-testing"
+      temp_location: "{TEMP_DIR_0}"
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromBigQuery
+          config:
+            table: "{BQ_TABLE}_US"
+        - type: AssertEqual
+          config:
+            elements:
+              - {id: 1, name: "Alice", country: "US"}
+              - {id: 4, name: "David", country: "US"}
+    options:
+      project: "apache-beam-testing"
+      temp_location: "{TEMP_DIR_0}"
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromBigQuery
+          config:
+            table: "{BQ_TABLE}_UK"
+        - type: AssertEqual
+          config:
+            elements:
+              - {id: 2, name: "Bob", country: "UK"}
+    options:
+      project: "apache-beam-testing"
+      temp_location: "{TEMP_DIR_0}"
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromBigQuery
+          config:
+            table: "{BQ_TABLE}_CN"
+        - type: AssertEqual
+          config:
+            elements:
+              - {id: 3, name: "Charlie", country: "CN"}
+    options:
+      project: "apache-beam-testing"
+      temp_location: "{TEMP_DIR_0}"