This is an automated email from the ASF dual-hosted git repository.
eladkal 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 bc78f0436b3 Add performance_target parameter to
DatabricksSubmitRunOperator (#71374)
bc78f0436b3 is described below
commit bc78f0436b3ec95ea38a3d103caee3b308b85861
Author: li-xiang-db <[email protected]>
AuthorDate: Mon Aug 17 18:26:10 2026 +0200
Add performance_target parameter to DatabricksSubmitRunOperator (#71374)
The Databricks jobs/runs/submit API accepts a performance_target field
that selects the serverless compute mode (PERFORMANCE_OPTIMIZED for fast
startup, STANDARD for cost-efficient execution). The operator had no named
parameter for it, so users could only set it through the raw json payload.
Expose it as a first-class templated named parameter so it can be set
directly and merged into the submit request like the other task fields.
---
providers/databricks/docs/operators/submit_run.rst | 1 +
.../providers/databricks/operators/databricks.py | 11 +++++++++++
.../unit/databricks/operators/test_databricks.py | 21 +++++++++++++++++++++
3 files changed, 33 insertions(+)
diff --git a/providers/databricks/docs/operators/submit_run.rst
b/providers/databricks/docs/operators/submit_run.rst
index 82919d982c2..0e85b7dd8f6 100644
--- a/providers/databricks/docs/operators/submit_run.rst
+++ b/providers/databricks/docs/operators/submit_run.rst
@@ -80,6 +80,7 @@ Currently the named parameters that
``DatabricksSubmitRunOperator`` supports are
- ``libraries``
- ``run_name``
- ``timeout_seconds``
+ - ``performance_target``
.. code-block:: python
diff --git
a/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
b/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
index 34a0c47e511..6f587f39824 100644
---
a/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
+++
b/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
@@ -704,6 +704,13 @@ class DatabricksSubmitRunOperator(ResumableJobMixin,
BaseOperator):
:param do_xcom_push: Whether we should push run_id and run_page_url to
xcom.
:param git_source: Optional specification of a remote git repository from
which
supported task types are retrieved.
+ :param performance_target: Optional performance mode for the run on
serverless compute.
+ Either ``PERFORMANCE_OPTIMIZED`` (prioritizes fast startup and
execution) or
+ ``STANDARD`` (enables cost-efficient execution of serverless
workloads). This field
+ will be templated.
+
+ .. seealso::
+ https://docs.databricks.com/api/workspace/jobs/submit
:param deferrable: Run operator in the deferrable mode.
.. seealso::
@@ -750,6 +757,7 @@ class DatabricksSubmitRunOperator(ResumableJobMixin,
BaseOperator):
"idempotency_token",
"access_control_list",
"git_source",
+ "performance_target",
"databricks_conn_id",
)
template_ext: Sequence[str] = (".json-tpl",)
@@ -784,6 +792,7 @@ class DatabricksSubmitRunOperator(ResumableJobMixin,
BaseOperator):
access_control_list: list[dict[str, str]] | None = None,
wait_for_termination: bool = True,
git_source: dict[str, str] | None = None,
+ performance_target: str | None = None,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
openlineage_inject_parent_job_info: bool = conf.getboolean(
"openlineage", "spark_inject_parent_job_info", fallback=False
@@ -816,6 +825,7 @@ class DatabricksSubmitRunOperator(ResumableJobMixin,
BaseOperator):
self.idempotency_token = idempotency_token
self.access_control_list = access_control_list
self.git_source = git_source
+ self.performance_target = performance_target
self.databricks_conn_id = databricks_conn_id
self.polling_period_seconds = polling_period_seconds
self.databricks_retry_limit = databricks_retry_limit
@@ -847,6 +857,7 @@ class DatabricksSubmitRunOperator(ResumableJobMixin,
BaseOperator):
"idempotency_token": self.idempotency_token,
"access_control_list": self.access_control_list,
"git_source": self.git_source,
+ "performance_target": self.performance_target,
}
def _get_merged_json(self) -> dict[str, Any]:
diff --git
a/providers/databricks/tests/unit/databricks/operators/test_databricks.py
b/providers/databricks/tests/unit/databricks/operators/test_databricks.py
index a581bc117b3..4a5dc4bd43f 100644
--- a/providers/databricks/tests/unit/databricks/operators/test_databricks.py
+++ b/providers/databricks/tests/unit/databricks/operators/test_databricks.py
@@ -748,6 +748,27 @@ class TestDatabricksSubmitRunOperator:
assert expected == utils.normalise_json_content(op._get_merged_json())
+ def test_init_with_performance_target_named_parameter(self):
+ """
+ Test the initializer merges ``performance_target`` into the submit
payload.
+ """
+ op = DatabricksSubmitRunOperator(
+ task_id=TASK_ID,
+ new_cluster=NEW_CLUSTER,
+ notebook_task=NOTEBOOK_TASK,
+ performance_target="PERFORMANCE_OPTIMIZED",
+ )
+ expected = utils.normalise_json_content(
+ {
+ "new_cluster": NEW_CLUSTER,
+ "notebook_task": NOTEBOOK_TASK,
+ "performance_target": "PERFORMANCE_OPTIMIZED",
+ "run_name": TASK_ID,
+ }
+ )
+
+ assert expected == utils.normalise_json_content(op._get_merged_json())
+
def test_init_with_spark_python_task_named_parameters(self):
"""
Test the initializer with the named parameters.
