[DO NOT MERGE] Repro 71d95f0503 state for bisect baseline

[TimothySeah]

Why

Repro of the test branch state at commit 71d95f05030b1a2892981ee4e48549617619e386,
which was the last known-passing run of the multitenancy release test
(heterogeneous_memory_batch_inference_multitenancy) at ~5% overhead.

The branch's master base is 750ef4e50632b95b3ae84ab51c4c3182bac2631c
(2026-05-29). All autoscaler-subcluster work in this branch corresponds to
PR #63375 which has since landed on master as 5d2c4e709b. This PR is a
duplicate vs current master on purpose — its sole use is to trigger the
release test on the historic state for bisect comparison.

Do not merge. Close after the release test result is captured.

Test plan

• Trigger the heterogeneous_memory_batch_inference_multitenancy release
  test on this PR's CI.
• Compare per-tenant ds.stats() output (esp. Cluster memory: Spilled to disk) and overhead ratio against the current branch's failing run.

[gemini-code-assist]

Code Review

This pull request introduces subcluster-based isolation for Ray Data's cluster autoscaler to support multitenancy, ensuring that resource requests and allocations are scoped to specific subclusters. The feedback highlights a few critical issues: a potential bypass of cross-subcluster checks when subcluster_selector is falsy, false-positive errors when comparing entire selector dicts instead of just the subcluster key, a PEP 8 violation from assigning a lambda to a variable, and the risk of discarding other configured labels by overwriting the label_selector dict instead of merging it.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The check for cross-subcluster requests is skipped if subcluster_selector is falsy (e.g., None or {}). This allows a requester in the default/None subcluster to submit bundles targeting a specific subcluster (e.g., training), which is a cross-subcluster request and should be rejected. Additionally, if subcluster_selector is truthy but does not contain SUBCLUSTER_LABEL_KEY (e.g., {"other_key": "val"}), it will incorrectly raise a ValueError for valid requests because req_subcluster is None while bundle_subcluster is not None. Simplifying the check to retrieve req_subcluster safely and checking against bundle_subcluster only when bundle_subcluster is explicitly set resolves both issues.

Suggested change

	if subcluster_selector and label_selectors:
	req_subcluster = subcluster_selector.get(SUBCLUSTER_LABEL_KEY)
	for i, sel in enumerate(label_selectors):
	bundle_subcluster = sel.get(SUBCLUSTER_LABEL_KEY)
	if (
	bundle_subcluster is not None
	and bundle_subcluster != req_subcluster
	):
	raise ValueError(
	f"Bundle {i} label_selector targets subcluster "
	f"{bundle_subcluster!r}, but requester is registered to "
	f"{req_subcluster!r}. Per-bundle cross-subcluster "
	f"allocation is not supported."
	)
	req_subcluster = (subcluster_selector or {}).get(SUBCLUSTER_LABEL_KEY, DEFAULT_SUBCLUSTER)
	if label_selectors:
	for i, sel in enumerate(label_selectors):
	bundle_subcluster = sel.get(SUBCLUSTER_LABEL_KEY, DEFAULT_SUBCLUSTER)
	if (
	bundle_subcluster is not None
	and bundle_subcluster != req_subcluster
	):
	raise ValueError(
	f"Bundle {i} label_selector targets subcluster "
	f"{bundle_subcluster!r}, but requester is registered to "
	f"{req_subcluster!r}. Per-bundle cross-subcluster "
	f"allocation is not supported."
	)

[gemini-code-assist]

Comparing the entire subcluster_selector dict directly can lead to false-positive ValueError exceptions. For example, if the selector changes from None to {} (which are semantically identical for subcluster routing), the check will fail. Additionally, if other non-subcluster labels in the selector change (e.g., zone or instance type), it shouldn't block the request. We should only validate that the actual subcluster value itself has not changed.

                old_subcluster = (self._subcluster_selectors.get(requester_id) or {}).get(SUBCLUSTER_LABEL_KEY, DEFAULT_SUBCLUSTER)
                new_subcluster = (subcluster_selector or {}).get(SUBCLUSTER_LABEL_KEY, DEFAULT_SUBCLUSTER)
                if old_subcluster != new_subcluster:
                    raise ValueError(
                        "Cannot change subcluster of an ongoing request "
                        f"from {old_subcluster!r} to {new_subcluster!r}."
                    )

[gemini-code-assist]

Assigning a lambda expression to a local variable violates PEP 8 (E731). Instead of using a lambda and suppressing the warning with # noqa: E731, use a nested def statement to define the helper function.

            def get_node_counts():
                return _get_node_resource_spec_and_count(subcluster=subcluster)

References

1. PEP 8: Always use a def statement instead of an assignment statement that binds a lambda expression directly to an identifier. ^(link)

[gemini-code-assist]

Overwriting the entire label_selector dict with {"subcluster": subcluster} will discard any other labels that might have been configured in execution_options.label_selector. To prevent breaking other label-based scheduling constraints, we should merge the new label into the existing dict.

    if subcluster is not None:
        # Set on the process-global DataContext first: some operator paths
        # read via DataContext.get_current() at task-submission time, which
        # is thread-local in places, so the per-Dataset setting alone isn't
        # sufficient when callers run on a non-driver thread.
        ctx = ray.data.DataContext.get_current()
        current_selector = ctx.execution_options.label_selector or {}
        ctx.execution_options.label_selector = {
            **current_selector,
            "subcluster": subcluster,
        }

    ds = ray.data.range(num_rows)

    if subcluster is not None:
        # Also pin on the Dataset's own context so chained ops inherit it.
        current_selector = ds.context.execution_options.label_selector or {}
        ds.context.execution_options.label_selector = {
            **current_selector,
            "subcluster": subcluster,
        }