Validate DatabricksCopyIntoOperator template fields after rendering

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,13 @@ 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.
+        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()

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():

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