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

Code before:
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_info_message(
    title: str, entries: List[str], truncate_num_mismatched_fields_to: int
) -> str:
    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 _find_schemas_mismatch(
    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)
        ]
        return _format_info_message(
            "Operator produced a RefBundle with an empty/unknown schema.",
            old_fields_info,
            truncate_num_mismatched_fields_to=truncation_length,
        )

    # 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]

Review comment (suggestion):
I felt confused by this name. I think the name `_find_schemas_mismatch` indicates that the function returns a list of "mismatches" or a "mismatch", but really this more like "get me the warning message for the mismatched schemas."

Code after:
    ranks = ranker.rank_operators(eligible_ops, topology, resource_manager)

    assert len(eligible_ops) == len(ranks), (eligible_ops, ranks)

    next_op, _ = min(zip(eligible_ops, ranks), key=lambda t: t[1])

    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 = [