fix: auto-detect dataset location for BigQuery analytics view creation

The plugin previously passed a configured location (default "US") to
bigquery.Client(location=...).  This only affects query job routing,
not table CRUD or the Storage Write API.  When the dataset lives in a
non-US region, view creation DDL queries silently fail with a location
mismatch — data writes succeed but analytics views are missing.

The fix auto-detects the dataset's actual location via get_dataset()
during initialization and passes it explicitly to client.query() for
view creation.  Falls back to the configured location when dataset
metadata cannot be resolved.

Fixes #5476

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: caohy1988

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -1990,6 +1990,7 @@ def __init__(
     self._setup_lock = None
     self._user_credentials = credentials
     self._credentials = credentials
+    self._resolved_location: Optional[str] = None
     self.client = None
     self._loop_state_by_loop: dict[asyncio.AbstractEventLoop, _LoopState] = {}
     self._write_stream_name = None  # Resolved stream name
@@ -2184,11 +2185,28 @@ async def _lazy_setup(self, **kwargs) -> None:
           self._executor,
           lambda: bigquery.Client(
               project=self.project_id,
-              location=self.location,
               credentials=self._credentials,
           ),
       )
 
+    # Auto-detect the dataset's location so view-creation DDL
+    # queries are routed to the correct region.  Table CRUD and the
+    # Storage Write API derive location from the dataset server-side,
+    # but client.query() uses the client's default location for job
+    # routing — a mismatch causes silent view-creation failure.
+    if self._resolved_location is None:
+      dataset_id = f"{self.project_id}.{self.dataset_id}"
+      try:
+        dataset = await loop.run_in_executor(
+            self._executor,
+            lambda: self.client.get_dataset(dataset_id),
+        )
+        self._resolved_location = dataset.location
+      except Exception:
+        # Fallback when dataset metadata cannot be resolved,
+        # preserving current behavior.
+        self._resolved_location = self.location
+
     self.full_table_id = f"{self.project_id}.{self.dataset_id}.{self.table_id}"
     if not self._schema:
       self._schema = _get_events_schema()
@@ -2447,7 +2465,7 @@ def _create_analytics_views(self) -> None:
           event_type=event_type,
       )
       try:
