Why of this read off ti, but other things just below are ti.dag_model.bundle*

Worth flagging that even off the same ti, version and version_data can disagree. Two cases:

1. Unpinned runs (disable_bundle_versioning=True): dag_run.bundle_version is None, but ti.dag_version.version_data may still carry a manifest -- so version=None, version_data={...}.
2. After _verify_integrity_if_dag_changed (scheduler_job_runner.py:2521-2530): TI's dag_version_id is bumped to the latest version while dag_run.bundle_version is left untouched, so version_data describes a newer version than version reports.

The scheduler picks a deliberate rule for bundle_version at scheduler_job_runner.py:1438-1442; worth deciding the equivalent rule for version_data here (e.g., is it valid to expose version_data when version is None?).

@ashb - What we're passing along here is the version that is being used and the bundle to use to grab it. The bundle itself isn't tracking the current version of the dag, that's on the dag_version. So once on the worker side, we'll ask the bundle to load the state of the bundle for the specific version the dag is trying to run with. So properties from these two things are combined later. Does that make sense?

@kaxil Fair point, I think the right rule is: version_data should only be populated when the run is pinned (i.e., dag_run.bundle_version is not None). If the run is unpinned, the worker should use the latest bundle state anyway, so sending stale version_data would be misleading. I'll add a guard:

version_data = None
if ti.dag_version is not None and ti.dag_run.bundle_version is not None:
    version_data = ti.dag_version.version_data 

This mirrors the existing rule for bundle_version at scheduler_job_runner.py:1438-1442.

For the second case the version_data will still come from the ti's dag_version (which may be newer than the run's), but since the run is pinned, the bundle will use version_data to check out the right code regardless. If we want stricter alignment (version_data always matches bundle_version), we'd need to look up the DagVersion by bundle_version string rather than using ti.dag_version but that adds another query and I think the current ti-based approach is correct for the worker's needs. Thoughts?

Hmm, joins aren't free, and this isn't used for most places.

I'm wondering if this needs to be based on what the bundle backend needs somehow?

selectinload uses a single additional SELECT batched across all tis in the result set (not n sized query), so the cost is one extra round-trip per scheduler loop iteration rather than per ti. It's hard to make it conditional because we're working with a bunch of tis that are mixed between versioned bundles and not. We could do a lazy load later, but then it becomes many requests, one for each TI lazy loaded. I think the impact here is pretty minimal given the single extra batched select (no new joins are happening here now)?

If the field is present it should be an object.

(I.e. its worker not sent, or its an {...} object )

The version field above uses the same approach which this uses? But happy to change it, I'll leave version as is though? @ashb?

This ended up needing a model serializer in workloads.base.BundleInfo to be able to get the field not passed over the wire if None/null. Seems maybe not worth it to me, and keeping it the same as version above seems fine. Thoughts @ashb?