# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
import ctypes
import datetime
import itertools
import os
import re
import threading
import time
from pathlib import Path
from typing import Any

import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
import pytest
from datafusion import (
    DataFrame,
    ExplainFormat,
    InsertOp,
    ParquetColumnOptions,
    ParquetWriterOptions,
    RecordBatch,
    SessionContext,
    WindowFrame,
    column,
    literal,
    udf,
)
from datafusion import (
    col as df_col,
)
from datafusion import (
    functions as f,
)
from datafusion.common import NullTreatment
from datafusion.dataframe import (
    DataFrameWriteOptions,
    ExplainAnalyzeLevel,
    ExplainMetricCategory,
)
from datafusion.dataframe_formatter import (
    DataFrameHtmlFormatter,
    configure_formatter,
    get_formatter,
    reset_formatter,
)
from datafusion.expr import EXPR_TYPE_ERROR, Window
from pyarrow.csv import write_csv

pa_cffi = pytest.importorskip("pyarrow.cffi")

MB = 1024 * 1024


@pytest.fixture
def ctx():
    return SessionContext()


@pytest.fixture
def df(ctx):
    # create a RecordBatch and a new DataFrame from it
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6]), pa.array([8, 5, 8])],
        names=["a", "b", "c"],
    )

    return ctx.from_arrow(batch)


@pytest.fixture
def large_df():
    ctx = SessionContext()

    rows = 100000
    data = {
        "a": list(range(rows)),
        "b": [f"s-{i}" for i in range(rows)],
        "c": [float(i + 0.1) for i in range(rows)],
    }
    batch = pa.record_batch(data)

    return ctx.from_arrow(batch)


@pytest.fixture
def large_multi_batch_df():
    """Create a DataFrame with multiple record batches for testing stream behavior.

    This fixture creates 10 batches of 10,000 rows each (100,000 rows total),
    ensuring the DataFrame spans multiple batches. This is essential for testing
    that memory limits actually cause early stream termination rather than
    truncating all collected data.
    """
    ctx = SessionContext()

    # Create multiple batches, each with 10,000 rows
    batches = []
    rows_per_batch = 10000
    num_batches = 10

    for batch_idx in range(num_batches):
        start_row = batch_idx * rows_per_batch
        end_row = start_row + rows_per_batch
        data = {
            "a": list(range(start_row, end_row)),
            "b": [f"s-{i}" for i in range(start_row, end_row)],
            "c": [float(i + 0.1) for i in range(start_row, end_row)],
        }
        batch = pa.record_batch(data)
        batches.append(batch)

    # Register as record batches to maintain multi-batch structure
    # Using [batches] wraps list in another list as required by register_record_batches
    ctx.register_record_batches("large_multi_batch_data", [batches])
    return ctx.table("large_multi_batch_data")


@pytest.fixture
def struct_df():
    ctx = SessionContext()

    # create a RecordBatch and a new DataFrame from it
    batch = pa.RecordBatch.from_arrays(
        [pa.array([{"c": 1}, {"c": 2}, {"c": 3}]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )

    return ctx.create_dataframe([[batch]])


@pytest.fixture
def nested_df():
    ctx = SessionContext()

    # create a RecordBatch and a new DataFrame from it
    # Intentionally make each array of different length
    batch = pa.RecordBatch.from_arrays(
        [pa.array([[1], [2, 3], [4, 5, 6], None]), pa.array([7, 8, 9, 10])],
        names=["a", "b"],
    )

    return ctx.create_dataframe([[batch]])


@pytest.fixture
def aggregate_df():
    ctx = SessionContext()
    ctx.register_csv("test", "testing/data/csv/aggregate_test_100.csv")
    return ctx.sql("select c1, sum(c2) from test group by c1")


@pytest.fixture
def partitioned_df():
    ctx = SessionContext()

    # create a RecordBatch and a new DataFrame from it
    batch = pa.RecordBatch.from_arrays(
        [
            pa.array([0, 1, 2, 3, 4, 5, 6]),
            pa.array([7, None, 7, 8, 9, None, 9]),
            pa.array(["A", "A", "A", "A", "B", "B", "B"]),
        ],
        names=["a", "b", "c"],
    )

    return ctx.create_dataframe([[batch]])


@pytest.fixture
def clean_formatter_state():
    """Reset the HTML formatter after each test."""
    reset_formatter()


@pytest.fixture
def null_df():
    """Create a DataFrame with null values of different types."""
    ctx = SessionContext()

    # Create a RecordBatch with nulls across different types
    batch = pa.RecordBatch.from_arrays(
        [
            pa.array([1, None, 3, None], type=pa.int64()),
            pa.array([4.5, 6.7, None, None], type=pa.float64()),
            pa.array(["a", None, "c", None], type=pa.string()),
            pa.array([True, None, False, None], type=pa.bool_()),
            pa.array(
                [10957, None, 18993, None], type=pa.date32()
            ),  # 2000-01-01, null, 2022-01-01, null
            pa.array(
                [946684800000, None, 1640995200000, None], type=pa.date64()
            ),  # 2000-01-01, null, 2022-01-01, null
        ],
        names=[
            "int_col",
            "float_col",
            "str_col",
            "bool_col",
            "date32_col",
            "date64_col",
        ],
    )

    return ctx.create_dataframe([[batch]])


# custom style for testing with html formatter
class CustomStyleProvider:
    def get_cell_style(self) -> str:
        return (
            "background-color: #f5f5f5; color: #333; padding: 8px; border: "
            "1px solid #ddd;"
        )

    def get_header_style(self) -> str:
        return (
            "background-color: #4285f4; color: white; font-weight: bold; "
            "padding: 10px; border: 1px solid #3367d6;"
        )


def count_table_rows(html_content: str) -> int:
    """Count the number of table rows in HTML content.
    Args:
        html_content: HTML string to analyze
    Returns:
        Number of table rows found (number of <tr> tags)
    """
    return len(re.findall(r"<tr", html_content))


def test_select(df):
    df_1 = df.select(
        column("a") + column("b"),
        column("a") - column("b"),
    )

    # execute and collect the first (and only) batch
    result = df_1.collect()[0]

    assert result.column(0) == pa.array([5, 7, 9])
    assert result.column(1) == pa.array([-3, -3, -3])

    df_2 = df.select("b", "a")

    # execute and collect the first (and only) batch
    result = df_2.collect()[0]

    assert result.column(0) == pa.array([4, 5, 6])
    assert result.column(1) == pa.array([1, 2, 3])


def test_select_exprs(df):
    df_1 = df.select_exprs(
        "a + b",
        "a - b",
    )

    # execute and collect the first (and only) batch
    result = df_1.collect()[0]

    assert result.column(0) == pa.array([5, 7, 9])
    assert result.column(1) == pa.array([-3, -3, -3])

    df_2 = df.select_exprs("b", "a")

    # execute and collect the first (and only) batch
    result = df_2.collect()[0]

    assert result.column(0) == pa.array([4, 5, 6])
    assert result.column(1) == pa.array([1, 2, 3])

    df_3 = df.select_exprs(
        "abs(a + b)",
        "abs(a - b)",
    )

    # execute and collect the first (and only) batch
    result = df_3.collect()[0]

    assert result.column(0) == pa.array([5, 7, 9])
    assert result.column(1) == pa.array([3, 3, 3])


def test_alias_self_join(df):
    left = df.alias("l")
    right = df.alias("r")
    joined = left.join(right, left_on="a", right_on="a").select(
        "a",
        column("l.b").alias("lb"),
        column("r.b").alias("rb"),
    )
    result = joined.sort(column("a")).collect()[0]
    assert result.column(0) == pa.array([1, 2, 3])
    assert result.column(1) == pa.array([4, 5, 6])
    assert result.column(2) == pa.array([4, 5, 6])


def test_drop_quoted_columns():
    ctx = SessionContext()
    batch = pa.RecordBatch.from_arrays([pa.array([1, 2, 3])], names=["ID_For_Students"])
    df = ctx.create_dataframe([[batch]])
    # here we must quote to match the original column name
    assert df.drop('"ID_For_Students"').schema().names == []

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])], names=["a", "b"]
    )
    df = ctx.create_dataframe([[batch]])
    # with a lower case column, both 'a' and '"a"' work
    assert df.drop("a").schema().names == ["b"]
    df = ctx.create_dataframe([[batch]])
    assert df.drop('"a"').schema().names == ["b"]


def test_select_mixed_expr_string(df):
    df = df.select(column("b"), "a")

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.column(0) == pa.array([4, 5, 6])
    assert result.column(1) == pa.array([1, 2, 3])


def test_select_unsupported(df):
    with pytest.raises(
        TypeError,
        match=f"Expected Expr or column name.*{re.escape(EXPR_TYPE_ERROR)}",
    ):
        df.select(1)


def test_filter(df):
    df1 = df.filter(column("a") > literal(2)).select(
        column("a") + column("b"),
        column("a") - column("b"),
    )

    # execute and collect the first (and only) batch
    result = df1.collect()[0]

    assert result.column(0) == pa.array([9])
    assert result.column(1) == pa.array([-3])

    df.show()
    # verify that if there is no filter applied, internal dataframe is unchanged
    df2 = df.filter()
    assert df.df == df2.df

    df3 = df.filter(column("a") > literal(1), column("b") != literal(6))
    result = df3.collect()[0]

    assert result.column(0) == pa.array([2])
    assert result.column(1) == pa.array([5])
    assert result.column(2) == pa.array([5])


def test_filter_string_predicates(df):
    df_str = df.filter("a > 2")
    result = df_str.collect()[0]

    assert result.column(0) == pa.array([3])
    assert result.column(1) == pa.array([6])
    assert result.column(2) == pa.array([8])

    df_mixed = df.filter("a > 1", column("b") != literal(6))
    result_mixed = df_mixed.collect()[0]

    assert result_mixed.column(0) == pa.array([2])
    assert result_mixed.column(1) == pa.array([5])
    assert result_mixed.column(2) == pa.array([5])

    df_strings = df.filter("a > 1", "b < 6")
    result_strings = df_strings.collect()[0]

    assert result_strings.column(0) == pa.array([2])
    assert result_strings.column(1) == pa.array([5])
    assert result_strings.column(2) == pa.array([5])


def test_parse_sql_expr(df):
    plan1 = df.filter(df.parse_sql_expr("a > 2")).logical_plan()
    plan2 = df.filter(column("a") > literal(2)).logical_plan()
    # object equality not implemented but string representation should match
    assert str(plan1) == str(plan2)

    df1 = df.filter(df.parse_sql_expr("a > 2")).select(
        column("a") + column("b"),
        column("a") - column("b"),
    )

    # execute and collect the first (and only) batch
    result = df1.collect()[0]

    assert result.column(0) == pa.array([9])
    assert result.column(1) == pa.array([-3])

    df.show()
    # verify that if there is no filter applied, internal dataframe is unchanged
    df2 = df.filter()
    assert df.df == df2.df

    df3 = df.filter(df.parse_sql_expr("a > 1"), df.parse_sql_expr("b != 6"))
    result = df3.collect()[0]

    assert result.column(0) == pa.array([2])
    assert result.column(1) == pa.array([5])
    assert result.column(2) == pa.array([5])


def test_show_empty(df, capsys):
    df_empty = df.filter(column("a") > literal(3))
    df_empty.show()
    captured = capsys.readouterr()
    assert "DataFrame has no rows" in captured.out


def test_show_on_explain(ctx, capsys):
    ctx.sql("explain select 1").show()
    captured = capsys.readouterr()
    assert "1 as Int64(1)" in captured.out

    ctx.sql("explain analyze select 1").show()
    captured = capsys.readouterr()
    assert "1 as Int64(1)" in captured.out


def test_sort(df):
    df = df.sort(column("b").sort(ascending=False))

    table = pa.Table.from_batches(df.collect())
    expected = {"a": [3, 2, 1], "b": [6, 5, 4], "c": [8, 5, 8]}

    assert table.to_pydict() == expected


def test_sort_string_and_expression_equivalent(df):
    from datafusion import col

    result_str = df.sort("a").to_pydict()
    result_expr = df.sort(col("a")).to_pydict()
    assert result_str == result_expr


def test_sort_unsupported(df):
    with pytest.raises(
        TypeError,
        match=f"Expected Expr or column name.*{re.escape(EXPR_TYPE_ERROR)}",
    ):
        df.sort(1)


def test_aggregate_string_and_expression_equivalent(df):
    from datafusion import col

    result_str = df.aggregate("a", [f.count()]).sort("a").to_pydict()
    result_expr = df.aggregate(col("a"), [f.count()]).sort("a").to_pydict()
    assert result_str == result_expr


def test_aggregate_tuple_group_by(df):
    result_list = df.aggregate(["a"], [f.count()]).sort("a").to_pydict()
    result_tuple = df.aggregate(("a",), [f.count()]).sort("a").to_pydict()
    assert result_tuple == result_list


def test_aggregate_none_group_by_equivalent_to_empty_list(df):
    result_none = df.aggregate(None, [f.count()]).to_pydict()
    result_empty = df.aggregate([], [f.count()]).to_pydict()
    assert result_none == result_empty


def test_aggregate_tuple_aggs(df):
    result_list = df.aggregate("a", [f.count()]).sort("a").to_pydict()
    result_tuple = df.aggregate("a", (f.count(),)).sort("a").to_pydict()
    assert result_tuple == result_list


def test_filter_string_equivalent(df):
    df1 = df.filter("a > 1").to_pydict()
    df2 = df.filter(column("a") > literal(1)).to_pydict()
    assert df1 == df2


def test_filter_string_invalid(df):
    with pytest.raises(Exception) as excinfo:
        df.filter("this is not valid sql").collect()
    assert "Expected Expr" not in str(excinfo.value)


def test_drop(df):
    df = df.drop("c")

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert df.schema().names == ["a", "b"]
    assert result.column(0) == pa.array([1, 2, 3])
    assert result.column(1) == pa.array([4, 5, 6])


def test_limit(df):
    df = df.limit(1)

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert len(result.column(0)) == 1
    assert len(result.column(1)) == 1


def test_limit_with_offset(df):
    # only 3 rows, but limit past the end to ensure that offset is working
    df = df.limit(5, offset=2)

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert len(result.column(0)) == 1
    assert len(result.column(1)) == 1


def test_head(df):
    df = df.head(1)

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.column(0) == pa.array([1])
    assert result.column(1) == pa.array([4])
    assert result.column(2) == pa.array([8])


def test_tail(df):
    df = df.tail(1)

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.column(0) == pa.array([3])
    assert result.column(1) == pa.array([6])
    assert result.column(2) == pa.array([8])


def test_with_column_sql_expression(df):
    df = df.with_column("c", "a + b")

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.schema.field(0).name == "a"
    assert result.schema.field(1).name == "b"
    assert result.schema.field(2).name == "c"

    assert result.column(0) == pa.array([1, 2, 3])
    assert result.column(1) == pa.array([4, 5, 6])
    assert result.column(2) == pa.array([5, 7, 9])


def test_with_column(df):
    df = df.with_column("c", column("a") + column("b"))

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.schema.field(0).name == "a"
    assert result.schema.field(1).name == "b"
    assert result.schema.field(2).name == "c"

    assert result.column(0) == pa.array([1, 2, 3])
    assert result.column(1) == pa.array([4, 5, 6])
    assert result.column(2) == pa.array([5, 7, 9])


def test_with_columns(df):
    df = df.with_columns(
        (column("a") + column("b")).alias("c"),
        (column("a") + column("b")).alias("d"),
        [
            (column("a") + column("b")).alias("e"),
            (column("a") + column("b")).alias("f"),
        ],
        g=(column("a") + column("b")),
    )

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.schema.field(0).name == "a"
    assert result.schema.field(1).name == "b"
    assert result.schema.field(2).name == "c"
    assert result.schema.field(3).name == "d"
    assert result.schema.field(4).name == "e"
    assert result.schema.field(5).name == "f"
    assert result.schema.field(6).name == "g"

    assert result.column(0) == pa.array([1, 2, 3])
    assert result.column(1) == pa.array([4, 5, 6])
    assert result.column(2) == pa.array([5, 7, 9])
    assert result.column(3) == pa.array([5, 7, 9])
    assert result.column(4) == pa.array([5, 7, 9])
    assert result.column(5) == pa.array([5, 7, 9])
    assert result.column(6) == pa.array([5, 7, 9])


