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

Code before:
        "Fields exclusive to the incoming schema", new_excl_fields_info
    )
    old_excl_fields_message = _format_info_message(
        "Fields exclusive to the old schema", old_excl_fields_info
    )
    changed_fields_message = _format_info_message(
        "Fields that have different types across the old and the incoming schemas",
        changed_fields_info,
    )

    disordered_message = ""
    if not new_exclusive_fields and not old_exclusive_fields and not changed_fields:
        assert old_pairs != new_pairs
        disordered_pairs = [
            old_pair
            for old_pair, new_pair in zip(old_pairs, new_pairs)
            if old_pair != new_pair
        ]
        disordered_pairs_info = list(
            map(
                lambda pair: f"{pair[0]}: {pair[1]}",
                disordered_pairs,
            )
        )
        disordered_message = _format_info_message(
            "Fields ordered differently across the old and the incoming schemas",
            disordered_pairs_info,
        )

    return (
        new_excl_fields_message
        + old_excl_fields_message
        + changed_fields_message
        + disordered_message
    )


def dedupe_schemas_with_validation(
    old_schema: Optional["Schema"],
    bundle: "RefBundle",
    warn: bool = True,
    enforce_schemas: bool = False,
) -> Tuple["RefBundle", bool]:
    """Unify/Dedupe two schemas, warning if warn=True

    Args:
        old_schema: The old schema to unify. This can be `None`, in which case
            the new schema will be used as the old schema.
        bundle: The new `RefBundle` to unify with the old schema.
        warn: Raise a warning if the schemas diverge.
        enforce_schemas: If `True`, allow the schemas to diverge and return unified schema.

Review comment (suggestion):
Sorry if i mentioned ordering mattered by accident, but right now Ray Data doesn't care about ordering of schemas, so this `if` branch can be removed

Code after:
            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
    ]
    changed_fields_info = [
        f"{field}: {old_fields[field]} => {new_fields[field]}"
        for field in changed_fields
    ]

    new_excl_fields_message = _format_schema_mismatch_section(
        "Fields exclusive to the incoming schema",
        new_excl_fields_info,
        truncate_num_mismatched_fields_to=truncation_length,
    )
    old_excl_fields_message = _format_schema_mismatch_section(
        "Fields exclusive to the old schema",
        old_excl_fields_info,
        truncate_num_mismatched_fields_to=truncation_length,
    )
    changed_fields_message = _format_schema_mismatch_section(
        "Fields that have different types across the old and the incoming schemas",
        changed_fields_info,
        truncate_num_mismatched_fields_to=truncation_length,
    )

    disordered_message = ""
    if not new_exclusive_fields and not old_exclusive_fields and not changed_fields:
        assert old_fields == new_fields
        disordered_message = "Some fields are ordered differently across the old and the incoming schemas.\n"

    return (
        "Operator produced a RefBundle with a different schema "