[Data] Decrease downstream backpressure ratio to 2.0

[bveeramani]

By default, Ray Data applies DownstreamCapacity backpressure once an operator queues up 10x as much data as the downstream operator uses. This conservative default causes Ray Data to queue up more objects than it would otherwise need to.

In this PR, I'm ratcheting down the default from 10 to 2.

@yuhuan130 performed a grid search over different values to find the value that uses the least amount of object store memory while not causing regressions. In his experiments, he found 1 to be the best value. You can see some of his results here: https://docs.google.com/spreadsheets/d/1_OGToIZpk-8HOy7fqn6QmuBRBisEBIRDOhYWgFL434U/edit?gid=0#gid=0.

Since there might be unknown unknown with changing this default, I'm more conservatively setting the value to 2 rather than 1.

Results

tl;dr: 10-25% peak memory usage across linear workloads with no meaningful runtime regressions.

Peak memory

Benchmark	Peak Δ	Runtime Δ
image_embedding_from_jsonl_fixed_size	−44.7%	−11.5%
large_image_embedding.ray	−43.4%	−61.7%
image_embedding_from_jsonl_autoscaling	−34.9%	+10.7%
mix.8ds_equal_random_mix	−31.8%	−0.1%
mix.8ds_power_law_random_mix	−31.2%	−11.5%
read_parquet_fixed_size	−26.2%	+12.7%
audio_transcription.ray	−24.2%	+10.4%
training_ingest_benchmark.s3_parquet_cpu	−23.4%	−0.3%
video_object_detection.ray	−21.7%	+8.2%
read_from_uris_autoscaling	−20.7%	+7.3%
iter_torch_batches	−18.2%	−3.9%

Runtime

Benchmark	Runtime Δ
large_image_embedding.ray	−61.7%
cross_az_map_batches_autoscaling.gce	−28.1%
distributed_training.chaos	−17.2%
worker_scaling_5000_tasks_1ops	−17.2%
flat_map	−16.5%
map	−15.3%
cross_az_map_batches_autoscaling.aws	−13.3%
image_embedding_from_jsonl_fixed_size	−11.5%
image_classification_from_parquet_autoscaling	−6.0%

Runtime regressions

I've manually verified that these are all noise (e.g., autoscaling and read tests have high variance).

Benchmark	Runtime Δ	Peak Δ
streaming_split.regular	+40.5%	+120.5%
read_from_uris_fixed_size	+18.3%	−18.5%
read_parquet_fixed_size	+12.7%	−26.2%
image_embedding_from_jsonl_autoscaling	+10.7%	−34.9%
read_tfrecords	+10.4%	+2.1%
audio_transcription.ray	+10.4%	−24.2%
count_parquet_autoscaling	+8.6%	−0.2%
video_object_detection.ray	+8.2%	−21.7%
read_from_uris_autoscaling	+7.3%	−20.7%

[gemini-code-assist]

Code Review

This pull request updates the default downstream capacity backpressure ratio (DEFAULT_DOWNSTREAM_CAPACITY_BACKPRESSURE_RATIO) in python/ray/data/context.py from 10.0 to 2.0. There are no review comments, so I have no feedback to provide.

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.

[justinvyu]

Checked the training ingest test results and they look good. High leverage change alert ⚠️ !

🚢

training_ingest_regression_test.peak_object_store_memory

Metric	Master	PR	Δ (%)	Verdict
peak obj store (GiB)	75.14 ± 1.88	70.65 ± 1.80	−4.49 (−6.0%)	win
throughput (rows/s)	473.2 ± 1.9	470.36 ± 1.74	−2.84 (−0.6%)	flat (noise)

training_ingest_regression_test.throughput

Metric	Master	PR	Δ (%)	Verdict
peak obj store (GiB)	58.09 ± 1.47	52.10 ± 0.76	−5.99 (−10.3%)	win
throughput (rows/s)	1279.6 ± 31.0	1376.0 ± 27.2	+96.4 (+7.5%)	win
next-batch (ms)	49.42 ± 19.93	39.22 ± 14.21	−10.2 (−20.6%)	favorable (noisy)

[justinvyu]

Before you merge, can you add another picture of the "operator time spent in downstream capacity backpressure" + "bytes output by operator" dashboard metrics before/after this PR for a release test that heavily improved? Ex: image_embedding_from_jsonl_fixed_size

