Repository: ray-project/ray
File: python/ray/data/_internal/execution/streaming_executor_state.py (Python)

Code before:

    return next_op


def _actor_info_summary_str(info: _ActorPoolInfo) -> str:
    total = info.running + info.pending + info.restarting
    base = f"Actors: {total}"

    if total == info.running:
        return base
    else:
        return f"{base} ({info})"


def _format_schema_mismatch_section(
    title: str, entries: List[str], truncate_num_mismatched_fields_to: int
) -> str:
    """Generate a titled section of the schema mismatch warning.

    Args:
        title: Section title describing the mismatched fields.
        entries: Field detail strings (name + type).
        truncate_num_mismatched_fields_to: Max entries shown before truncation.
    Returns:
        A formatted message for the warning section, or "" if entries is empty.

    """
    if not entries:
        return ""

    shown = entries[:truncate_num_mismatched_fields_to]
    remainder = len(entries) - truncate_num_mismatched_fields_to

    body = "\n".join(f"    {line}" for line in shown)
    suffix = f"\n    ... and {remainder} more" if remainder > 0 else ""

    return f"{title} ({len(entries)} total):\n{body}{suffix}\n"


def _build_schemas_mismatch_warning(
    old_schema: "Schema", new_schema: Optional["Schema"], truncation_length: int = 20
) -> str:
    from ray.data.block import _is_empty_schema

    if _is_empty_schema(new_schema):
        old_fields_info = [
            f"{name}: {str(t)}" for name, t in zip(old_schema.names, old_schema.types)
        ]
        is_empty_message = _format_schema_mismatch_section(
            "Operator produced a RefBundle with an empty/unknown schema.",
            old_fields_info,

Review comment (suggestion):
```suggestion
        truncate_num_mismatched_fields_to: Max entries shown before truncation.
        
    Returns:
        A formatted message for the warning section, or "" if entries is empty.
```

Code after:

    return next_op


def _actor_info_summary_str(info: _ActorPoolInfo) -> str:
    total = info.running + info.pending + info.restarting
    base = f"Actors: {total}"

    if total == info.running:
        return base
    else:
        return f"{base} ({info})"


def _format_schema_mismatch_section(
    title: str, entries: List[str], truncate_num_mismatched_fields_to: int
) -> str:
    """Generate a titled section of the schema mismatch warning.

    Args:
        title: Section title describing the mismatched fields.
        entries: Field detail strings (name + type).
        truncate_num_mismatched_fields_to: Max entries shown before truncation.

    Returns:
        A formatted message for the warning section, or "" if entries is empty.
    """
    if not entries:
        return ""

    shown = entries[:truncate_num_mismatched_fields_to]
    remainder = len(entries) - truncate_num_mismatched_fields_to

    body = "\n".join(f"    {line}" for line in shown)
    suffix = f"\n    ... and {remainder} more" if remainder > 0 else ""

    return f"{title} ({len(entries)} total):\n{body}{suffix}\n"


def _build_schemas_mismatch_warning(
    old_schema: "Schema", new_schema: Optional["Schema"], truncation_length: int = 20
) -> str:
    from ray.data.block import _is_empty_schema

    if _is_empty_schema(new_schema):
        old_fields_info = [
            f"{name}: {str(t)}" for name, t in zip(old_schema.names, old_schema.types)
        ]
        is_empty_message = _format_schema_mismatch_section(
            "Operator produced a RefBundle with an empty/unknown schema.",
            old_fields_info,