Skip to content

Emit GCSToBigQueryOperator deprecation warning after rendering #70542

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yuseok89 wants to merge 2 commits into
base: main
Choose a base branch
from
+104 −70
Open

Emit GCSToBigQueryOperator deprecation warning after rendering #70542

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
65d407a
Emit GCSToBigQueryOperator deprecation warning after rendering
yuseok89 Jul 27, 2026
7719794
Log the schema_object_bucket default resolved in execute
yuseok89 Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 23 additions & 15 deletions providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,8 +261,6 @@ def __init__(
self.configuration: dict[str, Any] = {}

# GCS config
if src_fmt_configs is None:
src_fmt_configs = {}
if time_partitioning is None:
time_partitioning = {}
if range_partitioning is None:
Expand All @@ -272,9 +270,6 @@ def __init__(
self.bucket = bucket
self.source_objects = source_objects
self.schema_object = schema_object

if schema_object_bucket is None:
schema_object_bucket = bucket
self.schema_object_bucket = schema_object_bucket

# BQ config
Expand Down Expand Up @@ -305,16 +300,6 @@ def __init__(

self.schema_update_options = schema_update_options
self.src_fmt_configs = src_fmt_configs
if src_fmt_configs:
warnings.warn(
"The 'src_fmt_configs' parameter is deprecated. Use 'extra_config' instead. "
"Note: 'extra_config' uses the fully-nested API structure, so format-specific "
"options must be nested under their parent key "
"(e.g., {'parquetOptions': {'enableListInference': True}} rather than "
"{'enableListInference': True}).",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
self.extra_config = extra_config
self.time_partitioning = time_partitioning
self.range_partitioning = range_partitioning
Expand Down Expand Up @@ -359,7 +344,30 @@ def _handle_job_error(job: BigQueryJob | UnknownJob) -> None:
if job.error_result:
raise AirflowException(f"BigQuery job {job.job_id} failed: {job.error_result}")

def _warn_on_deprecated_template_fields(self) -> None:
if self.src_fmt_configs:
warnings.warn(
"The 'src_fmt_configs' parameter is deprecated. Use 'extra_config' instead. "
"Note: 'extra_config' uses the fully-nested API structure, so format-specific "
"options must be nested under their parent key "
"(e.g., {'parquetOptions': {'enableListInference': True}} rather than "
"{'enableListInference': True}).",
AirflowProviderDeprecationWarning,
stacklevel=2,

@potiuk potiuk Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

stacklevel=2 is now vestigial, and this applies across the whole campaign rather than just here.

When the warning fired from __init__ it pointed at the Dag author's line, which is what made it actionable. Emitted from a helper called by execute, stacklevel 2 points at execute itself — worker code the user can't act on — and there is no stack level that reaches the Dag file, because the Dag file isn't on the stack at execution time.

The deprecation is still visible in the task log, which is arguably enough. But since the parameter no longer does what it was there for, either dropping it or naming the offending task in the message ("task %s: the 'src_fmt_configs' parameter is deprecated...") would give users something to grep. Worth deciding once for the campaign rather than per-PR.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

@yuseok89 yuseok89 Aug 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed it's vestigial.
Happy to follow whatever the campaign settles on.

)

def execute(self, context: Context):
# Template fields render after __init__, so defaults that depend on a template field
# (schema_object_bucket falls back to bucket) and the src_fmt_configs deprecation check
# must run here, against the rendered values.
if self.src_fmt_configs is None:
self.src_fmt_configs = {}
if self.schema_object_bucket is None:

@potiuk potiuk Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Worth being aware of a side effect: because these defaults are now resolved after rendering, the rendered-template view in the UI records schema_object_bucket=None even though the run actually used bucket. Someone debugging a failed schema download will see None in the UI and the real bucket in the logs.

Not a reason to go back — resolving before rendering was the bug — but if schema_object_bucket is in template_fields, a log line at resolution time ("schema_object_bucket not set, defaulting to %s") would close the gap cheaply.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

@yuseok89 yuseok89 Aug 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added the log as suggested.
The observable change here is the RTIF None, which is fully restorable by moving the fallback into render_template_fields, but it'll recur for every operator in the campaign, so render-vs-log is probably worth deciding once campaign-wide rather than per-PR.

self.schema_object_bucket = self.bucket
# Not captured in the rendered-template view (it defaults after rendering), so log it.
self.log.info("schema_object_bucket not set, defaulting to bucket %s", self.bucket)
self._warn_on_deprecated_template_fields()

hook = BigQueryHook(
gcp_conn_id=self.gcp_conn_id,
location=self.location,
Expand Down
135 changes: 81 additions & 54 deletions providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -812,6 +812,33 @@ def test_schema_obj_external_table_should_execute_successfully(self, bq_hook, gc
)
gcs_hook.return_value.download.assert_called_once_with(SCHEMA_BUCKET, SCHEMA_OBJECT)

@mock.patch(GCS_TO_BQ_PATH.format("GCSHook"))
@mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
def test_schema_object_bucket_defaults_to_bucket_when_omitted(self, bq_hook, gcs_hook):
# The schema_object_bucket -> bucket fallback now runs in execute() after rendering; when
# schema_object_bucket is omitted the schema download must still target the bucket.
bq_hook.return_value.insert_job.side_effect = [
MagicMock(job_id=REAL_JOB_ID, error_result=False),
REAL_JOB_ID,
]
bq_hook.return_value.generate_job_id.return_value = REAL_JOB_ID
bq_hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)
gcs_hook.return_value.download.return_value = bytes(json.dumps(SCHEMA_FIELDS), "utf-8")
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
schema_object=SCHEMA_OBJECT,
write_disposition=WRITE_DISPOSITION,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
external_table=True,
project_id=JOB_PROJECT_ID,
)

operator.execute(context=MagicMock())

gcs_hook.return_value.download.assert_called_once_with(TEST_BUCKET, SCHEMA_OBJECT)

@mock.patch(GCS_TO_BQ_PATH.format("GCSHook"))
@mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
def test_schema_obj_without_external_table_should_execute_successfully(self, bq_hook, gcs_hook):
Expand Down Expand Up @@ -1742,23 +1769,23 @@ def test_external_table_should_accept_parquet_format_and_options(self, hook):
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)

with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
schema_fields=SCHEMA_FIELDS,
write_disposition=WRITE_DISPOSITION,
external_table=True,
project_id=JOB_PROJECT_ID,
source_format="PARQUET",
src_fmt_configs={
"enableListInference": True,
},
)
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
schema_fields=SCHEMA_FIELDS,
write_disposition=WRITE_DISPOSITION,
external_table=True,
project_id=JOB_PROJECT_ID,
source_format="PARQUET",
src_fmt_configs={
"enableListInference": True,
},
)

