test_dataset.incn

"""Test: dataset types and RFC 001 contract (types, hierarchy, functions import)."""

from std.testing import fail_t, parametrize
from metadata import inql_version
from aggregate_builders import AggregateKind
from dataset import DataSet, BoundedDataSet, UnboundedDataSet, DataFrame, LazyFrame, DataStream, lazy_frame_named_table
from dataset.materialization import DataFrameMaterialization
from dataset.ops import filter_ds, join_ds
from functions import (
    always_false,
    always_true,
    avg,
    bool_lit,
    col,
    count,
    count_distinct,
    count_if,
    eq,
    explode,
    explode_outer,
    flatten,
    float_expr,
    int_expr,
    int_lit,
    inline,
    inline_outer,
    lit,
    max,
    min,
    mul,
    posexplode,
    posexplode_outer,
    row_number,
    str_expr,
    str_lit,
    sum,
    stack,
    window,
)
from projection_builders import ColumnExprKind, column_expr_kind, column_expr_name, with_column_assignment
from substrait.function_extensions import (
    EXPLODE_EXTENSION_URI,
    EXPLODE_OUTER_EXTENSION_URI,
    FLATTEN_EXTENSION_URI,
    FUNCTION_EXTENSION_URI,
    INLINE_EXTENSION_URI,
    INLINE_OUTER_EXTENSION_URI,
    POSEXPLODE_EXTENSION_URI,
    POSEXPLODE_OUTER_EXTENSION_URI,
    STACK_EXTENSION_URI,
)
from substrait.inspect import plan_contains_relation_kind, plan_has_extension_urn, relation_kind_name, root_rel
from substrait.plans import plan_encoded_len, plan_from_named_table, plan_from_root_relation
from substrait.relations import read_named_table_rel
from substrait.schema_registry import register_named_table_schema
from substrait.schema import RowColumnSpec, SubstraitPrimitiveKind


@derive(Clone)
model Order:
    id: int


@derive(Clone)
model OrderProjection:
    order_id: int


@derive(Clone)
model Event:
    id: int


def _register_order_schema(table_name: str) -> None:
    register_named_table_schema(table_name, [RowColumnSpec(name="id", kind=SubstraitPrimitiveKind.I64, nullable=false)])


def _column_expr_kind_name(kind: ColumnExprKind) -> str:
    match kind:
        ColumnExprKind.IntLiteral => return "IntLiteral"
        ColumnExprKind.FloatLiteral => return "FloatLiteral"
        ColumnExprKind.StringLiteral => return "StringLiteral"
        ColumnExprKind.BoolLiteral => return "BoolLiteral"
        _ => return "Other"


def _literal_helper_kind_name(helper_name: str) -> str:
    match helper_name:
        "int_expr" => return _column_expr_kind_name(column_expr_kind(int_expr(2)))
        "float_expr" => return _column_expr_kind_name(column_expr_kind(float_expr(2.5)))
        "str_expr" => return _column_expr_kind_name(column_expr_kind(str_expr("open")))
        "int_lit" => return _column_expr_kind_name(column_expr_kind(int_lit(3)))
        "str_lit" => return _column_expr_kind_name(column_expr_kind(str_lit("closed")))
        "bool_lit" => return _column_expr_kind_name(column_expr_kind(bool_lit(true)))
        _ => return fail_t(f"unknown literal helper: {helper_name}")


def _accept_data_set_generic[T with Clone](data: DataSet[T]) -> DataSet[T]:
    return data


def _accept_bounded_generic[T with Clone](data: BoundedDataSet[T]) -> BoundedDataSet[T]:
    return data


def _accept_unbounded_generic[T with Clone](data: UnboundedDataSet[T]) -> UnboundedDataSet[T]:
    return data


def _accept_data_frame_concrete[T with Clone](data: DataFrame[T]) -> DataFrame[T]:
    return data


def _accept_lazy_frame_concrete[T with Clone](data: LazyFrame[T]) -> LazyFrame[T]:
    return data


def _accept_data_stream_concrete[T with Clone](data: DataStream[T]) -> DataStream[T]:
    return data


def _upcast_data_frame_to_bounded[T with Clone](data: DataFrame[T]) -> BoundedDataSet[T]:
    return data


def _upcast_lazy_frame_to_bounded[T with Clone](data: LazyFrame[T]) -> BoundedDataSet[T]:
    return data


