[data] DataSourceV2: enable V2 by default

[goutamvenkat-anyscale]

Why

Turn on DataSource V2 by default for ray.data.read_parquet (and related reads that go through the same path), so the stack that landed in prior PRs is what users get unless they opt out via DataContext.get_current().use_datasource_v2 = False. Keeping the default flip in a dedicated change makes revert/revert-cherry-pick straightforward.

What

Default: DEFAULT_USE_DATASOURCE_V2 = True in python/ray/data/context.py.

Logical plan: ReadFiles no longer uses a bespoke InitVar + input_op pattern. It now takes input_dependencies like other operators, which removes the custom _apply_transform / input_dependency plumbing and simplifies every replace(...) that used to thread input_op through (read_operator.py, read_api.py, limit_pushdown.py, and unit tests). A per_block_limit field is declared on ReadFiles (always None here) so inherited AbstractMap machinery can resolve it; V2 limits still go through scanner.push_limit in LimitPushdownRule. The read stats metadata key is aligned with the operator name ("ReadFiles").

Tests and docs for V2-by-default: Several tests still assumed V1 behavior or patterns that V2 does not support the same way. Updates are grouped below. The read_parquet docstring example uses chained .filter(expr=...) instead of deprecated filter=.

Test changes (by reason)

V2 default + deprecated read API

• test_parquet.py — test_parquet_read_partitioned_with_filter: stop read_parquet(..., filter=pds.field(...)); use read_parquet(...).filter(expr=col(...) == lit(...)).
• test_parquet.py — test_count_with_filter: same (.filter(expr=col(...) < lit(...))).
• test_predicate_pushdown.py — test_filter_pushdown_source_and_op: remove read_parquet(..., filter=pc.greater(...)); chain .filter(expr=col("sepal.length") > lit(5.0)) before the string-expression filter; drop unused pyarrow.compute import.

include_row_hash + column projection (V2 ordering)

• test_parquet.py — test_include_row_hash_with_column_projection and test_include_row_hash_existing_column_with_projection: replace read_parquet(..., columns=[...], include_row_hash=True) with read_parquet(..., include_row_hash=True).select_columns([..., "row_hash"]) so row_hash remains in the schema after projection.

Empty input path (V2 fails fast)

• test_streaming_executor.py — test_execution_callbacks_executor_arg: write a one-row Parquet file under the input directory after makedirs. V2 raises "no files found" for an empty directory; V1 could return an empty dataset.

Nested ARROW-5030 fallback assertion (worker vs driver)

• test_parquet.py — test_read_parquet_nested_fallback_triggered_when_filter_references_nested_column: remove patch + assert_called() on _get_safe_batch_size_for_nested_types (called in a Ray worker, so a driver-side patch never fires). Assert with _resolve_read_columns(...) and _needs_nested_type_fallback(fragment, read_columns) on a real fragment from the fixture.

Logical ReadFiles constructor / graph shape

• test_read_files_logical.py: construct ReadFiles with input_dependencies=[list_files_op] instead of input_op=list_files_op.
• test_read_parquet_v2.py — test_read_parquet_builds_list_files_read_files_chain: assert dag.input_dependencies[0] is ListFiles instead of dag.input_dependency.

Performance tuning for the V2 reader

Several knobs in the V2 Parquet read path defaulted to conservative values that left throughput on the table for typical S3+Parquet workloads — the read_large_parquet_* release benchmark in particular saw underutilized CPU traceable to a mix of sequential fragment reads, small per-fragment batch readahead, and a tiny per-stream buffer. Bumped each default and gated every change behind an env var so the prior behavior is recoverable for memory-sensitive workloads (e.g. wide tensor columns).

Concurrent fragment reads (file_reader.py)

• New RAY_DATA_READ_FILES_NUM_THREADS env var, default 4. When > 1, _read_fragment_batches reads fragments concurrently via make_async_gen instead of the previous serial loop.
• Worker count is capped at runtime by min(env_default, len(fragments)) so single-fragment tasks don't spin up extra threads.
• Falls back to the sequential path when DataContext.execution_options.preserve_order is set.
• Uses preserve_ordering=True so block ordering is deterministic across task retries (required for safe block reconstruction).
• Per-worker loop extracted as _read_fragments_sequential so the threaded and sequential paths share the same fragment-read body.

