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

Code before:
        return f"{base} ({info})"


SCHEMA_MISMATCH_FIELDS_TO_SHOW = 20


def _schema_to_fields(schema: Optional["Schema"]) -> Dict[str, str]:
    # Helper function for converting Schemas into List of FieldRepr
    if schema is None:
        return {}

    import pyarrow as pa

    if isinstance(schema, pa.Schema):
        return {field.name: str(field.type) for field in schema}

    from ray.data._internal.pandas_block import PandasBlockSchema

    if isinstance(schema, PandasBlockSchema):
        return {name: str(t) for name, t in zip(schema.names, schema.types)}

    if isinstance(schema, type):
        return {"<plain Python type>": schema.__name__}

    # Fallback
    return {"<unknown schema type>": str(schema)}


def _format_info_message(title: str, entries: list[str]) -> str:
    if not entries:
        return ""

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

    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 _find_schemas_mismatch(
    old_schema: "Schema", new_schema: "Schema"
) -> Tuple[str, str, str]:
    # Transform both schemas into FieldRepr Lists
    old_fields = _schema_to_fields(old_schema)
    new_fields = _schema_to_fields(new_schema)

    new_exclusive_fields = [name for name in new_fields if name not in old_fields]
    old_exclusive_fields = [name for name in old_fields if name not in new_fields]
    changed_fields = [

Review comment (suggestion):
You should be able to assume that the schema is either pyarrow or pandasblockschema, so the fallback is unneccessary. What you can do is check `isinstance(schema, pa.Schema)`, then if it's not that `assert isinstance(schema, PandasBlockSchema)`

Code after:


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,
            truncate_num_mismatched_fields_to=truncation_length,
        )