def test_with_columns_str(df):
    df = df.with_columns(
        "a + b as c",
        "a + b as d",
        [
            "a + b as e",
            "a + b as f",
        ],
        g="a + b",
    )

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.schema.field(0).name == "a"
    assert result.schema.field(1).name == "b"
    assert result.schema.field(2).name == "c"
    assert result.schema.field(3).name == "d"
    assert result.schema.field(4).name == "e"
    assert result.schema.field(5).name == "f"
    assert result.schema.field(6).name == "g"

    assert result.column(0) == pa.array([1, 2, 3])
    assert result.column(1) == pa.array([4, 5, 6])
    assert result.column(2) == pa.array([5, 7, 9])
    assert result.column(3) == pa.array([5, 7, 9])
    assert result.column(4) == pa.array([5, 7, 9])
    assert result.column(5) == pa.array([5, 7, 9])
    assert result.column(6) == pa.array([5, 7, 9])


def test_cast(df):
    df = df.cast({"a": pa.float16(), "b": pa.list_(pa.uint32())})
    expected = pa.schema(
        [("a", pa.float16()), ("b", pa.list_(pa.uint32())), ("c", pa.int64())]
    )

    assert df.schema() == expected


def test_iter_batches(df):
    batches = []
    for batch in df:
        batches.append(batch)  # noqa: PERF402

    # Delete DataFrame to ensure RecordBatches remain valid
    del df

    assert len(batches) == 1

    batch = batches[0]
    assert isinstance(batch, RecordBatch)
    pa_batch = batch.to_pyarrow()
    assert pa_batch.column(0).to_pylist() == [1, 2, 3]
    assert pa_batch.column(1).to_pylist() == [4, 5, 6]
    assert pa_batch.column(2).to_pylist() == [8, 5, 8]


def test_iter_returns_datafusion_recordbatch(df):
    for batch in df:
        assert isinstance(batch, RecordBatch)


def test_execute_stream_basic(df):
    stream = df.execute_stream()
    batches = list(stream)

    assert len(batches) == 1
    assert isinstance(batches[0], RecordBatch)
    pa_batch = batches[0].to_pyarrow()
    assert pa_batch.column(0).to_pylist() == [1, 2, 3]
    assert pa_batch.column(1).to_pylist() == [4, 5, 6]
    assert pa_batch.column(2).to_pylist() == [8, 5, 8]


def test_with_column_renamed(df):
    df = df.with_column("c", column("a") + column("b")).with_column_renamed("c", "sum")

    result = df.collect()[0]

    assert result.schema.field(0).name == "a"
    assert result.schema.field(1).name == "b"
    assert result.schema.field(2).name == "sum"


def test_unnest(nested_df):
    nested_df = nested_df.unnest_columns("a")

    # execute and collect the first (and only) batch
    result = nested_df.collect()[0]

    assert result.column(0) == pa.array([1, 2, 3, 4, 5, 6, None])
    assert result.column(1) == pa.array([7, 8, 8, 9, 9, 9, 10])


def test_unnest_without_nulls(nested_df):
    nested_df = nested_df.unnest_columns("a", preserve_nulls=False)

    # execute and collect the first (and only) batch
    result = nested_df.collect()[0]

    assert result.column(0) == pa.array([1, 2, 3, 4, 5, 6])
    assert result.column(1) == pa.array([7, 8, 8, 9, 9, 9])


def test_join():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df = ctx.create_dataframe([[batch]], "l")

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2]), pa.array([8, 10])],
        names=["a", "c"],
    )
    df1 = ctx.create_dataframe([[batch]], "r")

    df2 = df.join(df1, on="a", how="inner")
    df2 = df2.sort(column("a"))
    table = pa.Table.from_batches(df2.collect())

    expected = {"a": [1, 2], "c": [8, 10], "b": [4, 5]}
    assert table.to_pydict() == expected

    # Test the default behavior for dropping duplicate keys
    # Since we may have a duplicate column name and pa.Table()
    # hides the fact, instead we need to explicitly check the
    # resultant arrays.
    df2 = df.join(
        df1, left_on="a", right_on="a", how="inner", coalesce_duplicate_keys=True
    )
    df2 = df2.sort(column("a"))
    result = df2.collect()[0]
    assert result.num_columns == 3
    assert result.column(0) == pa.array([1, 2], pa.int64())
    assert result.column(1) == pa.array([4, 5], pa.int64())
    assert result.column(2) == pa.array([8, 10], pa.int64())

    df2 = df.join(
        df1, left_on="a", right_on="a", how="inner", coalesce_duplicate_keys=False
    )
    df2 = df2.sort(column("l.a"))
    result = df2.collect()[0]
    assert result.num_columns == 4
    assert result.column(0) == pa.array([1, 2], pa.int64())
    assert result.column(1) == pa.array([4, 5], pa.int64())
    assert result.column(2) == pa.array([1, 2], pa.int64())
    assert result.column(3) == pa.array([8, 10], pa.int64())

    # Verify we don't make a breaking change to pre-43.0.0
    # where users would pass join_keys as a positional argument
    df2 = df.join(df1, (["a"], ["a"]), how="inner")
    df2 = df2.sort(column("a"))
    table = pa.Table.from_batches(df2.collect())

    expected = {"a": [1, 2], "c": [8, 10], "b": [4, 5]}
    assert table.to_pydict() == expected


def test_join_invalid_params():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df = ctx.create_dataframe([[batch]], "l")

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2]), pa.array([8, 10])],
        names=["a", "c"],
    )
    df1 = ctx.create_dataframe([[batch]], "r")

    with pytest.deprecated_call():
        df2 = df.join(df1, join_keys=(["a"], ["a"]), how="inner")
        df2.show()
        df2 = df2.sort(column("a"))
        table = pa.Table.from_batches(df2.collect())

        expected = {"a": [1, 2], "c": [8, 10], "b": [4, 5]}
        assert table.to_pydict() == expected

    with pytest.raises(
        ValueError, match=r"`left_on` or `right_on` should not provided with `on`"
    ):
        df2 = df.join(df1, on="a", how="inner", right_on="test")

    with pytest.raises(
        ValueError, match=r"`left_on` and `right_on` should both be provided."
    ):
        df2 = df.join(df1, left_on="a", how="inner")

    with pytest.raises(
        ValueError, match=r"either `on` or `left_on` and `right_on` should be provided."
    ):
        df2 = df.join(df1, how="inner")


def test_join_on():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df = ctx.create_dataframe([[batch]], "l")

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2]), pa.array([-8, 10])],
        names=["a", "c"],
    )
    df1 = ctx.create_dataframe([[batch]], "r")

    df2 = df.join_on(df1, column("l.a").__eq__(column("r.a")), how="inner")
    df2.show()
    df2 = df2.sort(column("l.a"))
    table = pa.Table.from_batches(df2.collect())

    expected = {"a": [1, 2], "c": [-8, 10], "b": [4, 5]}
    assert table.to_pydict() == expected

    df3 = df.join_on(
        df1,
        column("l.a").__eq__(column("r.a")),
        column("l.a").__lt__(column("r.c")),
        how="inner",
    )
    df3.show()
    df3 = df3.sort(column("l.a"))
    table = pa.Table.from_batches(df3.collect())
    expected = {"a": [2], "c": [10], "b": [5]}
    assert table.to_pydict() == expected


def test_join_full_with_drop_duplicate_keys():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 3, 5, 7, 9]), pa.array([True, True, True, True, True])],
        names=["log_time", "key_frame"],
    )
    key_frame = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([2, 4, 6, 8, 10])],
        names=["log_time"],
    )
    query_times = ctx.create_dataframe([[batch]])

    merged = query_times.join(
        key_frame,
        left_on="log_time",
        right_on="log_time",
        how="full",
        coalesce_duplicate_keys=True,
    )
    merged = merged.sort(column("log_time"))
    result = merged.collect()[0]

    assert result.num_columns == 2
    assert result.column(0).to_pylist() == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]


def test_join_on_invalid_expr():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2]), pa.array([4, 5])],
        names=["a", "b"],
    )
    df = ctx.create_dataframe([[batch]], "l")
    df1 = ctx.create_dataframe([[batch]], "r")

    with pytest.raises(
        TypeError, match=r"Use col\(\)/column\(\) or lit\(\)/literal\(\)"
    ):
        df.join_on(df1, "a")


def test_aggregate_invalid_aggs(df):
    with pytest.raises(
        TypeError, match=r"Use col\(\)/column\(\) or lit\(\)/literal\(\)"
    ):
        df.aggregate([], "a")


def test_distinct():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3, 1, 2, 3]), pa.array([4, 5, 6, 4, 5, 6])],
        names=["a", "b"],
    )
    df_a = ctx.create_dataframe([[batch]]).distinct().sort(column("a"))

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df_b = ctx.create_dataframe([[batch]]).sort(column("a"))

    assert df_a.collect() == df_b.collect()


data_test_window_functions = [
    (
        "row",
        f.row_number(
            order_by=[
                f.order_by(column("b"), nulls_first=True),
                column("a").sort(ascending=False),
            ]
        ),
        [4, 2, 3, 5, 7, 1, 6],
    ),
    (
        "row_w_params",
        f.row_number(
            order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
            partition_by=[column("c")],
        ),
        [2, 1, 3, 4, 2, 1, 3],
    ),
    (
        "row_w_params_no_lists",
        f.row_number(
            order_by=f.order_by(column("b"), nulls_first=True),
            partition_by=column("c"),
        ),
        [2, 1, 3, 4, 2, 1, 3],
    ),
    (
        "rank",
        f.rank(order_by=[f.order_by(column("b"), nulls_first=True)]),
        [3, 1, 3, 5, 6, 1, 6],
    ),
    (
        "rank_w_params",
        f.rank(
            order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
            partition_by=[column("c")],
        ),
        [2, 1, 3, 4, 2, 1, 3],
    ),
    (
        "rank_w_params_no_lists",
        f.rank(order_by=column("a"), partition_by=column("c")),
        [1, 2, 3, 4, 1, 2, 3],
    ),
    (
        "dense_rank",
        f.dense_rank(order_by=[f.order_by(column("b"), nulls_first=True)]),
        [2, 1, 2, 3, 4, 1, 4],
    ),
    (
        "dense_rank_w_params",
        f.dense_rank(
            order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
            partition_by=[column("c")],
        ),
        [2, 1, 3, 4, 2, 1, 3],
    ),
    (
        "dense_rank_w_params_no_lists",
        f.dense_rank(order_by=column("a"), partition_by=column("c")),
        [1, 2, 3, 4, 1, 2, 3],
    ),
    (
        "percent_rank",
        f.round(
            f.percent_rank(order_by=[f.order_by(column("b"), nulls_first=True)]),
            literal(3),
        ),
        [0.333, 0.0, 0.333, 0.667, 0.833, 0.0, 0.833],
    ),
    (
        "percent_rank_w_params",
        f.round(
            f.percent_rank(
                order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
                partition_by=[column("c")],
            ),
            literal(3),
        ),
        [0.333, 0.0, 0.667, 1.0, 0.5, 0.0, 1.0],
    ),
    (
        "percent_rank_w_params_no_lists",
        f.round(
            f.percent_rank(order_by=column("a"), partition_by=column("c")),
            literal(3),
        ),
        [0.0, 0.333, 0.667, 1.0, 0.0, 0.5, 1.0],
    ),
    (
        "cume_dist",
        f.round(
            f.cume_dist(order_by=[f.order_by(column("b"), nulls_first=True)]),
            literal(3),
        ),
        [0.571, 0.286, 0.571, 0.714, 1.0, 0.286, 1.0],
    ),
    (
        "cume_dist_w_params",
        f.round(
            f.cume_dist(
                order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
                partition_by=[column("c")],
            ),
            literal(3),
        ),
        [0.5, 0.25, 0.75, 1.0, 0.667, 0.333, 1.0],
    ),
    (
        "cume_dist_w_params_no_lists",
        f.round(
            f.cume_dist(order_by=column("a"), partition_by=column("c")),
            literal(3),
        ),
        [0.25, 0.5, 0.75, 1.0, 0.333, 0.667, 1.0],
    ),
    (
        "ntile",
        f.ntile(2, order_by=[f.order_by(column("b"), nulls_first=True)]),
        [1, 1, 1, 2, 2, 1, 2],
    ),
    (
        "ntile_w_params",
        f.ntile(
            2,
            order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
            partition_by=[column("c")],
        ),
        [1, 1, 2, 2, 1, 1, 2],
    ),
    (
        "ntile_w_params_no_lists",
        f.ntile(
            2,
            order_by=f.order_by(column("b"), nulls_first=True),
            partition_by=column("c"),
        ),
        [1, 1, 2, 2, 1, 1, 2],
    ),
    (
        "lead",
        f.lead(column("b"), order_by=[f.order_by(column("b"), nulls_first=True)]),
        [7, None, 8, 9, 9, 7, None],
    ),
    (
        "lead_w_params",
        f.lead(
            column("b"),
            shift_offset=2,
            default_value=-1,
            order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
            partition_by=[column("c")],
        ),
        [8, 7, -1, -1, -1, 9, -1],
    ),
    (
        "lead_w_params_no_lists",
        f.lead(
            column("b"),
            shift_offset=2,
            default_value=-1,
            order_by=f.order_by(column("b"), nulls_first=True),
            partition_by=column("c"),
        ),
        [8, 7, -1, -1, -1, 9, -1],
    ),
    (
        "lag",
        f.lag(column("b"), order_by=[f.order_by(column("b"), nulls_first=True)]),
        [None, None, 7, 7, 8, None, 9],
    ),
    (
        "lag_w_params",
        f.lag(
            column("b"),
            shift_offset=2,
            default_value=-1,
            order_by=[f.order_by(column("b"), nulls_first=True), column("a")],
            partition_by=[column("c")],
        ),
        [-1, -1, None, 7, -1, -1, None],
    ),
    (
        "lag_w_params_no_lists",
        f.lag(
            column("b"),
            shift_offset=2,
            default_value=-1,
            order_by=f.order_by(column("b"), nulls_first=True),
            partition_by=column("c"),
        ),
        [-1, -1, None, 7, -1, -1, None],
    ),
    (
        "lead_ignore_nulls",
        f.lead(
            column("b"),
            order_by=column("a"),
            partition_by=column("c"),
            null_treatment=NullTreatment.IGNORE_NULLS,
        ),
        [7, 7, 8, None, 9, 9, None],
    ),
    (
        "lag_ignore_nulls",
        f.lag(
            column("b"),
            order_by=column("a"),
            partition_by=column("c"),
            null_treatment=NullTreatment.IGNORE_NULLS,
        ),
        [None, 7, 7, 7, None, 9, 9],
    ),
    (
        "first_value",
        f.first_value(column("a")).over(
            Window(
                partition_by=[column("c")],
                order_by=[f.order_by(column("b"), nulls_first=True)],
            )
        ),
        [1, 1, 1, 1, 5, 5, 5],
    ),
    (
        "first_value_without_list_args",
        f.first_value(column("a")).over(
            Window(
                partition_by=column("c"),
                order_by=f.order_by(column("b"), nulls_first=True),
            )
        ),
        [1, 1, 1, 1, 5, 5, 5],
    ),
    (
        "first_value_order_by_string",
        f.first_value(column("a")).over(
            Window(
                partition_by=[column("c")],
                order_by=f.order_by(column("b"), nulls_first=True),
            )
        ),
        [1, 1, 1, 1, 5, 5, 5],
    ),
    (
        "last_value",
        f.last_value(column("a")).over(
            Window(
                partition_by=[column("c")],
                order_by=[f.order_by(column("b"), nulls_first=True)],
                window_frame=WindowFrame("rows", None, None),
            )
        ),
        [3, 3, 3, 3, 6, 6, 6],
    ),
    (
        "3rd_value",
        f.nth_value(column("b"), 3).over(
            Window(order_by=[f.order_by(column("a"), nulls_first=True)])
        ),
        [None, None, 7, 7, 7, 7, 7],
    ),
    (
        "avg",
        f.round(f.avg(column("b")).over(Window(order_by=[column("a")])), literal(3)),
        [7.0, 7.0, 7.0, 7.333, 7.75, 7.75, 8.0],
    ),
]


@pytest.mark.parametrize(("name", "expr", "result"), data_test_window_functions)
def test_window_functions(partitioned_df, name, expr, result):
    df = partitioned_df.select(
        column("a"), column("b"), column("c"), f.alias(expr, name)
    )
    df.sort(column("a")).show()
    table = pa.Table.from_batches(df.collect())

    expected = {
        "a": [0, 1, 2, 3, 4, 5, 6],
        "b": [7, None, 7, 8, 9, None, 9],
        "c": ["A", "A", "A", "A", "B", "B", "B"],
        name: result,
    }

    assert table.sort_by("a").to_pydict() == expected


@pytest.mark.parametrize("partition", ["c", df_col("c")])
def test_rank_partition_by_accepts_string(partitioned_df, partition):
    """Passing a string to partition_by should match using col()."""
    df = partitioned_df.select(
        f.rank(order_by=column("a"), partition_by=partition).alias("r")
    )
    table = pa.Table.from_batches(df.sort(column("a")).collect())
    assert table.column("r").to_pylist() == [1, 2, 3, 4, 1, 2, 3]


@pytest.mark.parametrize("partition", ["c", df_col("c")])
def test_window_partition_by_accepts_string(partitioned_df, partition):
    """Window.partition_by accepts string identifiers."""
    expr = f.first_value(column("a")).over(
        Window(
            partition_by=partition, order_by=f.order_by(column("b"), nulls_first=True)
        )
    )
    df = partitioned_df.select(expr.alias("fv"))
    table = pa.Table.from_batches(df.sort(column("a")).collect())
    assert table.column("fv").to_pylist() == [1, 1, 1, 1, 5, 5, 5]


@pytest.mark.parametrize("func", [f.lead, f.lag])
def test_lead_lag_default_null_treatment_keeps_column_name(partitioned_df, func):
    """Omitting null_treatment must not add RESPECT NULLS to the output name."""
    df = partitioned_df.select(func(column("b"), order_by=column("a")))
    name = df.schema().names[0]
    assert "RESPECT NULLS" not in name


@pytest.mark.parametrize(
    ("units", "start_bound", "end_bound"),
    [
        (units, start_bound, end_bound)
        for units in ("rows", "range")
        for start_bound in (None, 0, 1)
        for end_bound in (None, 0, 1)
    ]
    + [
        ("groups", 0, 0),
    ],
)
def test_valid_window_frame(units, start_bound, end_bound):
    WindowFrame(units, start_bound, end_bound)


@pytest.mark.parametrize(
    ("units", "start_bound", "end_bound"),
    [
        ("invalid-units", 0, None),
        ("invalid-units", None, 0),
        ("invalid-units", None, None),
        ("groups", None, 0),
        ("groups", 0, None),
        ("groups", None, None),
    ],
)
def test_invalid_window_frame(units, start_bound, end_bound):
    with pytest.raises(NotImplementedError, match=f"(?i){units}"):
        WindowFrame(units, start_bound, end_bound)


def test_window_frame_defaults_match_postgres(partitioned_df):
    col_a = column("a")

    # When order is not set, the default frame should be unbounded preceding to
    # unbounded following. When order is set, the default frame is unbounded preceding
    # to current row.
    no_order = f.avg(col_a).over(Window()).alias("over_no_order")
    with_order = f.avg(col_a).over(Window(order_by=[col_a])).alias("over_with_order")
    df = partitioned_df.select(col_a, no_order, with_order)

    expected = {
        "a": [0, 1, 2, 3, 4, 5, 6],
        "over_no_order": [3.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0],
        "over_with_order": [0.0, 0.5, 1.0, 1.5, 2.0, 2.5, 3.0],
    }

    assert df.sort(col_a).to_pydict() == expected


def _build_last_value_df(df):
    return df.select(
        f.last_value(column("a"))
        .over(
            Window(
                partition_by=[column("c")],
                order_by=[column("b")],
                window_frame=WindowFrame("rows", None, None),
            )
        )
        .alias("expr"),
        f.last_value(column("a"))
        .over(
            Window(
                partition_by=[column("c")],
                order_by="b",
                window_frame=WindowFrame("rows", None, None),
            )
        )
        .alias("str"),
    )


def _build_nth_value_df(df):
    return df.select(
        f.nth_value(column("b"), 3).over(Window(order_by=[column("a")])).alias("expr"),
        f.nth_value(column("b"), 3).over(Window(order_by="a")).alias("str"),
    )


def _build_rank_df(df):
    return df.select(
        f.rank(order_by=[column("b")]).alias("expr"),
        f.rank(order_by="b").alias("str"),
    )


def _build_array_agg_df(df):
    return df.aggregate(
        [column("c")],
        [
            f.array_agg(column("a"), order_by=[column("a")]).alias("expr"),
            f.array_agg(column("a"), order_by="a").alias("str"),
        ],
    ).sort(column("c"))


@pytest.mark.parametrize(
    ("builder", "expected"),
    [
        pytest.param(_build_last_value_df, [1, 1, 1, 1, 5, 5, 5], id="last_value"),
        pytest.param(_build_nth_value_df, [None, None, 7, 7, 7, 7, 7], id="nth_value"),
        pytest.param(_build_rank_df, [1, 1, 3, 4, 4, 6, 6], id="rank"),
        pytest.param(_build_array_agg_df, [[0, 1, 2, 3], [4, 5, 6]], id="array_agg"),
    ],
)
def test_order_by_string_equivalence(partitioned_df, builder, expected):
    df = builder(partitioned_df)
    table = pa.Table.from_batches(df.collect())
    assert table.column("expr").to_pylist() == expected
    assert table.column("expr").to_pylist() == table.column("str").to_pylist()


def test_html_formatter_cell_dimension(df, clean_formatter_state):
    """Test configuring the HTML formatter with different options."""
    # Configure with custom settings
    configure_formatter(
        max_width=500,
        max_height=200,
        enable_cell_expansion=False,
    )

    html_output = df._repr_html_()

    # Verify our configuration was applied
    assert "max-height: 200px" in html_output
    assert "max-width: 500px" in html_output
    # With cell expansion disabled, we shouldn't see expandable-container elements
    assert "expandable-container" not in html_output


def test_html_formatter_custom_style_provider(df, clean_formatter_state):
    """Test using custom style providers with the HTML formatter."""

    # Configure with custom style provider
    configure_formatter(style_provider=CustomStyleProvider())

    html_output = df._repr_html_()

    # Verify our custom styles were applied
    assert "background-color: #4285f4" in html_output
    assert "color: white" in html_output
    assert "background-color: #f5f5f5" in html_output


def test_html_formatter_type_formatters(df, clean_formatter_state):
    """Test registering custom type formatters for specific data types."""

    # Get current formatter and register custom formatters
    formatter = get_formatter()

    # Format integers with color based on value
    # Using int as the type for the formatter will work since we convert
    # Arrow scalar values to Python native types in _get_cell_value
    def format_int(value):
        return f'<span style="color: {"red" if value > 2 else "blue"}">{value}</span>'

    formatter.register_formatter(int, format_int)

    html_output = df._repr_html_()

    # Our test dataframe has values 1,2,3 so we should see:
    assert '<span style="color: blue">1</span>' in html_output


def test_html_formatter_custom_cell_builder(df, clean_formatter_state):
    """Test using a custom cell builder function."""

    # Create a custom cell builder with distinct styling for different value ranges
    def custom_cell_builder(value, row, col, table_id):
        try:
            num_value = int(value)
            if num_value > 5:  # Values > 5 get green background with indicator
                return (
                    '<td style="background-color: #d9f0d3" '
                    f'data-test="high">{value}-high</td>'
                )
            if num_value < 3:  # Values < 3 get blue background with indicator
                return (
                    '<td style="background-color: #d3e9f0" '
                    f'data-test="low">{value}-low</td>'
                )
        except (ValueError, TypeError):
            pass

        # Default styling for other cells (3, 4, 5)
        return f'<td style="border: 1px solid #ddd" data-test="mid">{value}-mid</td>'

    # Set our custom cell builder
    formatter = get_formatter()
    formatter.set_custom_cell_builder(custom_cell_builder)

    html_output = df._repr_html_()

    # Extract cells with specific styling using regex
    low_cells = re.findall(
        r'<td style="background-color: #d3e9f0"[^>]*>(\d+)-low</td>', html_output
    )
    mid_cells = re.findall(
        r'<td style="border: 1px solid #ddd"[^>]*>(\d+)-mid</td>', html_output
    )
    high_cells = re.findall(
        r'<td style="background-color: #d9f0d3"[^>]*>(\d+)-high</td>', html_output
    )

    # Sort the extracted values for consistent comparison
    low_cells = sorted(map(int, low_cells))
    mid_cells = sorted(map(int, mid_cells))
    high_cells = sorted(map(int, high_cells))

    # Verify specific values have the correct styling applied
    assert low_cells == [1, 2]  # Values < 3
    assert mid_cells == [3, 4, 5, 5]  # Values 3-5
    assert high_cells == [6, 8, 8]  # Values > 5

    # Verify the exact content with styling appears in the output
    assert (
        '<td style="background-color: #d3e9f0" data-test="low">1-low</td>'
        in html_output
    )
    assert (
        '<td style="background-color: #d3e9f0" data-test="low">2-low</td>'
        in html_output
    )
    assert (
        '<td style="border: 1px solid #ddd" data-test="mid">3-mid</td>' in html_output
    )
    assert (
        '<td style="border: 1px solid #ddd" data-test="mid">4-mid</td>' in html_output
    )
    assert (
        '<td style="background-color: #d9f0d3" data-test="high">6-high</td>'
        in html_output
    )
    assert (
        '<td style="background-color: #d9f0d3" data-test="high">8-high</td>'
        in html_output
    )

    # Count occurrences to ensure all cells are properly styled
    assert html_output.count("-low</td>") == 2  # Two low values (1, 2)
    assert html_output.count("-mid</td>") == 4  # Four mid values (3, 4, 5, 5)
    assert html_output.count("-high</td>") == 3  # Three high values (6, 8, 8)

    # Create a custom cell builder that changes background color based on value
    def custom_cell_builder(value, row, col, table_id):
        # Handle numeric values regardless of their exact type
        try:
            num_value = int(value)
            if num_value > 5:  # Values > 5 get green background
                return f'<td style="background-color: #d9f0d3">{value}</td>'
            if num_value < 3:  # Values < 3 get light blue background
                return f'<td style="background-color: #d3e9f0">{value}</td>'
        except (ValueError, TypeError):
            pass

        # Default styling for other cells
        return f'<td style="border: 1px solid #ddd">{value}</td>'

    # Set our custom cell builder
    formatter = get_formatter()
    formatter.set_custom_cell_builder(custom_cell_builder)

    html_output = df._repr_html_()

    # Verify our custom cell styling was applied
    assert "background-color: #d3e9f0" in html_output  # For values 1,2


def test_html_formatter_custom_header_builder(df, clean_formatter_state):
    """Test using a custom header builder function."""

    # Create a custom header builder with tooltips
    def custom_header_builder(field):
        tooltips = {
            "a": "Primary key column",
            "b": "Secondary values",
            "c": "Additional data",
        }
        tooltip = tooltips.get(field.name, "")
        return (
            f'<th style="background-color: #333; color: white" '
            f'title="{tooltip}">{field.name}</th>'
        )

    # Set our custom header builder
    formatter = get_formatter()
    formatter.set_custom_header_builder(custom_header_builder)

    html_output = df._repr_html_()

    # Verify our custom headers were applied
    assert 'title="Primary key column"' in html_output
    assert 'title="Secondary values"' in html_output
    assert "background-color: #333; color: white" in html_output


def test_html_formatter_complex_customization(df, clean_formatter_state):
    """Test combining multiple customization options together."""

    # Create a dark mode style provider
    class DarkModeStyleProvider:
        def get_cell_style(self) -> str:
            return (
                "background-color: #222; color: #eee; "
                "padding: 8px; border: 1px solid #444;"
            )

        def get_header_style(self) -> str:
            return (
                "background-color: #111; color: #fff; padding: 10px; "
                "border: 1px solid #333;"
            )

    # Configure with dark mode style
    configure_formatter(
        max_cell_length=10,
        style_provider=DarkModeStyleProvider(),
        custom_css="""
            .datafusion-table {
                font-family: monospace;
                border-collapse: collapse;
            }
            .datafusion-table tr:hover td {
                background-color: #444 !important;
            }
        """,
    )

    # Add type formatters for special formatting - now working with native int values
    formatter = get_formatter()
    formatter.register_formatter(
        int,
        lambda n: f'<span style="color: {"#5af" if n % 2 == 0 else "#f5a"}">{n}</span>',
    )

    html_output = df._repr_html_()

    # Verify our customizations were applied
    assert "background-color: #222" in html_output
    assert "background-color: #111" in html_output
    assert ".datafusion-table" in html_output
    assert "color: #5af" in html_output  # Even numbers


def test_html_formatter_memory(df, clean_formatter_state):
    """Test the memory and row control parameters in DataFrameHtmlFormatter."""
    configure_formatter(max_memory_bytes=10, min_rows=1)
    html_output = df._repr_html_()

    # Count the number of table rows in the output
    tr_count = count_table_rows(html_output)
    # With a tiny memory limit of 10 bytes, the formatter should display
    # the minimum number of rows (1) plus a message about truncation
    assert tr_count == 2  # 1 for header row, 1 for data row
    assert "data truncated" in html_output.lower()

    configure_formatter(max_memory_bytes=10 * MB, min_rows=1)
    html_output = df._repr_html_()
    # With larger memory limit and min_rows=2, should display all rows
    tr_count = count_table_rows(html_output)
    # Table should have header row (1) + 3 data rows = 4 rows
    assert tr_count == 4
    # No truncation message should appear
    assert "data truncated" not in html_output.lower()


def test_html_formatter_memory_boundary_conditions(large_df, clean_formatter_state):
    """Test memory limit behavior at boundary conditions with large dataset.

    This test validates that the formatter correctly handles edge cases when
    the memory limit is reached with a large dataset (100,000 rows), ensuring
    that min_rows constraint is properly respected while respecting memory limits.
    Uses large_df to actually test memory limit behavior with realistic data sizes.
    """

    # Get the raw size of the data to test boundary conditions
    # First, capture output with no limits
    # NOTE: max_rows=200000 is set well above the dataset size (100k rows) to ensure
    # we're testing memory limits, not row limits. Default max_rows=10 would
    # truncate before memory limit is reached.
    configure_formatter(max_memory_bytes=10 * MB, min_rows=1, max_rows=200000)
    unrestricted_output = large_df._repr_html_()
    unrestricted_rows = count_table_rows(unrestricted_output)

    # Test 1: Very small memory limit should still respect min_rows
    # With large dataset, this should definitely hit memory limit before min_rows
    configure_formatter(max_memory_bytes=10, min_rows=1)
    html_output = large_df._repr_html_()
    tr_count = count_table_rows(html_output)
    assert tr_count >= 2  # At least header + 1 data row (minimum)
    # Should show truncation since we limited memory so aggressively
    assert "data truncated" in html_output.lower()

    # Test 2: Memory limit at default size (2MB) should truncate the large dataset
    # Default max_rows would truncate at 10 rows, so we don't set it here to test
    # that memory limit is respected even with default row limit
    configure_formatter(max_memory_bytes=2 * MB, min_rows=1)
    html_output = large_df._repr_html_()
    tr_count = count_table_rows(html_output)
    assert tr_count >= 2  # At least header + min_rows
    # Should be truncated since full dataset is much larger than 2MB
    assert tr_count < unrestricted_rows

    # Test 3: Very large memory limit should show much more data
    # NOTE: max_rows=200000 is critical here - without it, default max_rows=10
    # would limit output to 10 rows even though we have 100MB of memory available
    configure_formatter(max_memory_bytes=100 * MB, min_rows=1, max_rows=200000)
    html_output = large_df._repr_html_()
    tr_count = count_table_rows(html_output)
    # Should show significantly more rows, possibly all
    assert tr_count > 100  # Should show substantially more rows

    # Test 4: Min rows should override memory limit
    # With tiny memory and larger min_rows, min_rows should win
    configure_formatter(max_memory_bytes=10, min_rows=2)
    html_output = large_df._repr_html_()
    tr_count = count_table_rows(html_output)
    assert tr_count >= 3  # At least header + 2 data rows (min_rows)
    # Should show truncation message despite min_rows being satisfied
    assert "data truncated" in html_output.lower()

    # Test 5: With reasonable memory and min_rows settings
    # NOTE: max_rows=200000 ensures we test memory limit behavior, not row limit
    configure_formatter(max_memory_bytes=2 * MB, min_rows=10, max_rows=200000)
    html_output = large_df._repr_html_()
    tr_count = count_table_rows(html_output)
    assert tr_count >= 11  # header + at least 10 data rows (min_rows)
    # Should be truncated due to memory limit
    assert tr_count < unrestricted_rows


