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

Code before:
                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,
        ),
        diverged,
    )


def format_op_state_summary(
    op_state: OpState, resource_manager: ResourceManager, verbose: bool = False
) -> str:
    """Get a formatted summary of the OpState for progress reporting."""
    # Active tasks
    active = op_state.op.num_active_tasks()
    desc = f"Tasks: {active}"

Review comment (bug):
The new, more concise warning message (including the "Previous schema mismatched fields" / "New schema mismatched fields" sections and the per-field formatting) is not covered by tests, so future changes could silently regress the message content or reintroduce overly verbose schema dumps without failing any tests. It would be good to add a test (e.g. in `test_deduping_schema.py`) that triggers a schema mismatch with `enforce_schemas=True` and asserts on the logged message, to lock in the intended behavior.

Code after:
    )
    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.
        enforce_schemas: If `True`, allow the schemas to diverge and return unified schema.
            If `False`, but keep the old schema.

    Returns:
        A ref bundle with the unified schema of the two input schemas.
    """

    # Note, often times the refbundles correspond to only one schema. We can reduce the
    # memory footprint of multiple schemas by keeping only one copy.
    diverged = False

    from ray.data.block import _is_empty_schema