[Data] Track p50/p90 of streaming scheduling-loop step duration

[xinyuangui2]

Summary

• Add Timer.percentile(p) backed by a DistributionTracker (KLL sketch, k=200). Memory is bounded (~few KB) regardless of sample count, so tracking is always on — no env-var gate.
• DatasetStatsSummary exposes streaming_exec_schedule_p50_s / _p90_s. collect_dataset_stats emits p50_scheduling_loop_duration_s / p90_scheduling_loop_duration_s unconditionally in release-test output.
• Worker_scaling release benchmark: ship the profiling/ package via runtime_env.py_modules so _UDFPySpyProfiler actors can deserialize on workers (they were silently failing ModuleNotFoundError, swallowed by a broad except).
• Drop --native from py-spy (broke on the Anyscale image).
• Two unit tests (test_streaming_exec_schedule_percentiles_*) now read stats from the materialize() return value — Dataset.materialize() runs on a deep copy so stats land on the MaterializedDataset, and the original ds would yield a fresh, never-added-to Timer.

Why

max_scheduling_loop_duration_s alone hides whether a regression is a true tail blow-up or just bursty noise. KLL gives bounded-memory approximate quantiles (~1.65% rank error at k=200), so we get a useful p50/p90 signal without the production cost concern that gated earlier revisions of this PR.

Confirmed on a worker_scaling release run (5000 tasks): avg 730 ms · p50 442 ms · p90 588 ms · max 12.3 s. The max is ~20× the p90 — exactly the tail outlier max alone over-reports as the "typical" loop step.

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request introduces percentile tracking for the streaming executor's scheduling loop duration by adding an optional fixed-bound histogram to the Timer class. This allows for O(1) memory overhead when computing approximate percentiles like p90, which is now exposed in the dataset stats summary and benchmark results. Feedback was provided regarding the percentile method, suggesting that the input parameter p should be validated or clamped to the [0, 1] range to prevent unexpected behavior.

[gemini-code-assist]

Code Review

This pull request introduces percentile tracking for the scheduling loop duration in Ray Data by enhancing the Timer class with an optional fixed-bound histogram. This allows for efficient, O(1) memory percentile approximations (specifically p90) without storing raw samples. The changes include the histogram implementation using bisect for binning, integration into the streaming executor's stats, and updates to the summary and benchmark outputs. Comprehensive unit tests for the histogram logic and end-to-end integration have also been added. I have no feedback to provide.

[iamjustinhsu]

I think if u go the route i mentioned above, u can remove this env flag since the DistributionTracker will bound the memory usage, and this should make the tests simpler

[cursor]

Every Timer allocates unused KLL sketch and redundant trackers

Low Severity

Every Timer() now unconditionally creates a DistributionTracker (including a kll_doubles_sketch(200)), but only one out of 14+ Timer instances per DatasetStats — streaming_exec_schedule_s — ever calls percentile(). The iteration timers (iter_wait_s, iter_next_batch_s, etc.) fire on the per-batch hot path, and each add() now redundantly runs Welford's algorithm plus a KLL sketch update via add_sample(), in addition to Timer's own count/min/max tracking. Making the DistributionTracker opt-in (e.g. a constructor flag) would avoid the wasted allocations and per-call overhead on the 13+ timers that never need percentiles.

Additional Locations (1)

• python/ray/data/_internal/stats.py#L202-L203

 

^{Reviewed by Cursor Bugbot for commit 7d5e6ec. Configure here.}

[cursor]

Timer accesses private _quantile method across module boundary

Low Severity

Timer.percentile() calls self._distribution._quantile(p), reaching into a private method of DistributionTracker from a different module. The single-underscore convention signals this method is an internal implementation detail subject to change. Since Timer now relies on it for production percentile queries, _quantile warrants promotion to a public quantile(q) method on DistributionTracker.

Additional Locations (1)

• python/ray/data/_internal/execution/interfaces/distribution_tracker.py#L74-L78

 

^{Reviewed by Cursor Bugbot for commit 775123e. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

There are 3 total unresolved issues (including 2 from previous reviews).

^{Reviewed by Cursor Bugbot for commit 3cc3e92. Configure here.}

[cursor]

Tests assert percentile > 0 without datasketches skip guard

Medium Severity

test_streaming_exec_schedule_percentiles_populated asserts p90 > 0, but Timer.percentile() returns 0 when datasketches is not installed (it's an optional dependency, guarded by a try/except ImportError). Unlike other sketch-dependent tests in test_distribution_tracker.py which use @pytest.mark.skipif(datasketches is None, ...), this test and the TestTimerPercentile class (e.g. test_single_sample asserts percentile(p) == approx(0.042)) have no such guard. These tests will fail in any environment where datasketches is absent.

Additional Locations (1)

• python/ray/data/tests/test_stats.py#L2339-L2344

 

^{Reviewed by Cursor Bugbot for commit 3cc3e92. Configure here.}