operator.execute(context=MagicMock())
with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator.execute(context=MagicMock())

hook.return_value.create_table.assert_called_once_with(
exists_ok=True,
Expand Down Expand Up @@ -1845,22 +1872,22 @@ def test_without_external_table_should_accept_parquet_format_and_options(self, h
]
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)
with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
write_disposition=WRITE_DISPOSITION,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
external_table=False,
project_id=JOB_PROJECT_ID,
source_format="PARQUET",
src_fmt_configs={
"enableListInference": True,
},
)
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
write_disposition=WRITE_DISPOSITION,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
external_table=False,
project_id=JOB_PROJECT_ID,
source_format="PARQUET",
src_fmt_configs={
"enableListInference": True,
},
)

operator.execute(context=MagicMock())
with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator.execute(context=MagicMock())

calls = [
call(
Expand Down Expand Up @@ -2054,18 +2081,18 @@ def test_src_fmt_configs_emits_deprecation_warning(self, hook):
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)

with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
write_disposition=WRITE_DISPOSITION,
project_id=JOB_PROJECT_ID,
src_fmt_configs={"skipLeadingRows": 1},
)
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
write_disposition=WRITE_DISPOSITION,
project_id=JOB_PROJECT_ID,
src_fmt_configs={"skipLeadingRows": 1},
)

operator.execute(context=MagicMock())
with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator.execute(context=MagicMock())

config = hook.return_value.insert_job.call_args[1]["configuration"]
assert config["load"]["skipLeadingRows"] == 1
Expand All @@ -2076,19 +2103,19 @@ def test_src_fmt_configs_and_extra_config_both_applied_with_precedence(self, hoo
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)

with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
write_disposition=WRITE_DISPOSITION,
project_id=JOB_PROJECT_ID,
src_fmt_configs={"skipLeadingRows": 1},
extra_config={"skipLeadingRows": 5, "columnNameCharacterMap": "STRICT"},
)
operator = GCSToBigQueryOperator(
task_id=TASK_ID,
bucket=TEST_BUCKET,
source_objects=TEST_SOURCE_OBJECTS,
destination_project_dataset_table=TEST_EXPLICIT_DEST,
write_disposition=WRITE_DISPOSITION,
project_id=JOB_PROJECT_ID,
src_fmt_configs={"skipLeadingRows": 1},
extra_config={"skipLeadingRows": 5, "columnNameCharacterMap": "STRICT"},
)

operator.execute(context=MagicMock())
with pytest.warns(AirflowProviderDeprecationWarning, match="src_fmt_configs"):
operator.execute(context=MagicMock())

config = hook.return_value.insert_job.call_args[1]["configuration"]
# extra_config wins for overlapping key
Expand Down
1 change: 0 additions & 1 deletion scripts/ci/prek/validate_operators_init_exemptions.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ providers/google/src/airflow/providers/google/cloud/operators/gcs.py::GCSFileTra
providers/google/src/airflow/providers/google/cloud/sensors/bigquery_dts.py::BigQueryDataTransferServiceTransferRunSensor
providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py::CloudComposerExternalTaskSensor
providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py::AzureFileShareToGCSOperator
providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py::GCSToBigQueryOperator
providers/google/src/airflow/providers/google/marketing_platform/operators/campaign_manager.py::GoogleCampaignManagerDeleteReportOperator
providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py::GCSToAzureBlobStorageOperator
providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py::PsrpOperator