SameerMesiah97 commented on code in PR #71675:
URL: https://github.com/apache/airflow/pull/71675#discussion_r3792288071
##########
providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py:
##########
@@ -2124,6 +2131,110 @@ def test_schema_fields_is_templated(self):
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, "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",
+ ),
+ ],
+ )
Review Comment:
Since autodetect is explicitly a three-state parameter, could we also cover
`autodetect=False `here?
##########
providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py:
##########
@@ -162,12 +164,21 @@ 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] Indicates whether the options and schema
should be inferred
+ from the source data for CSV and JSON sources. (Default: ``True``).
This parameter is
+ three-state:
+
+ * ``True`` - infer the schema from the source files. The inferred
schema is sent to
+ BigQuery as the load job's schema, so ``ignore_unknown_values`` has
no effect: no field
+ in the source data can be unknown to a schema derived from that same
data. If the
+ destination table already exists and the inferred schema differs
from it, the load
+ fails with a schema mismatch unless ``schema_update_options``
permits the change.
+ * ``False`` - do not infer the schema. One of ``schema_fields`` or
``schema_object``
+ must be provided, otherwise an exception is raised.
+ * ``None`` - neither infer a schema nor supply one, so the destination
table's own
+ schema is used. The table must already exist. This is the setting to
use when you want
+ ``ignore_unknown_values`` to drop fields that are present in the
source data but absent
+ from the destination table.
Review Comment:
This is too long. There is no need to go deep into the interaction between
this and the other parameters here. I would suggest the below instead:
`:param autodetect: [Optional] Whether to infer the schema from the source
data for CSV and JSON sources. If ``True``, the schema is inferred from the
source data. If ``False``, either ``schema_fields`` or ``schema_object`` must
be provided. If ``None``, no schema is supplied and the existing destination
table's schema is used. (Default: ``True``).`
##########
providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py:
##########
@@ -638,6 +649,23 @@ def _create_external_table(self):
self.log.info("External table created successfully: %s",
self.destination_project_dataset_table)
return table_obj_api_repr
+ def _warn_if_ignore_unknown_values_is_ineffective(self) -> None:
+ # Checked against the assembled load config rather than the operator
attributes, because
+ # src_fmt_configs and extra_config can still override any of these
three keys.
+ load_config = self.configuration["load"]
+ if (
+ load_config.get("autodetect")
+ and load_config.get("ignoreUnknownValues")
+ and "schema" not in load_config
+ ):
+ self.log.warning(
+ "`ignore_unknown_values` is set but will have no effect. With
`autodetect` enabled "
+ "and no schema supplied, the load job's schema is inferred
from the source data "
+ "itself, so no source field can be unknown to it. To drop
source fields that are "
+ "absent from an existing destination table, set
`autodetect=None` so that the "
+ "destination table's own schema is used instead."
+ )
Review Comment:
1) This helper is not needed as it is only called once. I would inline it.
2) The warning is too verbose. I would suggest the below:
```
self.log.warning(
"`ignore_unknown_values` has no effect when `autodetect=True` and no
schema is provided. "
"Set `autodetect=None` to use the existing destination table's schema
instead."
)
```
##########
providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py:
##########
@@ -2124,6 +2131,110 @@ def test_schema_fields_is_templated(self):
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, "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",
+ ),
+ ],
+ )
+ @mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
+ def test_ignore_unknown_values_no_op_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") as mock_warning:
+ operator.execute(context=MagicMock())
+
+ assert
mock_warning.call_args_list.count(call(IGNORE_UNKNOWN_VALUES_WARNING)) == (
+ 1 if expects_warning else 0
+ )
Review Comment:
Do these parametrized cases need to assert the complete warning text? The
behaviour under test seems to be whether the warning is emitted. Using
`assert_called_once() / assert_not_called()` would make these cases less
coupled to the wording, with the message itself asserted in a single test if we
want to protect it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
