[autoscaler] Clarify head node commands in ray up output

python/ray/autoscaler/BUILD.bazel

@@ -29,3 +29,12 @@ filegroup(
     ]),
     visibility = ["//:__pkg__"],
 )
+
+filegroup(
+    name = "cli_output_context_test_data",
+    srcs = [
+        "_private/cli_output_helpers.py",
+        "_private/commands.py",
+    ],
+    visibility = ["//python/ray/tests:__pkg__"],
+)

python/ray/autoscaler/_private/cli_output_helpers.py

@@ -0,0 +1,39 @@
+"""Lightweight CLI output helpers for cluster launcher context notes.
+
+These functions produce informational output that disambiguates head-node
+commands from local-machine commands.  They have NO dependency on the ``ray``
+C++ runtime and accept ``cli_logger`` / ``cf`` as explicit parameters so they
+can be tested in isolation.
+"""
+
+
+def print_next_steps_context_note(cli_logger, cf):
+    """Print a dimmed note at the top of the 'Next steps' block.
+
+    Informs the user that the commands below are intended for the head node
+    or for machines within the cluster network.
+    """
+    cli_logger.print(cf.dimmed("Note: The following commands are intended for use on"))
+    cli_logger.print(cf.dimmed("the head node or within the cluster network."))
+    cli_logger.newline()
+
+
+def print_head_node_context_separator(cli_logger, cf):
+    """Print a visual separator and context note after head-node output.
+
+    Used by ``ray up`` to separate the streamed ``ray start`` output from
+    the local-machine commands that follow.
+    """
+    cli_logger.print(cf.dimmed("-" * 60))
+    cli_logger.print(
+        cf.dimmed("Note: The output above is from the head node (via `ray start`).")
+    )
+    cli_logger.print(
+        cf.dimmed("  Commands shown in 'Next steps' may only work from the head node")
+    )
+    cli_logger.print(cf.dimmed("  or from within the cluster network."))
+    cli_logger.newline()
+
+
+# Group heading used by ``ray up`` for the local commands section.
+USEFUL_COMMANDS_HEADING = "Useful commands for your local machine:"

python/ray/autoscaler/_private/commands.py

@@ -22,6 +22,10 @@
 from ray.autoscaler._private import subprocess_output_util as cmd_output_util
 from ray.autoscaler._private.autoscaler import AutoscalerSummary
 from ray.autoscaler._private.cli_logger import cf, cli_logger
