This is an automated email from the ASF dual-hosted git repository.
shahar1 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 113bc123a6b Warn when ignore_unknown_values is a no-op in
GCSToBigQueryOperator (#71675)
113bc123a6b is described below
commit 113bc123a6b8c8aa5bad133eea803db8dcd8ced3
Author: Jonathan Gosling <[email protected]>
AuthorDate: Sun Oct 18 08:27:33 2026 +0100
Warn when ignore_unknown_values is a no-op in GCSToBigQueryOperator (#71675)
Clarify ignore_unknown_values schema behavior in GCSToBigQueryOperator
Warn when CSV or JSON loads use autodetect without allowing field
addition: values are checked against the inferred schema, not the
destination table schema. Recommend autodetect=None when the destination
table exists, or an explicit schema when it may not. Keep the guidance
accurate for supplied schemas and self-describing formats.
Read the warning conditions from the assembled load configuration and
test when the warning applies.
---------
Co-authored-by: Shahar Epstein <[email protected]>
---
.../google/cloud/transfers/gcs_to_bigquery.py | 44 ++++++--
.../google/cloud/transfers/test_gcs_to_bigquery.py | 124 +++++++++++++++++++++
2 files changed, 161 insertions(+), 7 deletions(-)
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 27c284a1848..5f59eff489d 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
@@ -120,6 +120,9 @@ class GCSToBigQueryOperator(BaseOperator):
If true, the extra values are ignored. If false, records with extra
columns
are treated as bad records, and if there are too many bad records, an
invalid error is returned in the job result.
+ For CSV and JSON sources with ``autodetect=True`` and no schema
provided, the values are
+ checked against the schema inferred from the source data rather than
the destination
+ table's schema - see ``autodetect``.
:param allow_quoted_newlines: Whether to allow quoted newlines (true) or
not (false).
:param allow_jagged_rows: Accept rows that are missing trailing optional
columns.
The missing values are treated as nulls. If false, records with
missing trailing
@@ -162,12 +165,11 @@ class GCSToBigQueryOperator(BaseOperator):
by one or more columns. BigQuery supports clustering for both
partitioned and
non-partitioned tables. The order of columns given determines the sort
order.
Not applicable for external tables.
- :param autodetect: [Optional] Indicates if we should automatically infer
the
- options and schema for CSV and JSON sources. (Default: ``True``).
- Parameter must be set to True if 'schema_fields' and 'schema_object'
are undefined.
- It is suggested to set to True if table are create outside of Airflow.
- If autodetect is None and no schema is provided (neither via
schema_fields
- nor a schema_object), assume the table already exists.
+ :param autodetect: [Optional] Whether to infer the schema from the source
data for CSV and
+ JSON sources when neither ``schema_fields`` nor ``schema_object`` is
provided. If ``True``,
+ the schema is inferred from the source data. If ``False``, either
``schema_fields`` or
+ ``schema_object`` must be provided. If ``None``, the existing
destination table's schema
+ is used. (Default: ``True``).
:param encryption_configuration: [Optional] Custom encryption
configuration (e.g., Cloud KMS keys).
.. code-block:: python
@@ -240,7 +242,7 @@ class GCSToBigQueryOperator(BaseOperator):
time_partitioning=None,
range_partitioning=None,
cluster_fields=None,
- autodetect=True,
+ autodetect: bool | None = True,
encryption_configuration=None,
location=None,
impersonation_chain: str | Sequence[str] | None = None,
@@ -774,6 +776,34 @@ class GCSToBigQueryOperator(BaseOperator):
if self.extra_config:
self.configuration["load"].update(self.extra_config)
+ # Checked against the assembled load config rather than the operator
attributes, because
+ # src_fmt_configs and extra_config can still override any of these
keys.
+ # Self-describing formats (Avro, Parquet, ORC, Datastore backups)
carry their own schema and
+ # ignoreUnknownValues still drops file fields missing from the table,
so only CSV and JSON apply.
+ load_config = self.configuration["load"]
+ if (
+ load_config.get("sourceFormat") in ("CSV",
"NEWLINE_DELIMITED_JSON")
+ and load_config.get("autodetect")
+ and load_config.get("ignoreUnknownValues")
+ and "schema" not in load_config
+ and "ALLOW_FIELD_ADDITION" not in
(load_config.get("schemaUpdateOptions") or [])
+ ):
+ # autodetect=None only works against a table that exists when the
load job runs,
+ # and force_delete deletes it first.
+ if self.force_delete:
+ fix = "Pass `schema_fields` or `schema_object` instead."
+ else:
+ fix = (
+ "If the destination table already exists, set
`autodetect=None` to use its "
+ "schema; otherwise pass `schema_fields` or
`schema_object`."
+ )
+ self.log.warning(
+ "`ignore_unknown_values` is checked against the schema
autodetected from the source "
+ "data, not against the destination table's schema, so fields
missing from the "
+ "destination table are not dropped. %s",
+ fix,
+ )
+
return self.configuration
def _validate_src_fmt_configs(
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 18007fff91a..496207fbf40 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
@@ -2151,6 +2151,130 @@ class TestGCSToBigQueryOperator:
assert operator.schema_fields == SCHEMA_FIELDS
+ @pytest.mark.parametrize(
+ ("operator_kwargs", "expects_warning"),
+ [
+ pytest.param({"ignore_unknown_values": True}, True,
id="autodetect_defaults_to_true"),
+ pytest.param({"ignore_unknown_values": True, "autodetect": True},
True, id="autodetect_true"),
+ pytest.param({"ignore_unknown_values": True, "autodetect": None},
False, id="autodetect_none"),
+ pytest.param(
+ {"ignore_unknown_values": True, "autodetect": False,
"schema_fields": SCHEMA_FIELDS},
+ False,
+ id="autodetect_false",
+ ),
+ pytest.param(
+ {"ignore_unknown_values": True, "schema_fields":
SCHEMA_FIELDS},
+ False,
+ id="schema_fields_supplied",
+ ),
+ pytest.param({"ignore_unknown_values": False}, False,
id="ignore_unknown_values_off"),
+ pytest.param(
+ {"ignore_unknown_values": True, "extra_config": {"autodetect":
None}},
+ False,
+ id="autodetect_cleared_by_extra_config",
+ ),
+ pytest.param(
+ {"ignore_unknown_values": True, "extra_config": {"schema":
{"fields": SCHEMA_FIELDS}}},
+ False,
+ id="schema_supplied_by_extra_config",
+ ),
+ pytest.param(
+ {"ignore_unknown_values": False, "extra_config":
{"ignoreUnknownValues": True}},
+ True,
+ id="ignore_unknown_values_set_by_extra_config",
+ ),
+ pytest.param(
+ {"ignore_unknown_values": True, "source_format":
"NEWLINE_DELIMITED_JSON"},
+ True,
+ id="json_source",
+ ),
+ pytest.param(
+ {"ignore_unknown_values": True, "source_format": "PARQUET"},
False, id="parquet_source"
+ ),
+ pytest.param({"ignore_unknown_values": True, "source_format":
"AVRO"}, False, id="avro_source"),
+ pytest.param({"ignore_unknown_values": True, "source_format":
"ORC"}, False, id="orc_source"),
+ pytest.param(
+ {"ignore_unknown_values": True, "extra_config":
{"sourceFormat": "PARQUET"}},
+ False,
+ id="source_format_set_by_extra_config",
+ ),
+ pytest.param(
+ {"ignore_unknown_values": True, "schema_update_options":
["ALLOW_FIELD_ADDITION"]},
+ False,
+ id="field_addition_allowed",
+ ),
+ pytest.param(
+ {"ignore_unknown_values": True, "schema_update_options":
["ALLOW_FIELD_RELAXATION"]},
+ True,
+ id="only_field_relaxation_allowed",
+ ),
+ ],
+ )
+ @mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
+ def test_ignore_unknown_values_autodetect_warning(self, bq_hook,
operator_kwargs, expects_warning):
+ 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)
+
+ 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,
+ **operator_kwargs,
+ )
+
+ with mock.patch.object(operator.log, "warning", autospec=True) as
mock_warning:
+ operator.execute(context=MagicMock())
+
+ if expects_warning:
+ mock_warning.assert_called_once()
+ else:
+ mock_warning.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("force_delete", "expected_fix", "unexpected_fix"),
+ [
+ pytest.param(False, "autodetect=None", None, id="existing_table"),
+ pytest.param(True, "schema_fields", "autodetect=None",
id="force_delete"),
+ ],
+ )
+ @mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
+ def test_ignore_unknown_values_autodetect_warning_names_the_fix(
+ self, bq_hook, force_delete, expected_fix, unexpected_fix
+ ):
+ 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)
+
+ 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,
+ ignore_unknown_values=True,
+ force_delete=force_delete,
+ )
+
+ with mock.patch.object(operator.log, "warning", autospec=True) as
mock_warning:
+ operator.execute(context=MagicMock())
+
+ mock_warning.assert_called_once()
+ message = mock_warning.call_args.args[0] %
mock_warning.call_args.args[1:]
+ assert expected_fix in message
+ if unexpected_fix:
+ assert unexpected_fix not in message
+
@pytest.fixture
def create_task_instance(create_task_instance_of_operator, session):
