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

Code before:
        truncation_length: When warning is active, limits the number of mismatched fields
            between the old schema and the incoming schema of the new `RefBundle`.

    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 warn and enforce_schemas:
        warning_message = _find_schemas_mismatch(
            old_schema, bundle.schema, truncation_length=truncation_length
        )
        logger.warning(f"{warning_message}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(
    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}"
    if (
        op_state.op._in_task_submission_backpressure

Review comment (refactor):
Flags like `warn` are generally a design smell, because they indicate that the abstraction might be trying to do too many things, and also it makes the abstraction harder to use.

Like, without reading the docstrings, I don't think it's intuitive that a `dedupe_schemas_with_validation` would have a `truncation_length` method.

Since we already return whether there's a divergence, would it make sense to do something like this in `OpState.add_output`?
```diff
        ref, diverged = dedupe_schemas_with_validation(
            self._schema,
            ref,
+           # dedupe_schema no longers directly emits warning
-           warn=not self._warned_on_schema_divergence,
            enforce_schemas=self.op.data_context.enforce_schemas,
        )
+       if diverged and not self._warned_on_schema_divergence:
+           # We introduce a new function just for warning
+           logger.warn(get_schema_mismatch_warning(...))
```


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