def _upcast_data_stream_to_unbounded[T with Clone](data: DataStream[T]) -> UnboundedDataSet[T]:
    return data


def _upcast_bounded_to_data_set[T with Clone](data: BoundedDataSet[T]) -> DataSet[T]:
    return data


def _upcast_unbounded_to_data_set[T with Clone](data: UnboundedDataSet[T]) -> DataSet[T]:
    return data


def _touch[T](x: T) -> None:
    """Consume a value so assignment chains are not unused-variable warnings."""
    pass


def _order_witness() -> list[Order]:
    return [Order(id=0)]


def _event_witness() -> list[Event]:
    return [Event(id=0)]


def _preview_materialization(columns: list[str], row_count: int, preview_text: str) -> DataFrameMaterialization:
    """Build one structured materialization fixture for direct DataFrame tests."""
    return DataFrameMaterialization(resolved_columns=columns, row_count=row_count, preview_text=preview_text)


def _register_projection_test_schema(table_name: str) -> None:
    register_named_table_schema(table_name, [RowColumnSpec(name="id", kind=SubstraitPrimitiveKind.I64, nullable=false)])


def _compile_hierarchy_assignability(
    order_frame: DataFrame[Order],
    order_lazy: LazyFrame[Order],
    event_stream: DataStream[Event],
) -> None:
    """Compile-time shape checks for concrete -> trait -> supertrait assignments."""
    sink0 = _upcast_bounded_to_data_set(_upcast_data_frame_to_bounded(order_frame))
    sink1 = _upcast_bounded_to_data_set(_upcast_lazy_frame_to_bounded(order_lazy))
    sink2 = _upcast_unbounded_to_data_set(_upcast_data_stream_to_unbounded(event_stream))
    _touch(sink0)
    _touch(sink1)
    _touch(sink2)


def test_smoke__dataset_types_are_published() -> None:
    """RFC 001: carrier types and aggregate helpers are importable."""
    # -- Arrange --
    amount = col("amount")

    # -- Act --
    sum_result = sum(amount)
    count_result = count()
    expression_count_result = count(amount)
    distinct_count_result = count_distinct(amount)
    filtered_count_result = count_if(eq(col("status"), str_lit("paid")))
    avg_result = avg(amount)
    min_result = min(amount)
    max_result = max(amount)

    # -- Assert --
    assert column_expr_name(sum_result.expr) == "amount", "sum should preserve the selected input column expression"
    assert not count_result.has_expr, "count should preserve row-count measures when called without an expression"
    assert len(count_result.arguments) == 0, "row-count measures should not carry synthetic expression arguments"
    assert expression_count_result.has_expr, "count(expr) should mark expression-count measures explicitly"
    assert column_expr_name(expression_count_result.expr) == "amount", "count(expr) should preserve the selected input column expression"
    assert distinct_count_result.is_distinct, "count_distinct should build a distinct aggregate measure"
    assert filtered_count_result.has_filter, "count_if should build a filtered aggregate measure"
    assert avg_result.kind == AggregateKind.Avg, "avg should build an aggregate measure"
    assert min_result.kind == AggregateKind.Min, "min should build an aggregate measure"
    assert max_result.kind == AggregateKind.Max, "max should build an aggregate measure"


@parametrize("helper_name, expected_kind", [("int_expr", "IntLiteral"), ("float_expr", "FloatLiteral"), ("str_expr", "StringLiteral"), ("int_lit", "IntLiteral"), ("str_lit", "StringLiteral"), ("bool_lit", "BoolLiteral")])
def test_scalar_literals__filter_helper_aliases_share_column_expr_model(helper_name: str, expected_kind: str) -> None:
    """Typed filter literal helper aliases should build canonical column expressions."""
    # -- Arrange --

    # -- Act --
    actual_kind = _literal_helper_kind_name(helper_name)

    # -- Assert --
    assert actual_kind == expected_kind, f"{helper_name} should build a {expected_kind} scalar literal"