+from ray.autoscaler._private.cli_output_helpers import (
+    USEFUL_COMMANDS_HEADING,
+    print_head_node_context_separator,
+)
 from ray.autoscaler._private.cluster_dump import (
     Archive,
     GetParameters,
@@ -932,7 +936,9 @@ def get_or_create_head_node(
         modifiers = ""
 
     cli_logger.newline()
-    with cli_logger.group("Useful commands:"):
+    if ray_start_commands:
+        print_head_node_context_separator(cli_logger, cf)
+    with cli_logger.group(USEFUL_COMMANDS_HEADING):
         printable_config_file = os.path.abspath(printable_config_file)
 
         cli_logger.print("To terminate the cluster:")

python/ray/data/tests/datasource/test_parquet.py

@@ -12,6 +12,7 @@
 import pandas as pd
 import pyarrow as pa
 import pyarrow.dataset as pds
+import pyarrow.fs as pafs
 import pyarrow.parquet as pq
 import pytest
 from packaging.version import parse as parse_version
@@ -24,6 +25,7 @@
     _MAX_PYARROW_TO_BATCHES_BATCH_SIZE,
     ParquetDatasource,
     _coerce_pyarrow_fragment_batch_size,
+    _infer_schema,
     _read_batches_from,
 )
 from ray.data._internal.execution.interfaces.ref_bundle import (
@@ -1779,7 +1781,6 @@ def test_read_null_data_in_first_file(
         {"data": "spam"},
     ]
 
-
 def test_read_parquet_does_not_call_infer_schema(
     tmp_path, monkeypatch, ray_start_regular_shared
 ):

python/ray/data/tests/unit/test_parquet_schema_inference.py

@@ -0,0 +1,68 @@
+import pytest
+import pyarrow as pa
+import pyarrow.dataset as pds
+import pyarrow.fs as pafs
+import pyarrow.parquet as pq
+from packaging.version import parse as parse_version
+
+from ray._common.utils import get_pyarrow_version
+from ray.data._internal.datasource.parquet_datasource import _infer_schema
+
+def test_read_parquet_memory_growth(tmp_path, monkeypatch):
+    """Schema inference should not inspect every fragment on PyArrow >= 22.
+
+    Regression test for a bug where _infer_schema fell back to reading every
+    fragment's physical_schema when the sampled fragment had a pa.null() column
+    (PyArrow < 22.0), causing O(N) metadata reads and memory usage.
+    """
+    if get_pyarrow_version() < parse_version("22.0.0"):
+        pytest.skip("Bounded permissive schema inspection requires PyArrow >= 22.0.0")
+
+    num_cols = 50
+    num_files = 1000
+    inspect_num_fragments = 1
+
+    def _write_files(directory, n_files):
+        directory.mkdir(exist_ok=True)
+        for i in range(n_files):
+            cols = {f"col_{j}": [0] for j in range(num_cols)}
+            # First file has a column of all nulls, which triggers the schema inference fallback.
+            if i == 0:
+                cols["null_col"] = pa.nulls(1)
+            else:
+                cols["null_col"] = [1]
+            pq.write_table(pa.table(cols), directory / f"part_{i:05d}.parquet")
+
+    _write_files(tmp_path, num_files)
+
+    inspect_calls = []
+    real_factory = pds.FileSystemDatasetFactory
+
+    # RSS deltas for this code path are sub-MiB in CI, so check the bounded
+    # schema-inspection behavior directly instead of comparing process memory.
+    class TrackingFactory:
+        def __init__(self, *args, **kwargs):
+            self._factory = real_factory(*args, **kwargs)
+
+        def inspect(self, **kwargs):
+            inspect_calls.append(kwargs)
+            return self._factory.inspect(**kwargs)
+
+        def finish(self, *args, **kwargs):
+            pytest.fail("Schema inference should not inspect every fragment")
+
+    monkeypatch.setattr(pds, "FileSystemDatasetFactory", TrackingFactory)
+
+    schema = _infer_schema(
+        [str(path) for path in sorted(tmp_path.iterdir())],
+        inspect_num_fragments=inspect_num_fragments,
+        filesystem=pafs.LocalFileSystem(),
+    )
+
+    assert inspect_calls == [
+        {
+            "fragments": inspect_num_fragments,
+            "promote_options": "permissive",
+        }
+    ]
+    assert "null_col" in schema.names

python/ray/scripts/scripts.py

@@ -41,6 +41,7 @@
     parse_resources_json,
 )
 from ray.autoscaler._private.cli_logger import add_click_logging_options, cf, cli_logger
+from ray.autoscaler._private.cli_output_helpers import print_next_steps_context_note
 from ray.autoscaler._private.commands import (
     RUN_ENV_TYPES,
     attach_cluster,
@@ -1082,6 +1083,7 @@ def start(
         cli_logger.success("-" * len(startup_msg))
         cli_logger.newline()
         with cli_logger.group("Next steps"):
+            print_next_steps_context_note(cli_logger, cf)
             dashboard_url = node.address_info["webui_url"]
             if ray_constants.ENABLE_RAY_CLUSTER:
                 cli_logger.print("To add another node to this Ray cluster, run")

python/ray/tests/BUILD.bazel

@@ -834,6 +834,22 @@ py_test_module_list(
     ],
 )
 
+py_test(
+    name = "test_cli_output_context",
+    size = "small",
+    srcs = ["test_cli_output_context.py"],
+    data = ["//python/ray/autoscaler:cli_output_context_test_data"],
+    tags = [
+        "exclusive",
+        "small_size_python_tests",
+        "team:core",
+    ],
+    deps = [
+        ":conftest",
+        "//:ray_lib",
+    ],
+)
+
 py_test_module_list(
     size = "small",
     files = [

python/ray/tests/test_cli_patterns/test_ray_start.txt

@@ -6,6 +6,9 @@ Ray runtime started.
 --------------------
 
 Next steps
+  Note: The following commands are intended for use on
+  the head node or within the cluster network\.
+
   To add another node to this Ray cluster, run
     ray start --address='.+'
 

python/ray/tests/test_cli_patterns/test_ray_start_windows_osx.txt

@@ -7,6 +7,9 @@ Ray runtime started.
 --------------------
 
 Next steps
+  Note: The following commands are intended for use on
+  the head node or within the cluster network\.
+
   To add another node to this Ray cluster, run
     RAY_ENABLE_WINDOWS_OR_OSX_CLUSTER=1 ray start --address='.+'
 

python/ray/tests/test_cli_patterns/test_ray_up.txt

@@ -41,7 +41,12 @@ Acquiring an up-to-date head node
   \[7/7\] Starting the Ray runtime
   New status: up-to-date
 
-Useful commands:
+-{60}
+Note: The output above is from the head node \(via `ray start`\)\.
+  Commands shown in 'Next steps' may only work from the head node
+  or from within the cluster network\.
+
+Useful commands for your local machine:
   To terminate the cluster:
     ray down .+
 

python/ray/tests/test_cli_patterns/test_ray_up_docker.txt

@@ -41,7 +41,12 @@ Acquiring an up-to-date head node
   \[7/7\] Starting the Ray runtime
   New status: up-to-date
 
-Useful commands:
+-{60}
+Note: The output above is from the head node \(via `ray start`\)\.
+  Commands shown in 'Next steps' may only work from the head node
+  or from within the cluster network\.
+
+Useful commands for your local machine:
   To terminate the cluster:
     ray down .+
 

python/ray/tests/test_cli_patterns/test_ray_up_record.txt

@@ -84,7 +84,11 @@
 .+\.py.*AWSNodeProvider: Set tag ray-runtime-config=.+ on \['.+'\]  \[LogTimer=.+\]
 .+\.py.*AWSNodeProvider: Set tag ray-file-mounts-contents=.+ on \['.+'\]  \[LogTimer=.+\]
 .+\.py.*New status: up-to-date
-.+\.py.*Useful commands:
+.+\.py.*-{60}
+.+\.py.*Note: The output above is from the head node \(via `ray start`\)\.
+.+\.py.*  Commands shown in 'Next steps' may only work from the head node
+.+\.py.*  or from within the cluster network\.
+.+\.py.*Useful commands for your local machine:
 .+\.py.*To terminate the cluster:
 .+\.py.*  ray down .+
 .+\.py.*To retrieve the IP address of the cluster head: