[Data] [DO NOT MERGE] arrow-rs Parquet reader A/B — treatment arm (release-test trigger)

ci/lint/pyrefly-excluded-files.txt

@@ -200,6 +200,7 @@ python/ray/data/tests/block_batching/test_iter_batches.py
 python/ray/data/tests/block_batching/test_util.py
 python/ray/data/tests/conftest.py
 python/ray/data/tests/datasource/test_arrow.py
+python/ray/data/tests/datasource/test_arrow_rs_parquet_reader.py
 python/ray/data/tests/datasource/test_bigquery.py
 python/ray/data/tests/datasource/test_clickhouse.py
 python/ray/data/tests/datasource/test_csv.py

docker/base-extra/Dockerfile

@@ -48,7 +48,7 @@ wget -O - https://packages.cloud.google.com/apt/doc/apt-key.gpg \
 # Add gdb since ray dashboard uses `memray attach`, which requires gdb.
 
 APT_PKGS=(
-    google-cloud-sdk
+    google-cloud-cli
     supervisor
     vim
     zsh

pyrefly.toml

@@ -45,6 +45,7 @@ ignore-missing-imports = [
     "pytest.*",
     "pytest_lazy_fixtures.*",
     "ray.core.generated.*",
+    "ray_data_arrow_rs.*",
     "raydp.*",
     "rich.*",
     "snappy.*",

python/ray/air/BUILD.bazel

@@ -1,6 +1,29 @@
 load("@rules_python//python:defs.bzl", "py_library", "py_test")
 load("//bazel:python.bzl", "doctest")
 
+# Shared env for the python/ray/air/tests targets. ``RAY_DATA_PARQUET_FOOTER_
+# NUM_ACTORS=1`` keeps the Parquet footer-reader pool tiny: the 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. Mirrored by a
+# setdefault in python/ray/air/tests/conftest.py for non-bazel pytest runs.
+_AIR_TEST_ENV = {
+    "RAY_DATA_PARQUET_FOOTER_NUM_ACTORS": "1",
+    "RAY_TRAIN_V2_ENABLED": "1",
+}
+
+# Same, for the targets that pin Train V1 telemetry/behaviour.
+_AIR_TEST_ENV_TRAIN_V1 = {
+    "RAY_DATA_PARQUET_FOOTER_NUM_ACTORS": "1",
+    "RAY_TRAIN_V2_ENABLED": "0",
+}
+
+# Same, plus legacy Keras for the Keras callback test.
+_AIR_TEST_ENV_TF = {
+    "RAY_DATA_PARQUET_FOOTER_NUM_ACTORS": "1",
+    "RAY_TRAIN_V2_ENABLED": "1",
+    "TF_USE_LEGACY_KERAS": "1",
+}
+
 doctest(
     name = "py_doctest[air]",
     env = {"RAY_TRAIN_V2_ENABLED": "1"},
@@ -31,7 +54,7 @@ py_test(
     size = "small",
     srcs = ["tests/test_air_usage.py"],
     # NOTE: This tests Train V1 telemetry.
-    env = {"RAY_TRAIN_V2_ENABLED": "0"},
+    env = _AIR_TEST_ENV_TRAIN_V1,
     tags = [
         "exclusive",
         "team:ml",
@@ -44,7 +67,7 @@ py_test(
     size = "large",
     srcs = ["tests/test_new_dataset_config.py"],
     # NOTE: Relevant tests moved to train/v2/tests/test_data_integration.py
-    env = {"RAY_TRAIN_V2_ENABLED": "0"},
+    env = _AIR_TEST_ENV_TRAIN_V1,
     tags = [
         "exclusive",
         "team:ml",
@@ -60,7 +83,7 @@ py_test(
         "tests/test_experiment_restore.py",
     ],
     # NOTE: This tests Tune and Train V1 restoration.
-    env = {"RAY_TRAIN_V2_ENABLED": "0"},
+    env = _AIR_TEST_ENV_TRAIN_V1,
     tags = [
         "exclusive",
         "team:ml",
@@ -73,7 +96,7 @@ py_test(
     size = "medium",
     srcs = ["tests/test_errors.py"],
     # NOTE: This tests Tune (Train V1) error propagation logic.
-    env = {"RAY_TRAIN_V2_ENABLED": "0"},
+    env = _AIR_TEST_ENV_TRAIN_V1,
     tags = [
         "exclusive",
         "team:ml",
@@ -86,7 +109,7 @@ py_test(
     size = "small",
     srcs = ["tests/test_integration_comet.py"],
     # NOTE: This tests the Tune Comet callback.
-    env = {"RAY_TRAIN_V2_ENABLED": "0"},
+    env = _AIR_TEST_ENV_TRAIN_V1,
     tags = [
         "exclusive",
         "team:ml",
@@ -99,7 +122,7 @@ py_test(
     size = "small",
     srcs = ["tests/test_integration_wandb.py"],
     # NOTE: This tests the Tune wandb callback.
-    env = {"RAY_TRAIN_V2_ENABLED": "0"},
+    env = _AIR_TEST_ENV_TRAIN_V1,
     tags = [
         "exclusive",
         "team:ml",
@@ -112,7 +135,7 @@ py_test(
     size = "medium",
     srcs = ["tests/test_integration_mlflow.py"],
     # NOTE: This tests the Tune mlflow callback.
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -124,10 +147,7 @@ py_test(
     name = "test_keras_callback",
     size = "medium",
     srcs = ["tests/test_keras_callback.py"],
-    env = {
-        "RAY_TRAIN_V2_ENABLED": "1",
-        "TF_USE_LEGACY_KERAS": "1",
-    },
+    env = _AIR_TEST_ENV_TF,
     tags = [
         "exclusive",
         "team:ml",
@@ -139,7 +159,7 @@ py_test(
     name = "test_remote_storage_hdfs",
     size = "small",
     srcs = ["tests/test_remote_storage_hdfs.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "hdfs",
@@ -152,7 +172,7 @@ py_test(
     name = "test_tracebacks",
     size = "small",
     srcs = ["tests/test_tracebacks.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -164,7 +184,7 @@ py_test(
     name = "test_utils",
     size = "small",
     srcs = ["tests/test_utils.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -183,7 +203,7 @@ py_test(
     name = "test_barrier",
     size = "small",
     srcs = ["tests/execution/test_barrier.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -195,7 +215,7 @@ py_test(
     name = "test_e2e_train_flow",
     size = "medium",
     srcs = ["tests/execution/test_e2e_train_flow.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -207,7 +227,7 @@ py_test(
     name = "test_e2e_tune_flow",
     size = "medium",
     srcs = ["tests/execution/test_e2e_tune_flow.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -219,7 +239,7 @@ py_test(
     name = "test_event_manager",
     size = "medium",
     srcs = ["tests/execution/test_event_manager.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -231,7 +251,7 @@ py_test(
     name = "test_resource_manager_fixed",
     size = "small",
     srcs = ["tests/execution/test_resource_manager_fixed.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -243,7 +263,7 @@ py_test(
     name = "test_resource_manager_placement_group",
     size = "medium",
     srcs = ["tests/execution/test_resource_manager_placement_group.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -255,7 +275,7 @@ py_test(
     name = "test_resource_request",
     size = "small",
     srcs = ["tests/execution/test_resource_request.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -267,7 +287,7 @@ py_test(
     name = "test_tracked_actor",
     size = "small",
     srcs = ["tests/execution/test_tracked_actor.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",
@@ -279,7 +299,7 @@ py_test(
     name = "test_tracked_actor_task",
     size = "small",
     srcs = ["tests/execution/test_tracked_actor_task.py"],
-    env = {"RAY_TRAIN_V2_ENABLED": "1"},
+    env = _AIR_TEST_ENV,
     tags = [
         "exclusive",
         "team:ml",

python/ray/air/tests/conftest.py

@@ -1,11 +1,18 @@
 # Trigger pytest hook to automatically zip test cluster logs to archive dir on failure
 import copy
+import os
 
 import pytest
 
 import ray
 from ray.tests.conftest import pytest_runtest_makereport  # noqa
 
+# Keep the Parquet footer-reader pool tiny for these tests. The production
+# default of 32 actors times out under CI parallelism; tests that need a larger
+# pool can override with monkeypatch.setenv. Mirrored in
+# python/ray/air/BUILD.bazel for bazel test targets.
+os.environ.setdefault("RAY_DATA_PARQUET_FOOTER_NUM_ACTORS", "1")
+
 
 @pytest.fixture
 def restore_data_context(request):

python/ray/data/.claude/CLAUDE.md

@@ -8,3 +8,39 @@
 
 ## Gotchas
 <!-- Non-obvious behaviors, common mistakes, things that break silently -->
+
+## Project docs (local-only, gitignored)
+
+Long-running investigations keep their notes in a directory under `python/ray/data/`
+excluded via `.git/info/exclude` — currently `arrow_rs_docs/` (the arrow-rs Parquet
+reader migration, draft PR #65117). **Start at that directory's `README.md`**: it holds
+the read order, the doc index with a staleness column, the ratio conventions, and the
+rules below. Do not read the other docs front to back; arrive at them from a link.
+
+## Doc rules for those directories
+
+Each rule exists because its absence already cost something real.
+
+1. **No markdown file over 1000 lines.** The moment one crosses it, move a whole section
+   out into an existing doc if one owns the topic, else a new file  leaving a stub that
+   says where it went and keeping section numbers unchanged so old cross-references still
+   resolve. Do it in the same edit that crosses the line. Two docs reached 1910 lines and
+   became unreadable and un-editable at once; splitting after the fact cost far more.
+2. **A measured number is a row in `findings.md`** — one tabular registry with permanent
+   IDs, a status column (LIVE / RE-MEASURE / DEAD / RETRACTED / OPEN) and a caveat column,
+   so a new confound is recorded by editing one row. Never renumber, never delete a row: a
+   wrong finding becomes RETRACTED with the reason, because the wrong version is what
+   older docs still say. The prose that qualifies a number goes in the topic doc, never in
+   the plan — a plan that quotes numbers goes stale silently.
+3. **Closing a work item splits it three ways, in the same edit:** numbers → `findings.md`,
+   prose → the topic doc, and the *decision* (one row: disposition, why, and for parked
+   items what would revive it) → `todo_archive.md` under its original number, which is
+   never reused. It leaves `TODO.md` entirely, which holds open work only. **Before
+   archiving, check the closed item for open work hiding inside it** and promote that —
+   this has now bitten four times, including a P0 that had already shipped and a stale
+   docstring nobody could see.
+4. **Verify against the tree before writing a doc claim.** Env vars, defaults, file paths
+   and which branch a script lives on all drift; each audit of these docs has found
+   documented knobs that do not exist and shipped defaults documented three values stale.
+5. After every message, check if there is anything to be updated/added/removed in any doc 
+   and make the apt change. 

python/ray/data/BUILD.bazel

@@ -1,5 +1,5 @@
-load("@rules_python//python:defs.bzl", "py_library", "py_test")
-load("//bazel:python.bzl", "doctest", "py_test_module_list")
+load("@rules_python//python:defs.bzl", "py_library")
+load(":test.bzl", "doctest", "py_test", "py_test_module_list")
 
 # Export pytest plugin so it can be used in the documentation tests.
 exports_files(
@@ -1030,7 +1030,7 @@ py_test(
 
 py_test(
     name = "test_predicate_pushdown",
-    size = "small",
+    size = "medium",
     srcs = ["tests/test_predicate_pushdown.py"],
     tags = [
         "exclusive",