def test_html_formatter_stream_early_termination(
    large_multi_batch_df, clean_formatter_state
):
    """Test that memory limits cause early stream termination with multi-batch data.

    This test specifically validates that the formatter stops collecting data when
    the memory limit is reached, rather than collecting all data and then truncating.
    The large_multi_batch_df fixture creates 10 record batches, allowing us to verify
    that not all batches are consumed when memory limit is hit.

    Key difference from test_html_formatter_memory_boundary_conditions:
    - Uses multi-batch DataFrame to verify stream termination behavior
    - Tests with memory limit exceeded by 2-3 batches but not 1 batch
    - Verifies partial data + truncation message + respects min_rows
    """

    # Get baseline: how much data fits without memory limit
    configure_formatter(max_memory_bytes=100 * MB, min_rows=1, max_rows=200000)
    unrestricted_output = large_multi_batch_df._repr_html_()
    unrestricted_rows = count_table_rows(unrestricted_output)

    # Test 1: Memory limit exceeded by ~2 batches (each batch ~10k rows)
    # With 1 batch (~1-2MB), we should have space. With 2-3 batches, we exceed limit.
    # Set limit to ~3MB to ensure we collect ~1 batch before hitting limit
    configure_formatter(max_memory_bytes=3 * MB, min_rows=1, max_rows=200000)
    html_output = large_multi_batch_df._repr_html_()
    tr_count = count_table_rows(html_output)

    # Should show significant truncation (not all 100k rows)
    assert tr_count < unrestricted_rows, "Should be truncated by memory limit"
    assert tr_count >= 2, "Should respect min_rows"
    assert "data truncated" in html_output.lower(), "Should indicate truncation"

    # Test 2: Very tight memory limit should still respect min_rows
    # Even with tiny memory (10 bytes), should show at least min_rows
    configure_formatter(max_memory_bytes=10, min_rows=5, max_rows=200000)
    html_output = large_multi_batch_df._repr_html_()
    tr_count = count_table_rows(html_output)

    assert tr_count >= 6, "Should show header + at least min_rows (5)"
    assert "data truncated" in html_output.lower(), "Should indicate truncation"

    # Test 3: Memory limit should take precedence over max_rows in early termination
    # With max_rows=100 but small memory limit, should terminate early due to memory
    configure_formatter(max_memory_bytes=2 * MB, min_rows=1, max_rows=100)
    html_output = large_multi_batch_df._repr_html_()
    tr_count = count_table_rows(html_output)

    # Should be truncated by memory limit (showing more than max_rows would suggest
    # but less than unrestricted)
    assert tr_count >= 2, "Should respect min_rows"
    assert tr_count < unrestricted_rows, "Should be truncated"
    # Output should indicate why truncation occurred
    assert "data truncated" in html_output.lower()


def test_html_formatter_max_rows(df, clean_formatter_state):
    configure_formatter(min_rows=2, max_rows=2)
    html_output = df._repr_html_()

    tr_count = count_table_rows(html_output)
    # Table should have header row (1) + 2 data rows = 3 rows
    assert tr_count == 3

    configure_formatter(min_rows=2, max_rows=3)
    html_output = df._repr_html_()

    tr_count = count_table_rows(html_output)
    # Table should have header row (1) + 3 data rows = 4 rows
    assert tr_count == 4


def test_html_formatter_validation():
    # Test validation for invalid parameters

    with pytest.raises(ValueError, match="max_cell_length must be a positive integer"):
        DataFrameHtmlFormatter(max_cell_length=0)

    with pytest.raises(ValueError, match="max_width must be a positive integer"):
        DataFrameHtmlFormatter(max_width=0)

    with pytest.raises(ValueError, match="max_height must be a positive integer"):
        DataFrameHtmlFormatter(max_height=0)

    with pytest.raises(ValueError, match="max_memory_bytes must be a positive integer"):
        DataFrameHtmlFormatter(max_memory_bytes=0)

    with pytest.raises(ValueError, match="max_memory_bytes must be a positive integer"):
        DataFrameHtmlFormatter(max_memory_bytes=-100)

    with pytest.raises(ValueError, match="min_rows must be a positive integer"):
        DataFrameHtmlFormatter(min_rows=0)

    with pytest.raises(ValueError, match="min_rows must be a positive integer"):
        DataFrameHtmlFormatter(min_rows=-5)

    with pytest.raises(ValueError, match="max_rows must be a positive integer"):
        DataFrameHtmlFormatter(max_rows=0)

    with pytest.raises(ValueError, match="max_rows must be a positive integer"):
        DataFrameHtmlFormatter(max_rows=-10)

    with pytest.raises(
        ValueError, match="min_rows must be less than or equal to max_rows"
    ):
        DataFrameHtmlFormatter(min_rows=5, max_rows=4)


def test_repr_rows_backward_compatibility(clean_formatter_state):
    """Test that repr_rows parameter still works as deprecated alias."""
    # Should work when not conflicting with max_rows
    with pytest.warns(DeprecationWarning, match="repr_rows parameter is deprecated"):
        formatter = DataFrameHtmlFormatter(repr_rows=15, min_rows=10)
    assert formatter.max_rows == 15
    assert formatter.repr_rows == 15

    # Should fail when conflicting with max_rows. The deprecation warning still
    # fires before the ValueError, so assert both.
    with (
        pytest.raises(ValueError, match="Cannot specify both repr_rows and max_rows"),
        pytest.warns(DeprecationWarning, match="repr_rows parameter is deprecated"),
    ):
        DataFrameHtmlFormatter(repr_rows=5, max_rows=10)

    # Setting repr_rows via property should warn
    formatter2 = DataFrameHtmlFormatter()
    with pytest.warns(DeprecationWarning, match="repr_rows is deprecated"):
        formatter2.repr_rows = 7
    assert formatter2.max_rows == 7
    assert formatter2.repr_rows == 7


def test_configure_formatter(df, clean_formatter_state):
    """Test using custom style providers with the HTML formatter and configured
    parameters."""

    # these are non-default values
    max_cell_length = 10
    max_width = 500
    max_height = 30
    max_memory_bytes = 3 * MB
    min_rows = 2
    max_rows = 2
    enable_cell_expansion = False
    show_truncation_message = False
    use_shared_styles = False

    reset_formatter()
    formatter_default = get_formatter()

    assert formatter_default.max_cell_length != max_cell_length
    assert formatter_default.max_width != max_width
    assert formatter_default.max_height != max_height
    assert formatter_default.max_memory_bytes != max_memory_bytes
    assert formatter_default.min_rows != min_rows
    assert formatter_default.max_rows != max_rows
    assert formatter_default.enable_cell_expansion != enable_cell_expansion
    assert formatter_default.show_truncation_message != show_truncation_message
    assert formatter_default.use_shared_styles != use_shared_styles

    # Configure with custom style provider and additional parameters
    configure_formatter(
        max_cell_length=max_cell_length,
        max_width=max_width,
        max_height=max_height,
        max_memory_bytes=max_memory_bytes,
        min_rows=min_rows,
        max_rows=max_rows,
        enable_cell_expansion=enable_cell_expansion,
        show_truncation_message=show_truncation_message,
        use_shared_styles=use_shared_styles,
    )
    formatter_custom = get_formatter()
    assert formatter_custom.max_cell_length == max_cell_length
    assert formatter_custom.max_width == max_width
    assert formatter_custom.max_height == max_height
    assert formatter_custom.max_memory_bytes == max_memory_bytes
    assert formatter_custom.min_rows == min_rows
    assert formatter_custom.max_rows == max_rows
    assert formatter_custom.enable_cell_expansion == enable_cell_expansion
    assert formatter_custom.show_truncation_message == show_truncation_message
    assert formatter_custom.use_shared_styles == use_shared_styles


def test_configure_formatter_invalid_params(clean_formatter_state):
    """Test that configure_formatter rejects invalid parameters."""
    with pytest.raises(ValueError, match="Invalid formatter parameters"):
        configure_formatter(invalid_param=123)

    # Test with multiple parameters, one valid and one invalid
    with pytest.raises(ValueError, match="Invalid formatter parameters"):
        configure_formatter(max_width=500, not_a_real_param="test")

    # Test with multiple invalid parameters
    with pytest.raises(ValueError, match="Invalid formatter parameters"):
        configure_formatter(fake_param1="test", fake_param2=456)


def test_get_dataframe(tmp_path):
    ctx = SessionContext()

    path = tmp_path / "test.csv"
    table = pa.Table.from_arrays(
        [
            [1, 2, 3, 4],
            ["a", "b", "c", "d"],
            [1.1, 2.2, 3.3, 4.4],
        ],
        names=["int", "str", "float"],
    )
    write_csv(table, path)

    ctx.register_csv("csv", path)

    df = ctx.table("csv")
    assert isinstance(df, DataFrame)


def test_struct_select(struct_df):
    df = struct_df.select(
        column("a")["c"] + column("b"),
        column("a")["c"] - column("b"),
    )

    # execute and collect the first (and only) batch
    result = df.collect()[0]

    assert result.column(0) == pa.array([5, 7, 9])
    assert result.column(1) == pa.array([-3, -3, -3])


def test_explain(df):
    df = df.select(
        column("a") + column("b"),
        column("a") - column("b"),
    )
    df.explain()


def test_logical_plan(aggregate_df):
    plan = aggregate_df.logical_plan()

    expected = "Projection: test.c1, sum(test.c2)"

    assert expected == plan.display()

    expected = (
        "Projection: test.c1, sum(test.c2)\n"
        "  Aggregate: groupBy=[[test.c1]], aggr=[[sum(test.c2)]]\n"
        "    TableScan: test"
    )

    assert expected == plan.display_indent()


def test_optimized_logical_plan(aggregate_df):
    plan = aggregate_df.optimized_logical_plan()

    expected = "Aggregate: groupBy=[[test.c1]], aggr=[[sum(test.c2)]]"

    assert expected == plan.display()

    expected = (
        "Aggregate: groupBy=[[test.c1]], aggr=[[sum(test.c2)]]\n"
        "  TableScan: test projection=[c1, c2]"
    )

    assert expected == plan.display_indent()


def test_execution_plan(aggregate_df):
    plan = aggregate_df.execution_plan()

    expected = (
        "AggregateExec: mode=FinalPartitioned, gby=[c1@0 as c1], aggr=[sum(test.c2)]\n"
    )

    assert expected == plan.display()

    # Check the number of partitions is as expected.
    assert isinstance(plan.partition_count, int)

    expected = (
        "ProjectionExec: expr=[c1@0 as c1, SUM(test.c2)@1 as SUM(test.c2)]\n"
        "  Aggregate: groupBy=[[test.c1]], aggr=[[SUM(test.c2)]]\n"
        "    TableScan: test projection=[c1, c2]"
    )

    indent = plan.display_indent()

    # indent plan will be different for everyone due to absolute path
    # to filename, so we just check for some expected content
    assert "AggregateExec:" in indent
    assert "RepartitionExec:" in indent
    assert "DataSourceExec:" in indent
    assert "file_type=csv" in indent

    ctx = SessionContext()
    rows_returned = 0
    for idx in range(plan.partition_count):
        stream = ctx.execute(plan, idx)
        try:
            batch = stream.next()
            assert batch is not None
            rows_returned += len(batch.to_pyarrow()[0])
        except StopIteration:
            # This is one of the partitions with no values
            pass
        with pytest.raises(StopIteration):
            stream.next()

    assert rows_returned == 5


@pytest.mark.asyncio
async def test_async_iteration_of_df(aggregate_df):
    rows_returned = 0
    async for batch in aggregate_df:
        assert batch is not None
        rows_returned += len(batch.to_pyarrow()[0])

    assert rows_returned == 5


def test_repartition(df):
    df.repartition(2)


def test_repartition_by_hash(df):
    df.repartition_by_hash(column("a"), num=2)


def test_repartition_by_hash_sql_expression(df):
    df.repartition_by_hash("a", num=2)


def test_repartition_by_hash_mix(df):
    df.repartition_by_hash(column("a"), "b", num=2)


def test_intersect():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df_a = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([3, 4, 5]), pa.array([6, 7, 8])],
        names=["a", "b"],
    )
    df_b = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([3]), pa.array([6])],
        names=["a", "b"],
    )
    df_c = ctx.create_dataframe([[batch]]).sort(column("a"))

    df_a_i_b = df_a.intersect(df_b).sort(column("a"))

    assert df_c.collect() == df_a_i_b.collect()


def test_except_all():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df_a = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([3, 4, 5]), pa.array([6, 7, 8])],
        names=["a", "b"],
    )
    df_b = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2]), pa.array([4, 5])],
        names=["a", "b"],
    )
    df_c = ctx.create_dataframe([[batch]]).sort(column("a"))

    df_a_e_b = df_a.except_all(df_b).sort(column("a"))

    assert df_c.collect() == df_a_e_b.collect()


def test_collect_partitioned():
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )

    assert [[batch]] == ctx.create_dataframe([[batch]]).collect_partitioned()


def test_collect_column(ctx: SessionContext):
    batch_1 = pa.RecordBatch.from_pydict({"a": [1, 2, 3]})
    batch_2 = pa.RecordBatch.from_pydict({"a": [4, 5, 6]})
    batch_3 = pa.RecordBatch.from_pydict({"a": [7, 8, 9]})

    ctx.register_record_batches("t", [[batch_1, batch_2], [batch_3]])

    result = ctx.table("t").sort(column("a")).collect_column("a")
    expected = pa.array([1, 2, 3, 4, 5, 6, 7, 8, 9])
    assert result == expected


def test_union(ctx):
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df_a = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([3, 4, 5]), pa.array([6, 7, 8])],
        names=["a", "b"],
    )
    df_b = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3, 3, 4, 5]), pa.array([4, 5, 6, 6, 7, 8])],
        names=["a", "b"],
    )
    df_c = ctx.create_dataframe([[batch]]).sort(column("a"))

    df_a_u_b = df_a.union(df_b).sort(column("a"))

    assert df_c.collect() == df_a_u_b.collect()


def test_union_distinct(ctx):
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )
    df_a = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([3, 4, 5]), pa.array([6, 7, 8])],
        names=["a", "b"],
    )
    df_b = ctx.create_dataframe([[batch]])

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3, 4, 5]), pa.array([4, 5, 6, 7, 8])],
        names=["a", "b"],
    )
    df_c = ctx.create_dataframe([[batch]]).sort(column("a"))

    df_a_u_b = df_a.union(df_b, distinct=True).sort(column("a"))

    assert df_c.collect() == df_a_u_b.collect()
    assert df_c.collect() == df_a_u_b.collect()


def test_cache(df):
    assert df.cache().collect() == df.collect()


def test_count(df):
    # Get number of rows
    assert df.count() == 3


def test_to_pandas(df):
    # Skip test if pandas is not installed
    pd = pytest.importorskip("pandas")

    # Convert datafusion dataframe to pandas dataframe
    pandas_df = df.to_pandas()
    assert isinstance(pandas_df, pd.DataFrame)
    assert pandas_df.shape == (3, 3)
    assert set(pandas_df.columns) == {"a", "b", "c"}


def test_empty_to_pandas(df):
    # Skip test if pandas is not installed
    pd = pytest.importorskip("pandas")

    # Convert empty datafusion dataframe to pandas dataframe
    pandas_df = df.limit(0).to_pandas()
    assert isinstance(pandas_df, pd.DataFrame)
    assert pandas_df.shape == (0, 3)
    assert set(pandas_df.columns) == {"a", "b", "c"}


def test_to_polars(df):
    # Skip test if polars is not installed
    pl = pytest.importorskip("polars")

    # Convert datafusion dataframe to polars dataframe
    polars_df = df.to_polars()
    assert isinstance(polars_df, pl.DataFrame)
    assert polars_df.shape == (3, 3)
    assert set(polars_df.columns) == {"a", "b", "c"}


def test_empty_to_polars(df):
    # Skip test if polars is not installed
    pl = pytest.importorskip("polars")

    # Convert empty datafusion dataframe to polars dataframe
    polars_df = df.limit(0).to_polars()
    assert isinstance(polars_df, pl.DataFrame)
    assert polars_df.shape == (0, 3)
    assert set(polars_df.columns) == {"a", "b", "c"}


def test_to_arrow_table(df):
    # Convert datafusion dataframe to pyarrow Table
    pyarrow_table = df.to_arrow_table()
    assert isinstance(pyarrow_table, pa.Table)
    assert pyarrow_table.shape == (3, 3)
    assert set(pyarrow_table.column_names) == {"a", "b", "c"}


def test_parquet_non_null_column_to_pyarrow(ctx, tmp_path):
    path = tmp_path.joinpath("t.parquet")

    ctx.sql("create table t_(a int not null)").collect()
    ctx.sql("insert into t_ values (1), (2), (3)").collect()
    ctx.sql(f"copy (select * from t_) to '{path}'").collect()

    ctx.register_parquet("t", path)
    pyarrow_table = ctx.sql("select max(a) as m from t").to_arrow_table()
    assert pyarrow_table.to_pydict() == {"m": [3]}


def test_parquet_empty_batch_to_pyarrow(ctx, tmp_path):
    path = tmp_path.joinpath("t.parquet")

    ctx.sql("create table t_(a int not null)").collect()
    ctx.sql("insert into t_ values (1), (2), (3)").collect()
    ctx.sql(f"copy (select * from t_) to '{path}'").collect()

    ctx.register_parquet("t", path)
    pyarrow_table = ctx.sql("select * from t limit 0").to_arrow_table()
    assert pyarrow_table.schema == pa.schema(
        [
            pa.field("a", pa.int32(), nullable=False),
        ]
    )


def test_parquet_null_aggregation_to_pyarrow(ctx, tmp_path):
    path = tmp_path.joinpath("t.parquet")

    ctx.sql("create table t_(a int not null)").collect()
    ctx.sql("insert into t_ values (1), (2), (3)").collect()
    ctx.sql(f"copy (select * from t_) to '{path}'").collect()

    ctx.register_parquet("t", path)
    pyarrow_table = ctx.sql(
        "select max(a) as m from (select * from t where a < 0)"
    ).to_arrow_table()
    assert pyarrow_table.to_pydict() == {"m": [None]}
    assert pyarrow_table.schema == pa.schema(
        [
            pa.field("m", pa.int32(), nullable=True),
        ]
    )


def test_execute_stream(df):
    stream = df.execute_stream()
    assert all(batch is not None for batch in stream)
    assert not list(stream)  # after one iteration the generator must be exhausted


@pytest.mark.asyncio
async def test_execute_stream_async(df):
    stream = df.execute_stream()
    batches = [batch async for batch in stream]

    assert all(batch is not None for batch in batches)

    # After consuming all batches, the stream should be exhausted
    remaining_batches = [batch async for batch in stream]
    assert not remaining_batches


@pytest.mark.parametrize("schema", [True, False])
def test_execute_stream_to_arrow_table(df, schema):
    stream = df.execute_stream()

    if schema:
        pyarrow_table = pa.Table.from_batches(
            (batch.to_pyarrow() for batch in stream), schema=df.schema()
        )
    else:
        pyarrow_table = pa.Table.from_batches(batch.to_pyarrow() for batch in stream)

    assert isinstance(pyarrow_table, pa.Table)
    assert pyarrow_table.shape == (3, 3)
    assert set(pyarrow_table.column_names) == {"a", "b", "c"}


@pytest.mark.asyncio
@pytest.mark.parametrize("schema", [True, False])
async def test_execute_stream_to_arrow_table_async(df, schema):
    stream = df.execute_stream()

    if schema:
        pyarrow_table = pa.Table.from_batches(
            [batch.to_pyarrow() async for batch in stream], schema=df.schema()
        )
    else:
        pyarrow_table = pa.Table.from_batches(
            [batch.to_pyarrow() async for batch in stream]
        )

    assert isinstance(pyarrow_table, pa.Table)
    assert pyarrow_table.shape == (3, 3)
    assert set(pyarrow_table.column_names) == {"a", "b", "c"}


def test_execute_stream_partitioned(df):
    streams = df.execute_stream_partitioned()
    assert all(batch is not None for stream in streams for batch in stream)
    assert all(
        not list(stream) for stream in streams
    )  # after one iteration all generators must be exhausted


@pytest.mark.asyncio
async def test_execute_stream_partitioned_async(df):
    streams = df.execute_stream_partitioned()

    for stream in streams:
        batches = [batch async for batch in stream]
        assert all(batch is not None for batch in batches)

        # Ensure the stream is exhausted after iteration
        remaining_batches = [batch async for batch in stream]
        assert not remaining_batches


def test_empty_to_arrow_table(df):
    # Convert empty datafusion dataframe to pyarrow Table
    pyarrow_table = df.limit(0).to_arrow_table()
    assert isinstance(pyarrow_table, pa.Table)
    assert pyarrow_table.shape == (0, 3)
    assert set(pyarrow_table.column_names) == {"a", "b", "c"}


def test_iter_batches_dataframe(fail_collect):
    ctx = SessionContext()

    batch1 = pa.record_batch([pa.array([1])], names=["a"])
    batch2 = pa.record_batch([pa.array([2])], names=["a"])
    df = ctx.create_dataframe([[batch1], [batch2]])

    expected = [batch1, batch2]
    results = [b.to_pyarrow() for b in df]

    assert len(results) == len(expected)
    for exp in expected:
        assert any(got.equals(exp) for got in results)


def test_arrow_c_stream_to_table_and_reader(fail_collect):
    ctx = SessionContext()

    # Create a DataFrame with two separate record batches
    batch1 = pa.record_batch([pa.array([1])], names=["a"])
    batch2 = pa.record_batch([pa.array([2])], names=["a"])
    df = ctx.create_dataframe([[batch1], [batch2]])

    table = pa.Table.from_batches(batch.to_pyarrow() for batch in df)
    batches = table.to_batches()

    assert len(batches) == 2
    expected = [batch1, batch2]
    for exp in expected:
        assert any(got.equals(exp) for got in batches)
    assert table.schema == df.schema()
    assert table.column("a").num_chunks == 2

    reader = pa.RecordBatchReader.from_stream(df)
    assert isinstance(reader, pa.RecordBatchReader)
    reader_table = pa.Table.from_batches(reader)
    expected = pa.Table.from_batches([batch1, batch2])
    assert reader_table.equals(expected)


def test_arrow_c_stream_order():
    ctx = SessionContext()

    batch1 = pa.record_batch([pa.array([1])], names=["a"])
    batch2 = pa.record_batch([pa.array([2])], names=["a"])

    df = ctx.create_dataframe([[batch1, batch2]])

    table = pa.Table.from_batches(batch.to_pyarrow() for batch in df)
    expected = pa.Table.from_batches([batch1, batch2])

    assert table.equals(expected)
    col = table.column("a")
    assert col.chunk(0)[0].as_py() == 1
    assert col.chunk(1)[0].as_py() == 2


def test_arrow_c_stream_schema_selection(fail_collect):
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [
            pa.array([1, 2]),
            pa.array([3, 4]),
            pa.array([5, 6]),
        ],
        names=["a", "b", "c"],
    )
    df = ctx.create_dataframe([[batch]])

    requested_schema = pa.schema([("c", pa.int64()), ("a", pa.int64())])

    c_schema = pa_cffi.ffi.new("struct ArrowSchema*")
    address = int(pa_cffi.ffi.cast("uintptr_t", c_schema))
    requested_schema._export_to_c(address)
    capsule_new = ctypes.pythonapi.PyCapsule_New
    capsule_new.restype = ctypes.py_object
    capsule_new.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_void_p]

    reader = pa.RecordBatchReader.from_stream(df, schema=requested_schema)

    assert reader.schema == requested_schema

    batches = list(reader)

    assert len(batches) == 1
    expected_batch = pa.record_batch(
        [pa.array([5, 6]), pa.array([1, 2])], names=["c", "a"]
    )
    assert batches[0].equals(expected_batch)


def test_arrow_c_stream_schema_mismatch(fail_collect):
    ctx = SessionContext()

    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2]), pa.array([3, 4])], names=["a", "b"]
    )
    df = ctx.create_dataframe([[batch]])

    bad_schema = pa.schema([("a", pa.string())])

    c_schema = pa_cffi.ffi.new("struct ArrowSchema*")
    address = int(pa_cffi.ffi.cast("uintptr_t", c_schema))
    bad_schema._export_to_c(address)

    capsule_new = ctypes.pythonapi.PyCapsule_New
    capsule_new.restype = ctypes.py_object
    capsule_new.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_void_p]
    bad_capsule = capsule_new(ctypes.c_void_p(address), b"arrow_schema", None)

    with pytest.raises(Exception, match="Fail to merge schema"):
        df.__arrow_c_stream__(bad_capsule)


def test_to_pylist(df):
    # Convert datafusion dataframe to Python list
    pylist = df.to_pylist()
    assert isinstance(pylist, list)
    assert pylist == [
        {"a": 1, "b": 4, "c": 8},
        {"a": 2, "b": 5, "c": 5},
        {"a": 3, "b": 6, "c": 8},
    ]


def test_to_pydict(df):
    # Convert datafusion dataframe to Python dictionary
    pydict = df.to_pydict()
    assert isinstance(pydict, dict)
    assert pydict == {"a": [1, 2, 3], "b": [4, 5, 6], "c": [8, 5, 8]}


def test_describe(df):
    # Calculate statistics
    df = df.describe()

    # Collect the result
    result = df.to_pydict()

    assert result == {
        "describe": [
            "count",
            "null_count",
            "mean",
            "std",
            "min",
            "max",
            "median",
        ],
        "a": [3.0, 0.0, 2.0, 1.0, 1.0, 3.0, 2.0],
        "b": [3.0, 0.0, 5.0, 1.0, 4.0, 6.0, 5.0],
        "c": [3.0, 0.0, 7.0, 1.7320508075688772, 5.0, 8.0, 8.0],
    }


@pytest.mark.parametrize("path_to_str", [True, False])
def test_write_csv(ctx, df, tmp_path, path_to_str):
    path = str(tmp_path) if path_to_str else tmp_path

    df.write_csv(path, with_header=True)

    ctx.register_csv("csv", path)
    result = ctx.table("csv").to_pydict()
    expected = df.to_pydict()

    assert result == expected


def generate_test_write_params() -> list[tuple]:
    # Overwrite and Replace are not implemented for many table writers
    insert_ops = [InsertOp.APPEND, None]
    sort_by_cases = [
        (None, [1, 2, 3], "unsorted"),
        (column("c"), [2, 1, 3], "single_column_expr"),
        (column("a").sort(ascending=False), [3, 2, 1], "single_sort_expr"),
        ([column("c"), column("b")], [2, 1, 3], "list_col_expr"),
        (
            [column("c").sort(ascending=False), column("b").sort(ascending=False)],
            [3, 1, 2],
            "list_sort_expr",
        ),
    ]

    formats = ["csv", "json", "parquet", "table"]

    return [
        pytest.param(
            output_format,
            insert_op,
            sort_by,
            expected_a,
            id=f"{output_format}_{test_id}",
        )
        for output_format, insert_op, (
            sort_by,
            expected_a,
            test_id,
        ) in itertools.product(formats, insert_ops, sort_by_cases)
    ]


@pytest.mark.parametrize(
    ("output_format", "insert_op", "sort_by", "expected_a"),
    generate_test_write_params(),
)
def test_write_files_with_options(
    ctx, df, tmp_path, output_format, insert_op, sort_by, expected_a
) -> None:
    write_options = DataFrameWriteOptions(insert_operation=insert_op, sort_by=sort_by)

    if output_format == "csv":
        df.write_csv(tmp_path, with_header=True, write_options=write_options)
        ctx.register_csv("test_table", tmp_path)
    elif output_format == "json":
        df.write_json(tmp_path, write_options=write_options)
        ctx.register_json("test_table", tmp_path)
    elif output_format == "parquet":
        df.write_parquet(tmp_path, write_options=write_options)
        ctx.register_parquet("test_table", tmp_path)
    elif output_format == "table":
        batch = pa.RecordBatch.from_arrays([[], [], []], schema=df.schema())
        ctx.register_record_batches("test_table", [[batch]])
        ctx.table("test_table").show()
        df.write_table("test_table", write_options=write_options)

    result = ctx.table("test_table").to_pydict()["a"]
    ctx.table("test_table").show()

    assert result == expected_a


@pytest.mark.parametrize("path_to_str", [True, False])
def test_write_json(ctx, df, tmp_path, path_to_str):
    path = str(tmp_path) if path_to_str else tmp_path

    df.write_json(path)

    ctx.register_json("json", path)
    result = ctx.table("json").to_pydict()
    expected = df.to_pydict()

    assert result == expected


@pytest.mark.parametrize("path_to_str", [True, False])
def test_write_parquet(df, tmp_path, path_to_str):
    path = str(tmp_path) if path_to_str else tmp_path

    df.write_parquet(str(path))
    result = pq.read_table(str(path)).to_pydict()
    expected = df.to_pydict()

    assert result == expected


def test_write_parquet_writer_options_keeps_write_options(ctx, tmp_path):
    """``write_parquet`` honours ``write_options`` alongside ``ParquetWriterOptions``.

    The ``ParquetWriterOptions`` branch delegates to
    :py:meth:`DataFrame.write_parquet_with_options`, which takes ``write_options``
    too, so ``partition_by`` must still reach the writer.
    """
    df = ctx.from_pydict({"part": ["a", "a", "b"], "v": [1, 2, 3]})
    path = tmp_path / "partitioned"

    df.write_parquet(
        path,
        ParquetWriterOptions(),
        write_options=DataFrameWriteOptions(partition_by="part"),
    )

    assert sorted(p.name for p in path.iterdir()) == ["part=a", "part=b"]


@pytest.mark.parametrize(
    ("compression", "compression_level"),
    [("gzip", 6), ("brotli", 7), ("zstd", 15)],
)
def test_write_compressed_parquet(df, tmp_path, compression, compression_level):
    path = tmp_path

    df.write_parquet(
        str(path), compression=compression, compression_level=compression_level
    )

    # test that the actual compression scheme is the one written
    for _root, _dirs, files in os.walk(path):
        for file in files:
            if file.endswith(".parquet"):
                metadata = pq.ParquetFile(tmp_path / file).metadata.to_dict()
                for row_group in metadata["row_groups"]:
                    for columns in row_group["columns"]:
                        assert columns["compression"].lower() == compression

    result = pq.read_table(str(path)).to_pydict()
    expected = df.to_pydict()

    assert result == expected


@pytest.mark.parametrize(
    ("compression", "compression_level"),
    [("gzip", 12), ("brotli", 15), ("zstd", 23), ("wrong", 12)],
)
def test_write_compressed_parquet_wrong_compression_level(
    df, tmp_path, compression, compression_level
):
    path = tmp_path

    with pytest.raises(ValueError):
        df.write_parquet(
            str(path),
            compression=compression,
            compression_level=compression_level,
        )


@pytest.mark.parametrize("compression", ["wrong"])
def test_write_compressed_parquet_invalid_compression(df, tmp_path, compression):
    path = tmp_path

    with pytest.raises(ValueError):
        df.write_parquet(str(path), compression=compression)


# not testing lzo because it it not implemented yet
# https://github.com/apache/arrow-rs/issues/6970
@pytest.mark.parametrize("compression", ["zstd", "brotli", "gzip"])
def test_write_compressed_parquet_default_compression_level(df, tmp_path, compression):
    # Test write_parquet with zstd, brotli, gzip default compression level,
    # ie don't specify compression level
    # should complete without error
    path = tmp_path

    df.write_parquet(str(path), compression=compression)


def test_write_parquet_with_options_default_compression(df, tmp_path):
    """Test that the default compression is ZSTD."""
    df.write_parquet(tmp_path)

    for file in tmp_path.rglob("*.parquet"):
        metadata = pq.ParquetFile(file).metadata.to_dict()
        for row_group in metadata["row_groups"]:
            for col in row_group["columns"]:
                assert col["compression"].lower() == "zstd"


@pytest.mark.parametrize(
    "compression",
    ["gzip(6)", "brotli(7)", "zstd(15)", "snappy", "uncompressed"],
)
def test_write_parquet_with_options_compression(df, tmp_path, compression):
    import re

    path = tmp_path
    df.write_parquet_with_options(
        str(path), ParquetWriterOptions(compression=compression)
    )

    # test that the actual compression scheme is the one written
    for _root, _dirs, files in os.walk(path):
        for file in files:
            if file.endswith(".parquet"):
                metadata = pq.ParquetFile(tmp_path / file).metadata.to_dict()
                for row_group in metadata["row_groups"]:
                    for col in row_group["columns"]:
                        assert col["compression"].lower() == re.sub(
                            r"\(\d+\)", "", compression
                        )

    result = pq.read_table(str(path)).to_pydict()
    expected = df.to_pydict()

    assert result == expected