def test_type_contracts__signature_tiers_compile() -> None:
    """RFC 001: DataSet / BoundedDataSet / UnboundedDataSet accept concrete carriers."""
    # -- Arrange --
    row = Order(id=1)
    df: DataFrame[Order] = DataFrame(
        _type_witness=_order_witness(),
        _materialization=DataFrameMaterialization.empty(),
        _substrait_rel=read_named_table_rel("orders"),
        _planned_columns=[],
    )
    st: DataStream[Event] = DataStream(
        _type_witness=_event_witness(),
        _substrait_rel=read_named_table_rel("events"),
        _planned_columns=[],
    )

    # -- Act --
    any_df = _accept_data_set_generic(df)
    bounded_df = _accept_bounded_generic(df)
    any_st = _accept_data_set_generic(st)
    unbounded_st = _accept_unbounded_generic(st)

    # -- Assert --
    _touch(row)
    _touch(any_df)
    _touch(bounded_df)
    _touch(any_st)
    _touch(unbounded_st)


def test_type_contracts__concrete_carriers_compile() -> None:
    """RFC 001: concrete carriers flow through generic helper signatures."""
    # -- Arrange --
    df: DataFrame[Order] = DataFrame(
        _type_witness=_order_witness(),
        _materialization=DataFrameMaterialization.empty(),
        _substrait_rel=read_named_table_rel("orders"),
        _planned_columns=[],
    )
    lf: LazyFrame[Order] = lazy_frame_named_table("orders")
    st: DataStream[Event] = DataStream(
        _type_witness=_event_witness(),
        _substrait_rel=read_named_table_rel("events"),
        _planned_columns=[],
    )

    # -- Act --
    concrete_df = _accept_data_frame_concrete(df)
    concrete_lf = _accept_lazy_frame_concrete(lf)
    concrete_st = _accept_data_stream_concrete(st)

    # -- Assert --
    _touch(concrete_df)
    _touch(concrete_lf)
    _touch(concrete_st)


def test_hierarchy__concrete_and_supertrait_assignability() -> None:
    """RFC 001: concrete -> bounded/unbounded trait -> DataSet chains compile."""
    # -- Arrange --
    df: DataFrame[Order] = DataFrame(
        _type_witness=_order_witness(),
        _materialization=DataFrameMaterialization.empty(),
        _substrait_rel=read_named_table_rel("orders"),
        _planned_columns=[],
    )
    lf: LazyFrame[Order] = lazy_frame_named_table("orders")
    st: DataStream[Event] = DataStream(
        _type_witness=_event_witness(),
        _substrait_rel=read_named_table_rel("events"),
        _planned_columns=[],
    )

    # -- Act --
    _compile_hierarchy_assignability(df, lf, st)

    # -- Assert --
    _touch(df)
    _touch(lf)
    _touch(st)


def test_type_contracts__concrete_and_trait_types_match_generic_arguments() -> None:
    """Generic parameter T matches through supported callable trait boundaries."""
    # -- Arrange --
    df: DataFrame[Order] = DataFrame(
        _type_witness=_order_witness(),
        _materialization=DataFrameMaterialization.empty(),
        _substrait_rel=read_named_table_rel("orders"),
        _planned_columns=[],
    )
    lf: LazyFrame[Order] = lazy_frame_named_table("orders")
    st: DataStream[Event] = DataStream(
        _type_witness=_event_witness(),
        _substrait_rel=read_named_table_rel("events"),
        _planned_columns=[],
    )

    # -- Act --
    bounded_df = _accept_bounded_generic(df)
    bounded_lf = _accept_bounded_generic(lf)
    ub = _accept_unbounded_generic(st)

    # -- Assert --
    _touch(bounded_df)
    _touch(bounded_lf)
    _touch(ub)


def test_version__inql_version_is_published() -> None:
    """InQL version should be available from metadata."""
    # -- Arrange --
    expected_version = "0.1.0"

    # -- Act --
    version = inql_version()

    # -- Assert --
    assert version == expected_version, "InQL version should be 0.1.0"


def test_dataset_ops__method_wrapper_matches_canonical_function() -> None:
    """Method wrappers should preserve canonical dataset_ops semantics."""
    # -- Arrange --
    base_df = DataFrame(
        _type_witness=_order_witness(),
        _materialization=DataFrameMaterialization.empty(),
        _substrait_rel=read_named_table_rel("orders"),
        _planned_columns=[],
    )

    # -- Act --
    via_function = filter_ds(read_named_table_rel("orders"), always_true())
    via_method = base_df.filter(always_true())

    # -- Assert --
    assert relation_kind_name(via_function) == "FilterRel", "canonical function should produce FilterRel root"
    assert plan_encoded_len(via_method.to_substrait_plan()) > 0, "method wrapper should still emit a real proto plan"
    assert relation_kind_name(root_rel(via_method.to_substrait_plan())) == "FilterRel", "method wrapper plan should preserve filter root"