-        self.client.query(sql).result()
+        self.client.query(sql, location=self._resolved_location).result()
       except cloud_exceptions.Conflict:
         logger.debug(
             "View %s was updated concurrently by another process.",
@@ -2546,6 +2564,8 @@ def __getstate__(self):
     state["_startup_error"] = None
     state["_is_shutting_down"] = False
     state["_init_pid"] = 0
+    # Re-detect dataset location after unpickle.
+    state["_resolved_location"] = None
     # _credentials is always runtime-resolved; clear unconditionally.
     state["_credentials"] = None
     # Preserve _user_credentials if they are picklable (e.g.,
@@ -2567,6 +2587,7 @@ def __setstate__(self, state):
     state.setdefault("_init_pid", 0)
     state.setdefault("_user_credentials", None)
     state.setdefault("_credentials", None)
+    state.setdefault("_resolved_location", None)
     # Restore _credentials from _user_credentials if available so
     # _create_loop_state uses the user's identity.  When both are
     # None (non-picklable credentials were dropped), ADC is used.
@@ -2616,6 +2637,7 @@ def _reset_runtime_state(self) -> None:
 
     # Clear all runtime state.
     self._setup_lock = None
+    self._resolved_location = None
     self.client = None
     self._loop_state_by_loop = {}
     self._write_stream_name = None

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -7246,3 +7246,130 @@ async def test_no_a2a_interaction_for_no_metadata(
 
     await asyncio.sleep(0.05)
     assert mock_write_client.append_rows.call_count == 0
+
+
+# ================================================================
+# TEST CLASS: Dataset location auto-detection (Issue #5476)
+# ================================================================
+class TestDatasetLocationDetection:
+  """Tests for auto-detecting the dataset location."""
+
+  @pytest.mark.asyncio
+  async def test_location_detected_from_dataset(
+      self,
+      mock_auth_default,
+      mock_to_arrow_schema,
+      mock_asyncio_to_thread,
+  ):
+    """Resolved location comes from get_dataset, not constructor."""
+    mock_dataset = mock.MagicMock()
+    mock_dataset.location = "europe-west1"
+
+    with mock.patch.object(bigquery, "Client", autospec=True) as mock_bq_cls:
+      mock_bq_cls.return_value.get_dataset.return_value = mock_dataset
+      mock_bq_cls.return_value.get_table.side_effect = (
+          cloud_exceptions.NotFound("table")
+      )
+      mock_bq_cls.return_value.create_table.return_value = None
+
+      async with managed_plugin(
+          project_id=PROJECT_ID,
+          dataset_id=DATASET_ID,
+          table_id=TABLE_ID,
+          config=bigquery_agent_analytics_plugin.BigQueryLoggerConfig(
+              create_views=False,
+          ),
+      ) as plugin:
+        await plugin._ensure_started()
+        assert plugin._resolved_location == "europe-west1"
+
+  @pytest.mark.asyncio
+  async def test_location_falls_back_on_get_dataset_failure(
+      self,
+      mock_auth_default,
+      mock_to_arrow_schema,
+      mock_asyncio_to_thread,
+  ):
+    """Falls back to configured location when get_dataset fails."""
+    with mock.patch.object(bigquery, "Client", autospec=True) as mock_bq_cls:
+      mock_bq_cls.return_value.get_dataset.side_effect = Exception("not found")
+      mock_bq_cls.return_value.get_table.side_effect = (
+          cloud_exceptions.NotFound("table")
+      )
+      mock_bq_cls.return_value.create_table.return_value = None
+
+      async with managed_plugin(
+          project_id=PROJECT_ID,
+          dataset_id=DATASET_ID,
+          table_id=TABLE_ID,
+          location="asia-northeast1",
+          config=bigquery_agent_analytics_plugin.BigQueryLoggerConfig(
+              create_views=False,
+          ),
+      ) as plugin:
+        await plugin._ensure_started()
+        assert plugin._resolved_location == "asia-northeast1"
+
+  @pytest.mark.asyncio
+  async def test_views_created_with_resolved_location(
+      self,
+      mock_auth_default,
+      mock_to_arrow_schema,
+      mock_asyncio_to_thread,
+  ):
+    """View creation DDL passes resolved location to client.query()."""
+    mock_dataset = mock.MagicMock()
+    mock_dataset.location = "europe-west1"
+
+    with mock.patch.object(bigquery, "Client", autospec=True) as mock_bq_cls:
+      mock_client = mock_bq_cls.return_value
+      mock_client.get_dataset.return_value = mock_dataset
+      mock_client.get_table.return_value = mock.MagicMock()
+      mock_client.query.return_value.result.return_value = None
+
+      async with managed_plugin(
+          project_id=PROJECT_ID,
+          dataset_id=DATASET_ID,
+          table_id=TABLE_ID,
+          config=bigquery_agent_analytics_plugin.BigQueryLoggerConfig(
+              create_views=True,
+          ),
+      ) as plugin:
+        await plugin._ensure_started()
+
+        # Verify query() was called with location kwarg
+        assert mock_client.query.call_count > 0
+        for call in mock_client.query.call_args_list:
+          _, kwargs = call
+          assert kwargs.get("location") == "europe-west1"
+
+  @pytest.mark.asyncio
+  async def test_view_error_still_logged(
+      self,
+      mock_auth_default,
+      mock_to_arrow_schema,
+      mock_asyncio_to_thread,
+  ):
+    """View creation errors are logged but not raised."""
+    mock_dataset = mock.MagicMock()
+    mock_dataset.location = "US"
+
+    with mock.patch.object(bigquery, "Client", autospec=True) as mock_bq_cls:
+      mock_client = mock_bq_cls.return_value
+      mock_client.get_dataset.return_value = mock_dataset
+      mock_client.get_table.return_value = mock.MagicMock()
+      mock_client.query.return_value.result.side_effect = Exception(
+          "view error"
+      )
+
+      # Should not raise
+      async with managed_plugin(
+          project_id=PROJECT_ID,
+          dataset_id=DATASET_ID,
+          table_id=TABLE_ID,
+          config=bigquery_agent_analytics_plugin.BigQueryLoggerConfig(
+              create_views=True,
+          ),
+      ) as plugin:
+        await plugin._ensure_started()
+        assert plugin._started