@pytest.mark.parametrize(
    "compression",
    ["gzip(12)", "brotli(15)", "zstd(23)"],
)
def test_write_parquet_with_options_wrong_compression_level(df, tmp_path, compression):
    path = tmp_path

    with pytest.raises(Exception, match=r"valid compression range .*? exceeded."):
        df.write_parquet_with_options(
            str(path), ParquetWriterOptions(compression=compression)
        )


@pytest.mark.parametrize("compression", ["wrong", "wrong(12)"])
def test_write_parquet_with_options_invalid_compression(df, tmp_path, compression):
    path = tmp_path

    with pytest.raises(Exception, match="Unknown or unsupported parquet compression"):
        df.write_parquet_with_options(
            str(path), ParquetWriterOptions(compression=compression)
        )


@pytest.mark.parametrize(
    ("writer_version", "format_version"),
    [("1.0", "1.0"), ("2.0", "2.6"), (None, "1.0")],
)
def test_write_parquet_with_options_writer_version(
    df, tmp_path, writer_version, format_version
):
    """Test the Parquet writer version. Note that writer_version=2.0 results in
    format_version=2.6"""
    if writer_version is None:
        df.write_parquet_with_options(tmp_path, ParquetWriterOptions())
    else:
        df.write_parquet_with_options(
            tmp_path, ParquetWriterOptions(writer_version=writer_version)
        )

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()
        assert metadata["format_version"] == format_version


@pytest.mark.parametrize("writer_version", ["1.2.3", "custom-version", "0"])
def test_write_parquet_with_options_wrong_writer_version(df, tmp_path, writer_version):
    """Test that invalid writer versions in Parquet throw an exception."""
    with pytest.raises(Exception, match="Invalid parquet writer version"):
        df.write_parquet_with_options(
            tmp_path, ParquetWriterOptions(writer_version=writer_version)
        )


@pytest.mark.parametrize("dictionary_enabled", [True, False, None])
def test_write_parquet_with_options_dictionary_enabled(
    df, tmp_path, dictionary_enabled
):
    """Test enabling/disabling the dictionaries in Parquet."""
    df.write_parquet_with_options(
        tmp_path, ParquetWriterOptions(dictionary_enabled=dictionary_enabled)
    )
    # by default, the dictionary is enabled, so None results in True
    result = dictionary_enabled if dictionary_enabled is not None else True

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()

        for row_group in metadata["row_groups"]:
            for col in row_group["columns"]:
                assert col["has_dictionary_page"] == result


@pytest.mark.parametrize(
    ("statistics_enabled", "has_statistics"),
    [("page", True), ("chunk", True), ("none", False), (None, True)],
)
def test_write_parquet_with_options_statistics_enabled(
    df, tmp_path, statistics_enabled, has_statistics
):
    """Test configuring the statistics in Parquet. In pyarrow we can only check for
    column-level statistics, so "page" and "chunk" are tested in the same way."""
    df.write_parquet_with_options(
        tmp_path, ParquetWriterOptions(statistics_enabled=statistics_enabled)
    )

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()

        for row_group in metadata["row_groups"]:
            for col in row_group["columns"]:
                if has_statistics:
                    assert col["statistics"] is not None
                else:
                    assert col["statistics"] is None


@pytest.mark.parametrize("max_row_group_size", [1000, 5000, 10000, 100000])
def test_write_parquet_with_options_max_row_group_size(
    large_df, tmp_path, max_row_group_size
):
    """Test configuring the max number of rows per group in Parquet. These test cases
    guarantee that the number of rows for each row group is max_row_group_size, given
    the total number of rows is a multiple of max_row_group_size."""
    path = f"{tmp_path}/t.parquet"
    large_df.write_parquet_with_options(
        path, ParquetWriterOptions(max_row_group_size=max_row_group_size)
    )

    parquet = pq.ParquetFile(path)
    metadata = parquet.metadata.to_dict()
    for row_group in metadata["row_groups"]:
        assert row_group["num_rows"] == max_row_group_size


@pytest.mark.parametrize("created_by", ["datafusion", "datafusion-python", "custom"])
def test_write_parquet_with_options_created_by(df, tmp_path, created_by):
    """Test configuring the created by metadata in Parquet."""
    df.write_parquet_with_options(tmp_path, ParquetWriterOptions(created_by=created_by))

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()
        assert metadata["created_by"] == created_by


@pytest.mark.parametrize("statistics_truncate_length", [5, 25, 50])
def test_write_parquet_with_options_statistics_truncate_length(
    df, tmp_path, statistics_truncate_length
):
    """Test configuring the truncate limit in Parquet's row-group-level statistics."""
    ctx = SessionContext()
    data = {
        "a": [
            "a_the_quick_brown_fox_jumps_over_the_lazy_dog",
            "m_the_quick_brown_fox_jumps_over_the_lazy_dog",
            "z_the_quick_brown_fox_jumps_over_the_lazy_dog",
        ],
        "b": ["a_smaller", "m_smaller", "z_smaller"],
    }
    df = ctx.from_arrow(pa.record_batch(data))
    df.write_parquet_with_options(
        tmp_path,
        ParquetWriterOptions(statistics_truncate_length=statistics_truncate_length),
    )

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()

        for row_group in metadata["row_groups"]:
            for col in row_group["columns"]:
                statistics = col["statistics"]
                assert len(statistics["min"]) <= statistics_truncate_length
                assert len(statistics["max"]) <= statistics_truncate_length


def test_write_parquet_with_options_default_encoding(tmp_path):
    """Test that, by default, Parquet files are written with dictionary encoding.
    Note that dictionary encoding is not used for boolean values, so it is not tested
    here."""
    ctx = SessionContext()
    data = {
        "a": [1, 2, 3],
        "b": ["1", "2", "3"],
        "c": [1.01, 2.02, 3.03],
    }
    df = ctx.from_arrow(pa.record_batch(data))
    df.write_parquet_with_options(tmp_path, ParquetWriterOptions())

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()

        for row_group in metadata["row_groups"]:
            for col in row_group["columns"]:
                assert col["encodings"] == ("PLAIN", "RLE", "RLE_DICTIONARY")


@pytest.mark.parametrize(
    ("encoding", "data_types", "result"),
    [
        ("plain", ["int", "float", "str", "bool"], ("PLAIN", "RLE")),
        ("rle", ["bool"], ("RLE",)),
        ("delta_binary_packed", ["int"], ("RLE", "DELTA_BINARY_PACKED")),
        ("delta_length_byte_array", ["str"], ("RLE", "DELTA_LENGTH_BYTE_ARRAY")),
        ("delta_byte_array", ["str"], ("RLE", "DELTA_BYTE_ARRAY")),
        ("byte_stream_split", ["int", "float"], ("RLE", "BYTE_STREAM_SPLIT")),
    ],
)
def test_write_parquet_with_options_encoding(tmp_path, encoding, data_types, result):
    """Test different encodings in Parquet in their respective support column types."""
    ctx = SessionContext()

    data = {}
    for data_type in data_types:
        if data_type == "int":
            data["int"] = [1, 2, 3]
        elif data_type == "float":
            data["float"] = [1.01, 2.02, 3.03]
        elif data_type == "str":
            data["str"] = ["a", "b", "c"]
        elif data_type == "bool":
            data["bool"] = [True, False, True]

    df = ctx.from_arrow(pa.record_batch(data))
    df.write_parquet_with_options(
        tmp_path, ParquetWriterOptions(encoding=encoding, dictionary_enabled=False)
    )

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()

        for row_group in metadata["row_groups"]:
            for col in row_group["columns"]:
                assert col["encodings"] == result


@pytest.mark.parametrize("encoding", ["bit_packed"])
def test_write_parquet_with_options_unsupported_encoding(df, tmp_path, encoding):
    """Test that unsupported Parquet encodings do not work."""
    # BaseException is used since this throws a Rust panic: https://github.com/PyO3/pyo3/issues/3519
    with pytest.raises(BaseException, match=r"Encoding .*? is not supported"):
        df.write_parquet_with_options(tmp_path, ParquetWriterOptions(encoding=encoding))


@pytest.mark.parametrize("encoding", ["non_existent", "unknown", "plain123"])
def test_write_parquet_with_options_invalid_encoding(df, tmp_path, encoding):
    """Test that invalid Parquet encodings do not work."""
    with pytest.raises(Exception, match="Unknown or unsupported parquet encoding"):
        df.write_parquet_with_options(tmp_path, ParquetWriterOptions(encoding=encoding))


@pytest.mark.parametrize("encoding", ["plain_dictionary", "rle_dictionary"])
def test_write_parquet_with_options_dictionary_encoding_fallback(
    df, tmp_path, encoding
):
    """Test that the dictionary encoding cannot be used as fallback in Parquet."""
    # BaseException is used since this throws a Rust panic: https://github.com/PyO3/pyo3/issues/3519
    with pytest.raises(
        BaseException, match="Dictionary encoding can not be used as fallback encoding"
    ):
        df.write_parquet_with_options(tmp_path, ParquetWriterOptions(encoding=encoding))


def test_write_parquet_with_options_bloom_filter(df, tmp_path):
    """Test Parquet files with and without (default) bloom filters. Since pyarrow does
    not expose any information about bloom filters, the easiest way to confirm that they
    are actually written is to compare the file size."""
    path_no_bloom_filter = tmp_path / "1"
    path_bloom_filter = tmp_path / "2"

    df.write_parquet_with_options(path_no_bloom_filter, ParquetWriterOptions())
    df.write_parquet_with_options(
        path_bloom_filter, ParquetWriterOptions(bloom_filter_on_write=True)
    )

    size_no_bloom_filter = 0
    for file in path_no_bloom_filter.rglob("*.parquet"):
        size_no_bloom_filter += Path(file).stat().st_size

    size_bloom_filter = 0
    for file in path_bloom_filter.rglob("*.parquet"):
        size_bloom_filter += Path(file).stat().st_size

    assert size_no_bloom_filter < size_bloom_filter


def test_write_parquet_with_options_column_options(df, tmp_path):
    """Test writing Parquet files with different options for each column, which replace
    the global configs (when provided)."""
    data = {
        "a": [1, 2, 3],
        "b": ["a", "b", "c"],
        "c": [False, True, False],
        "d": [1.01, 2.02, 3.03],
        "e": [4, 5, 6],
    }

    column_specific_options = {
        "a": ParquetColumnOptions(statistics_enabled="none"),
        "b": ParquetColumnOptions(encoding="plain", dictionary_enabled=False),
        "c": ParquetColumnOptions(
            compression="snappy", encoding="rle", dictionary_enabled=False
        ),
        "d": ParquetColumnOptions(
            compression="zstd(6)",
            encoding="byte_stream_split",
            dictionary_enabled=False,
            statistics_enabled="none",
        ),
        # column "e" will use the global configs
    }

    results = {
        "a": {
            "statistics": False,
            "compression": "brotli",
            "encodings": ("PLAIN", "RLE", "RLE_DICTIONARY"),
        },
        "b": {
            "statistics": True,
            "compression": "brotli",
            "encodings": ("PLAIN", "RLE"),
        },
        "c": {
            "statistics": True,
            "compression": "snappy",
            "encodings": ("RLE",),
        },
        "d": {
            "statistics": False,
            "compression": "zstd",
            "encodings": ("RLE", "BYTE_STREAM_SPLIT"),
        },
        "e": {
            "statistics": True,
            "compression": "brotli",
            "encodings": ("PLAIN", "RLE", "RLE_DICTIONARY"),
        },
    }

    ctx = SessionContext()
    df = ctx.from_arrow(pa.record_batch(data))
    df.write_parquet_with_options(
        tmp_path,
        ParquetWriterOptions(
            compression="brotli(8)", column_specific_options=column_specific_options
        ),
    )

    for file in tmp_path.rglob("*.parquet"):
        parquet = pq.ParquetFile(file)
        metadata = parquet.metadata.to_dict()

        for row_group in metadata["row_groups"]:
            for col in row_group["columns"]:
                column_name = col["path_in_schema"]
                result = results[column_name]
                assert (col["statistics"] is not None) == result["statistics"]
                assert col["compression"].lower() == result["compression"].lower()
                assert col["encodings"] == result["encodings"]


def test_write_parquet_options(df, tmp_path):
    options = ParquetWriterOptions(compression="gzip", compression_level=6)
    df.write_parquet(str(tmp_path), options)

    result = pq.read_table(str(tmp_path)).to_pydict()
    expected = df.to_pydict()

    assert result == expected


def test_write_parquet_options_error(df, tmp_path):
    options = ParquetWriterOptions(compression="gzip", compression_level=6)
    with pytest.raises(ValueError):
        df.write_parquet(str(tmp_path), options, compression_level=1)


def test_write_table(ctx, df):
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3])],
        names=["a"],
    )

    ctx.register_record_batches("t", [[batch]])

    df = ctx.table("t").with_column("a", column("a") * literal(-1))

    ctx.table("t").show()

    df.write_table("t")
    result = ctx.table("t").sort(column("a")).collect()[0][0].to_pylist()
    expected = [-3, -2, -1, 1, 2, 3]

    assert result == expected


def test_dataframe_export(df) -> None:
    # Guarantees that we have the canonical implementation
    # reading our dataframe export
    table = pa.table(df)
    assert table.num_columns == 3
    assert table.num_rows == 3

    desired_schema = pa.schema([("a", pa.int64())])

    # Verify we can request a schema
    table = pa.table(df, schema=desired_schema)
    assert table.num_columns == 1
    assert table.num_rows == 3

    # Expect a table of nulls if the schema don't overlap
    desired_schema = pa.schema([("g", pa.string())])
    table = pa.table(df, schema=desired_schema)
    assert table.num_columns == 1
    assert table.num_rows == 3
    for i in range(3):
        assert table[0][i].as_py() is None

    # Expect an error when we cannot convert schema
    desired_schema = pa.schema([("a", pa.float32())])
    failed_convert = False
    try:
        table = pa.table(df, schema=desired_schema)
    except Exception:
        failed_convert = True
    assert failed_convert

    # Expect an error when we have a not set non-nullable
    desired_schema = pa.schema([("g", pa.string(), False)])
    failed_convert = False
    try:
        table = pa.table(df, schema=desired_schema)
    except Exception:
        failed_convert = True
    assert failed_convert


def test_dataframe_transform(df):
    def add_string_col(df_internal) -> DataFrame:
        return df_internal.with_column("string_col", literal("string data"))

    def add_with_parameter(df_internal, value: Any) -> DataFrame:
        return df_internal.with_column("new_col", literal(value))

    df = df.transform(add_string_col).transform(add_with_parameter, 3)

    result = df.to_pydict()

    assert result["a"] == [1, 2, 3]
    assert result["string_col"] == ["string data" for _i in range(3)]
    assert result["new_col"] == [3 for _i in range(3)]


def test_dataframe_repr_html_structure(df, clean_formatter_state) -> None:
    """Test that DataFrame._repr_html_ produces expected HTML output structure."""

    output = df._repr_html_()

    # Since we've added a fair bit of processing to the html output, lets just verify
    # the values we are expecting in the table exist. Use regex and ignore everything
    # between the <th></th> and <td></td>. We also don't want the closing > on the
    # td and th segments because that is where the formatting data is written.

    headers = ["a", "b", "c"]
    headers = [f"<th(.*?)>{v}</th>" for v in headers]
    header_pattern = "(.*?)".join(headers)
    header_matches = re.findall(header_pattern, output, re.DOTALL)
    assert len(header_matches) == 1

    # Update the pattern to handle values that may be wrapped in spans
    body_data = [[1, 4, 8], [2, 5, 5], [3, 6, 8]]

    body_lines = [
        f"<td(.*?)>(?:<span[^>]*?>)?{v}(?:</span>)?</td>"
        for inner in body_data
        for v in inner
    ]
    body_pattern = "(.*?)".join(body_lines)

    body_matches = re.findall(body_pattern, output, re.DOTALL)

    assert len(body_matches) == 1, "Expected pattern of values not found in HTML output"


