Providers: Allow opting out of forwarding Dag-level parameters in DatabricksRunNowOperator

providers/databricks/docs/operators/run_now.rst

@@ -60,7 +60,14 @@ for this run.
 If ``job_parameters`` is not set in ``json`` and the operator's ``params`` dict is
 non-empty, ``params`` is forwarded as ``job_parameters`` as-is, so Airflow Dag params can
 be passed dynamically to a run without hardcoding them in ``json``. If ``json`` already
-contains ``job_parameters``, it is left untouched.
+contains ``job_parameters``, it is left untouched. You can set ``forward_dag_params=False`` to
+disable this parameter forwarding behavior.
+
+.. note::
+  The Databricks API does not permit ``job_parameters`` to be used in combination with
+  ``notebook_params``, ``python_params``, ``jar_params``, ``spark_submit_params``,
+  ``python_named_params``, or ``dbt_commands``. Auto-forwarding is automatically skipped
+  when any of those parameter slots are used.
 
 .. code-block:: python
 

providers/databricks/src/airflow/providers/databricks/operators/databricks.py

@@ -303,6 +303,16 @@ def _handle_deferrable_databricks_operator_completion(event: dict, log: Logger)
     "run_job_task": "job_parameters",
 }
 
+# Parameter slots in run-now payload that Databricks API rejects when combined with job_parameters
+_RUN_NOW_PARAM_SLOTS_CONFLICTING_WITH_JOB_PARAMETERS = (
+    "notebook_params",
+    "python_params",
+    "jar_params",
+    "spark_submit_params",
+    "python_named_params",
+    "dbt_commands",
+)
+
 
 def _inject_airflow_params_into_task(task: dict, params: dict) -> None:
     """Set dict-shaped per-task parameter fields from ``params`` if they are not already set."""
@@ -1167,12 +1177,18 @@ class DatabricksRunNowOperator(ResumableJobMixin, BaseOperator):
         before polling begins so that a worker crash and retry reconnects to the existing run
         instead of triggering a duplicate run of the same job. Set to ``False`` to always trigger a
         fresh run on retry. Requires Airflow 3.3+; on earlier versions it is silently ignored.
+    :param forward_dag_params: Whether to forward Dag-level params as ``job_parameters``
+        when no ``job_parameters`` are specified. (default: ``True``)
 
     .. note::
         If ``job_parameters`` is not set in ``json`` and the operator's ``params`` dict is
         non-empty, the operator's ``params`` are automatically forwarded as ``job_parameters``
         so that Airflow Dag params can be passed dynamically to Databricks runs without
-        hardcoding them in ``json``.
+        hardcoding them in ``json``. Set ``forward_dag_params=False`` to disable this.
+        Note that the Databricks API does not permit ``job_parameters`` to be used in combination
+        with ``notebook_params``, ``python_params``, ``jar_params``, ``spark_submit_params``,
+        ``python_named_params``, or ``dbt_commands``; auto-forwarding is automatically skipped
+        when any of those parameters are set.
     """
 
     external_id_key = "databricks_run_now_id"
@@ -1223,6 +1239,7 @@ def __init__(
         repair_run: bool = False,
         databricks_repair_reason_new_settings: dict[str, Any] | None = None,
         cancel_previous_runs: bool = False,
+        forward_dag_params: bool = True,
         **kwargs,
     ) -> None:
         """Create a new ``DatabricksRunNowOperator``."""
@@ -1248,6 +1265,7 @@ def __init__(
         self.repair_run = repair_run
         self.databricks_repair_reason_new_settings = databricks_repair_reason_new_settings or {}
         self.cancel_previous_runs = cancel_previous_runs
+        self.forward_dag_params = forward_dag_params
 
         # This variable will be used in case our task gets killed.
         self.run_id: int | None = None
@@ -1315,7 +1333,12 @@ def _build_run_now_payload(self) -> dict[str, Any]:
             json["job_id"] = job_id
             del json["job_name"]
 
-        if not json.get("job_parameters") and self.params:
+        if (
+            self.forward_dag_params
+            and not json.get("job_parameters")
+            and self.params
+            and not any(k in json for k in _RUN_NOW_PARAM_SLOTS_CONFLICTING_WITH_JOB_PARAMETERS)
+        ):
             json["job_parameters"] = dict(self.params)
 
         return json

providers/databricks/tests/unit/databricks/operators/test_databricks.py

@@ -2028,7 +2028,7 @@ def test_execute_does_not_mutate_json_template_field(self, db_mock_class):
             durable=False,
             task_id=TASK_ID,
             job_id=JOB_ID,
-            json={"notebook_params": {"a": "b"}},
+            json={"idempotency_token": "token_123"},
             params={"env": "prod"},
         )
         op.render_template_fields(context={"ds": DATE})
@@ -2925,6 +2925,62 @@ def test_run_now_does_not_override_existing_job_parameters(self, db_mock_class):
         actual = db_mock.run_now.call_args.args[0]
         assert actual["job_parameters"] == {"explicit": "value"}
 
+    @mock.patch("airflow.providers.databricks.operators.databricks.DatabricksHook")
+    def test_run_now_does_not_inject_airflow_params_when_forward_dag_params_is_false(self, db_mock_class):
+        """
+        When ``forward_dag_params`` is False, the operator's ``params`` should
+        not be forwarded as ``job_parameters``.
+        """
+        op = DatabricksRunNowOperator(
+            durable=False,
+            task_id=TASK_ID,
+            job_id=JOB_ID,
+            forward_dag_params=False,
+            params={"env": "prod", "batch_size": 100},
+        )
+        db_mock = db_mock_class.return_value
+        db_mock.run_now.return_value = RUN_ID
+        db_mock.get_run = make_run_with_state_mock("TERMINATED", "SUCCESS")
+
+        op.execute(None)
+
+        actual = db_mock.run_now.call_args.args[0]
+        assert "job_parameters" not in actual
+
+    @pytest.mark.parametrize(
+        ("slot_param", "slot_val"),
+        [
+            ("notebook_params", {"foo": "bar"}),
+            ("python_params", ["foo", "bar"]),
+            ("jar_params", ["foo", "bar"]),
+            ("spark_submit_params", ["--class", "Foo"]),
+            ("python_named_params", {"foo": "bar"}),
+            ("dbt_commands", ["dbt deps", "dbt run"]),
+        ],
+    )
+    @mock.patch("airflow.providers.databricks.operators.databricks.DatabricksHook")
+    def test_run_now_skips_param_injection_with_legacy_param_slots(self, db_mock_class, slot_param, slot_val):
+        """
+        When any legacy param slot (notebook_params, etc.) is set, auto-injection of ``self.params``
+        into ``job_parameters`` must be skipped because the Databricks API rejects combining them.
+        """
+        op = DatabricksRunNowOperator(
+            durable=False,
+            task_id=TASK_ID,
+            job_id=JOB_ID,
+            params={"env": "prod"},
+            **{slot_param: slot_val},
+        )
+        db_mock = db_mock_class.return_value
+        db_mock.run_now.return_value = RUN_ID
+        db_mock.get_run = make_run_with_state_mock("TERMINATED", "SUCCESS")
+
+        op.execute(None)
+
+        actual = db_mock.run_now.call_args.args[0]
+        assert slot_param in actual
+        assert "job_parameters" not in actual
+
 
 @pytest.mark.skipif(
     not AIRFLOW_V_3_3_PLUS, reason="task_state_store (durable execution) requires Airflow 3.3+"