fix(bigquery-analytics): make GCS offload projection-aware (google#321)

Content parsing/offload ran before row projection, so denying content_parts
(which holds the offload object reference) could still upload the payload to
GCS with no retained reference -- a payload leak + cost. And denying both
content and content_parts still did the full parse/offload for a row that
keeps neither payload column.

- When content_parts is denied, do not construct the GCS offloader (large /
  binary content is kept inline + truncated instead of uploaded); log a
  warning so the disabled offload is visible.
- When both content and content_parts are denied, skip content parsing
  entirely (no inline summary, no parts, no offload).

Adds 2 tests asserting the storage upload mock is not called for
payload_column_denylist=["content_parts"] and ["content","content_parts"]
with gcs_bucket_name set. Full plugin suite: 294 passed, 6 skipped.

Author: caohy1988

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -2545,14 +2545,28 @@ async def _lazy_setup(self, **kwargs) -> None:
 
       self.offloader = None
       if self.config.gcs_bucket_name:
-        self.offloader = GCSOffloader(
-            self.project_id,
-            self.config.gcs_bucket_name,
-            self._executor,
-            storage_client=storage.Client(
-                project=self.project_id, credentials=self._credentials
-            ),
-        )
+        if "content_parts" in self._denied_columns:
+          # #321: GCS offload stores its object reference in the
+          # ``content_parts`` column. With ``content_parts`` projected out,
+          # an upload would be orphaned -- payload leaks to GCS and incurs
+          # cost with no retained reference. Disable offload and keep
+          # content inline (truncated) instead.
+          logger.warning(
+              "GCS offload disabled: payload_column_denylist drops"
+              " 'content_parts', which holds the offloaded object reference;"
+              " large/binary content is kept inline (truncated) instead of"
+              " being uploaded to %s.",
+              self.config.gcs_bucket_name,
+          )
+        else:
+          self.offloader = GCSOffloader(
+              self.project_id,
+              self.config.gcs_bucket_name,
+              self._executor,
+              storage_client=storage.Client(
+                  project=self.project_id, credentials=self._credentials
+              ),
+          )
 
       self.parser = HybridContentParser(
           self.offloader,
@@ -3419,12 +3433,18 @@ async def _log_event(
       logger.warning("Parser not initialized; skipping event %s.", event_type)
       return
 
-    # Update parser's trace/span IDs for GCS pathing (reuse instance)
-    self.parser.trace_id = trace_id or "no_trace"
-    self.parser.span_id = span_id or "no_span"
-    content_json, content_parts, parser_truncated = await self.parser.parse(
-        raw_content
-    )
+    # #321: when both payload columns are projected out, skip content parsing
+    # entirely -- no inline summary, no parts, and (critically) no GCS offload
+    # work for a row that retains neither payload column.
+    if {"content", "content_parts"} <= self._denied_columns:
+      content_json, content_parts, parser_truncated = None, [], False
+    else:
+      # Update parser's trace/span IDs for GCS pathing (reuse instance)
+      self.parser.trace_id = trace_id or "no_trace"
+      self.parser.span_id = span_id or "no_span"
+      content_json, content_parts, parser_truncated = await self.parser.parse(
+          raw_content
+      )
     is_truncated = is_truncated or parser_truncated
 
     latency_json = self._extract_latency(event_data)

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -9210,3 +9210,80 @@ def test_enrich_attributes_skips_otel_when_attributes_denied(callback_context):
         bigquery_agent_analytics_plugin.EventData(), callback_context
     )
   assert "otel" not in attrs
+
+
+@pytest.mark.asyncio
+async def test_content_parts_denied_disables_gcs_offload(
+    mock_write_client,
+    callback_context,
+    mock_auth_default,
+    mock_bq_client,
+    mock_to_arrow_schema,
+    dummy_arrow_schema,
+    mock_storage_client,
+):
+  # #321: denying content_parts (which holds the offload object reference)
+  # must disable GCS offload, otherwise the payload is uploaded with no
+  # retained reference (leak + cost).
+  config = bigquery_agent_analytics_plugin.BigQueryLoggerConfig(
+      gcs_bucket_name="test-bucket",
+      payload_column_denylist=["content_parts"],
+  )
+  async with managed_plugin(
+      PROJECT_ID, DATASET_ID, table_id=TABLE_ID, config=config
+  ) as plugin:
+    await plugin._ensure_started(
+        storage_client=mock_storage_client.return_value
+    )
+    assert plugin.offloader is None
+    mock_blob = (
+        mock_storage_client.return_value.bucket.return_value.blob.return_value
+    )
+    large_text = "A" * (32 * 1024 + 1)
+    llm_request = llm_request_lib.LlmRequest(
+        model="gemini-pro",
+        contents=[types.Content(parts=[types.Part(text=large_text)])],
+    )
+    await plugin.before_model_callback(
+        callback_context=callback_context, llm_request=llm_request
+    )
+    await plugin.flush()
+    mock_blob.upload_from_string.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_both_payload_columns_denied_skips_parse_and_offload(
+    mock_write_client,
+    callback_context,
+    mock_auth_default,
+    mock_bq_client,
+    mock_to_arrow_schema,
+    dummy_arrow_schema,
+    mock_storage_client,
+):
+  # #321: with both content and content_parts denied, parsing is skipped
+  # entirely -- no inline summary, no parts, and no GCS upload work.
+  config = bigquery_agent_analytics_plugin.BigQueryLoggerConfig(
+      gcs_bucket_name="test-bucket",
+      payload_column_denylist=["content", "content_parts"],
+  )
+  async with managed_plugin(
+      PROJECT_ID, DATASET_ID, table_id=TABLE_ID, config=config
+  ) as plugin:
+    await plugin._ensure_started(
+        storage_client=mock_storage_client.return_value
+    )
+    assert plugin.offloader is None
+    mock_blob = (
+        mock_storage_client.return_value.bucket.return_value.blob.return_value
+    )
+    large_text = "A" * (32 * 1024 + 1)
+    llm_request = llm_request_lib.LlmRequest(
+        model="gemini-pro",
+        contents=[types.Content(parts=[types.Part(text=large_text)])],
+    )
+    await plugin.before_model_callback(
+        callback_context=callback_context, llm_request=llm_request
+    )
+    await plugin.flush()
+    mock_blob.upload_from_string.assert_not_called()