Purpose is to sanity check the expected behavior change from this change.

[bveeramani]

Before you merge, can you add another picture of the "operator time spent in downstream capacity backpressure" + "bytes output by operator" dashboard metrics before/after this PR for a release test that heavily improved? Ex: image_embedding_from_jsonl_fixed_size

Purpose is to sanity check the expected behavior change from this change.

Here're the metrics for large_image_embedding

Master

Backpressure is conservative and the cluster is large, so the system doesn't backpressure the read and runs the operator to completion.

2026-06-28 23:38:53,016 INFO logging_progress.py:174 -- ======= Running Dataset: dataset_5_0 =======
2026-06-28 23:38:53,016 INFO logging_progress.py:225 -- Total Progress: 19/1041
2026-06-28 23:38:53,016 INFO logging_progress.py:227 -- Active & requested resources: 0/2208 CPU, 1/40 GPU, 790.6GiB/2.6TiB object store (pending: 1 GPU)
2026-06-28 23:38:53,016 INFO logging_progress.py:181 -- 
2026-06-28 23:38:53,016 INFO logging_progress.py:231 -- ReadParquet->Map(decode)->Map(preprocess): 1431167/1431167
2026-06-28 23:38:53,016 INFO logging_progress.py:233 --   Tasks: 0; Actors: 0; Queued blocks: 0 (0.0B); Resources: 0.0 CPU, 790.6GiB object store
2026-06-28 23:38:53,016 INFO logging_progress.py:231 -- MapBatches(Infer): 21575/1181901
2026-06-28 23:38:53,016 INFO logging_progress.py:233 --   Tasks: 2; Actors: 2 (running=1, restarting=0, pending=1, active=1, idle=0, util=1.000, tasks_in_flight=2); Queued blocks: 9078 (789.3GiB); Resources: 0.0 CPU, 1.0 GPU, 4.4MiB object store; [3/21 objects local]
2026-06-28 23:38:53,017 INFO logging_progress.py:231 -- Write: 19/1041
2026-06-28 23:38:53,017 INFO logging_progress.py:233 --   Tasks: 0; Actors: 0; Queued blocks: 0 (0.0B); Resources: 0.0 CPU, 0.0B object store

This PR

Backpressure kicks in once the operator hits the 50% budget threshold and prevents the read from producing more data.

We can improve this workload further by ratcheting down or removing the 50% threshold, but that's out of scope for this change.

2026-06-25 14:23:51,086 INFO logging_progress.py:174 -- ======= Running Dataset: dataset_5_0 =======
2026-06-25 14:23:51,086 INFO logging_progress.py:225 -- Total Progress: 24/912
2026-06-25 14:23:51,086 INFO logging_progress.py:227 -- Active & requested resources: 618/2208 CPU, 53.6GiB/10.3TiB memory, 3/40 GPU, 503.3GiB/2.6TiB object store (pending: 4.4MiB memory, 1 GPU)
2026-06-25 14:23:51,086 INFO logging_progress.py:181 -- 
2026-06-25 14:23:51,086 INFO logging_progress.py:231 -- ReadParquet->Map(decode)->Map(preprocess): 813940/1405256
2026-06-25 14:23:51,086 INFO logging_progress.py:233 --   Tasks: 618 [backpressured:tasks(DownstreamCapacity),outputs(DownstreamCapacity)]; Actors: 0; Queued blocks: 3174 (0.0B); Resources: 618.0 CPU, 53.6GiB memory, 503.3GiB object store
2026-06-25 14:23:51,087 INFO logging_progress.py:231 -- MapBatches(Infer): 27077/1029171
2026-06-25 14:23:51,087 INFO logging_progress.py:233 --   Tasks: 6; Actors: 4 (running=3, restarting=0, pending=1, active=3, idle=0, util=1.500, tasks_in_flight=6); Queued blocks: 4932 (437.7GiB); Resources: 0.0 CPU, 4.3MiB memory, 3.0 GPU, 13.3MiB object store; [3/30 objects local]
2026-06-25 14:23:51,087 INFO logging_progress.py:231 -- Write: 24/912
2026-06-25 14:23:51,087 INFO logging_progress.py:233 --   Tasks: 0; Actors: 0; Queued blocks: 0 (0.0B); Resources: 0.0 CPU, 0.0B object store
2026-06-25 14:23:51,087 INFO logging_progress.py:192 -- ============================================