Skip to content

Databricks tasks generates wrong depends_on task_key adding the task itself as dependency #48623

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pacmora wants to merge 5 commits into from
Closed

Databricks tasks generates wrong depends_on task_key adding the task itself as dependency #48623

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d4846f7
Fixed dependency issue taking the object itself instead of task_id
Apr 1, 2025
fc5fc93
Added empty line
Apr 1, 2025
c23f923
Merge branch 'main' into issues_47614
pacmora Apr 4, 2025
51c0548
Merge branch 'main' into issues_47614
pacmora Apr 8, 2025
30231f4
Merge branch 'main' into issues_47614
pacmora Apr 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions providers/databricks/src/airflow/providers/databricks/operators/databricks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,8 @@
from airflow.utils.task_group import TaskGroup

if AIRFLOW_V_3_0_PLUS:
from airflow.sdk import BaseOperatorLink
from airflow.sdk import

from airflow.sdk.execution_time.xcom import XCom
else:
from airflow.models import XCom # type: ignore[no-redef]
Expand Down Expand Up @@ -1361,9 +1362,9 @@ def _convert_to_databricks_workflow_task(
result = {
"task_key": self.databricks_task_key,
"depends_on": [
{"task_key": self._generate_databricks_task_key(task_id)}
for task_id in self.upstream_task_ids
if task_id in relevant_upstreams
{"task_key": upstream_task.databricks_task_key}
for upstream_task in relevant_upstreams
if isinstance(upstream_task, DatabricksTaskBaseOperator) and upstream_task.task_id in self.upstream_task_ids
],
**base_task_json,
}
Expand Down
6 changes: 3 additions & 3 deletions providers/databricks/src/airflow/providers/databricks/operators/databricks_workflow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,8 +114,8 @@ def __init__(
self.job_clusters = job_clusters or []
self.max_concurrent_runs = max_concurrent_runs
self.notebook_params = notebook_params or {}
self.tasks_to_convert = tasks_to_convert or []
self.relevant_upstreams = [task_id]
self.tasks_to_convert = tasks_to_convert or []
self.relevant_upstreams: list[BaseOperator] = [self]
self.workflow_run_metadata: WorkflowRunMetadata | None = None
super().__init__(task_id=task_id, **kwargs)

Expand Down Expand Up @@ -333,7 +333,7 @@ def __exit__(
)

task.workflow_run_metadata = create_databricks_workflow_task.output
create_databricks_workflow_task.relevant_upstreams.append(task.task_id)
create_databricks_workflow_task.relevant_upstreams.append(task)
create_databricks_workflow_task.add_task(task)

for root_task in roots:
Expand Down