fix(bigquery): default Arrow pagination stream location to US instead of global (#14458)

When both jobId and BigQueryOptions locations are unspecified, default
the Storage Read API stream resource location in ArrowQueryPageFetcher
to US instead of global to prevent 404 errors.

From
https://docs.cloud.google.com/bigquery/docs/locations#default_location:
> If the location isn't explicitly specified, and it can't be determined
from the resources in the request, the default location is used. If
default location isn't set, the job runs in the US multi-region.

Author: jinseopkim0

java-bigquery/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryImpl.java

@@ -379,18 +379,19 @@ public Page<FieldValueList> getNextPage() {
       List<FieldValueList> rowBatch = new ArrayList<>((int) Math.min(pageSize, 10000L));
 
       try {
-        // Resolve job location in order: JobId location -> BigQueryOptions location -> "global"
+        // Resolve job location in order: JobId location -> BigQueryOptions location -> "US"
         // default.
         // The Storage Read API stream resource name requires a location component (e.g.
         // projects/{project}/locations/{location}/jobs/{job}/streams/_default). If no specific
-        // location was provided on the job or service options, defaulting to "global" allows
-        // queries created without an explicit location to still stream results without failing.
+        // location was provided on the job or service options, defaulting to "US" (the standard
+        // BigQuery default multi-region) allows queries created without an explicit location
+        // to stream results from default datasets without failing.
         String location = jobId.getLocation();
         if (location == null) {
           location = serviceOptions.getLocation();
         }
         if (location == null) {
-          location = "global";
+          location = "US";
         }
 
         if (streamIterator == null) {
@@ -2919,12 +2920,13 @@ && getOptions().getOpenTelemetryTracer() != null) {
           if (jobLocation == null) {
             jobLocation = getOptions().getLocation();
           }
-          if (jobLocation != null) {
-            streamName =
-                String.format(
-                    "projects/%s/locations/%s/jobs/%s/streams/_default",
-                    jobProject, jobLocation, actualJobId.getJob());
+          if (jobLocation == null) {
+            jobLocation = "US";
           }
+          streamName =
+              String.format(
+                  "projects/%s/locations/%s/jobs/%s/streams/_default",
+                  jobProject, jobLocation, actualJobId.getJob());
         }
 
         BigQueryReadClient client = null;

java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ArrowQueryPageFetcherTest.java

@@ -289,8 +289,8 @@ void testGetNextPage_respectsMaxResults() throws IOException {
   }
 
   @Test
-  void testGetNextPage_missingLocationDefaultsToGlobal() throws IOException {
-    byte[] batchBytes = createBatchBytes(ImmutableList.of(10L));
+  void testGetNextPage_missingLocationDefaultsToUS() throws IOException {
+    byte[] batchBytes = createBatchBytes(ImmutableList.of(1L));
     ReadRowsResponse response =
         ReadRowsResponse.newBuilder()
             .setArrowRecordBatch(
@@ -331,7 +331,7 @@ void testGetNextPage_missingLocationDefaultsToGlobal() throws IOException {
     Page<FieldValueList> page = fetcher.getNextPage();
     assertNotNull(page);
     assertEquals(
-        "projects/" + PROJECT + "/locations/global/jobs/" + JOB + "/streams/_default",
+        "projects/" + PROJECT + "/locations/US/jobs/" + JOB + "/streams/_default",
         requestCapture.getValue().getReadStream());
   }
 

java-bigquery/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/BigQueryImplTest.java

@@ -2954,6 +2954,65 @@ void testQueryArrowDefaultsToJobCreationOptional() throws IOException, Interrupt
     assertEquals("ARROW", requestPb.getQueryResultsFormat());
   }
 
+  @Test
+  void testQueryArrowDefaultsToUSLocationWhenUnspecified() throws Exception {
+    org.apache.arrow.vector.types.pojo.Schema arrowSchema =
+        new org.apache.arrow.vector.types.pojo.Schema(
+            ImmutableList.of(
+                org.apache.arrow.vector.types.pojo.Field.nullable(
+                    "id", new ArrowType.Int(64, true))));
+
+    byte[] schemaBytes;
+    try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+      MessageSerializer.serialize(new WriteChannel(Channels.newChannel(out)), arrowSchema);
+      schemaBytes = out.toByteArray();
+    }
+
+    QueryJobConfiguration config =
+        QueryJobConfiguration.newBuilder("SELECT 1")
+            .setQueryResultsFormat(QueryResultsFormat.ARROW)
+            .build();
+    com.google.api.services.bigquery.model.JobReference jobRef =
+        new com.google.api.services.bigquery.model.JobReference()
+            .setProjectId(PROJECT)
+            .setJobId(JOB); // No location set
+    com.google.api.services.bigquery.model.QueryResponse queryResponsePb =
+        new com.google.api.services.bigquery.model.QueryResponse()
+            .setJobReference(jobRef)
+            .setJobComplete(true)
+            .setTotalRows(BigInteger.ONE)
+            .setArrowSchema(
+                new com.google.api.services.bigquery.model.ArrowSchema()
+                    .setSerializedSchema(BaseEncoding.base64().encode(schemaBytes)));
+
+    BigQueryReadClient mockReadClient =
+        mock(BigQueryReadClient.class, withSettings().withoutAnnotations());
+    @SuppressWarnings("unchecked")
+    ServerStreamingCallable<ReadRowsRequest, ReadRowsResponse> mockCallable =
+        mock(ServerStreamingCallable.class, withSettings().withoutAnnotations());
+    @SuppressWarnings("unchecked")
+    ServerStream<ReadRowsResponse> mockServerStream =
+        mock(ServerStream.class, withSettings().withoutAnnotations());
+    ArgumentCaptor<ReadRowsRequest> requestCapture = ArgumentCaptor.forClass(ReadRowsRequest.class);
+    when(mockCallable.call(requestCapture.capture())).thenReturn(mockServerStream);
+    when(mockServerStream.iterator()).thenReturn(Collections.emptyIterator());
+    when(mockReadClient.readRowsCallable()).thenReturn(mockCallable);
+
+    when(bigqueryRpcMock.queryRpcSkipExceptionTranslation(eq(PROJECT), any(QueryRequest.class)))
+        .thenReturn(queryResponsePb);
+
+    bigquery = options.getService();
+    ((BigQueryImpl) bigquery).setBigQueryReadClient(mockReadClient);
+
+    ArrowQueryResult result = bigquery.queryArrow(config);
+    assertNotNull(result);
+    result.iterator().hasNext();
+
+    assertEquals(
+        "projects/" + PROJECT + "/locations/US/jobs/" + JOB + "/streams/_default",
+        requestCapture.getValue().getReadStream());
+  }
+
   @Test
   void testQueryWithArrowFormatSlowPathFallback() throws Exception {
     JobId queryJob = JobId.of(PROJECT, JOB).toBuilder().setLocation(LOCATION).build();