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

Code before:
                if (
                    (old_idx > 0)
                    and (new_idx > 0)
                    and (old_fields[old_idx - 1] == new_fields[new_idx - 1])
                ):
                    old_idx -= 1
                    new_idx -= 1
                elif (new_idx == 0) or (
                    old_idx > 0 and dp[old_idx - 1][new_idx] >= dp[old_idx][new_idx - 1]
                ):
                    old_mismatch.append(old_fields[old_idx - 1])
                    old_idx -= 1
                else:
                    new_mismatch.append(new_fields[new_idx - 1])
                    new_idx -= 1
            old_mismatch.reverse()
            new_mismatch.reverse()

            old_fields_info = "\n".join(
                map(lambda field: f"    {field.name}: {field.type_desc}", old_mismatch)
            )
            new_fields_info = "\n".join(
                map(lambda field: f"    {field.name}: {field.type_desc}", new_mismatch)
            )

            return old_fields_info, new_fields_info

        old_fields_info, new_fields_info = find_schemas_mismatch(
            old_schema, bundle.schema
        )

        logger.warning(
            f"Operator produced a RefBundle with a different schema "
            f"than the previous one. This may lead to unexpected behavior.\n"
            f"Previous schema mismatched fields:\n"
            f"{old_fields_info}"
            f"\nNew schema mismatched fields:\n"
            f"{new_fields_info}"
        )

    if enforce_schemas:
        old_schema = unify_schemas_with_validation([old_schema, bundle.schema])

    return (
        RefBundle(
            bundle.blocks,
            schema=old_schema,
            owns_blocks=bundle.owns_blocks,
            output_split_idx=bundle.output_split_idx,
            _cached_object_meta=bundle._cached_object_meta,
            _cached_preferred_locations=bundle._cached_preferred_locations,

Review comment (performance):
The new warning formatting logic builds a full O(N*M) DP table to compute the longest common subsequence of fields just to identify mismatches, which adds quadratic time and memory overhead in the number of fields and significantly increases the complexity of this logging path. Since this code is only used to format a warning, a simpler linear-time diff (for example, comparing by field name and type) would be easier to maintain and scale better for very wide schemas, while still meeting the goal of showing only mismatched fields.

Code after:
    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 "
        "than the previous one.\n"
        + new_excl_fields_message
        + old_excl_fields_message
        + changed_fields_message
        + disordered_message
        + "This may lead to unexpected behavior."
    )


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

    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.