[data][doc] Document 2 datasets in 1 cluster usage

[TimothySeah]

Summary

Document how to run 2 datasets in 1 cluster for various use cases including training + validation.

I'm not sure whether it's better to recommend that users set the subcluster label in:

1. the global datacontext, which is more confusing (we would rather users mutate individual datasets' contexts) but achieves slightly better isolation (creation-time dataset tasks will go to the right subcluster)
2. each dataset's datacontext, which is less confusing (don't need to worry about multiple datasets concurrently using global state) but has worse isolation (creation-time dataset tasks may go to wrong subcluster)

[gemini-code-assist]

Code Review

This pull request introduces documentation and code examples for running multiple Ray Datasets concurrently on a single cluster using subclusters and label selectors. This includes a new guide on concurrent dataset execution and updates to the asynchronous validation guide to demonstrate pinning training and validation datasets to specific subclusters. The review feedback highlights that mutating the global DataContext in-place can cause unexpected side-effects and race conditions, and strongly recommends copying the context and applying it temporarily using the DataContext.current() context manager instead.

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]

Mutating the global DataContext in-place (via ray.data.DataContext.get_current()) introduces a permanent side-effect that persists for the rest of the driver's execution. This can easily lead to unexpected behavior in subsequent dataset operations.

Instead, copy the current context, customize it, and apply it temporarily using the DataContext.current() context manager. This ensures thread-safety and avoids polluting the global state.

Suggested change

	# Set the global DataContext first.
	ctx = ray.data.DataContext.get_current()
	ctx.execution_options.label_selector = {"ray-subcluster": "tenant_a"}
	# Now create the Dataset. ``Dataset.context`` is a deep copy of the global
	# ``DataContext`` taken at construction time, so it inherits the selector
	# automatically — you don't need to also set ``dataset.context``.
	dataset = ray.data.read_parquet("s3://my-bucket/tenant_a/")
	# Copy and customize the DataContext, then apply it using the context manager.
	ctx = ray.data.DataContext.get_current().copy()
	ctx.execution_options.label_selector = {"ray-subcluster": "tenant_a"}
	with ray.data.DataContext.current(ctx):
	# Dataset.context inherits the selector automatically from the active context.
	dataset = ray.data.read_parquet("s3://my-bucket/tenant_a/")

[gemini-code-assist]

Mutating the global DataContext in-place sequentially is prone to race conditions if multiple datasets are created concurrently or in different threads. Using the DataContext.current() context manager with a copied context is much safer, cleaner, and eliminates temporal coupling.

Suggested change

	ctx = ray.data.DataContext.get_current()
	# Tenant A: subcluster "tenant_a".
	ctx.execution_options.label_selector = {"ray-subcluster": "tenant_a"}
	ds_a = ray.data.read_parquet("s3://my-bucket/tenant_a/")
	# Tenant B: subcluster "tenant_b". Set the global context again
	# right before creating ds_b.
	ctx.execution_options.label_selector = {"ray-subcluster": "tenant_b"}
	ds_b = ray.data.read_parquet("s3://my-bucket/tenant_b/")
	# Tenant A: subcluster "tenant_a".
	ctx_a = ray.data.DataContext.get_current().copy()
	ctx_a.execution_options.label_selector = {"ray-subcluster": "tenant_a"}
	with ray.data.DataContext.current(ctx_a):
	ds_a = ray.data.read_parquet("s3://my-bucket/tenant_a/")
	# Tenant B: subcluster "tenant_b".
	ctx_b = ray.data.DataContext.get_current().copy()
	ctx_b.execution_options.label_selector = {"ray-subcluster": "tenant_b"}
	with ray.data.DataContext.current(ctx_b):
	ds_b = ray.data.read_parquet("s3://my-bucket/tenant_b/")

[gemini-code-assist]

Mutating the global DataContext in-place permanently affects any subsequent dataset operations in the same driver process. It is safer and more idiomatic to copy the context and apply it temporarily using the DataContext.current() context manager.

    # Set the global ``DataContext`` to the "training" subcluster BEFORE
    # creating the training Dataset so reads/schema-inference tasks land
    # on training nodes.
    ctx = ray.data.DataContext.get_current().copy()
    ctx.execution_options.label_selector = {"ray-subcluster": "training"}
    with ray.data.DataContext.current(ctx):
        train_dataset = ray.data.read_parquet(...)

[justinvyu]

I think this is better^