def test_dataset_ops__data_frame_transforms_invalidate_stale_materialization() -> None:
    """DataFrame transforms should not retain payload-derived schema once the logical relation changes."""
    # -- Arrange --
    base_df = DataFrame(
        _type_witness=_order_witness(),
        _materialization=_preview_materialization(["id"], 1, "| id |\n| 1 |"),
        _substrait_rel=read_named_table_rel("orders"),
        _planned_columns=[],
    )

    # -- Act --
    transformed = base_df.filter(always_true())

    # -- Assert --
    assert transformed.preview_text() == "", "transformed DataFrame should clear stale preview text"
    assert transformed.row_count() == 0, "transformed DataFrame should clear stale collected row-count metadata"
    assert len(transformed.resolved_columns()) == 0, "transformed DataFrame should clear resolved columns derived from stale materialization"
    assert relation_kind_name(root_rel(transformed.to_substrait_plan())) == "FilterRel", "transformed DataFrame should still preserve the updated logical relation"


def test_dataset_ops__all_carriers_emit_real_plans() -> None:
    # -- Arrange --
    df = DataFrame(
        _type_witness=_order_witness(),
        _materialization=DataFrameMaterialization.empty(),
        _substrait_rel=read_named_table_rel("orders"),
        _planned_columns=[],
    )
    lf: LazyFrame[Order] = lazy_frame_named_table("orders")
    st = DataStream(_type_witness=_event_witness(), _substrait_rel=read_named_table_rel("events"), _planned_columns=[])

    # -- Act --
    df_plan = df.to_substrait_plan()
    lf_plan = lf.to_substrait_plan()
    st_plan = st.to_substrait_plan()

    # -- Assert --
    assert plan_encoded_len(df_plan) > 0, "DataFrame should emit a real plan"
    assert plan_encoded_len(lf_plan) > 0, "LazyFrame should emit a real plan"
    assert plan_encoded_len(st_plan) > 0, "DataStream should emit a real plan"


def test_dataset_ops__with_column_updates_lazy_planned_columns_with_add_or_replace_semantics() -> None:
    """with_column should append new names and replace existing names in place at lazy plan time."""
    # -- Arrange --
    _register_projection_test_schema("orders_projection")
    base: LazyFrame[Order] = lazy_frame_named_table("orders_projection")

    # -- Act --
    augmented = base.with_column("double_id", mul(col("id"), 2))
    replaced = augmented.with_column("id", lit(0))

    # -- Assert --
    augmented_cols = augmented.planned_columns()
    replaced_cols = replaced.planned_columns()
    assert len(augmented_cols) == 2, "with_column should append one new projected output column"
    assert augmented_cols[0] == "id", "existing source columns should stay ahead of appended projections"
    assert augmented_cols[1] == "double_id", "with_column should append the requested output name"
    assert len(replaced_cols) == 2, "replacing an existing name should preserve output column count"
    assert replaced_cols[0] == "id", "replacing an existing name should preserve that slot's ordinal position"
    assert replaced_cols[1] == "double_id", "later projected columns should keep their relative order after replacement"


def test_dataset_ops__select_retargets_output_row_type_while_preserving_carrier_kind() -> None:
    """SELECT is the schema-changing boundary used by RFC003 query blocks."""
    # -- Arrange --
    _register_projection_test_schema("orders_projection_types")
    assignments = [with_column_assignment("order_id", col("id"))]
    base_df: DataFrame[Order] = DataFrame(
        _type_witness=_order_witness(),
        _materialization=DataFrameMaterialization.empty(),
        _substrait_rel=read_named_table_rel("orders_projection_types"),
        _planned_columns=[],
    )
    base_lazy: LazyFrame[Order] = lazy_frame_named_table("orders_projection_types")
    base_stream: DataStream[Event] = DataStream(
        _type_witness=_event_witness(),
        _substrait_rel=read_named_table_rel("orders_projection_types"),
        _planned_columns=[],
    )

    # -- Act --
    projected_df: DataFrame[OrderProjection] = base_df.select(assignments)
    projected_lazy: LazyFrame[OrderProjection] = base_lazy.select(assignments)
    projected_stream: DataStream[OrderProjection] = base_stream.select(assignments)

    # -- Assert --
    assert projected_df.planned_columns() == ["order_id"], "DataFrame SELECT should expose the projected row shape"
    assert projected_lazy.planned_columns() == ["order_id"], "LazyFrame SELECT should expose the projected row shape"
    assert projected_stream.planned_columns() == ["order_id"], "DataStream SELECT should expose the projected row shape"


