Repository: ray-project/ray
File: python/ray/data/tests/unit/test_deduping_schema.py (Python)

Code before:
            [
                pa.field("foo", pa.int64()),
                pa.field("bar", pa.string()),
                pa.field("baz", pa.bool8()),
                pa.field("qux", pa.float64()),
            ]
        )

        msg = "\n".join(caplog.messages)
        assert (
            """Operator produced a RefBundle with a different schema than the previous one.
Fields exclusive to the incoming schema (1 total):
    qux: double
Fields exclusive to the old schema (1 total):
    bar: string
Fields that have different types across the old and the incoming schemas (1 total):
    foo: int32 => int64
This may lead to unexpected behavior."""
            == msg
        )
    else:
        assert out_bundle.schema == old_schema
        assert not caplog.records


def test_dedupe_schema_many_fields_warning(caplog, propagate_logs):
    old_schema = pa.schema([pa.field(f"foo{i}", pa.int64()) for i in range(30)])
    incoming_schema = pa.schema(
        [
            pa.field("foo0", pa.int64()),
        ]
    )

    incoming_bundle = RefBundle([], owns_blocks=False, schema=incoming_schema)

    # Capture warnings
    with caplog.at_level(
        logging.WARNING,
    ):
        out_bundle, diverged = dedupe_schemas_with_validation(
            old_schema, incoming_bundle, enforce_schemas=True, warn=True
        )

    assert diverged

    assert out_bundle.schema == pa.schema(
        [pa.field(f"foo{i}", pa.int64()) for i in range(30)]
    )

    # Warning message should have truncated 9 schema fields
    msg = "\n".join(caplog.messages)

Review comment (suggestion):
can this test and test_dedupe_schema_mismatch_warning be `pytest.parameterized(truncation_length)`? (to avoid duplication)

Code after:
    bar: string
Fields that have different types across the old and the incoming schemas (4 total):
    foo: int32 => int64
    quux: uint16 => uint32
    corge: int64 => int32
    ... and 1 more
This may lead to unexpected behavior."""
            == msg
        )
    else:
        assert (
            """Operator produced a RefBundle with a different schema than the previous one.
Fields exclusive to the incoming schema (1 total):
    qux: double
Fields exclusive to the old schema (1 total):
    bar: string
Fields that have different types across the old and the incoming schemas (4 total):
    foo: int32 => int64
    quux: uint16 => uint32
    ... and 2 more
This may lead to unexpected behavior."""
            == msg
        )


def test_build_mismatch_warning_disordered():
    old_schema = pa.schema(
        [
            pa.field("foo", pa.int32()),
            pa.field("bar", pa.string()),
            pa.field("baz", pa.bool_()),
            pa.field("qux", pa.float64()),
        ]
    )
    incoming_schema = pa.schema(
        [
            pa.field("foo", pa.int32()),
            pa.field("baz", pa.bool_()),
            pa.field("bar", pa.string()),
            pa.field("qux", pa.float64()),
        ]
    )

    msg = _build_schemas_mismatch_warning(old_schema, incoming_schema)

    assert (
        """Operator produced a RefBundle with a different schema than the previous one.
Some fields are ordered differently across the old and the incoming schemas.
This may lead to unexpected behavior."""
        == msg
    )