From 455c4f68d25265aa130cbdea38e8ce58d8c61266 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 23 Jul 2026 15:09:59 -0700 Subject: [PATCH 1/3] Validate DatabricksCopyIntoOperator template fields after rendering files, table_name and file_location are template fields, rendered after __init__ runs. The constructor checked files/pattern mutual exclusivity and rejected empty table_name/file_location there, acting on the un-rendered Jinja expressions. Move those three checks into execute(); the file_format check reads no template field and stays in __init__. related: #70296 Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../databricks/operators/databricks_sql.py | 14 ++++++++------ .../operators/test_databricks_sql.py | 19 +++++++++++++++++++ .../validate_operators_init_exemptions.txt | 1 - 3 files changed, 27 insertions(+), 7 deletions(-) diff --git a/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py b/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py index f025bc0962211..0ceac229a1785 100644 --- a/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py +++ b/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py @@ -405,12 +405,6 @@ def __init__( ) -> None: """Create a new ``DatabricksCopyIntoOperator``.""" super().__init__(**kwargs) - if files is not None and pattern is not None: - raise AirflowException("Only one of 'pattern' or 'files' should be specified") - if table_name == "": - raise AirflowException("table_name shouldn't be empty") - if file_location == "": - raise AirflowException("file_location shouldn't be empty") if file_format not in COPY_INTO_APPROVED_FORMATS: raise AirflowException(f"file_format '{file_format}' isn't supported") self.files = files @@ -540,6 +534,14 @@ def _get_query_tags(self, context: Context) -> dict[str, str | None] | None: return build_query_tags(context, self.query_tags, self.include_airflow_query_tags) def execute(self, context: Context) -> Any: + # files/table_name/file_location are template fields; validate after rendering here rather + # than in __init__, where they are still the un-rendered Jinja expressions. + if self.files is not None and self._pattern is not None: + raise AirflowException("Only one of 'pattern' or 'files' should be specified") + if self.table_name == "": + raise AirflowException("table_name shouldn't be empty") + if self.file_location == "": + raise AirflowException("file_location shouldn't be empty") self._sql = self._create_sql_query() self.log.info("Executing: %s", self._sql) hook = self._get_hook() diff --git a/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py b/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py index 1f4513b886a2f..d48fa45719d33 100644 --- a/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py +++ b/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py @@ -610,3 +610,22 @@ def test_execute_sets_query_tags_on_hook(self): op.execute(None) assert mock_hook.query_tags == {"env": "prod"} + + +class TestDatabricksCopyIntoOperatorValidation: + """files/table_name/file_location are template fields; their checks run at execute.""" + + def test_files_and_pattern_rejected_at_execute(self): + from airflow.providers.common.compat.sdk import AirflowException + from airflow.providers.databricks.operators.databricks_sql import DatabricksCopyIntoOperator + + op = DatabricksCopyIntoOperator( + task_id=TASK_ID, + table_name="test_table", + file_location="s3://bucket/path", + file_format="CSV", + files=["a.csv"], + pattern="*.csv", + ) + with pytest.raises(AirflowException, match="Only one of 'pattern' or 'files'"): + op.execute(context={}) diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index 01fd5bc56dbb7..ca23584fc2833 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -39,7 +39,6 @@ providers/common/ai/src/airflow/providers/common/ai/operators/document_loader.py providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py::DatabricksReposCreateOperator providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py::DatabricksReposDeleteOperator providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py::DatabricksReposUpdateOperator -providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py::DatabricksCopyIntoOperator providers/databricks/src/airflow/providers/databricks/sensors/databricks.py::DatabricksSQLStatementsSensor providers/dbt/cloud/src/airflow/providers/dbt/cloud/operators/dbt.py::DbtCloudGetJobRunArtifactOperator providers/docker/src/airflow/providers/docker/operators/docker.py::DockerOperator From af59be9cd9e20e2aa27859e6ad8da42855522f3b Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 23 Jul 2026 15:55:10 -0700 Subject: [PATCH 2/3] Update DatabricksCopyIntoOperator param-validation tests for execute-time checks The files/pattern, empty table_name and empty file_location checks moved from __init__ to execute(); trigger them via execute() in test_databricks_copy.py and drop the now-redundant duplicate added in test_databricks_sql.py. Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../operators/test_databricks_copy.py | 43 ++++++++++--------- .../operators/test_databricks_sql.py | 19 -------- 2 files changed, 23 insertions(+), 39 deletions(-) diff --git a/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py b/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py index f00653f8b22c8..1d369a3343b05 100644 --- a/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py +++ b/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py @@ -195,37 +195,40 @@ def test_copy_with_validate_N_rows(): def test_incorrect_params_files_patterns(): exception_message = "Only one of 'pattern' or 'files' should be specified" + op = DatabricksCopyIntoOperator( + task_id=TASK_ID, + file_location=COPY_FILE_LOCATION, + file_format="JSON", + table_name="test", + files=["file1", "file2", "file3"], + pattern="abc", + ) with pytest.raises(AirflowException, match=exception_message): - DatabricksCopyIntoOperator( - task_id=TASK_ID, - file_location=COPY_FILE_LOCATION, - file_format="JSON", - table_name="test", - files=["file1", "file2", "file3"], - pattern="abc", - ) + op.execute(context={}) def test_incorrect_params_emtpy_table(): exception_message = "table_name shouldn't be empty" + op = DatabricksCopyIntoOperator( + task_id=TASK_ID, + file_location=COPY_FILE_LOCATION, + file_format="JSON", + table_name="", + ) with pytest.raises(AirflowException, match=exception_message): - DatabricksCopyIntoOperator( - task_id=TASK_ID, - file_location=COPY_FILE_LOCATION, - file_format="JSON", - table_name="", - ) + op.execute(context={}) def test_incorrect_params_emtpy_location(): exception_message = "file_location shouldn't be empty" + op = DatabricksCopyIntoOperator( + task_id=TASK_ID, + file_location="", + file_format="JSON", + table_name="abc", + ) with pytest.raises(AirflowException, match=exception_message): - DatabricksCopyIntoOperator( - task_id=TASK_ID, - file_location="", - file_format="JSON", - table_name="abc", - ) + op.execute(context={}) def test_incorrect_params_wrong_format(): diff --git a/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py b/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py index d48fa45719d33..1f4513b886a2f 100644 --- a/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py +++ b/providers/databricks/tests/unit/databricks/operators/test_databricks_sql.py @@ -610,22 +610,3 @@ def test_execute_sets_query_tags_on_hook(self): op.execute(None) assert mock_hook.query_tags == {"env": "prod"} - - -class TestDatabricksCopyIntoOperatorValidation: - """files/table_name/file_location are template fields; their checks run at execute.""" - - def test_files_and_pattern_rejected_at_execute(self): - from airflow.providers.common.compat.sdk import AirflowException - from airflow.providers.databricks.operators.databricks_sql import DatabricksCopyIntoOperator - - op = DatabricksCopyIntoOperator( - task_id=TASK_ID, - table_name="test_table", - file_location="s3://bucket/path", - file_format="CSV", - files=["a.csv"], - pattern="*.csv", - ) - with pytest.raises(AirflowException, match="Only one of 'pattern' or 'files'"): - op.execute(context={}) From db723beb2e25a52bebde093e539d905da1cada6e Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 23 Jul 2026 23:14:03 -0700 Subject: [PATCH 3/3] Tighten DatabricksCopyIntoOperator validation comment Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../airflow/providers/databricks/operators/databricks_sql.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py b/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py index 0ceac229a1785..f44278982aaa1 100644 --- a/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py +++ b/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py @@ -534,8 +534,7 @@ def _get_query_tags(self, context: Context) -> dict[str, str | None] | None: return build_query_tags(context, self.query_tags, self.include_airflow_query_tags) def execute(self, context: Context) -> Any: - # files/table_name/file_location are template fields; validate after rendering here rather - # than in __init__, where they are still the un-rendered Jinja expressions. + # files/table_name/file_location are template fields; validate after rendering. if self.files is not None and self._pattern is not None: raise AirflowException("Only one of 'pattern' or 'files' should be specified") if self.table_name == "":