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

Code before:
        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

    if _is_empty_schema(old_schema):
        return bundle, diverged

    # This check is fast assuming pyarrow schemas
    if old_schema == bundle.schema:
        return bundle, diverged

    diverged = True
    if warn and enforce_schemas:
        if _is_empty_schema(bundle.schema):
            old_fields = [
                (name, str(t)) for name, t in zip(old_schema.names, old_schema.types)
            ]
            warning_message = _format_info_message(
                "Operator produced a RefBundle with an empty/unknown schema.",
                list(map(lambda field: f"{field[0]}: {field[1]}", old_fields)),
            )
        else:
            warning_message = _find_schemas_mismatch(old_schema, bundle.schema)
        logger.warning(
            f"Operator produced a RefBundle with a different schema "
            f"than the previous one.\n"
            f"{warning_message}"
            f"This may lead to unexpected behavior."
        )
    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(

Review comment (suggestion):
This case can be handled in `_find_schemas_mismatch` so that the logic is consolidated

Code after:
        + 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

    if _is_empty_schema(old_schema):
        return bundle, diverged

    # This check is fast assuming pyarrow schemas
    if old_schema == bundle.schema:
        return bundle, diverged

    diverged = True
    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,