Add argument checks and tests for BQ StorageAPI sinks.

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

@@ -3138,16 +3138,44 @@ public WriteResult expand(PCollection<T> input) {
                 : getTriggeringFrequency();
         checkArgument(
             triggeringFrequency != null,
-            "When writing an unbounded PCollection via FILE_LOADS or STORAGE_API_WRITES, "
+            "When writing an unbounded PCollection via FILE_LOADS or STORAGE_WRITE_API, "
                 + "triggering frequency must be specified");
       } else {
         checkArgument(
-            getTriggeringFrequency() == null && getNumFileShards() == 0,
-            "Triggering frequency or number of file shards can be specified only when writing an"
-                + " unbounded PCollection via FILE_LOADS or STORAGE_API_WRITES, but: the collection"
-                + " was %s and the method was %s",
+            getTriggeringFrequency() == null,
+            "Triggering frequency can be specified only when writing an unbounded PCollection via"
+                + " FILE_LOADS or STORAGE_WRITE_API, but the collection was %s and the method was"
+                + " %s.",
             input.isBounded(),
             method);
+
+        if (method == Method.STORAGE_WRITE_API) {
+          BigQueryOptions bqOptions = input.getPipeline().getOptions().as(BigQueryOptions.class);
+          if (getStorageApiTriggeringFrequency(bqOptions) != null) {
+            LOG.warn(
+                "The setting of storageApiTriggeringFrequency in BigQueryOptions is ignored."
+                    + " It is only supported when writing an unbounded PCollection via"
+                    + " STORAGE_WRITE_API, but the collection was {} and the method was {}.",
+                input.isBounded(),
+                method);
+          }
+        }
+
+        checkArgument(
+            (getNumFileShards() == 0),
+            "Number of file shards can be specified only when writing an unbounded PCollection via"
+                + " FILE_LOADS, but the collection was %s and the method was %s",
+            input.isBounded(),
+            method);
+
+        if (getNumStorageWriteApiStreams() != 0) {
+          LOG.warn(
+              "The setting of numStorageWriteApiStreams is ignored. It can be specified only"
+                  + " when writing an unbounded PCollection via STORAGE_WRITE_API, but the collection"
+                  + " was {} and the method was {}.",
+              input.isBounded(),
+              method);
+        }
       }
 
       if (method != Method.STORAGE_WRITE_API && method != Method.STORAGE_API_AT_LEAST_ONCE) {
@@ -3174,6 +3202,27 @@ public WriteResult expand(PCollection<T> input) {
 
       if (input.isBounded() == IsBounded.BOUNDED) {
         checkArgument(!getAutoSharding(), "Auto-sharding is only applicable to unbounded input.");
+      } else {
+        if (method == Method.STORAGE_WRITE_API) {
+          if (getNumStorageWriteApiStreams() > 0 && getAutoSharding()) {
+            LOG.warn(
+                "The setting of numStorageWriteApiStreams is ignored. It is only supported when"
+                    + " autoSharding is not enabled.");
+          }
+        } else if (method == Method.FILE_LOADS) {
+          if (getNumFileShards() > 0 && getAutoSharding()) {
+            LOG.warn(
+                "The setting of numFileShards is ignored. It is only supported when autoSharding is"
+                    + " not enabled.");
+          }
+        } else if (method != Method.STREAMING_INSERTS) {
+          LOG.warn(
+              "The setting of auto-sharding is ignored. It is only supported when writing an"
+                  + " unbounded PCollection via FILE_LOADS, STREAMING_INSERTS or"
+                  + " STORAGE_WRITE_API, but the collection was {} and the method was {}.",
+              input.isBounded(),
+              method);
+        }
       }
 
       if (getJsonTimePartitioning() != null) {

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

@@ -570,4 +570,17 @@ public Table getTableResource(String projectId, String datasetId, String tableId
             MAX_QUERY_RETRIES, tableId),
         lastException);
   }
+
+  public void updateTableSchema(
+      String projectId, String datasetId, String tableId, TableSchema newSchema) {
+    try {
+      this.bqClient
+          .tables()
+          .patch(projectId, datasetId, tableId, new Table().setSchema(newSchema))
+          .execute();
+      LOG.info("Successfully updated the schema of table: " + tableId);
+    } catch (Exception e) {
+      LOG.debug("Exceptions caught when updating table schema: " + e.getMessage());
+    }
+  }
 }