Batch readahead (file_reader.py)

• _ARROW_SCANNER_BATCH_READAHEAD is now env_integer("RAY_DATA_ARROW_SCANNER_BATCH_READAHEAD", 8) (was hardcoded 1). The previous value was set to bound peak memory on jumbo tensor columns; 8 keeps decode pipelined for non-tensor workloads while remaining tunable down via env var.

Parquet fragment buffer size (parquet_file_reader.py)

• New RAY_DATA_PARQUET_FRAGMENT_BUFFER_SIZE env var, default 8 MiB. Passed as ParquetFragmentScanOptions.buffer_size so use_buffered_stream=True issues meaningfully sized range requests against S3 instead of pyarrow's small default (~8 KiB).

Testing

Ran the updated Parquet, predicate pushdown, streaming executor, and V2 read unit tests in .venv (including nested fallback and row-hash cases touched above). Performance defaults are exercised by the existing V2 read tests; release benchmarks (read_large_parquet_fixed_size / _autoscaling) verify the wall-clock improvement on the full S3 dataset.

[gemini-code-assist]

Code Review

This pull request enables DataSourceV2 by default and ports the nested-type fallback (ARROW-5030) to the V2 Parquet reader. It refactors predicate handling to use Ray Expr objects until the scanner boundary, allowing for better introspection. Key changes include the implementation of _iter_fragment_tables in ParquetFileReader to handle large nested columns, schema evolution support via null-filling, and per-fragment retry logic in FileReader. Feedback focuses on robustly handling schema evolution in the Parquet fallback path, specifically ensuring all unified schema columns are tracked, filtering missing columns before calling iter_batches, and selecting correct columns before casting to avoid mismatches. A typo in a static analysis comment was also noted.

[cursor]

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

^{Reviewed by Cursor Bugbot for commit 6657b2d. Configure here.}

[richardliaw]

Are we actually ready to flip this to default? Should we let it bake for a while before flipping, and raise a tip/warning to mention users to flip this on themselves?

1. Are there any behavior changes to note?
2. Are there any API changes to note?
3. Are there any performance changes to note?

[goutamvenkat-anyscale]

Are we actually ready to flip this to default? Should we let it bake for a while before flipping, and raise a tip/warning to mention users to flip this on themselves?

1. Are there any behavior changes to note?
2. Are there any API changes to note?
3. Are there any performance changes to note?

• 1
  Reading is broken up into Listing and Reading as separate operators. So listing stage determines the parallelism, memory hints etc. of the reading stage

• 2
  Yes a couple of API diffs. filter and columns are not supported as params in read_parquet (it will raise an error.) The expectation is to use ds.filter(...) and ds.select_columns(...) methods

-3
Based on the large parquet release test there seems to be a ~10% regression compared to V1. Still investigating where the delta is coming from.

Actually nvm. Seeing no regression on the latest run: https://console.anyscale-staging.com/cld_kvedZWag2qA8i5BjxUevf5i7/prj_92c7b71w55flm6gv6imv4m6vqg/jobs/prodjob_eb3ym8edqcj6h76r1llrr47ptv/data?job-logs-section-tabs=application_logs&job-tab=ray-turbo-dashboard

[richardliaw]

No perf regression is good, but API changes is more scary. Can we soft deprecate the API changes instead of hard break?

[goutamvenkat-anyscale]

No perf regression is good, but API changes is more scary. Can we soft deprecate the API changes instead of hard break?

So filter and columns are already deprecated from before.

[iamjustinhsu]

Hmm it's not clear to me what the relationship is between preserve_ordering=True, bounding something by manifest size, and fragment-object constructions?

[goutamvenkat-anyscale]

Cleaned up comment

[iamjustinhsu]

Why is identical block sequences required? Like what happens when preserve_order=False?

[iamjustinhsu]

btw, what did u have to do to get the read performance up?

[goutamvenkat-anyscale]

btw, what did u have to do to get the read performance up?

Bumped up batch_readahead and increase I/O buffer size to 8 MiB