[Data] [DO NOT MERGE] arrow-rs Parquet reader A/B — treatment arm (release-test trigger) - #65406
Open
AarryaSaraf wants to merge 137 commits into
Open
[Data] [DO NOT MERGE] arrow-rs Parquet reader A/B — treatment arm (release-test trigger)#65406AarryaSaraf wants to merge 137 commits into
AarryaSaraf wants to merge 137 commits into
Conversation
…et V2 Replace the blind, size-based ParquetFileChunker + RoundRobinPartitioner path for Parquet V2 reads with a FooterFileIndexer that reads each file's Parquet footer on a pool of FooterReader actors (SPREAD across the cluster) and packs row groups into read units with an online, locality-preserving bin packer. This makes read units row-group-accurate instead of size-estimated, and adds predicate pushdown + row-group skipping and limit pushdown into listing: the optimizer rules mirror the pushed predicate/projection/limit from the ReadFiles scanner onto the upstream ListFiles (sync_list_files_pushdown), so the footer actors prune row groups by statistics, size only projected columns, and stop listing early under a limit. Optional row-group coalescing + split-packing are supported and gated (no-op by default). The footer indexer reuses the generic FileIndexer abstraction (ListFiles drives it via the ordinary list_files transform, yields_read_units=True so it lists in a single task and skips the partitioner). The blind ParquetFileChunker / ParquetFileChunkMetadata and their size-based row-group range helpers are removed; the reader now reads exactly the row groups a bin names. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
LimitPushdownRule._push_limit_down returned early ("no row-preserving ops")
when a Limit was directly above ReadFiles, so it never applied the per-block
limit -> the scanner limit and (footer path) ListFiles.limit were left unset.
The footer indexer then read every file's footer regardless of the limit.
Push the per-block limit into a directly-attached ReadFiles (keeping the Limit
on top for exact enforcement), which also mirrors the limit onto ListFiles via
sync_list_files_pushdown so the footer indexer stops listing early. Scoped to
ReadFiles to keep other sources unchanged.
read_parquet(s3://...).limit(10_000) on a 260-file dataset: ~9.8s -> ~3.1s.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Signed-off-by: Goutam <goutam@anyscale.com>
Follow-up refinements on the footer-based row-group chunking path: - Carry projection-scoped uncompressed_size on ParquetRowGroupChunkMetadata so the reader sizes batches from footer stats ListFiles already read, instead of re-reading the footer in the read task. - Read the exact physical row groups a bin assigned via explicit row_group_ids fragments (_fragments_from_row_group_ids), scanning a file's groups together unless row hashing needs per-group offsets. - Size Arrow's process-wide IO/CPU thread pools per read task so a scan can issue many concurrent column/range fetches against S3 (RAY_DATA_PARQUET_READER_IO_THREAD_COUNT / _CPU_COUNT). - Add RAY_DATA_PARQUET_FOOTER_MAX_INFLIGHT_BATCHES to bound the in-flight footer-batch window independently of actor count. - Use a fixed fallback bin budget (RAY_DATA_PARQUET_FOOTER_BIN_BYTES, default 1 GiB) for read-task packing. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
A shared bin at or over cap can never accept another positive-size item, so leaving it in the pool just burns one of the max_shared_open_bins slots and delays draining. Seal and evict it immediately from both the whole-item fast path and the best-fit split loop. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
`PushdownCountFiles` rebuilt `ListFiles` by deep-copying the existing indexer and overwriting its `_file_chunker`. `FooterFileIndexer` subclasses `NonSamplingFileIndexer`, so the `isinstance` assert passed -- but it overrides `list_files` outright and never reads the chunker, so the swap was a silent no-op. The result on `count_parquet_fixed_size` (10k files): listing still footer-swept every file, and because `count()` projects to zero columns every row group measured 0 bytes, so the bin cap was never reached and the whole dataset collapsed into one bin -- one manifest block, one count task that then re-read all 10k footers serially. 30.5s -> 219.8s. Build the plain indexer explicitly via a new `NonSamplingFileIndexer.as_whole_file_indexer()`, and bail out rather than assert when the indexer isn't one (the assert was stripped under -O, which is the same failure mode as this bug). This restores one manifest row per file and parallel count tasks, and removes a latent over-count: the bin packer emits one row per path per bin, so a file spanning bins would have been counted more than once. Adds the first test coverage for the rule, including an exact-type assertion on the rebuilt indexer -- an isinstance check is what let the footer indexer through. Also in this commit: - lower the bin-packing default from 1 GiB to 64 MiB - expose `max_shared_open_bins` via `RAY_DATA_PARQUET_BIN_PACKING_MAX_SHARED_OPEN_BINS` - experiment env vars on the `wide_schema_pipeline` release test - `map_benchmark` reads via V2 instead of pinning V1 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…gets ## Listing parallelism `ListFiles` forced single-task listing whenever the indexer sets `yields_read_units` -- i.e. always, for Parquet V2 -- on the grounds that the bin packer "must see the whole file stream to pack globally". But the V1 partitioner it replaced also runs inside each listing task, so global packing was never the status quo, and the cost is real: a read that hands in explicit paths loses listing parallelism entirely. `distributed_training` passes 200 explicit parquet paths. Master sharded them across 200 listing tasks (3.00s); the footer path did all 200 footer reads in one (5.81s), ~65% of that test's +15% regression, with block count and block size otherwise identical to master. Drop `yields_read_units` from the `should_parallelize` condition. Only the shuffle-RNG case still needs a single task. Reads that pass one directory -- the common case -- shard to one task either way and are unaffected. The footer reader pool is provisioned per `list_files` call, so it is now per listing task rather than cluster-wide. Documented as such on `RAY_DATA_PARQUET_FOOTER_NUM_ACTORS`, and `distributed_training` sets it to 1 since each of its tasks reads a single footer. ## Per-test bin budgets The 64 MiB default is right for some datasets and badly wrong for others, because the useful budget tracks per-file projected bytes -- which ranges from 0.45 MiB to 3.87 GiB across the suite, and varies with the query's projection on the same table. Measured from parquet footers and verified against the V1 partitioner's block counts: - read_large_parquet 4 GiB (3.87 GiB/file; 64 MiB -> 5768 tasks) - write_parquet, joins, 1.25 GiB (~1.17 GiB/file, full schema) aggregate_groups, map_groups - tpch_q20 192 MiB (160 MiB/file, 4-column projection) - streaming_split 36 MiB (V1 landed on ~31 MB blocks here) The map_groups value also fixes four outright failures, not just a slowdown: `map_groups_*_sort_shuffle_pull_based` OOM'd the driver at 64 MiB. Pull-based shuffle holds `input_blocks * output_blocks` object refs on the driver, so at 2500 blocks that is 17.5 GiB (observed 21.6 GiB RSS, cgroup kill). At 1.25 GiB it is ~95 blocks and 0.03 GiB. `wide_schema` gets no budget: those datasets are 10-40 MB total across 9-27 single-row-group files, so every budget from 64 MiB up packs them into one bin. They need a minimum-block-count floor instead. `mix` gets no budget either -- it already matches V1 granularity -- only a narrower reader pool, since it opens 8 datasets over the same 513-file path. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…tunable ## Revert: parallel listing for the footer indexer 71a1bf2 dropped `yields_read_units` from `should_parallelize` so a read handing in explicit paths would shard listing across tasks. Release build 102070 shows that was wrong. `FooterFileIndexer.list_files` pays a fixed cost *per call* -- provisioning a reader-actor pool, building a packer, serializing manifests -- so N tasks pay it N times instead of splitting it. On `distributed_training` (200 explicit paths) each of 200 listing tasks took 55.9s, 50.7s of it in block generation, against 5.35s for one task doing all 200. Aggregate listing work went 5.35 -> 11,180 task-seconds. `distributed_training.chaos` went +21% -> +463% vs master; `.regular` +15% -> +28%. Restore the original condition. Parallel listing here needs a pool shared across tasks, not one per task, which is a larger change. ## RAY_DATA_PARQUET_FOOTER_RESULT_BATCH_SIZE `read_footers` has always taken `result_batch_size` to amortize driver-side object-store fetches, but the indexer never passed it, so it defaulted to 1: a directory of N files costs N fetches on the single listing task (7.4k for the imagenet dataset). Plumb it through as an env-backed setting, default 1 so behavior is unchanged. Note it batches *within* one `read_footers` call, so it saturates at `RAY_DATA_PARQUET_FOOTER_BATCH_SIZE`; both have to move together. ## Release test config - `streaming_split`: batch sizes 50/50, cutting driver fetches ~7.4k -> ~149. ListFiles dominates this test (15.5s of a 26.6s run) and its cost tracks results pulled back, not footer IO -- raising the actor pool 32 -> 128 left block generation flat at ~7 ms/block. Bin budget reverted to the default: 36 MiB was matching V1's block size, but V1 only lands on ~31 MB blocks because its 5x encoding estimate over-counts by 5x on image data that does not compress. Matching it doubles the block count and makes the dominant cost worse. - `distributed_training`: footer actors 1 -> 20. 200 paths at a batch size of 10 is 20 batches, so actors past the 20th are never dispatched to. Also carried: `@ray.method(num_returns="streaming")` on `read_footers`, re-reading the actor count at construction so env overrides apply after import, a `None`-size guard in `sample_files`, and `None`-safety in the bin packer's heavy path. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
Release-test tuning for the Parquet V2 footer read path, plus one real regression the benchmark itself introduced. map_batches: 26cdef2 added ``num_cpus=0.99`` to ``read_parquet`` in map_benchmark.py. The fusion rule canonicalizes an unspecified ``num_cpus`` to 1, so 0.99 makes the read incompatible with the downstream map_batches ops and they stop fusing. Unfused, every read output crosses the object store: 1.13 TiB on sf1000 lineitem and 484.5s of block serialization against 0.6s fused, taking the test from 141s to 335s vs master. Drop the arg and pin a 1 GiB bin budget -- the 64 MiB default is worst-case on this table (three 24.36 MiB row groups exceed it, so bins take two and sit 24% empty, giving 24,432 read tasks against master's 1,000). aggregate_groups: 32 MiB on both shuffle strategies. First setting to move fixed_size/84-groups/sort_shuffle, stuck at 220-259s across four builds under every other budget; 32 MiB brings it to 192s. This is the opposite direction from map_groups, which wants 1.25 GiB on the same dataset. tpch_q1: 192 MiB, matching q20 on the same table. Also corrects comments that described master's read path as V1. Master is DSv2 with the round-robin partitioner for every suite except map_batches, which is V1 there because DEFAULT_USE_DATASOURCE_V2 is False on master and ray-project#64821 (branch-only) flips it. Tests: pytest python/ray/data/tests/datasource/test_parquet.py::test_parquet_read_spread python/ray/data/tests/test_predicate_pushdown.py -- 51 passed. AI assistance was used for this change. Signed-off-by: Goutam <goutam@anyscale.com>
aggregate_groups: move the bin budget from the block to a per-variant setting via matrix.adjustments, with bin_packing_bytes as a fourth matrix variable. Only fixed_size/sort_shuffle/84-groups needed retuning -- it ran 211s at 32 MiB in release build #102329 against master's ~145s, and 137s at 256 MiB. The other seven variants keep 32 MiB, where they were measured healthy in that same build (+10.2% to -30.7%), rather than move onto an unmeasured value. The cost being tuned is per-block sort-shuffle overhead, not bytes: at 32 MiB the scan emitted 4x the ListFiles outputs and ~1.5x the blocks into the shuffle (Sort Sample 3050 vs master's 2000) for an identical 20.8GB object-store peak and identical row count. cross_az_map_batches_autoscaling_iptable_failure_injection: pin 1 GiB. These tests moved from the V1 to the V2 read path when the use_datasource_v2=False pin came out of map_benchmark.py; on V2 at the 64 MiB default the object-store peak doubled (310GB -> 616GB) and runtime went 304s -> 725s. The larger budget is a mitigation and is not yet measured. Also request num_cpus=0 for FooterReaderActor. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
Conflict in release/release_data_tests.yaml: master added a third shuffle_strategy (hash_shuffle_v2) to the aggregate_groups matrix, while this branch had converted that block from matrix.setup to matrix.adjustments so each variant can carry its own bin budget. Kept both -- the four new hash_shuffle_v2 variants are enumerated as adjustments alongside the existing eight. The new variants get the 64 MiB branch default rather than the 32 MiB the other variants carry: 32 MiB is a deliberate pin for variants that were measured at it in release build #102329, and hash_shuffle_v2 has never run on this branch. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…get to 128MiB The Parquet footer indexer's default 32-actor pool times out (and trips Ray's "too many worker processes" warning, polluting doctest output) when many test targets run in parallel under CI. Rather than sprinkling per-module env writes, set the pool size to 1 in one place per test surface: - New python/ray/data/test.bzl wraps py_test/py_test_module_list/doctest to inject RAY_DATA_PARQUET_FOOTER_NUM_ACTORS=1 into every Data test target's env (caller-supplied env still wins). - python/ray/data/tests/conftest.py and doctest_pytest_plugin.py setdefault the same var so non-bazel pytest runs match. - python/ray/train/v2 gets shared _TRAIN_V2_TEST_ENV dicts plus a conftest setdefault, replacing the repeated per-target env literals. - Drops the now-redundant os.environ write from test_predicate_pushdown.py. Also raises the fallback bin-packing budget from 64MiB to 128MiB when target_max_block_size is unset, and fixes the write_lance docstring example to pass SaveMode.OVERWRITE so repeated doctest runs don't fail on an existing dataset. AI assistance was used for this change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
Signed-off-by: Goutam <goutam@anyscale.com>
…ault Release build #102597 raised the footer indexer's fallback bin budget from 64 to 128 MiB. Comparing each test's delta-vs-its-own-master between #102329 and #102597, 24 of the 38 tests running on the fallback default got worse and 14 improved. This pins the 24 back to 64 MiB explicitly; the 14 keep following the raised default. The split is directional rather than arbitrary: among Parquet tests the regressions are read-dominated (read_parquet_autoscaling -20.9% -> +3.1%, read_large_parquet_autoscaling -8.1% -> +32.9%, count_parquet_autoscaling -15.4% -> -1.7%) while the improvements are shuffle/sort-heavy (random_shuffle_chaos +67.8% -> +14.9%, sort_fixed_size +22.2% -> +2.1%). Same tension the aggregate_groups and map_groups comments already describe. read_from_uris, iceberg_benchmark and iter_batches are split out of their matrices so only the affected variant is pinned and the others keep following the default. training_ingest_benchmark is pinned at block level with s3_read_images_gpu exempted. Caveats recorded in the per-test comments: - Eight of the 24 pins are inert. read_images_*, read_tfrecords and the s3_url_image_* / s3_read_images_cpu ingest variants do not read Parquet. mix reads 513 single-row-group files of ~88 MB, which pack one file per bin at both 64 and 128 MiB. These tests did move, but not because of this knob. - mix and distributed_training.chaos define their own runtime_env, which replaces the DEFAULTS list, so they already ran without the RAYTEST_* guards (and chaos without FOOTER_NUM_ACTORS=20). Left as-is rather than changed under cover of this commit. - The two builds differ by 46 merged master commits, so these deltas are not a controlled A/B of the budget alone. Verified by expanding the yaml through the same DEFAULTS/variations/matrix logic ray_release.config uses, before and after: test name set identical at 199, all 24 targets at 64 MiB, no non-target budget changed, no test lost or had an env var altered. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…moke test Extends the pattern from 74e599b to two more test surfaces. The Parquet footer indexer's production default of 32 actors times out and trips Ray's "too many worker processes" warning when many test targets run in parallel under CI. - python/ray/air/BUILD.bazel gets three shared env dicts (_AIR_TEST_ENV, _AIR_TEST_ENV_TRAIN_V1, _AIR_TEST_ENV_TF) replacing the repeated per-target literals, so all 20 py_test targets under python/ray/air/tests get RAY_DATA_PARQUET_FOOTER_NUM_ACTORS=1. The three dicts preserve the existing RAY_TRAIN_V2_ENABLED 1-vs-0 split, which six of those targets depend on. - python/ray/air/tests/conftest.py setdefaults the same var so non-bazel pytest runs match. This also covers tests/execution, which has no conftest of its own. - //release:xgboost_train_batch_inference_benchmark_smoke_test gets the var inline; a shared dict would be indirection for a single target. Verified by resolving each air target's env before and after: 20/20 now set the var and no target had any other env value change. Not included: the py_doctest[air] target, whose glob excludes tests/**, and the four sibling smoke tests in release/BUILD.bazel that carry the same RAY_TRAIN_V2_ENABLED=1 env (air_benchmark_gpu_batch_inference_parquet in particular reads Parquet and will likely want the same treatment). AI assistance was used for this change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…ld for map_batches release test Signed-off-by: Goutam <goutam@anyscale.com>
ListFiles.predicate / projected_columns / limit were mirrored onto the op by hand from three separate rules (predicate, projection and limit pushdown). Nothing enforced the invariant those copies rest on -- that ListFiles never prunes by more than the downstream ReadFiles actually applies -- so a future rule that rewrote or dropped the scanner's predicate without re-syncing would leave ListFiles pruning row groups the reader would have kept, dropping rows with no error. Replace the mirroring with DeriveListFilesPushdown, which recomputes every ListFiles' state from its consuming ReadFiles scanner. It runs in LogicalOptimizer._post_optimize, after the rule loop reaches a fixed point, and derives unconditionally in both directions: a ListFiles whose consumer is not a ReadFiles is reset to no constraints. The worst a future rule can now cause is listing more than it needs to. Also replace the getattr(scanner, "predicate", None) access with abstract pushed_predicate() / pushed_limit() accessors on the scanner mixins, mirroring the existing pruned_column_names(), so a field rename is a type error rather than a silent no-op. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…e_env deep_update merges dicts but overwrites lists, so a test that declares its own cluster.byod.runtime_env replaces the DEFAULTS list rather than extending it. Pinning a bin budget therefore silently dropped RAY_DATA_DEBUG_RESOURCE_MANAGER and all three RAYTEST_FAIL_ON_* flags: 27 tests had stopped failing on worker OOM, dead nodes and spilling. It also split failure semantics across halves of the same test -- read_large_parquet_autoscaling kept the flags while its fixed_size twin lost them. Re-list the four inherited entries in every affected block, leaving the two that override them deliberately (heterogeneous_memory_batch_inference_ multitenancy and cross_az_...) alone. Verified by expanding both configs: the resolved failure flags and cluster_compute of all 199 tests now match master exactly. Also drop the redundant cluster_compute on iter_batches_pyarrow. It pinned the same value DEFAULTS supplies, so the baseline was never broken, but keeping it would stop the variant tracking its former matrix siblings if DEFAULTS moves. Document the list-replacement trap at the DEFAULTS block. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
Nothing exercised the two push-downs together, and nothing used nulls. The combination is where the fully_matched classification matters: listing stops early once the num_rows of fully-matched row groups reaches the pushed limit, so counting a non-surviving row would make Limit return fewer rows than asked for, with no error. Parquet min/max statistics are computed over non-null values only, which makes a group whose non-null values all satisfy the filter look fully matched by bounds alone. Arrow's row-group pruning turns out to be null-aware -- it declines to prune such a group by ~filter, so the classification is already correct -- but that was untested. Add one parameterized end-to-end test. The fixture is deliberately lopsided at 10 survivors per 100 rows: the early stop is evaluated per file, so a fixture with a small shortfall passes even when the classification is wrong, the last file's overshoot covering the deficit. With the misclassification injected, the 100- and 200-row limits fail. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…ion test pyrefly resolved `replace(read_files.input_dependencies[0], predicate=...)` against `LogicalOperator` and `read_files.scanner.push_filters(...)` against the base `Scanner`, neither of which carries those fields or methods. Route both through narrowing helpers, and take `Plan` in `_list_files_of` since that is what `Optimizer.optimize` returns. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
…ests A matrix cannot vary ``byod.runtime_env`` per setup value, so the variants that need their own tuning are spelled out. Rendered test names are unchanged. - q1 autoscaling: keep the 192 MiB bin budget of the fixed_size variants and scale up at 60% utilization instead of the 0.75 default. Both autoscaling variants land at 364-385s regardless of shuffle strategy while master runs hash_shuffle in 249s, which points at scale-up latency, not shuffle cost. - q17 autoscaling: 64 MiB. Smaller bins mean more read tasks, which give the autoscaler demand to act on earlier; the fixed_size variant is ~5x faster on the same query at the default budget. - q22 fixed_size/hash_shuffle: 16 MiB. At the 128 MiB default this sf100 scan collapses into a handful of bins and starves a 23-second query of read parallelism (+76.7% vs master in #102732, +72.0% in #102650). Each spelled-out entry re-lists the four inherited RAYTEST_FAIL_ON_* / RAY_DATA_DEBUG_RESOURCE_MANAGER vars, since a test-level runtime_env replaces the DEFAULTS list rather than extending it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
The target ran two full ``TorchTrainer.fit()`` runs (2 workers, 3 epochs) plus a ``ray.init`` per parameterization inside small's 60s budget, and timed out at 60.3s on both attempts in CI. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
``ParquetFileFragment.subset`` and ``.metadata`` open the file to read its footer, so on remote storage they fail with the same transient errors the rest of the read path already retries -- an sf1000 lineitem read died on "AWS Error NETWORK_CONNECTION ... curlCode: 28" inside ``subset``. Route both through ``call_with_retry`` with the context's ``retried_io_errors``. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
tpch_q18_autoscaling_hash_shuffle_v2 has failed two builds running: in #102732 the workload completed in 2075.9s and was then failed by RAYTEST_FAIL_ON_SPILLING on 4.62 GiB spilled against a 17.75 GiB object-store peak (61.6% utilization) -- spill under moderate pressure rather than a cluster out of memory, which is the shape a smaller bin budget is most likely to help. Smaller bins also give the autoscaler demand to act on earlier, as for q17. tpch_q21_autoscaling_hash_shuffle_v2 fails the same way and is left at the default as a control. The matrix cannot vary ``byod.runtime_env`` per setup value, so the two scaling variants are spelled out; rendered test names are unchanged, and the autoscaling entry re-lists the four inherited RAYTEST_FAIL_ON_* / RAY_DATA_DEBUG_RESOURCE_MANAGER vars. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Goutam <goutam@anyscale.com>
Signed-off-by: Goutam <goutam@anyscale.com>
Introduce an optional native (Rust/PyO3) Parquet reader for the DataSource V2 read path, gated by a new DataContext flag `use_arrow_rs_parquet_reader` (default False, active only under `use_datasource_v2`). When enabled, `ParquetScanner.create_reader()` returns `ArrowRsParquetFileReader`, which footer-reads each file via the `ray_data_arrow_rs` crate and decodes supported row groups natively, falling back to the PyArrow reader for unsupported files/types/ filesystems. This targets scheduler-invisible Parquet decode memory (ray#49158): the native path bounds the decode working set instead of materializing whole decoded row groups. Changes: - New Rust crate `ray_data_arrow_rs` (maturin/PyO3, out of the Bazel build) exposing native row-group reads over an Arrow C-stream. - New `ArrowRsParquetFileReader` + `native_metadata` helpers. - Base `FileReader` refactor (behavior-preserving): extract `_split_columns`/`_postprocess`/`_dispatch_fragment_reads` for reuse. - `DataContext.use_arrow_rs_parquet_reader` flag + scanner toggle. - Full parity/fallback test suite. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
Review fixes: - predicate.rs: mixed int/float comparison returns None (incomparable, keep the row group) when the integer is outside f64's exactly- representable range, so a precision-lossy promotion can't wrongly prune a matching row group. - lib.rs: release the GIL (py.allow_threads) around blocking footer / metadata I/O in read_row_groups, read_row_groups_s3, read_metadata, read_metadata_s3, and select_row_groups so Ray's fragment-pool threads decode/fetch in parallel instead of serializing on the GIL. - native_metadata.py / arrow_rs reader: strip a leading "s3://" before the bucket/key split (defensive; pyarrow paths are scheme-less). - parquet_file_reader.py: restore the arrow_rs_* tuning-kwarg pop + validation in __init__ (dropped when the file was reconciled to upstream) so ArrowRsParquetFileReader._tuning works and typo'd knobs raise a clear ValueError. - arrow_rs reader: _filesystem_supported now treats filesystem=None as the default local filesystem, matching the metadata/read paths, so eligible local reads no longer silently fall back to PyArrow. Lint: apply ruff import-ordering fixes in the reader and its test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
stat_min_max mapped every INT32/INT64 page statistic to a signed Value::Int, ignoring the column's logical type. For a UINT_32/UINT_64 column, Parquet orders stats by unsigned comparison, so a value with the high bit set (e.g. u32 max) is stored as a negative i32/i64 and reads back inverting min/max — can_match could then prune a row group that actually contains matching rows, silently dropping data (the pruning layer's soundness contract is "never prune a group that might match"). DECIMAL (unscaled int != the decimal literal) and DATE/TIME/TIMESTAMP (encoding may not match the predicate literal) have the same hazard. Gate the INT32/INT64 arms on is_plain_signed_int(descr), which consults the ColumnDescriptor's logical/converted type and only allows genuine signed integers; everything else returns no bound (conservative keep). Adds Rust unit tests covering signed/unsigned/date cases. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
end-of-file-fixer (pre-commit) requires exactly one trailing newline. Ran the full pre-commit suite over every file in the PR this time so no later hook stage surprises CI again. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
goutamvenkat-anyscale
added a commit
that referenced
this pull request
Sep 2, 2026
## Why The last step of the split of #64985: flip Parquet V2 to the footer read path, delete the blind one, and tune the release tests so this can be validated by a real release run. Everything it depends on has landed — footer types and the bin packer (#65210), the `FooterReader` actor pool (#65273), the indexer plus `count()` fix and packer-as-partitioner (#65596), reader IO tuning (#65806), and test env pinning (#65169). The path has been reachable behind `RAY_DATA_PARQUET_ENABLE_FOOTER_INDEXER` since #65596; this makes it the only path. ## Flow ```mermaid flowchart LR RP[read_parquet] --> FF["FooterFileIndexer<br/>+ OnlineBinPacker"] RP -.->|deleted| PC["ParquetFileChunker<br/>size-estimated chunks<br/>+ parquet_chunker_target_chunk_size"] style PC stroke-dasharray: 4 4 ``` ## What changed **The flip** — `RAY_DATA_PARQUET_ENABLE_FOOTER_INDEXER` is gone from both call sites. `_get_file_indexer()` always returns `FooterFileIndexer`; `get_file_partitioner()` always returns `OnlineBinPacker`. **The deletions** — `ParquetFileChunker`, `ParquetFileChunkMetadata`, `_calculate_row_group_range`, `_fragments_from_chunk_metadata`, and the tests covering only that path. **The config it left behind** — `DataContext.parquet_chunker_target_chunk_size` and `DEFAULT_PARQUET_CHUNKER_TARGET_CHUNK_SIZE` were read *only* by `ParquetFileChunker.__init__`, so they would otherwise stay declared and documented while doing nothing. Verified dead by grep across `.py` / `.yaml` / `.rst` / `.md`. **Reader pool defaults** — `RAY_DATA_PARQUET_READER_IO_THREAD_COUNT` 128, `RAY_DATA_PARQUET_READER_CPU_COUNT` 32. ## Release tests tuned Budgets are per-test overrides of `RAY_DATA_PARQUET_BIN_PACKING_BYTES` (uncompressed bytes of row-group data per read task; default 128 MiB). The shape of the tuning: reads over few large groups want **large** bins, because small bins shatter the read into short tasks the executor cannot dispatch fast enough and the cluster idles; shuffle-bound reads over many groups want **small** bins. Rows whose variants all share the same tuning are collapsed and marked. | release test | bin budget | other tuning | | --- | --- | --- | | `aggregate_groups_fixed_size_hash_shuffle_column02 column14` | 32 MiB | — | | `aggregate_groups_fixed_size_hash_shuffle_column08 column13 column14` | 128 MiB | — | | `aggregate_groups_fixed_size_shuffle_v2_column02 column14` | 64 MiB | — | | `aggregate_groups_fixed_size_shuffle_v2_column08 column13 column14` | 512 MiB | — | | `aggregate_groups_fixed_size_sort_shuffle_pull_based_column02 column14` | 32 MiB | — | | `aggregate_groups_fixed_size_sort_shuffle_pull_based_column08 column13 column14` | 256 MiB | — | | `count_parquet_fixed_size` | 64 MiB | — | | `cross_az_map_batches_autoscaling_iptable_failure_injection` | 1 GiB | — | | `distributed_training` | — | footer actors `20` | | `flat_map` | 64 MiB | — | | `iceberg_benchmark_overwrite` | 64 MiB | — | | `iter_batches_pyarrow` | 64 MiB | — | | `iter_torch_batches` | 64 MiB | — | | `joins_{{dataset}}_{{join_type}} *(all 4 variants)*` | 1.25 GiB | — | | `map` | 64 MiB | — | | `map_batches_fixed_size_{{compute}}_{{format}}_{{repeat_map_batches}} *(all 7 variants)*` | 1 GiB | scale-up threshold `0.6` | | `map_groups_fixed_size_{{shuffle_strategy}}_{{columns}} *(all 6 variants)*` | 1.25 GiB | — | | `mix` | — | footer actors `4` | | `random_shuffle_fixed_size` | 64 MiB | — | | `read_large_parquet_fixed_size` | 1 GiB | — | | `read_parquet_fixed_size` | 64 MiB | — | | `read_tfrecords` | 64 MiB | — | | `streaming_split` | 64 MiB | footer batch `50`, result batch `50` | | `to_tf` | 64 MiB | — | | `tpch_q1_fixed_size_{{shuffle_strategy}} *(all 2 variants)*` | 192 MiB | — | | `training_ingest_benchmark` | 64 MiB | — | | `wide_schema_pipeline_nested_structs` | 2 MiB | footer batch `1`, footer actors `21`, io threads `5000` | | `wide_schema_pipeline_objects` | 4.2 MiB | footer batch `1`, footer actors `9`, io threads `5000` | | `wide_schema_pipeline_primitives` | 256 KiB | footer batch `1`, footer actors `27`, io threads `5000` | | `wide_schema_pipeline_tensors` | 39 MiB | footer batch `1`, footer actors `22`, io threads `5000` | | `write_parquet` | 1.25 GiB | — | **Three matrices are split**, because a matrix cannot vary `byod.runtime_env` per value and their variants want different budgets: `wide_schema_pipeline_{{data_type}}` into four entries, and one variant each out of `iceberg_benchmark_{{mode}}` and `iter_batches_{{format}}`. `aggregate_groups` keeps its matrix and carries the budget as a matrix dimension with explicit adjustments. **Rendered test names are unchanged — 121 before and after** — so release baselines still line up. **Not carried over** from the original branch: 7 previously-pinned entries no longer exist — the autoscaling variants removed in #65506, plus `tpch_q20` and `tpch_q22`. Any entry gaining its own `runtime_env` re-lists the four `DEFAULTS` guards verbatim: `deep_update` overwrites lists rather than merging them, so declaring one silently drops the OOM / dead-node / object-store checks. (17 entries on master are already missing guards for this reason; pre-existing, not touched here.) `map_benchmark.py` also drops its `use_datasource_v2 = False` pin, which existed because V2 was spilling. > Budgets originally measured against the pre-decoupling design and refined since. Packing moved between stages in #65596 but the packer and its inputs did not, so bins should be unchanged — a release run against this PR is what confirms that. The wide-schema tests are the first place to look. ## Behavior change worth reviewing **Read-task sizing no longer comes from `override_num_blocks` / `read_op_min_num_blocks` for Parquet.** Bins are sized by `RAY_DATA_PARQUET_BIN_PACKING_BYTES`, which is *also* not derived from `DataContext.target_max_block_size`. That decoupling is deliberate — read units are row-group-accurate rather than size-estimated — but it means the two knobs users reach for first no longer affect Parquet read parallelism. Deriving the budget from `target_max_block_size` is worth considering as a follow-up; the per-test overrides above are the evidence that a single global default does not fit every workload. ## Testing ``` $ python -m pytest -q python/ray/data/tests/datasource/test_parquet.py 286 passed, 6 skipped in 521.25s $ python -m pytest -q python/ray/data/tests/datasource/test_read_parquet_v2.py 32 passed $ python -m pytest -q python/ray/data/tests/unit/datasource_v2/ python/ray/data/_internal/datasource_v2/tests/ 181 passed $ python -m pytest -q python/ray/data/tests/test_predicate_pushdown.py \ python/ray/data/tests/test_execution_optimizer_limit_pushdown.py 78 passed ``` The 286-test parquet suite is the one that matters — it runs the full read surface with the footer path as the default rather than behind a flag. The release YAML is additionally checked for parse validity, unresolved `{{...}}` in any pin, duplicate `byod:` keys, and rendered-name stability. ## Stack Final step of an 11-step split of #64985 (43 files / +3234 −719, not reviewable as one unit). - #65167 `[1]` · #65168 `[2]` · #65214 `[3]` · #65210 `[4]` · #65273 `[5]` · #65596 `[6]` · #65806 `[8]` · #65169 `[9]` — all merged - `[7]` (count push-down fix) was absorbed into #65596 - `[10]` (release tuning) is folded into this PR, so a release run exercises the flip and its tuning together - **this PR** `[11/11]` --- Not a duplicate. File-level overlap check: #63158 none; #65142 overlaps `file_indexer.py` / `context.py`; #65406 / #65407 (arrow-rs reader A/B) overlap broadly but are explicitly `[DO NOT MERGE]` benchmark branches carrying the footer work out-of-tree. AI assistance was used; every line reviewed by me and tests run locally. --------- Signed-off-by: Goutam <goutam@anyscale.com>
…e read loss Ray+real-S3 sweep of read-task concurrency through the box's core count, both readers, with a /proc runnable-thread sampler — discriminates CPU/thread oversubscription (runnable ~= HW threads at the cliff) from stalling (runnable low while wall explodes) and from no-repro. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
The reader is chosen from the driver DataContext singleton, fixed at first ray.data import and serialized to tasks (it survives ray.shutdown() and overrides worker runtime_env env vars), so the in-process arm loop ran every cell with the first arm's reader. Each cell now runs in a fresh interpreter with the flag env set before import plus an explicit DataContext set; RAY_ADDRESS is stripped so cells on an Anyscale workspace start a local Ray from this checkout instead of joining the platform cluster. Also hold the aggregated dataset so read-task counts survive into .stats() (ds.sum() discards the executed plan's stats). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
… workspace raylet Plain ray.init() auto-discovers a running cluster through /tmp/ray/ray_current_cluster even with RAY_ADDRESS unset; on an Anyscale workspace that is the platform raylet (Ray 2.57/Py3.11) and the cell dies with a version mismatch against this checkout's venv. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…able The synthetic fixtures didn't reproduce M92 (150 bin-packed ~2s tasks vs the release's 5780 ~0.9s row-group tasks; both arms network-capped at ~1.9 GB/s above N~36, rs 0.94-1.36 with runnable p90 <=36). Next cell = the actual large-parquet dataset (104 files x 56 x ~69 MB row groups = the release granularity), which needs a schema-appropriate consume column. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…decode The sum consume gets projection-pushed into the read (the rlp sweep fetched one double column, ~40 of 402 GB uncompressed). The release read_large_parquet tests consume via --iter-bundles with no projection; mirror that. The 64 MiB bin variant additionally needs RAY_DATA_PARQUET_BIN_PACKING_BYTES=67108864 in the environment. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
At 64 MiB bin granularity (the release read_large_parquet_autoscaling regime) every single-row-group read task built a fresh object_store client — a cold connection pool paying DNS + full TLS handshakes before the footer fetch. Stack sampling on an m5.24xlarge showed read workers blocked in NativeS3Store.open_file -> load_meta_s3 for ~2/3 of task wall while the box sat idle (runnable p50 = 3/96), walls 2.88x PyArrow at 48-way concurrency: 5769 cold client builds for 104 distinct files. connect_native_s3 now caches clients per process, keyed by (bucket, full connection config incl. credentials) — the same lifetime pyarrow's serialized S3FileSystem client has in a reused Ray worker. Rotated credentials change the key, so staleness is structural; the table is size-capped. Multi-file bins behave as before (first open builds, rest reuse) plus cross-task reuse. RAY_DATA_ARROW_RS_S3_CLIENT_CACHE=0 restores the uncached per-call behavior. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
… 31) Fresh 2-CPU local cluster per arm in a fresh subprocess, one-file S3 read at concurrency=1; the read worker's USS (smaps_rollup) idle / 20 Hz peak / settled after two consecutive reads. The rs-pa delta on the settled floor is the M84 ~90-105 MB constant; read ray-project#2 separates a one-time constant from per-task retention growth. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
… sampler Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
| # auto-discover a workspace raylet via /tmp/ray/ray_current_cluster). | ||
| ray.init( | ||
| address="local", num_cpus=2, include_dashboard=False, logging_level="ERROR" | ||
| ) |
There was a problem hiding this comment.
Footer pool can stall two-CPU cluster
Medium Severity
The cell pins a 2-CPU cluster but leaves RAY_DATA_PARQUET_FOOTER_NUM_ACTORS at the default of 32 one-CPU actors. ListFiles already holds one CPU while creating that pool, so the remaining CPU can go to a footer actor that never receives the single-file batch, blocking listing and leaving the read unscheduled.
Reviewed by Cursor Bugbot for commit b3583bf. Configure here.
Review find (medium, valid): the worker census ran once after a single dummy task, so a read task landing on a worker forked afterwards was never sampled — idle/peak/after USS would come from an idle worker. The M99 run self-certifies (chosen pid's after-read-1 USS 353.6/503.5 MB vs a ~52 MB idle floor proves the censused pid did both reads), but the planned multi-read soak would be far more exposed. Fix: the sampler now rescans /proc every 0.5 s for new ray:: pids, records first-seen USS as each pid's baseline, and the result carries two integrity flags: read_worker_late_spawn (baseline may include read work) and reads_on_same_worker (read-1/read-2 attribution is only per-worker-meaningful when one pid grew on both). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
Second review find (high, valid): growth-based selection (settled USS minus first-seen baseline) cannot identify a worker first seen mid-read - its baseline already holds decode buffers, so after the buffers drop its delta reads ~0 or negative and a prestarted idle pid's noise wins the argmax, reporting an idle process as the read worker. Fix: positive identification. Ray retitles an executing worker ray::<TaskName> (read tasks: ray::ReadFilesParquetV2), and the sampler already polls cmdlines every 0.5 s - it now records every title a pid ever shows, and the read worker is any pid that bore the read-op title. Growth selection survives only as a fallback for a sub-0.5 s read the title poll could miss, and the result reports read_worker_id_method plus n_read_task_workers so a fallback or multi-worker cell is visible. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…OW_RS_MALLOC_TRIM_EOS) Second allocator lever for the arrow-rs Parquet reader, alongside the existing mallopt one (RAY_DATA_ARROW_RS_MALLOC_TRIM). Off by default. Why a second lever: the box soak (arrow_rs_docs findings M48) showed glibc keeping the reader's freed decode heap resident under task churn (idle-worker USS +492 MiB over ~100 tasks). mallopt(M_TRIM_THRESHOLD, 0) collapses that floor (R 0.08) but costs 24-36% wall (M61), and the threshold value is a dead knob because any explicit setting disables glibc's dynamic mmap threshold (M64). Calling malloc_trim(0) ONCE at the end of each read task's stream returns the same retained pages while the allocator behaves normally during decode. This is the ship candidate the next release A/B (4 arms: pa / rs / rs+trim / rs+eos) measures. Mechanics: DataContext.arrow_rs_malloc_trim_eos (env RAY_DATA_ARROW_RS_MALLOC_TRIM_EOS); read() now wraps the former body (_read_split) in try/finally so the trim fires exactly once per stream, whichever path (native or fallback) served it, including early consumer close. libc symbol resolved once per worker via ctypes; non-Linux and non-glibc are no-ops with a one-time warning. Test: knob off never trims; knob on trims once per read() over 2 files and once on an early-closed generator. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…eos)
gen_2x2_release_tests.py grows a second matrix, --matrix alloc (now the
default; --matrix 2x2 still emits the original one byte-for-byte):
{pa, rs, rstrim, rseos} x original fleet over the 8 memory, 8 sustained
and 3 control targets, plus 3 wall targets kept as the release-scale
confirmation of the M97/M98 wall fix; single-node cells only for
read_large_parquet_autoscaling (M75/M86) and map_groups col02 (M90).
22 targets -> 88 multi + 8 single = 96 manual entries.
Arms are one env prefix each on the run script, like the reader flag:
rstrim RAY_DATA_ARROW_RS_MALLOC_TRIM=1 box-confirmed mechanism
probe, known wall price
rseos RAY_DATA_ARROW_RS_MALLOC_TRIM_EOS=1 ship candidate
Both knobs are read only by the arrow-rs reader, so they are inert on
the pa path by construction (validate() also refuses them without the
reader flag).
Names stay <parent>_2x2_<arm>_<topology>, so the pa/rs cells continue the
results-DB history from build 105711 and one filter (name:.*_2x2_.*)
selects the whole matrix. The block header records which matrix it
holds so --check validates against the right expected set.
release_data_tests.yaml regenerated (marker block only).
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
Under RAY_DATA_ARROW_RS_MALLOC_TRIM_EOS the col02 map_groups fleet cell ran 2.37x pyarrow wall in build 106096 (findings M107) while the same lever closed every retention row elsewhere. One run cannot tell "the trim slowed the tasks" from "the autoscaler packed 44 workers/node", so surface the trim's cost per task: - _maybe_trim_at_stream_end() returns the seconds the malloc_trim(0) took; ArrowRsParquetFileReader accumulates it and hands it over through a new generic Reader.pop_task_stats() hook (default: nothing). - ReadFiles transform folds it into ReadFilesTaskStats.trim_wall_s (a new field; contained in decode_wall_s since the finalizer runs inside the iterator's final next()). - OpRuntimeMetrics gains the read_task_trim_wall_s per-task distribution. Tests: metrics fold + reader pop/drain semantics. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…s arm
Three instrument defects found reading build 106096 (arrow_rs_docs
2026-09-04.md), fixed so the next build answers them:
- benchmark.collect_operator_metrics took the FIRST operator containing
"Read" as the headline; TPC-H q17 has two reads and the headline was the
2-task `part` read. Now: `read_operators` lists every Read op compactly
(tasks, wall, bytes, max_uss/decoded/peak_batch/trim per-task quantiles)
and the `read_*` headline is the Read op with the most tasks (ties: plan
order). Also exports the new read_task_trim_wall_s distribution.
- node_memory_monitor summed worker USS per sample, so the sustained (p50)
column on a 10 s test tracked how many workers were alive, not what each
held. Now also reports, for the peak-USS node, workers_p50/max and the
per-worker mean USS p50, plus a compact per-node list and the sampler
interval.
- release_regression_probe / tpch_probe had the arms hardcoded ("pa","rs").
Now --arms pa,rs,rseos[,rstrim] (env sets in ARMS), --cpus 24,48,96 runs
each non-tpch cell once per local num_cpus (the M107 concurrency sweep),
--monitor-interval sets the sampler period; the table prints one row per
arm with a trim p50 column. run_release_regressions.sh passes ARMS, CPUS,
MONITOR_INTERVAL through.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…table is yielded ReadFilesTaskStats.trim_wall_s is only known when the reader's stream ends (the arrow-rs eos malloc_trim runs in the reader's finalizer, i.e. inside the final next()), which is AFTER the last table was yielded. A block's stats snapshot is pickled when the block leaves the task, and a table that completes a block (buffer >= target_max_block_size but under the 1.5x slice limit -> emitted whole, no remainder) is followed by no flush block, so the _update after the loop reached the driver only when the shaping buffer happened to hold a remainder. read_task_trim_wall_s then read 0 while the trim actually ran, hiding exactly the cost the metric exists to expose. Fix: one-table lookahead in the per-manifest loop. The next table is pulled before the previous one is yielded, so the StopIteration (and the reader's pop_task_stats drain) is folded into task_stats before the final yield. Memory-neutral: the previous table stayed bound in this frame during the next decode already. Regression test: one numeric file under the 2048-row batch floor (one table, one task), target = 0.8x its bytes so that table completes the only block; asserts num_blocks == 1 and read_task_trim_wall_s.max > 0. Fails on the old loop with 0.0 > 0.0. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
There are 13 total unresolved issues (including 12 from previous reviews).
Reviewed by Cursor Bugbot for commit 650e98f. Configure here.
The col02 concurrency sweep (findings M111) showed arrow-rs read tasks running 1.4-3.9x longer than PyArrow's at 24-96 concurrent tasks per node while using 0.64x the CPU, with the reader iterator itself a flat ~3 s per task and the rest of the task duration unaccounted for. Split it: - ReadFilesTaskStats.yield_wall_s: wall spent inside the planner's yield (output-buffer shaping, block build, object-store put, streaming-generator backpressure). Disjoint from decode_wall_s. - ReadFilesTaskStats.first_table_wall_s: task start to first decoded table (reader construction + first next()), the per-task fixed cost. Both flow through OpRuntimeMetrics distributions (read_task_yield_wall_s, read_task_first_table_wall_s) into the benchmark result.json per-task dists and read_operators p50/max. Task duration - decode - yield = start-up and teardown. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
release_regression_probe folds TEST_OUTPUT_JSON into each cell record as rec['bench']; tpch_probe only kept the CELL_JSON walls, so its summary.json had no per-operator wall/cpu, per-task dists or node-mem fields even though every tpch cell writes them (2026-09-04 q17 leg). Same fold now. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
… propagates The one-table lookahead (650e98f) only yielded a decoded table after the NEXT next() returned or the stream ended. A later decode failure (IO error, corrupt page) therefore dropped a table that had already decoded fine, whereas before the lookahead that table had already left the task. Observable under DataContext.max_errored_blocks, where a tolerated task keeps its earlier blocks: the held table was missing from the output. Catch the non-StopIteration exception, yield the pending table (timed into yield_wall_s like every other yield), then re-raise. Regression test: one file, a 32-row first row group of a dictionary-encoded constant string (one arrow-rs batch, above a 0.9x block target so it completes a block on its own) followed by a row group whose first data page is zeroed behind an intact footer; with max_errored_blocks=1 the 32 rows must survive. arrow-rs only: pyarrow's scanner reads row groups concurrently and raises before the first batch, so no table precedes the error on that arm. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…ry target) The fixed-size twin of read_large_parquet is the one release read shape with no concurrent stage, i.e. the fleet counterpart of the box's clean read_parquet_binned win (M116) and the rlp re-baseline row (M75). It was missing from the alloc matrix; adding it as a memory target gives the flip gate a fixed-pool read control next to the autoscaling one. Regenerated release_data_tests.yaml: 4 additive entries (pa / rs / rstrim / rseos, multi), 100 generated cells, --check passes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
TODO 34f: the col02 rseos fleet cell ran 2.37 once, and one run cannot separate "the trim slowed the tasks" from "the autoscaler packed 44 workers per node". The release form has no repeat field; the runner reads a per-test `repeated_run` key (release/ray_release/buildkite/step.py), so the generator now emits it for the alloc-matrix gate cells: col02 (multi + single) and the new read_large_parquet_fixed_size twin, 3 runs each. 12 cells x 3 runs; everything else stays at 1. --check passes, 100 generated entries unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
… form
Parent test names carry spaces ("column02 column14"); TARGETS and
ALLOC_REPEATED use the '+' form, so the col02 cells missed the repeat in
61ef8bf (4 of 12 cells tagged). Compare on base_name. Now 12 cells carry
repeated_run: 3 -- col02 multi + single and read_large_parquet_fixed_size,
4 arms each.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
…e walls The per-task reader timers (decode / first_table / yield wall) leave ~5 s of a read task's duration unexplained on the sf10 lineitem shape — outside the reader iterator, not CPU (M117-M119). Ray Data already tracks block generation and serialization wall (object_creation_dur_s per output) and the operator's submission / output backpressure wall; collect_operator_metrics now records those op totals next to the per-task distributions so the residual splits into serialization vs the rest with no reader change. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
Two single-file probes behind findings M120: plasma_ser_micro.py times Ray serialization + ray.put of the same Parquet row groups decoded by the crate and by PyArrow in one process; plasma_put_micro.py times ray.put inside N concurrent Ray tasks with rusage fault/CPU counters, arms run back-to-back in one Ray instance. They showed the per-task block serialization cost is the fresh object store's first-touch page faults (7 s cold, 0.2 s warm, either arm), not the arrow-rs block. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: Aarrya <aarrya.saraf@anyscale.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Draft PR whose only purpose is to trigger the release pipeline for the multi-node
arrow-rs vs PyArrow A/B (the reader from #65117, ported onto #64985's planner).
This branch is stacked on unmerged #64985, so its diff includes that PR's commits —
do not review or merge. Baseline arm: see the companion draft PR for
arrow-rs-ab-baseline-64985.🤖 Generated with Claude Code