This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a76701fb474 Make schema_fields templated in GCSToBigQueryOperator
(#69108)
a76701fb474 is described below
commit a76701fb4747563586df7ac2fac5e6f5900d8ca7
Author: Nikolaus Schuetz <[email protected]>
AuthorDate: Sun Jul 19 15:16:05 2026 -0700
Make schema_fields templated in GCSToBigQueryOperator (#69108)
schema_fields was not a template field, so a templated value (a Variable
or XComArg resolving to a schema) passed to GCSToBigQueryOperator was never
rendered and reached BigQuery unresolved. Mark it templated, matching how
BigQueryUpdateTableSchemaOperator already templates schema_fields_updates.
---
.../google/cloud/transfers/gcs_to_bigquery.py | 2 ++
.../google/cloud/transfers/test_gcs_to_bigquery.py | 31 ++++++++++++++++++++++
2 files changed, 33 insertions(+)
diff --git
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
index e57fc20110f..68a137d7d05 100644
---
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
+++
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
@@ -196,6 +196,7 @@ class GCSToBigQueryOperator(BaseOperator):
template_fields: Sequence[str] = (
"bucket",
"source_objects",
+ "schema_fields",
"schema_object",
"schema_object_bucket",
"destination_project_dataset_table",
@@ -203,6 +204,7 @@ class GCSToBigQueryOperator(BaseOperator):
"src_fmt_configs",
"extra_config",
)
+ template_fields_renderers = {"schema_fields": "json"}
template_ext: Sequence[str] = (".sql",)
ui_color = "#f0eee4"
operator_extra_links = (BigQueryTableLink(),)
diff --git
a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
index be2f5ed52e5..b71456dd658 100644
--- a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
+++ b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import functools
import json
+from datetime import datetime
from unittest import mock
from unittest.mock import MagicMock, call
@@ -27,6 +28,7 @@ from google.cloud.bigquery import DEFAULT_RETRY, Table
from google.cloud.exceptions import Conflict
from sqlalchemy import select
+from airflow import DAG
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.models.trigger import Trigger
from airflow.providers.common.compat.openlineage.facet import (
@@ -2093,6 +2095,35 @@ class TestGCSToBigQueryOperator:
assert config["load"]["skipLeadingRows"] == 5
assert config["load"]["columnNameCharacterMap"] == "STRICT"
+ def test_schema_fields_is_templated(self):
+ """Regression test for #31481.
+
+ ``schema_fields`` must be a template field so a value supplied at
runtime
+ (the issue used ``.expand()``) is resolved before the operator runs.
Before
+ the fix the field was absent from ``template_fields``, so the value
was never
+ rendered and reached BigQuery verbatim.
+ """
+ assert "schema_fields" in GCSToBigQueryOperator.template_fields
+ assert
GCSToBigQueryOperator.template_fields_renderers["schema_fields"] == "json"
+
+ with DAG(
+ dag_id="test_gcs_to_bq_schema_fields_templating",
+ start_date=datetime(2024, 1, 1),
+ render_template_as_native_obj=True,
+ ) as dag:
+ operator = GCSToBigQueryOperator(
+ task_id=TASK_ID,
+ bucket=TEST_BUCKET,
+ source_objects=TEST_SOURCE_OBJECTS,
+ destination_project_dataset_table=TEST_EXPLICIT_DEST,
+ schema_fields="{{ var.value.schema_fields }}",
+ dag=dag,
+ )
+
+ operator.render_template_fields({"var": {"value": {"schema_fields":
SCHEMA_FIELDS}}})
+
+ assert operator.schema_fields == SCHEMA_FIELDS
+
@pytest.fixture
def create_task_instance(create_task_instance_of_operator, session):