def test_dataset_ops__api_lowered_boundary_facts_stay_stable() -> None:
    # -- Arrange --
    left = read_named_table_rel("orders")
    right = read_named_table_rel("orders_archive")

    # -- Act --
    joined_plan = plan_from_root_relation(join_ds(left, right, always_true()), ["id"])

    # -- Assert --
    assert relation_kind_name(root_rel(joined_plan)) == "JoinRel", "canonical join function should still lower to a JoinRel root"


def test_lazy_frame__immutable_branching_and_origin_mapping_hold() -> None:
    # -- Arrange --
    base: LazyFrame[Order] = lazy_frame_named_table("orders")

    # -- Act --
    filtered: LazyFrame[Order] = base.filter(always_false())

    # -- Assert --
    assert plan_encoded_len(base.to_substrait_plan()) > 0, "base lazy carrier should still emit a real proto-backed plan"
    assert relation_kind_name(root_rel(filtered.to_substrait_plan())) == "FilterRel", "Prism-backed LazyFrame should still lower to FilterRel"
    assert plan_contains_relation_kind(filtered.to_substrait_plan(), "ReadRel"), "derived filter plans should still preserve the read root"


def test_lazy_frame__independent_roots_can_join_and_lower() -> None:
    # -- Arrange --
    left: LazyFrame[Order] = lazy_frame_named_table("orders").filter(always_false())
    right: LazyFrame[Order] = lazy_frame_named_table("orders_archive").filter(always_false())

    # -- Act --
    joined: LazyFrame[Order] = left.join(right, always_true())
    plan = joined.to_substrait_plan()

    # -- Assert --
    assert relation_kind_name(root_rel(plan)) == "JoinRel", "joined lazy carriers should lower to a JoinRel root"
    assert plan_contains_relation_kind(plan, "FilterRel"), "joined lazy carriers should preserve nested filters"
    assert plan_contains_relation_kind(plan, "ReadRel"), "joined lazy carriers should still reach the original read roots"


