AIP-72 Add Task Scheduling Metadata to TaskInstance

[jscheffl]

This is a sugar topping to PR #44982:
It adds support for scheduling relevant task instance fields "pool slots", "priority weight" and "queue".

Note: Only the last commit of this PR is relevant, the branch builds on top of #44982 - will rebase once the other PR is merged.

[Copilot]

Copilot reviewed 5 out of 17 changed files in this pull request and generated 4 comments.

Files not reviewed (12)

• docs/apache-airflow-providers-edge/edge_executor.rst: Language not supported
• providers/src/airflow/providers/edge/CHANGELOG.rst: Language not supported
• airflow/jobs/scheduler_job_runner.py: Evaluated as low risk
• providers/src/airflow/providers/edge/worker_api/routes/jobs.py: Evaluated as low risk
• airflow/executors/base_executor.py: Evaluated as low risk
• airflow/executors/local_executor.py: Evaluated as low risk
• airflow/executors/workloads.py: Evaluated as low risk
• providers/tests/edge/plugins/test_edge_executor_plugin.py: Evaluated as low risk
• providers/src/airflow/providers/edge/worker_api/datamodels.py: Evaluated as low risk
• providers/src/airflow/providers/edge/init.py: Evaluated as low risk
• providers/src/airflow/providers/edge/provider.yaml: Evaluated as low risk
• providers/src/airflow/providers/edge/worker_api/routes/_v2_compat.py: Evaluated as low risk

Comments suppressed due to low confidence (3)

providers/src/airflow/providers/edge/cli/edge_command.py:145

• The method poll should be called on the Popen process to update the returncode before checking it. Without calling poll, the returncode may not be updated correctly.

return self.process.returncode is None

providers/tests/edge/cli/test_edge_command.py:222

• The attribute 'generated_returncode' does not exist on the process object. This will cause an AttributeError during test execution.

job.process.generated_returncode = 0  # type: ignore[union-attr]

providers/src/airflow/providers/edge/executors/edge_executor.py:141

• The type check for 'workloads.ExecuteTask' should be done using 'issubclass' instead of 'isinstance' to properly handle cases where 'workload' might be a subclass of 'ExecuteTask'.

if not isinstance(workload, workloads.ExecuteTask):

[ashb]

@kaxil Should this be here, or in the context returned from the /run endpoint?

[jscheffl]

At least that are attributes on TaskInstance in the ORM model. Just adding these three lines filled it over. No objection moving it somewhere else but as here it is consistent with the backend.

[kaxil]

I think it will be needed for the executor to route to correct pool before the task is marked as queued

[ashb]

If this is referring to a TI it should also have the UUID column from task_instance.id

[jscheffl]

I need to switch the model - but rather in a later PR. As atm this is not critical and Airflow 2 does not have the UUID. But once we can drop Airflow 2 support that totally makes sense.

[ashb]

(I have only skimmed the surface of this PR)

[ashb]

Oh just notices this is a chained PR, I'll look at the base one instead.