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

Code before:
    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 = [
        name
        for name in new_fields
        if name in old_fields and new_fields[name] != old_fields[name]
    ]

    new_excl_fields_info = list(
        map(lambda field: f"{field}: {new_fields[field]}", new_exclusive_fields)
    )
    old_excl_fields_info = list(
        map(lambda field: f"{field}: {old_fields[field]}", old_exclusive_fields)
    )
    changed_fields_info = list(
        map(
            lambda field: f"{field}: {old_fields[field]} => {new_fields[field]}",
            changed_fields,
        )
    )

    new_excl_fields_message = _format_info_message(
        "Fields exclusive to the incoming schema", new_excl_fields_info
    )

Review comment (suggestion):
I think it's ok to inline `_schema_to_fields` since `names` and `types` properties are pretty standard. Also, the type of `old_schema` and `new_schema` should be the same, so you don't need to check for types of both if you know one 

Code after:

    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,
        )
        return is_empty_message + "This may lead to unexpected behavior."

    # We assume old_schema and new_schema have the same underlying type
    # and can only either be PyArrow schemas or PandasBlockSchema
    old_fields = {name: str(t) for name, t in zip(old_schema.names, old_schema.types)}
    new_fields = {name: str(t) for name, t in zip(new_schema.names, new_schema.types)}

    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 = [
        name
        for name in new_fields
        if name in old_fields and new_fields[name] != old_fields[name]
    ]

    new_excl_fields_info = [
        f"{field}: {new_fields[field]}" for field in new_exclusive_fields
    ]
    old_excl_fields_info = [
        f"{field}: {old_fields[field]}" for field in old_exclusive_fields
    ]