def test_lazy_frame__native_prism_ops_preserve_current_boundary_shapes() -> None:
    # -- Arrange --
    _register_order_schema("orders")
    register_named_table_schema(
        "orders_generator_dataset",
        [RowColumnSpec(name="id", kind=SubstraitPrimitiveKind.I64, nullable=false), RowColumnSpec(
            name="line_items",
            kind=SubstraitPrimitiveKind.String,
            nullable=true,
        )],
    )

    # -- Act --
    projected: LazyFrame[Order] = lazy_frame_named_table[Order]("orders").select()
    grouped: LazyFrame[Order] = lazy_frame_named_table("orders").group_by([col("id")])

    aggregated = grouped.agg([count()])
    ordered: LazyFrame[Order] = lazy_frame_named_table("orders").order_by([col("id")])
    limited: LazyFrame[Order] = lazy_frame_named_table("orders").limit(10)
    generated: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        explode(col("line_items"), "line_item"),
    )
    generated_outer: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        explode_outer(col("line_items"), "line_item"),
    )
    generated_positional: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        posexplode(col("line_items"), "position", "line_item"),
    )
    generated_positional_outer: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        posexplode_outer(col("line_items"), "position", "line_item"),
    )
    generated_inline: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        inline(col("line_items"), ["sku", "quantity"]),
    )
    generated_inline_outer: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        inline_outer(col("line_items"), ["sku", "quantity"]),
    )
    generated_flatten: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        flatten(col("line_items"), "line_item"),
    )
    generated_stack: LazyFrame[Order] = lazy_frame_named_table("orders_generator_dataset").generate(
        stack(2, [str_lit("sku_a"), int_lit(1), str_lit("sku_b"), int_lit(2)], ["sku", "quantity"]),
    )
    windowed: LazyFrame[Order] = lazy_frame_named_table("orders").with_window_column(
        "row_num",
        row_number().over(window().order_by([col("id")])),
    )

    # -- Assert --
    assert relation_kind_name(root_rel(projected.to_substrait_plan())) == "ProjectRel", "select should lower through the project boundary shape"
    assert relation_kind_name(root_rel(grouped.to_substrait_plan())) == "AggregateRel", "group_by should lower to the aggregate scaffold boundary shape"
    assert relation_kind_name(root_rel(aggregated.to_substrait_plan())) == "AggregateRel", "agg should lower to AggregateRel"
    assert relation_kind_name(root_rel(ordered.to_substrait_plan())) == "SortRel", "order_by should lower to SortRel"
    assert relation_kind_name(root_rel(limited.to_substrait_plan())) == "FetchRel", "limit should lower to FetchRel"
    assert relation_kind_name(root_rel(generated.to_substrait_plan())) == "ExtensionSingleRel", "generate should lower through the relation extension boundary"
    assert generated.planned_columns() == ["id", "line_items", "line_item"], "generate should append declared output aliases"
    assert plan_has_extension_urn(generated_outer.to_substrait_plan(), EXPLODE_OUTER_EXTENSION_URI), "outer explode should use its relation extension URI"
    assert plan_has_extension_urn(generated_positional.to_substrait_plan(), POSEXPLODE_EXTENSION_URI), "posexplode should use its relation extension URI"
    assert plan_has_extension_urn(generated_positional_outer.to_substrait_plan(), POSEXPLODE_OUTER_EXTENSION_URI), "posexplode_outer should use its relation extension URI"
    assert generated_inline.planned_columns() == ["id", "line_items", "sku", "quantity"], "inline should append all declared struct output aliases"
    assert plan_has_extension_urn(generated_inline.to_substrait_plan(), INLINE_EXTENSION_URI), "inline should use its relation extension URI"
    assert plan_has_extension_urn(generated_inline_outer.to_substrait_plan(), INLINE_OUTER_EXTENSION_URI), "inline_outer should use its relation extension URI"
    assert plan_has_extension_urn(generated_flatten.to_substrait_plan(), FLATTEN_EXTENSION_URI), "flatten should use its relation extension URI"
    assert plan_has_extension_urn(generated_stack.to_substrait_plan(), STACK_EXTENSION_URI), "stack should use its relation extension URI"
    assert relation_kind_name(root_rel(windowed.to_substrait_plan())) == "ConsistentPartitionWindowRel", "with_window_column should lower through the window boundary shape"
    assert windowed.planned_columns() == ["id", "row_num"], "window projections should append declared output aliases"
    assert plan_has_extension_urn(windowed.to_substrait_plan(), FUNCTION_EXTENSION_URI), "window plans should use the shared function extension URI"


def test_lazy_frame__deeper_independent_roots_still_lower_with_stable_shapes() -> None:
    # -- Arrange --
    _register_order_schema("orders")
    _register_order_schema("orders_archive")
    left: LazyFrame[Order] = lazy_frame_named_table("orders").filter(always_false())
    right_base: LazyFrame[Order] = lazy_frame_named_table("orders_archive")

    # -- Act --
    right_joined: LazyFrame[Order] = right_base
        .filter(always_false())
        .order_by([col("id")])
        .join(right_base.filter(always_false()).order_by([col("id")]), always_true())
    joined: LazyFrame[Order] = left.join(right_joined, always_true())
    plan = joined.to_substrait_plan()

    # -- Assert --
    assert relation_kind_name(root_rel(plan)) == "JoinRel", "deeper independent-root join compositions should still lower to JoinRel"
    assert plan_contains_relation_kind(plan, "SortRel"), "deeper branch compositions should preserve sort nodes"
    assert plan_contains_relation_kind(plan, "FilterRel"), "deeper branch compositions should preserve filter nodes"
    assert plan_contains_relation_kind(plan, "ReadRel"), "deeper branch compositions should still preserve read roots"


def test_lazy_frame__canonical_rewrite_removes_filter_true_and_preserves_boundary_validity() -> None:
    # -- Arrange --
    simplified: LazyFrame[Order] = lazy_frame_named_table("orders").filter(always_true())

    # -- Act --
    plan = simplified.to_substrait_plan()

    # -- Assert --
    assert plan_encoded_len(plan) > 0, "canonicalized plans should still encode to a non-empty Substrait payload"
    assert relation_kind_name(root_rel(plan)) == "ReadRel", "filter(true) should canonicalize away before RFC 002 lowering"
    assert plan_contains_relation_kind(plan, "ReadRel"), "canonicalized plans should still preserve the read root"