def test_dataframe_repr_html_values(df, clean_formatter_state):
    """Test that DataFrame._repr_html_ contains the expected data values."""
    html = df._repr_html_()
    assert html is not None

    # Create a more flexible pattern that handles values being wrapped in spans
    # This pattern will match the sequence of values 1,4,8,2,5,5,3,6,8 regardless
    # of formatting
    pattern = re.compile(
        r"<td[^>]*?>(?:<span[^>]*?>)?1(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?4(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?8(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?2(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?5(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?5(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?3(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?6(?:</span>)?</td>.*?"
        r"<td[^>]*?>(?:<span[^>]*?>)?8(?:</span>)?</td>",
        re.DOTALL,
    )

    # Print debug info if the test fails
    matches = re.findall(pattern, html)
    if not matches:
        print(f"HTML output snippet: {html[:500]}...")  # noqa: T201

    assert len(matches) > 0, "Expected pattern of values not found in HTML output"


def test_html_formatter_shared_styles(df, clean_formatter_state):
    """Test that shared styles work correctly across multiple tables."""

    # First, ensure we're using shared styles
    configure_formatter(use_shared_styles=True)

    html_first = df._repr_html_()
    html_second = df._repr_html_()

    assert "<script>" in html_first
    assert "df-styles" in html_first
    assert "<script>" in html_second
    assert "df-styles" in html_second
    assert "<style>" not in html_first
    assert "<style>" not in html_second


def test_html_formatter_no_shared_styles(df, clean_formatter_state):
    """Test that styles are always included when shared styles are disabled."""

    # Configure formatter to NOT use shared styles
    configure_formatter(use_shared_styles=False)

    html_first = df._repr_html_()
    html_second = df._repr_html_()

    assert "<script>" in html_first
    assert "<script>" in html_second
    assert "df-styles" in html_first
    assert "df-styles" in html_second
    assert "<style>" not in html_first
    assert "<style>" not in html_second


def test_html_formatter_manual_format_html(clean_formatter_state):
    """Test direct usage of format_html method with shared styles."""

    # Create sample data
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([4, 5, 6])],
        names=["a", "b"],
    )

    formatter = get_formatter()

    html_first = formatter.format_html([batch], batch.schema)
    html_second = formatter.format_html([batch], batch.schema)

    assert "<script>" in html_first
    assert "<script>" in html_second
    assert "df-styles" in html_first
    assert "df-styles" in html_second
    assert "<style>" not in html_first
    assert "<style>" not in html_second

    # Create a new formatter with shared_styles=False
    local_formatter = DataFrameHtmlFormatter(use_shared_styles=False)

    # Both calls should include styles
    local_html_1 = local_formatter.format_html([batch], batch.schema)
    local_html_2 = local_formatter.format_html([batch], batch.schema)

    assert "<script>" in local_html_1
    assert "<script>" in local_html_2
    assert "df-styles" in local_html_1
    assert "df-styles" in local_html_2
    assert "<style>" not in local_html_1
    assert "<style>" not in local_html_2


def test_html_formatter_backward_compatibility_repr_rows(df, clean_formatter_state):
    """Test backward compatibility with custom formatter using deprecated repr_rows.

    This test validates that the Rust code correctly handles custom formatter
    implementations that only have the deprecated `repr_rows` attribute.

    This is critical for supporting custom formatters created before `max_rows`
    was added. Users should be able to pass their custom formatter objects
    without breaking the rendering pipeline.
    """

    # Create a custom formatter class that ONLY has repr_rows (simulating old code)
    class LegacyCustomFormatter:
        """Simulates a custom formatter created before max_rows existed."""

        def __init__(self):
            # Only set repr_rows, not max_rows (as old formatters would)
            self.repr_rows = 5
            self.max_memory_bytes = 2 * MB
            self.min_rows = 2

        def format_html(self, batches, schema):
            """Minimal format_html implementation for testing."""
            # Just return valid HTML to pass validation
            return "<table><tr><td>test</td></tr></table>"

    # Use the legacy formatter with DataFusion (currently unused)
    _ = LegacyCustomFormatter()

    # This should not raise an error even though max_rows doesn't exist
    # The Rust code should fall back to repr_rows
    html_output = df._repr_html_()

    # Verify that rendering succeeded
    assert isinstance(html_output, str)
    assert len(html_output) > 0

    # Verify it's valid HTML (basic check)
    assert "<table" in html_output.lower() or "data truncated" in html_output.lower()


def test_fill_null_basic(null_df):
    """Test basic fill_null functionality with a single value."""
    # Fill all nulls with 0
    filled_df = null_df.fill_null(0)

    result = filled_df.collect()[0]

    # Check that nulls were filled with 0 (or equivalent)
    assert result.column(0) == pa.array([1, 0, 3, 0])
    assert result.column(1) == pa.array([4.5, 6.7, 0.0, 0.0])
    # String column should be filled with "0"
    assert result.column(2) == pa.array(["a", "0", "c", "0"])
    # Boolean column should be filled with False (0 converted to bool)
    assert result.column(3) == pa.array([True, False, False, False])


def test_fill_null_subset(null_df):
    """Test filling nulls only in a subset of columns."""
    # Fill nulls only in numeric columns
    filled_df = null_df.fill_null(0, subset=["int_col", "float_col"])

    result = filled_df.collect()[0]

    # Check that nulls were filled only in specified columns
    assert result.column(0) == pa.array([1, 0, 3, 0])
    assert result.column(1) == pa.array([4.5, 6.7, 0.0, 0.0])
    # These should still have nulls
    assert None in result.column(2).to_pylist()
    assert None in result.column(3).to_pylist()


def test_fill_null_str_column(null_df):
    """Test filling nulls in string columns with different values."""
    # Fill string nulls with a replacement string
    filled_df = null_df.fill_null("N/A", subset=["str_col"])

    result = filled_df.collect()[0]

    # Check that string nulls were filled with "N/A"
    assert result.column(2).to_pylist() == ["a", "N/A", "c", "N/A"]

    # Other columns should be unchanged
    assert None in result.column(0).to_pylist()
    assert None in result.column(1).to_pylist()
    assert None in result.column(3).to_pylist()

    # Fill with an empty string
    filled_df = null_df.fill_null("", subset=["str_col"])
    result = filled_df.collect()[0]
    assert result.column(2).to_pylist() == ["a", "", "c", ""]


def test_fill_null_bool_column(null_df):
    """Test filling nulls in boolean columns with different values."""
    # Fill bool nulls with True
    filled_df = null_df.fill_null(value=True, subset=["bool_col"])

    result = filled_df.collect()[0]

    # Check that bool nulls were filled with True
    assert result.column(3).to_pylist() == [True, True, False, True]

    # Other columns should be unchanged
    assert None in result.column(0).to_pylist()

    # Fill bool nulls with False
    filled_df = null_df.fill_null(value=False, subset=["bool_col"])
    result = filled_df.collect()[0]
    assert result.column(3).to_pylist() == [True, False, False, False]


def test_fill_null_date32_column(null_df):
    """Test filling nulls in date32 columns."""

    # Fill date32 nulls with a specific date (1970-01-01)
    epoch_date = datetime.date(1970, 1, 1)
    filled_df = null_df.fill_null(epoch_date, subset=["date32_col"])

    result = filled_df.collect()[0]

    # Check that date32 nulls were filled with epoch date
    dates = result.column(4).to_pylist()
    assert dates[0] == datetime.date(2000, 1, 1)  # Original value
    assert dates[1] == epoch_date  # Filled value
    assert dates[2] == datetime.date(2022, 1, 1)  # Original value
    assert dates[3] == epoch_date  # Filled value

    # Other date column should be unchanged
    assert None in result.column(5).to_pylist()


def test_fill_null_date64_column(null_df):
    """Test filling nulls in date64 columns."""

    # Fill date64 nulls with a specific date (1970-01-01)
    epoch_date = datetime.date(1970, 1, 1)
    filled_df = null_df.fill_null(epoch_date, subset=["date64_col"])

    result = filled_df.collect()[0]

    # Check that date64 nulls were filled with epoch date
    dates = result.column(5).to_pylist()
    assert dates[0] == datetime.date(2000, 1, 1)  # Original value
    assert dates[1] == epoch_date  # Filled value
    assert dates[2] == datetime.date(2022, 1, 1)  # Original value
    assert dates[3] == epoch_date  # Filled value

    # Other date column should be unchanged
    assert None in result.column(4).to_pylist()


def test_fill_null_type_coercion(null_df):
    """Test type coercion when filling nulls with values of different types."""
    # Try to fill string nulls with a number
    filled_df = null_df.fill_null(42, subset=["str_col"])

    result = filled_df.collect()[0]

    # String nulls should be filled with string representation of the number
    assert result.column(2).to_pylist() == ["a", "42", "c", "42"]

    # Try to fill bool nulls with a string that converts to True
    filled_df = null_df.fill_null("true", subset=["bool_col"])
    result = filled_df.collect()[0]

    # This behavior depends on the implementation - check it works without error
    # but don't make assertions about exact conversion behavior
    assert None not in result.column(3).to_pylist()


def test_fill_null_multiple_date_columns(null_df):
    """Test filling nulls in both date column types simultaneously."""

    # Fill both date column types with the same date
    test_date = datetime.date(2023, 12, 31)
    filled_df = null_df.fill_null(test_date, subset=["date32_col", "date64_col"])

    result = filled_df.collect()[0]

    # Check both date columns were filled correctly
    date32_vals = result.column(4).to_pylist()
    date64_vals = result.column(5).to_pylist()

    assert None not in date32_vals
    assert None not in date64_vals

    assert date32_vals[1] == test_date
    assert date32_vals[3] == test_date
    assert date64_vals[1] == test_date
    assert date64_vals[3] == test_date


def test_fill_null_specific_types(null_df):
    """Test filling nulls with type-appropriate values."""
    # Fill with type-specific values
    filled_df = null_df.fill_null("missing")

    result = filled_df.collect()[0]

    # Check that nulls were filled appropriately by type

    assert result.column(0).to_pylist() == [1, None, 3, None]
    assert result.column(1).to_pylist() == [4.5, 6.7, None, None]
    assert result.column(2).to_pylist() == ["a", "missing", "c", "missing"]
    assert result.column(3).to_pylist() == [True, None, False, None]  # Bool gets False
    assert result.column(4).to_pylist() == [
        datetime.date(2000, 1, 1),
        None,
        datetime.date(2022, 1, 1),
        None,
    ]
    assert result.column(5).to_pylist() == [
        datetime.date(2000, 1, 1),
        None,
        datetime.date(2022, 1, 1),
        None,
    ]


@pytest.mark.parametrize(
    ("subset", "expected_cols"),
    [
        pytest.param([], [], id="empty list"),
        pytest.param(np.array([], dtype=str), [], id="empty numpy"),
        pytest.param(np.array(["int_col"]), ["int_col"], id="numpy"),
    ],
)
def test_fill_null_subset_forms(null_df, subset, expected_cols):
    # An empty subset fills nothing rather than everything, and a numpy array
    # is accepted without being tested for truthiness.
    result = null_df.fill_null(0, subset=subset).to_pydict()
    original = null_df.to_pydict()
    for name, values in result.items():
        if name in expected_cols:
            assert None not in values
        else:
            assert values == original[name]


def test_fill_null_immutability(null_df):
    """Test that original DataFrame is unchanged after fill_null."""
    # Get original values with nulls
    original = null_df.collect()[0]
    original_int_nulls = original.column(0).to_pylist().count(None)

    # Apply fill_null
    _filled_df = null_df.fill_null(0)

    # Check that original is unchanged
    new_original = null_df.collect()[0]
    new_original_int_nulls = new_original.column(0).to_pylist().count(None)

    assert original_int_nulls == new_original_int_nulls
    assert original_int_nulls > 0  # Ensure we actually had nulls in the first place


def test_fill_null_empty_df(ctx):
    """Test fill_null on empty DataFrame."""
    # Create an empty DataFrame with schema
    batch = pa.RecordBatch.from_arrays(
        [pa.array([], type=pa.int64()), pa.array([], type=pa.string())],
        names=["a", "b"],
    )
    empty_df = ctx.create_dataframe([[batch]])

    # Fill nulls (should work without errors)
    filled_df = empty_df.fill_null(0)

    # Should still be empty but with same schema
    result = filled_df.collect()[0]
    assert len(result.column(0)) == 0
    assert len(result.column(1)) == 0
    assert result.schema.field(0).name == "a"
    assert result.schema.field(1).name == "b"


def test_fill_null_all_null_column(ctx):
    """Test fill_null on a column with all nulls."""
    # Create DataFrame with a column of all nulls
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3]), pa.array([None, None, None], type=pa.string())],
        names=["a", "b"],
    )
    all_null_df = ctx.create_dataframe([[batch]])

    # Fill nulls with a value
    filled_df = all_null_df.fill_null("filled")

    # Check that all nulls were filled
    result = filled_df.collect()[0]
    assert result.column(1).to_pylist() == ["filled", "filled", "filled"]


def _nan_df(ctx):
    nan = float("nan")
    batch = pa.RecordBatch.from_arrays(
        [
            pa.array([1.0, nan, None], type=pa.float64()),
            pa.array([nan, 2.0, 3.0], type=pa.float32()),
            pa.array([1, 2, 3]),
            pa.array(["x", "nan", None]),
        ],
        names=["f64", "f32", "i", "s"],
    )
    return ctx.create_dataframe([[batch]])


def _is_nan(v):
    return v is not None and v != v  # noqa: PLR0124


def test_fill_nan_all_columns(ctx):
    df = _nan_df(ctx)
    assert df.fill_nan(0.0).schema() == df.schema()
    result = df.fill_nan(0.0).to_pydict()
    # NaN replaced in both float widths; null is not NaN and stays null.
    assert result["f64"] == [1.0, 0.0, None]
    assert result["f32"] == [0.0, 2.0, 3.0]
    # Non-float columns are untouched.
    assert result["i"] == [1, 2, 3]
    assert result["s"] == ["x", "nan", None]


def test_fill_nan_subset(ctx):
    result = _nan_df(ctx).fill_nan(-1.0, subset=["f32"]).to_pydict()
    assert result["f32"] == [-1.0, 2.0, 3.0]
    assert _is_nan(result["f64"][1])


@pytest.mark.parametrize(
    ("subset", "f64_filled", "f32_filled"),
    [
        pytest.param([], False, False, id="empty list"),
        pytest.param(np.array([], dtype=str), False, False, id="empty numpy"),
        pytest.param(np.array(["f32"]), False, True, id="numpy"),
    ],
)
def test_fill_nan_subset_forms(ctx, subset, f64_filled, f32_filled):
    # An empty subset fills nothing rather than everything, and a numpy array
    # is accepted without being tested for truthiness.
    result = _nan_df(ctx).fill_nan(0.0, subset=subset).to_pydict()
    assert _is_nan(result["f64"][1]) != f64_filled
    assert _is_nan(result["f32"][0]) != f32_filled


_slow_udf_started = threading.Event()


@udf([pa.int64()], pa.int64(), "immutable")
def slow_udf(x: pa.Array) -> pa.Array:
    _slow_udf_started.set()
    # Sleep in small increments so Python's eval loop checks for pending
    # async exceptions (like KeyboardInterrupt via PyThreadState_SetAsyncExc)
    # between iterations. A single long time.sleep() is a C call where async
    # exceptions are not checked on all Python versions (notably 3.11).
    for _ in range(200):
        time.sleep(0.01)
    return x


