[Data] Make PandasBlock.size_bytes deterministic

[dragongu]

Description

PandasBlockAccessor.size_bytes() estimates string/object column sizes by sampling up to 200 rows via pandas.DataFrame.sample(n=...) without a seed, so two blocks holding identical data report different sizes. That estimate drives block splitting in streaming generators, so a re-executed task can emit a different number of objects than the original attempt — which leaves downstream consumers silently hanging or silently dropping data (the companion #64394 fails fast when that mismatch occurs). This PR fixes the sampling non-determinism by passing random_state=0, and skips sampling entirely when the column fits in the sample. User UDFs remain responsible for their own determinism.

Symptom. A streaming generator task re-executed for lineage reconstruction can leave the pipeline silently hanging or silently dropping data, with no error logged.

Root cause. PandasBlockAccessor.size_bytes() estimates string/object columns by sampling up to 200 rows via pandas.DataFrame.sample(n=...). No seed is passed, so each call samples different rows and two blocks holding identical data report different sizes.

How it turns into a bug. That estimate drives block splitting:

1. OutputBuffer.next() calls size_bytes() to compute target_num_rows, so a task can split the same output into a different number of blocks each time it runs.
2. The first successful attempt pins the end-of-stream index to the object count it produced.
3. On re-execution, a different split produces a different object count:
   • fewer → downstream consumers block on indices that are never produced (silent hang);
   • more → the extras past the pinned end-of-stream index are dropped by ObjectRefStream::InsertToStream (silent data loss).

Fix. Pass random_state=0 to sample(), and skip sampling when the column has fewer than sample_size rows. Identical input now always yields the same size estimate. This removes one Ray-internal source of streaming-generator object-count non-determinism; user UDFs remain responsible for their own determinism.

Scope. Only PandasBlockAccessor.size_bytes() is affected — the Arrow equivalent already uses table.nbytes and is deterministic. Pandas blocks are produced along several common paths:

• a UDF that returns a pandas.DataFrame (e.g. map_batches(fn, batch_format="pandas")), kept as a pandas block by BlockAccessor.batch_to_block;
• batches that Arrow cannot convert (object/ragged columns), which fall back to pandas blocks;
• split_repartition preserving a pandas input schema; and
• the built-in preprocessors (encoder, vectorizer, imputer).

On released branches these paths default to pandas, so the bug is broadly reachable. On master, #61733 added a default that converts many UDF outputs to Arrow, but the fallback and pandas-schema paths above still build pandas blocks.

Related issues

None.

Additional information

Reproduction (before this PR, the two printed sizes frequently differ):

import pandas as pd
from ray.data._internal.pandas_block import PandasBlockAccessor

data = [f"str_{i}" for i in range(10_000)]
b1 = pd.DataFrame({"col": pd.Series(data, dtype="string")})
b2 = pd.DataFrame({"col": pd.Series(data, dtype="string")})

print(PandasBlockAccessor.for_block(b1).size_bytes())
print(PandasBlockAccessor.for_block(b2).size_bytes())

• Test: added TestSizeBytes.test_deterministic_across_blocks, asserting that two blocks holding identical data report the same size_bytes().

[gemini-code-assist]

Code Review

This pull request ensures that size estimation in PandasBlockAccessor.size_bytes() is deterministic by using a fixed random_state=0 when sampling columns. This prevents issues like silent hangs or data loss during lineage reconstruction. A corresponding unit test was added to verify determinism. The reviewer suggested a performance optimization to bypass sampling when the sample size equals the total size.