fix: use compute instead of concurrency for ray data

[nightcityblade]

Description

closes #1520

Switch the Ray Data adapter from the deprecated concurrency argument to compute for actor-backed map_batches() calls, and add a focused adapter test to guard the forwarded kwargs.

Usage

# Existing Ray Data pipelines continue to work; the adapter now forwards `compute=`
# instead of the deprecated `concurrency=` when building actor-backed map_batches calls.

Checklist

• The title of the PR starts with fix or feat, if this applies to your PR.
• Changes are manually tested and documented in the PR if automated coverage is not practical.
• New or Existing tests cover these changes.
• The documentation is up to date with these changes.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[greptile-apps]

Greptile Summary

This PR intends to fix actor-backed map_batches calls by swapping the concurrency kwarg for compute, but the direction is reversed: the Ray 2.54.1 docs explicitly mark compute as deprecated for map_batches and direct callers to use concurrency instead (map is the method where concurrency was deprecated in favour of compute).

• The adapter now passes compute= to map_batches, which is the deprecated path and may also produce incorrect behaviour since the old compute API expected an ActorPoolStrategy object rather than a plain int/tuple.
• The new test guards the wrong kwarg (compute) and would not catch a correct fix that uses concurrency=.

Confidence Score: 2/5

Not safe to merge — the core change introduces the deprecated Ray parameter for map_batches when the correct parameter was already in use.

There is a concrete P1 defect: the PR swaps concurrency for compute in map_batches calls, but Ray 2.54.1 documents compute as deprecated for that method and concurrency as the correct kwarg. The test reinforces the wrong behavior. Both files need to be corrected.

nemo_curator/backends/experimental/ray_data/adapter.py (kwarg choice) and tests/backends/experimental/ray_data/test_adapter.py (assertion on wrong kwarg).

Important Files Changed

Filename	Overview
nemo_curator/backends/experimental/ray_data/adapter.py	Switches actor-backed map_batches calls from concurrency= to compute=, but per Ray 2.54.1 docs compute is the deprecated parameter for map_batches — the change is backwards.
tests/backends/experimental/ray_data/test_adapter.py	Adds two focused adapter tests; the first asserts compute is forwarded, which mirrors the incorrect kwarg choice in the adapter and would fail if the adapter is corrected to use concurrency.

Sequence Diagram

sequenceDiagram
    participant PD as process_dataset()
    participant Ray as ray.data.Dataset.map_batches()

    PD->>PD: is_actor_stage? → True
    PD->>PD: calculate_concurrency_for_actors_for_stage() → (min, max)
    Note over PD: PR sets compute=(min,max) ❌<br/>Should set concurrency=(min,max) ✅
    PD->>Ray: map_batches(fn, batch_size=N, compute=(min,max))
    Note over Ray: compute is DEPRECATED for map_batches<br/>(Ray 2.54.1 docs)

Loading

_{Reviews (1): Last reviewed commit: "fix: use compute instead of concurrency ..." | Re-trigger Greptile}

[greptile-apps]

Wrong direction: compute is deprecated for map_batches, not concurrency

According to the Ray 2.54.1 docs, the signature for map_batches explicitly states: "compute – This argument is deprecated. Use concurrency argument." The map method is the one where concurrency was deprecated in favour of compute — but this code calls map_batches (line 111), so the PR has the direction backwards. The original use of concurrency= was correct for map_batches, and this change reintroduces the deprecated parameter.

Additionally, the old compute parameter expected a ray.data.ActorPoolStrategy object, not a plain int or (min, max) tuple. Passing compute=(1, 4) may silently mis-behave or raise at runtime when Ray processes the argument.

Suggested change

	map_batches_kwargs = {
	"compute": calculate_concurrency_for_actors_for_stage(
	self.stage, ignore_head_node=ignore_head_node
	),
	}
	map_batches_kwargs = {
	"concurrency": calculate_concurrency_for_actors_for_stage(
	self.stage, ignore_head_node=ignore_head_node
	),
	}

[greptile-apps]

Test asserts the wrong keyword argument

This assertion validates that compute is forwarded, but since map_batches should receive concurrency (not compute), the test is guarding the wrong behavior. If the adapter is corrected to use concurrency=, this assertion should match:

Suggested change

	_, kwargs = dataset.map_batches.call_args
	assert kwargs["compute"] == (1, 4)
	assert "concurrency" not in kwargs
	assert kwargs["concurrency"] == (1, 4)
	assert "compute" not in kwargs