@pytest.mark.parametrize(
    ("slow_query", "as_c_stream"),
    [
        (True, True),
        (True, False),
        (False, True),
        (False, False),
    ],
)
def test_collect_or_stream_interrupted(slow_query, as_c_stream):  # noqa: C901 PLR0915
    """Ensure collection responds to ``KeyboardInterrupt`` signals.

    This test issues a long-running query, and consumes the results via
    either a collect() call or ``__arrow_c_stream__``. It raises
    ``KeyboardInterrupt`` in the main thread and verifies that the
    process has interrupted.

    The `slow_query` determines if the query itself is slow via a
    UDF with a timeout or if it is a fast query that generates many
    results so it takes a long time to iterate through them all.
    """

    ctx = SessionContext()
    df = ctx.sql("select * from generate_series(1, 1000000000000000000)")
    if slow_query:
        df = ctx.from_pydict({"a": [1, 2, 3]}).select(slow_udf(column("a")))

    if as_c_stream:
        reader = pa.RecordBatchReader.from_stream(df)

    _slow_udf_started.clear()
    read_started = threading.Event()
    read_exception = []
    read_thread_id = None
    max_wait_time = 5.0

    def trigger_interrupt():
        """Wait for read to start, then raise KeyboardInterrupt in read thread."""
        if not read_started.wait(timeout=max_wait_time):
            msg = f"Read operation did not start within {max_wait_time} seconds"
            raise RuntimeError(msg)

        # For slow_query tests, wait until the UDF is actually executing Python
        # bytecode before sending the interrupt. PyThreadState_SetAsyncExc only
        # delivers exceptions when the thread is in the Python eval loop, not
        # while in native (Rust/C) code.
        if slow_query and not _slow_udf_started.wait(timeout=max_wait_time):
            msg = f"UDF did not start within {max_wait_time} seconds"
            raise RuntimeError(msg)

        if read_thread_id is None:
            msg = "Cannot get read thread ID"
            raise RuntimeError(msg)

        exception = ctypes.py_object(KeyboardInterrupt)
        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
            ctypes.c_long(read_thread_id), exception
        )
        if res != 1:
            ctypes.pythonapi.PyThreadState_SetAsyncExc(
                ctypes.c_long(read_thread_id), ctypes.py_object(0)
            )
            msg = "Failed to raise KeyboardInterrupt in read thread"
            raise RuntimeError(msg)

    def read_stream():
        """Consume the reader, which should be interrupted."""
        nonlocal read_thread_id
        read_thread_id = threading.get_ident()
        try:
            read_started.set()
            if as_c_stream:
                reader.read_all()
            else:
                df.collect()
            # If we get here, the read completed without interruption
            read_exception.append(RuntimeError("Read completed without interruption"))
        except KeyboardInterrupt:
            read_exception.append(KeyboardInterrupt)
        except Exception as e:
            read_exception.append(e)

    read_thread = threading.Thread(target=read_stream)
    read_thread.daemon = True
    read_thread.start()

    interrupt_thread = threading.Thread(target=trigger_interrupt)
    interrupt_thread.daemon = True
    interrupt_thread.start()

    # Wait for the read operation with a timeout
    read_thread.join(timeout=10.0)

    if read_thread.is_alive():
        pytest.fail("Stream read operation timed out after 10 seconds")

    # Verify we got the expected KeyboardInterrupt
    if not read_exception:
        pytest.fail("No exception was raised during stream read")

    # Check if we got KeyboardInterrupt directly or wrapped in another exception
    exception = read_exception[0]
    if not (
        isinstance(exception, type(KeyboardInterrupt))
        or "KeyboardInterrupt" in str(exception)
    ):
        pytest.fail(f"Expected KeyboardInterrupt, got: {exception}")

    interrupt_thread.join(timeout=1.0)


def test_show_select_where_no_rows(capsys) -> None:
    ctx = SessionContext()
    df = ctx.sql("SELECT 1 WHERE 1=0")
    df.show()
    out = capsys.readouterr().out
    assert "DataFrame has no rows" in out


def test_show_from_empty_batch(capsys) -> None:
    ctx = SessionContext()
    batch = pa.record_batch([pa.array([], type=pa.int32())], names=["a"])
    ctx.create_dataframe([[batch]]).show()
    out = capsys.readouterr().out
    assert "| a |" in out


@pytest.mark.parametrize("file_sort_order", [[["a"]], [[df_col("a")]]])
def test_register_parquet_file_sort_order(ctx, tmp_path, file_sort_order):
    table = pa.table({"a": [1, 2]})
    path = tmp_path / "file.parquet"
    pa.parquet.write_table(table, path)
    ctx.register_parquet("t", path, file_sort_order=file_sort_order)
    assert "t" in ctx.catalog().schema().names()


@pytest.mark.parametrize("file_sort_order", [[["a"]], [[df_col("a")]]])
def test_register_listing_table_file_sort_order(ctx, tmp_path, file_sort_order):
    table = pa.table({"a": [1, 2]})
    dir_path = tmp_path / "dir"
    dir_path.mkdir()
    pa.parquet.write_table(table, dir_path / "file.parquet")
    ctx.register_listing_table(
        "t", dir_path, schema=table.schema, file_sort_order=file_sort_order
    )
    assert "t" in ctx.catalog().schema().names()


@pytest.mark.parametrize("file_sort_order", [[["a"]], [[df_col("a")]]])
def test_read_parquet_file_sort_order(tmp_path, file_sort_order):
    ctx = SessionContext()
    table = pa.table({"a": [1, 2]})
    path = tmp_path / "data.parquet"
    pa.parquet.write_table(table, path)
    df = ctx.read_parquet(path, file_sort_order=file_sort_order)
    assert df.collect()[0].column(0).to_pylist() == [1, 2]


@pytest.mark.parametrize(
    ("df1_data", "df2_data", "method", "kwargs", "expected_a", "expected_b"),
    [
        pytest.param(
            {"a": [1, 2, 3, 1], "b": [10, 20, 30, 10]},
            {"a": [1, 2], "b": [10, 20]},
            "except_all",
            {"distinct": True},
            [3],
            [30],
            id="except_all(distinct=True): removes matching rows and deduplicates",
        ),
        pytest.param(
            {"a": [1, 2, 3, 1], "b": [10, 20, 30, 10]},
            {"a": [1, 4], "b": [10, 40]},
            "intersect",
            {"distinct": True},
            [1],
            [10],
            id="intersect(distinct=True): keeps common rows and deduplicates",
        ),
        pytest.param(
            {"a": [1], "b": [10]},
            {"b": [20], "a": [2]},  # reversed column order tests matching by name
            "union_by_name",
            {},
            [1, 2],
            [10, 20],
            id="union_by_name: matches columns by name not position",
        ),
    ],
)
def test_set_operations_distinct(
    df1_data, df2_data, method, kwargs, expected_a, expected_b
):
    ctx = SessionContext()
    df1 = ctx.from_pydict(df1_data)
    df2 = ctx.from_pydict(df2_data)
    result = (
        getattr(df1, method)(df2, **kwargs)
        .sort(column("a").sort(ascending=True))
        .collect()[0]
    )
    assert result.column(0).to_pylist() == expected_a
    assert result.column(1).to_pylist() == expected_b


def test_union_by_name_distinct():
    ctx = SessionContext()
    df1 = ctx.from_pydict({"a": [1, 1], "b": [10, 10]})
    df2 = ctx.from_pydict({"b": [10], "a": [1]})
    result = df1.union_by_name(df2, distinct=True).collect()[0]
    assert result.column(0).to_pylist() == [1]
    assert result.column(1).to_pylist() == [10]


def test_column_qualified():
    """DataFrame.column() returns a qualified column expression."""
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1, 2], "b": [3, 4]})
    expr = df.column("a")
    result = df.select(expr).collect()[0]
    assert result.column(0).to_pylist() == [1, 2]


def test_column_not_found():
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1]})
    with pytest.raises(Exception, match="not found"):
        df.column("z")


def test_column_ambiguous():
    """After a join, duplicate column names that cannot be resolved raise an error."""
    ctx = SessionContext()
    left = ctx.from_pydict({"id": [1, 2], "val": [10, 20]})
    right = ctx.from_pydict({"id": [1, 2], "val": [30, 40]})
    joined = left.join(right, on="id", how="inner")
    with pytest.raises(Exception, match="not found"):
        joined.column("val")


def test_column_after_join():
    """Qualified column works for non-ambiguous columns after a join."""
    ctx = SessionContext()
    left = ctx.from_pydict({"id": [1, 2], "x": [10, 20]})
    right = ctx.from_pydict({"id": [1, 2], "y": [30, 40]})
    joined = left.join(right, on="id", how="inner")
    expr = joined.column("y")
    result = joined.select("id", expr).sort("id").collect()[0]
    assert result.column(0).to_pylist() == [1, 2]
    assert result.column(1).to_pylist() == [30, 40]


def test_col_join_disambiguate():
    """Use col() to disambiguate and select columns after a join."""
    ctx = SessionContext()
    df1 = ctx.from_pydict({"foo": [1, 2, 3], "bar": [5, 6, 7]})
    df2 = ctx.from_pydict({"foo": [1, 2, 3], "baz": [8, 9, 10]})
    joined = df1.join_on(df2, df1.col("foo") == df2.col("foo"), how="inner")
    result = (
        joined.select(df1.col("foo"), df1.col("bar"), df2.col("baz"))
        .sort(df1.col("foo"))
        .to_pydict()
    )
    assert result["bar"] == [5, 6, 7]
    assert result["baz"] == [8, 9, 10]


def test_find_qualified_columns():
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1, 2], "b": [3, 4], "c": [5, 6]})
    exprs = df.find_qualified_columns("a", "c")
    assert len(exprs) == 2
    result = df.select(*exprs).collect()[0]
    assert result.column(0).to_pylist() == [1, 2]
    assert result.column(1).to_pylist() == [5, 6]


def test_find_qualified_columns_not_found():
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1]})
    with pytest.raises(Exception, match="not found"):
        df.find_qualified_columns("a", "z")


def test_distinct_on():
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1, 1, 2, 2], "b": [10, 20, 30, 40]})
    result = (
        df.distinct_on(
            [column("a")],
            [column("a"), column("b")],
            [column("a").sort(ascending=True), column("b").sort(ascending=True)],
        )
        .sort(column("a").sort(ascending=True))
        .collect()[0]
    )
    # Keeps the first row per group (smallest b per a)
    assert result.column(0).to_pylist() == [1, 2]
    assert result.column(1).to_pylist() == [10, 30]


@pytest.mark.parametrize(
    ("input_values", "expected"),
    [
        ([3, 1, 2], [1, 2, 3]),
        ([1, 2, 3], [1, 2, 3]),
        ([3, None, 1, 2], [1, 2, 3, None]),
    ],
)
def test_sort_by(input_values, expected):
    """sort_by always sorts ascending with nulls last regardless of input order."""
    ctx = SessionContext()
    df = ctx.from_pydict({"a": input_values})
    result = df.sort_by(column("a")).collect()[0]
    assert result.column(0).to_pylist() == expected


@pytest.mark.parametrize(
    ("fmt", "verbose", "analyze", "expected_substring"),
    [
        pytest.param(None, False, False, None, id="default format"),
        pytest.param(ExplainFormat.TREE, False, False, "---", id="tree format"),
        pytest.param(
            ExplainFormat.INDENT, True, True, None, id="indent verbose+analyze"
        ),
        pytest.param(ExplainFormat.PGJSON, False, False, '"Plan"', id="pgjson format"),
        pytest.param(
            ExplainFormat.GRAPHVIZ, False, False, "digraph", id="graphviz format"
        ),
    ],
)
def test_explain_with_format(capsys, fmt, verbose, analyze, expected_substring):
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1]})
    df.explain(verbose=verbose, analyze=analyze, format=fmt)
    captured = capsys.readouterr()
    assert "plan_type" in captured.out
    if expected_substring is not None:
        assert expected_substring in captured.out


def _explain_output(capsys, **kwargs):
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1, 2]}).filter(column("a") > literal(1))
    df.explain(**kwargs)
    return capsys.readouterr().out


@pytest.mark.parametrize(
    ("kwargs", "present", "absent"),
    [
        pytest.param({}, [], ["statistics="], id="default_no_statistics"),
        pytest.param(
            {"show_statistics": True}, ["statistics=[Rows="], [], id="show_statistics"
        ),
        pytest.param(
            {"analyze": True, "analyze_level": ExplainAnalyzeLevel.DEV},
            ["output_rows=", "output_batches="],
            [],
            id="analyze_level_dev",
        ),
        pytest.param(
            {"analyze": True, "analyze_level": ExplainAnalyzeLevel.SUMMARY},
            ["output_rows="],
            ["output_batches="],
            id="analyze_level_summary",
        ),
        pytest.param(
            {
                "analyze": True,
                "analyze_categories": [ExplainMetricCategory.ROWS],
            },
            ["output_rows="],
            ["elapsed_compute=", "output_bytes="],
            id="analyze_categories_rows",
        ),
        pytest.param(
            {"analyze": True, "analyze_categories": []},
            ["FilterExec: a@0 > 1, metrics=[]"],
            ["output_rows="],
            id="analyze_categories_empty_suppresses_metrics",
        ),
    ],
)
def test_explain_options(capsys, kwargs, present, absent):
    out = _explain_output(capsys, **kwargs)
    for text in present:
        assert text in out
    for text in absent:
        assert text not in out


@pytest.mark.parametrize(
    ("kwargs", "message"),
    [
        pytest.param(
            {"analyze": True, "show_statistics": True},
            "show_statistics cannot be combined with analyze",
            id="show_statistics_true_with_analyze",
        ),
        pytest.param(
            {"analyze": True, "show_statistics": False},
            "show_statistics cannot be combined with analyze",
            id="show_statistics_false_with_analyze",
        ),
        pytest.param(
            {"analyze_level": ExplainAnalyzeLevel.DEV},
            "analyze_level requires analyze",
            id="analyze_level_without_analyze",
        ),
        pytest.param(
            {"analyze_categories": [ExplainMetricCategory.ROWS]},
            "analyze_categories requires analyze",
            id="analyze_categories_without_analyze",
        ),
    ],
)
def test_explain_rejects_options_that_would_be_ignored(capsys, kwargs, message):
    # Upstream would silently ignore these; SQL's EXPLAIN rejects them.
    with pytest.raises(ValueError, match=message):
        _explain_output(capsys, **kwargs)


@pytest.mark.parametrize(
    ("window_exprs", "expected_columns"),
    [
        pytest.param(
            lambda: [
                f.row_number(partition_by=[column("b")], order_by=[column("a")]).alias(
                    "rn"
                ),
            ],
            {"rn": [1, 2, 1]},
            id="single window expression",
        ),
        pytest.param(
            lambda: [
                f.row_number(partition_by=[column("b")], order_by=[column("a")]).alias(
                    "rn"
                ),
                f.rank(partition_by=[column("b")], order_by=[column("a")]).alias("rnk"),
            ],
            {"rn": [1, 2, 1], "rnk": [1, 2, 1]},
            id="multiple window expressions",
        ),
    ],
)
def test_window(window_exprs, expected_columns):
    ctx = SessionContext()
    df = ctx.from_pydict({"a": [1, 2, 3], "b": ["x", "x", "y"]})
    result = (
        df.window(*window_exprs()).sort(column("a").sort(ascending=True)).collect()[0]
    )
    for col_name, expected_values in expected_columns.items():
        assert col_name in result.schema.names
        assert (
            result.column(result.schema.get_field_index(col_name)).to_pylist()
            == expected_values
        )


@pytest.mark.parametrize(
    ("input_data", "recursions", "expected_a"),
    [
        pytest.param(
            {"a": [[1, 2], [3]], "b": ["x", "y"]},
            None,
            [1, 2, 3],
            id="basic unnest without recursions",
        ),
        pytest.param(
            {"a": [[1, 2], [3]], "b": ["x", "y"]},
            [("a", "a", 1)],
            [1, 2, 3],
            id="explicit depth 1 matches basic unnest",
        ),
        pytest.param(
            {"a": [[[1, 2], [3]], [[4]]], "b": ["x", "y"]},
            [("a", "a", 1)],
            [[1, 2], [3], [4]],
            id="depth 1 on nested lists keeps inner lists",
        ),
        pytest.param(
            {"a": [[[1, 2], [3]], [[4]]], "b": ["x", "y"]},
            [("a", "a", 2)],
            [1, 2, 3, 4],
            id="depth 2 fully flattens nested lists",
        ),
    ],
)
def test_unnest_columns_with_recursions(input_data, recursions, expected_a):
    ctx = SessionContext()
    df = ctx.from_pydict(input_data)
    kwargs = {}
    if recursions is not None:
        kwargs["recursions"] = recursions
    result = df.unnest_columns("a", **kwargs).collect()[0]
    assert result.column(0).to_pylist() == expected_a


def test_sort_default_null_behavior():
    ctx = SessionContext()
    ctx.sql("create table t (a int)").collect()
    ctx.sql("insert into t values (3), (null), (1), (null), (4), (2)").collect()

    # sort
    result_sort = ctx.table("t").sort(column("a")).to_pydict()

    # sort_by
    result_sort_by = ctx.table("t").sort_by(column("a")).to_pydict()

    # sql
    result_sql = ctx.sql("select * from t order by a").to_pydict()

    # order_by function
    result_order_by = ctx.table("t").sort(f.order_by(column("a"))).to_pydict()

    assert result_sort == result_sort_by == result_sql == result_order_by
    assert result_sort == {"a": [1, 2, 3, 4, None, None]}