import json
import os
import pathlib
import re
from io import BytesIO
from itertools import product
from packaging.version import Version

import numpy as np
from pandas import ArrowDtype, DataFrame, Index, Series
from pandas import read_parquet as pd_read_parquet

import shapely
from shapely.geometry import LineString, MultiPolygon, Point, Polygon, box

import geopandas
from geopandas import GeoDataFrame, read_feather, read_file, read_parquet
from geopandas._compat import HAS_PYPROJ
from geopandas.array import to_wkb
from geopandas.io.arrow import (
    METADATA_VERSION,
    SUPPORTED_VERSIONS,
    _convert_bbox_to_parquet_filter,
    _create_metadata,
    _decode_metadata,
    _encode_metadata,
    _geopandas_to_arrow,
    _get_filesystem_path,
    _remove_id_from_member_of_ensembles,
    _validate_dataframe,
    _validate_geo_metadata,
)

import pytest
from geopandas.testing import assert_geodataframe_equal, assert_geoseries_equal
from geopandas.tests.util import mock
from pandas.testing import assert_frame_equal

DATA_PATH = pathlib.Path(os.path.dirname(__file__)) / "data"


# Skip all tests in this module if pyarrow is not available
pyarrow = pytest.importorskip("pyarrow")

import pyarrow.compute as pc
import pyarrow.parquet as pq
from pyarrow import feather


@pytest.fixture(params=["parquet", pytest.param("feather")])
def file_format(request):
    if request.param == "parquet":
        return read_parquet, GeoDataFrame.to_parquet
    elif request.param == "feather":
        return read_feather, GeoDataFrame.to_feather


def test_create_metadata(naturalearth_lowres):
    df = read_file(naturalearth_lowres)
    metadata = _create_metadata(df, geometry_encoding={"geometry": "WKB"})

    assert isinstance(metadata, dict)
    assert metadata["version"] == METADATA_VERSION
    assert metadata["primary_column"] == "geometry"
    assert "geometry" in metadata["columns"]
    if HAS_PYPROJ:
        crs_expected = df.crs.to_json_dict()
        _remove_id_from_member_of_ensembles(crs_expected)
        assert metadata["columns"]["geometry"]["crs"] == crs_expected
    assert metadata["columns"]["geometry"]["encoding"] == "WKB"
    assert metadata["columns"]["geometry"]["geometry_types"] == [
        "MultiPolygon",
        "Polygon",
    ]

    assert np.array_equal(
        metadata["columns"]["geometry"]["bbox"], df.geometry.total_bounds
    )

    assert metadata["creator"]["library"] == "geopandas"
    assert metadata["creator"]["version"] == geopandas.__version__

    # specifying non-WKB encoding sets default schema to 1.1.0
    metadata = _create_metadata(df, geometry_encoding={"geometry": "point"})
    assert metadata["version"] == "1.1.0"
    assert metadata["columns"]["geometry"]["encoding"] == "point"

    # check that providing no geometry encoding defaults to WKB
    metadata = _create_metadata(df)
    assert metadata["columns"]["geometry"]["encoding"] == "WKB"


def test_create_metadata_with_z_geometries():
    geometry_types = [
        "Point",
        "Point Z",
        "LineString",
        "LineString Z",
        "Polygon",
        "Polygon Z",
        "MultiPolygon",
        "MultiPolygon Z",
    ]
    df = geopandas.GeoDataFrame(
        {
            "geo_type": geometry_types,
            "geometry": [
                Point(1, 2),
                Point(1, 2, 3),
                LineString([(0, 0), (1, 1), (2, 2)]),
                LineString([(0, 0, 1), (1, 1, 2), (2, 2, 3)]),
                Polygon([(0, 0), (0, 1), (1, 1), (1, 0)]),
                Polygon([(0, 0, 0), (0, 1, 0.5), (1, 1, 1), (1, 0, 0.5)]),
                MultiPolygon(
                    [
                        Polygon([(0, 0), (0, 1), (1, 1), (1, 0)]),
                        Polygon([(0.5, 0.5), (0.5, 1.5), (1.5, 1.5), (1.5, 0.5)]),
                    ]
                ),
                MultiPolygon(
                    [
                        Polygon([(0, 0, 0), (0, 1, 0.5), (1, 1, 1), (1, 0, 0.5)]),
                        Polygon(
                            [
                                (0.5, 0.5, 1),
                                (0.5, 1.5, 1.5),
                                (1.5, 1.5, 2),
                                (1.5, 0.5, 1.5),
                            ]
                        ),
                    ]
                ),
            ],
        },
    )
    metadata = _create_metadata(df, geometry_encoding={"geometry": "WKB"})
    assert sorted(metadata["columns"]["geometry"]["geometry_types"]) == sorted(
        geometry_types
    )
    # only 3D geometries
    metadata = _create_metadata(df.iloc[1::2], geometry_encoding={"geometry": "WKB"})
    assert all(
        geom_type.endswith(" Z")
        for geom_type in metadata["columns"]["geometry"]["geometry_types"]
    )

    metadata = _create_metadata(df.iloc[5:7], geometry_encoding={"geometry": "WKB"})
    assert metadata["columns"]["geometry"]["geometry_types"] == [
        "MultiPolygon",
        "Polygon Z",
    ]


def test_crs_metadata_datum_ensemble():
    pyproj = pytest.importorskip("pyproj")
    # compatibility for older PROJ versions using PROJJSON with datum ensembles
    # https://github.com/geopandas/geopandas/pull/2453
    crs = pyproj.CRS("EPSG:4326")
    crs_json = crs.to_json_dict()
    check_ensemble = False
    if "datum_ensemble" in crs_json:
        # older version of PROJ don't yet have datum ensembles
        check_ensemble = True
        assert "id" in crs_json["datum_ensemble"]["members"][0]
    _remove_id_from_member_of_ensembles(crs_json)
    if check_ensemble:
        assert "id" not in crs_json["datum_ensemble"]["members"][0]
    # ensure roundtrip still results in an equivalent CRS
    assert pyproj.CRS(crs_json) == crs


def test_write_metadata_invalid_spec_version(tmp_path):
    gdf = geopandas.GeoDataFrame(geometry=[box(0, 0, 10, 10)], crs="EPSG:4326")
    with pytest.raises(ValueError, match="schema_version must be one of"):
        _create_metadata(gdf, schema_version="invalid")

    with pytest.raises(
        ValueError,
        match="'geoarrow' encoding is only supported with schema version >= 1.1.0",
    ):
        gdf.to_parquet(tmp_path, schema_version="1.0.0", geometry_encoding="geoarrow")


def test_encode_metadata():
    metadata = {"a": "b"}

    expected = b'{"a": "b"}'
    assert _encode_metadata(metadata) == expected


def test_decode_metadata():
    metadata_str = b'{"a": "b"}'

    expected = {"a": "b"}
    assert _decode_metadata(metadata_str) == expected

    assert _decode_metadata(None) is None


def test_validate_dataframe(naturalearth_lowres):
    df = read_file(naturalearth_lowres)

    # valid: should not raise ValueError
    _validate_dataframe(df)
    _validate_dataframe(df.set_index("iso_a3"))

    # add column with non-string type
    df[0] = 1

    # invalid: should raise ValueError
    with pytest.raises(ValueError):
        _validate_dataframe(df)

    with pytest.raises(ValueError):
        _validate_dataframe(df.set_index(0))

    # not a DataFrame: should raise ValueError
    with pytest.raises(ValueError):
        _validate_dataframe("not a dataframe")


def test_validate_geo_metadata_valid():
    _validate_geo_metadata(
        {
            "primary_column": "geometry",
            "columns": {"geometry": {"crs": None, "encoding": "WKB"}},
            "schema_version": "0.1.0",
        }
    )

    _validate_geo_metadata(
        {
            "primary_column": "geometry",
            "columns": {"geometry": {"crs": None, "encoding": "WKB"}},
            "version": "<version>",
        }
    )

    _validate_geo_metadata(
        {
            "primary_column": "geometry",
            "columns": {
                "geometry": {
                    "crs": {
                        # truncated PROJJSON for testing, as PROJJSON contents
                        # not validated here
                        "id": {"authority": "EPSG", "code": 4326},
                    },
                    "encoding": "point",
                }
            },
            "version": "0.4.0",
        }
    )


@pytest.mark.parametrize(
    "metadata,error",
    [
        (None, "Missing or malformed geo metadata in Parquet/Feather file"),
        ({}, "Missing or malformed geo metadata in Parquet/Feather file"),
        # missing "version" key:
        (
            {"primary_column": "foo", "columns": None},
            "'geo' metadata in Parquet/Feather file is missing required key",
        ),
        # missing "columns" key:
        (
            {"primary_column": "foo", "version": "<version>"},
            "'geo' metadata in Parquet/Feather file is missing required key:",
        ),
        # missing "primary_column"
        (
            {"columns": [], "version": "<version>"},
            "'geo' metadata in Parquet/Feather file is missing required key:",
        ),
        (
            {"primary_column": "foo", "columns": [], "version": "<version>"},
            "'columns' in 'geo' metadata must be a dict",
        ),
        # missing "encoding" for column
        (
            {"primary_column": "foo", "columns": {"foo": {}}, "version": "<version>"},
            (
                "'geo' metadata in Parquet/Feather file is missing required key "
                "'encoding' for column 'foo'"
            ),
        ),
        # invalid column encoding
        (
            {
                "primary_column": "foo",
                "columns": {"foo": {"crs": None, "encoding": None}},
                "version": "<version>",
            },
            "Only WKB geometry encoding",
        ),
        (
            {
                "primary_column": "foo",
                "columns": {"foo": {"crs": None, "encoding": "BKW"}},
                "version": "<version>",
            },
            "Only WKB geometry encoding",
        ),
    ],
)
def test_validate_geo_metadata_invalid(metadata, error):
    with pytest.raises(ValueError, match=error):
        _validate_geo_metadata(metadata)


def test_validate_geo_metadata_edges():
    metadata = {
        "primary_column": "geometry",
        "columns": {"geometry": {"crs": None, "encoding": "WKB", "edges": "spherical"}},
        "version": "1.0.0-beta.1",
    }
    with pytest.warns(
        UserWarning,
        match="The geo metadata indicate that column 'geometry' has spherical edges",
    ):
        _validate_geo_metadata(metadata)


def test_to_parquet_fails_on_invalid_engine(tmpdir):
    df = GeoDataFrame(data=[[1, 2, 3]], columns=["a", "b", "a"], geometry=[Point(1, 1)])

    with pytest.raises(
        ValueError,
        match=(
            "GeoPandas only supports using pyarrow as the engine for "
            "to_parquet: 'fastparquet' passed instead."
        ),
    ):
        df.to_parquet(tmpdir / "test.parquet", engine="fastparquet")


@mock.patch("geopandas.io.arrow._to_parquet")
def test_to_parquet_does_not_pass_engine_along(mock_to_parquet):
    df = GeoDataFrame(data=[[1, 2, 3]], columns=["a", "b", "a"], geometry=[Point(1, 1)])
    df.to_parquet("", engine="pyarrow")
    # assert that engine keyword is not passed through to _to_parquet (and thus
    # parquet.write_table)
    mock_to_parquet.assert_called_with(
        df,
        "",
        compression="snappy",
        geometry_encoding="WKB",
        index=None,
        schema_version=None,
        write_covering_bbox=False,
    )


# TEMPORARY: used to determine if pyarrow fails for roundtripping pandas data
# without geometries
def test_pandas_parquet_roundtrip1(tmpdir):
    df = DataFrame({"a": [1, 2, 3], "b": ["a", "b", "c"]})

    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename)

    pq_df = pd_read_parquet(filename)

    assert_frame_equal(df, pq_df)


@pytest.mark.parametrize(
    "test_dataset", ["naturalearth_lowres", "naturalearth_cities", "nybb_filename"]
)
def test_pandas_parquet_roundtrip2(test_dataset, tmpdir, request):
    path = request.getfixturevalue(test_dataset)
    df = DataFrame(read_file(path).drop(columns=["geometry"]))

    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename)

    pq_df = pd_read_parquet(filename)

    assert_frame_equal(df, pq_df)


@pytest.mark.parametrize(
    "test_dataset", ["naturalearth_lowres", "naturalearth_cities", "nybb_filename"]
)
def test_roundtrip(tmpdir, file_format, test_dataset, request):
    """Writing to parquet should not raise errors, and should not alter original
    GeoDataFrame
    """
    path = request.getfixturevalue(test_dataset)
    reader, writer = file_format

    df = read_file(path)
    orig = df.copy()

    filename = os.path.join(str(tmpdir), "test.pq")

    writer(df, filename)

    assert os.path.exists(filename)

    # make sure that the original data frame is unaltered
    assert_geodataframe_equal(df, orig)

    # make sure that we can roundtrip the data frame
    pq_df = reader(filename)

    assert isinstance(pq_df, GeoDataFrame)
    assert_geodataframe_equal(df, pq_df)


@pytest.mark.parametrize(
    "test_dataset", ["naturalearth_lowres", "naturalearth_cities", "nybb_filename"]
)
def test_roundtrip_bytesio(file_format, test_dataset, request):
    """Should be able to roundtrip in BytesIO."""
    path = request.getfixturevalue(test_dataset)
    reader, writer = file_format

    df = read_file(path)
    orig = df.copy()

    buf = BytesIO()

    writer(df, buf)
    buf.seek(0)

    # make sure that the original data frame is unaltered
    assert_geodataframe_equal(df, orig)

    # make sure that we can roundtrip the data frame
    pq_df = reader(BytesIO(buf.getvalue()))

    assert isinstance(pq_df, GeoDataFrame)
    assert_geodataframe_equal(df, pq_df)


def test_index(tmpdir, file_format, naturalearth_lowres):
    """Setting index=`True` should preserve index in output, and
    setting index=`False` should drop index from output.
    """
    reader, writer = file_format

    df = read_file(naturalearth_lowres).set_index("iso_a3")

    filename = os.path.join(str(tmpdir), "test_with_index.pq")
    writer(df, filename, index=True)
    pq_df = reader(filename)
    assert_geodataframe_equal(df, pq_df)

    filename = os.path.join(str(tmpdir), "drop_index.pq")
    writer(df, filename, index=False)
    pq_df = reader(filename)
    assert_geodataframe_equal(df.reset_index(drop=True), pq_df)


def test_column_order(tmpdir, file_format, naturalearth_lowres):
    """The order of columns should be preserved in the output."""
    reader, writer = file_format

    df = read_file(naturalearth_lowres)
    df = df.set_index("iso_a3")
    df["geom2"] = df.geometry.representative_point()
    table = _geopandas_to_arrow(df)
    custom_column_order = [
        "iso_a3",
        "geom2",
        "pop_est",
        "continent",
        "name",
        "geometry",
        "gdp_md_est",
    ]
    table = table.select(custom_column_order)

    if reader is read_parquet:
        filename = os.path.join(str(tmpdir), "test_column_order.pq")
        pq.write_table(table, filename)
    else:
        filename = os.path.join(str(tmpdir), "test_column_order.feather")
        feather.write_feather(table, filename)

    result = reader(filename)
    assert list(result.columns) == custom_column_order[1:]
    assert_geodataframe_equal(result, df[custom_column_order[1:]])


@pytest.mark.parametrize(
    "compression", ["snappy", "gzip", "brotli", "lz4", "zstd", None]
)
def test_parquet_compression(compression, tmpdir, naturalearth_lowres):
    """Using compression options should not raise errors, and should
    return identical GeoDataFrame.
    """

    df = read_file(naturalearth_lowres)

    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, compression=compression)
    pq_df = read_parquet(filename)

    assert isinstance(pq_df, GeoDataFrame)
    assert_geodataframe_equal(df, pq_df)


@pytest.mark.parametrize("compression", ["uncompressed", "lz4", "zstd"])
def test_feather_compression(compression, tmpdir, naturalearth_lowres):
    """Using compression options should not raise errors, and should
    return identical GeoDataFrame.
    """

    df = read_file(naturalearth_lowres)

    filename = os.path.join(str(tmpdir), "test.feather")
    df.to_feather(filename, compression=compression)
    pq_df = read_feather(filename)

    assert isinstance(pq_df, GeoDataFrame)
    assert_geodataframe_equal(df, pq_df)


def test_parquet_multiple_geom_cols(tmpdir, file_format, naturalearth_lowres):
    """If multiple geometry columns are present when written to parquet,
    they should all be returned as such when read from parquet.
    """
    reader, writer = file_format

    df = read_file(naturalearth_lowres)
    df["geom2"] = df.geometry.copy()

    filename = os.path.join(str(tmpdir), "test.pq")
    writer(df, filename)

    assert os.path.exists(filename)

    pq_df = reader(filename)

    assert isinstance(pq_df, GeoDataFrame)
    assert_geodataframe_equal(df, pq_df)

    assert_geoseries_equal(df.geom2, pq_df.geom2, check_geom_type=True)


def test_parquet_missing_metadata(tmpdir, naturalearth_lowres):
    """Missing geo metadata, such as from a parquet file created
    from a pandas DataFrame, will raise a ValueError.
    """

    df = read_file(naturalearth_lowres)

    # convert to DataFrame
    df = DataFrame(df)

    # convert the geometry column so we can extract later
    df["geometry"] = to_wkb(df["geometry"].values)

    filename = os.path.join(str(tmpdir), "test.pq")

    # use pandas to_parquet (no geo metadata)
    df.to_parquet(filename)

    # missing metadata will raise ValueError
    with pytest.raises(
        ValueError, match="Missing geo metadata in Parquet/Feather file."
    ):
        read_parquet(filename)


def test_parquet_missing_metadata2(tmpdir):
    """Missing geo metadata, such as from a parquet file created
    from a pyarrow Table (which will also not contain pandas metadata),
    will raise a ValueError.
    """
    import pyarrow.parquet as pq

    table = pyarrow.table({"a": [1, 2, 3]})
    filename = os.path.join(str(tmpdir), "test.pq")

    # use pyarrow.parquet write_table (no geo metadata, but also no pandas metadata)
    pq.write_table(table, filename)

    # missing metadata will raise ValueError
    with pytest.raises(
        ValueError, match="Missing geo metadata in Parquet/Feather file."
    ):
        read_parquet(filename)


@pytest.mark.parametrize(
    "geo_meta,error",
    [
        ({"geo": b""}, "Missing or malformed geo metadata in Parquet/Feather file"),
        (
            {"geo": _encode_metadata({})},
            "Missing or malformed geo metadata in Parquet/Feather file",
        ),
        (
            {"geo": _encode_metadata({"foo": "bar"})},
            "'geo' metadata in Parquet/Feather file is missing required key",
        ),
    ],
)
def test_parquet_invalid_metadata(tmpdir, geo_meta, error, naturalearth_lowres):
    """Has geo metadata with missing required fields will raise a ValueError.

    This requires writing the parquet file directly below, so that we can
    control the metadata that is written for this test.
    """

    from pyarrow import Table, parquet

    df = read_file(naturalearth_lowres)

    # convert to DataFrame and encode geometry to WKB
    df = DataFrame(df)
    df["geometry"] = to_wkb(df["geometry"].values)

    table = Table.from_pandas(df)
    metadata = table.schema.metadata
    metadata.update(geo_meta)
    table = table.replace_schema_metadata(metadata)

    filename = os.path.join(str(tmpdir), "test.pq")
    parquet.write_table(table, filename)

    with pytest.raises(ValueError, match=error):
        read_parquet(filename)


def test_subset_columns(tmpdir, file_format, naturalearth_lowres):
    """Reading a subset of columns should correctly decode selected geometry
    columns.
    """
    reader, writer = file_format

    df = read_file(naturalearth_lowres)

    filename = os.path.join(str(tmpdir), "test.pq")
    writer(df, filename)
    pq_df = reader(filename, columns=["name", "geometry"])

    assert_geodataframe_equal(df[["name", "geometry"]], pq_df)

    with pytest.raises(
        ValueError, match="No geometry columns are included in the columns read"
    ):
        reader(filename, columns=["name"])


def test_promote_secondary_geometry(tmpdir, file_format, naturalearth_lowres):
    """Reading a subset of columns that does not include the primary geometry
    column should promote the first geometry column present.
    """
    reader, writer = file_format

    df = read_file(naturalearth_lowres)
    df["geom2"] = df.geometry.copy()

    filename = os.path.join(str(tmpdir), "test.pq")
    writer(df, filename)
    pq_df = reader(filename, columns=["name", "geom2"])

    assert_geodataframe_equal(df.set_geometry("geom2")[["name", "geom2"]], pq_df)

    df["geom3"] = df.geometry.copy()

    writer(df, filename)
    with pytest.warns(
        UserWarning,
        match="Multiple non-primary geometry columns read from Parquet/Feather file.",
    ):
        pq_df = reader(filename, columns=["name", "geom2", "geom3"])

    assert_geodataframe_equal(
        df.set_geometry("geom2")[["name", "geom2", "geom3"]], pq_df
    )


def test_columns_no_geometry(tmpdir, file_format, naturalearth_lowres):
    """Reading a parquet file that is missing all of the geometry columns
    should raise a ValueError"""
    reader, writer = file_format

    df = read_file(naturalearth_lowres)

    filename = os.path.join(str(tmpdir), "test.pq")
    writer(df, filename)

    with pytest.raises(ValueError):
        reader(filename, columns=["name"])


def test_missing_crs(tmpdir, file_format, naturalearth_lowres):
    """If CRS is `None`, it should be properly handled
    and remain `None` when read from parquet`.
    """
    reader, writer = file_format

    df = read_file(naturalearth_lowres)
    df.geometry.array.crs = None

    filename = os.path.join(str(tmpdir), "test.pq")
    writer(df, filename)
    pq_df = reader(filename)

    assert pq_df.crs is None

    assert_geodataframe_equal(df, pq_df, check_crs=True)


def test_default_geo_col_writes(tmp_path):
    # edge case geo col name None writes successfully
    df = GeoDataFrame({"a": [1, 2]})
    df.to_parquet(tmp_path / "test.pq")
    # cannot be round tripped as gdf due to invalid geom col
    pq_df = pd_read_parquet(tmp_path / "test.pq")
    assert_frame_equal(df, pq_df)


def test_fsspec_url(naturalearth_lowres):
    _ = pytest.importorskip("fsspec")
    import fsspec.implementations.memory

    class MyMemoryFileSystem(fsspec.implementations.memory.MemoryFileSystem):
        # Simple fsspec filesystem that adds a required keyword.
        # Attempting to use this filesystem without the keyword will raise an exception.
        def __init__(self, is_set, *args, **kwargs):
            self.is_set = is_set
            super().__init__(*args, **kwargs)

    fsspec.register_implementation("memory", MyMemoryFileSystem, clobber=True)
    memfs = MyMemoryFileSystem(is_set=True)

    df = read_file(naturalearth_lowres)

    with memfs.open("data.parquet", "wb") as f:
        df.to_parquet(f)

    result = read_parquet("memory://data.parquet", storage_options={"is_set": True})
    assert_geodataframe_equal(result, df)

    result = read_parquet("memory://data.parquet", filesystem=memfs)
    assert_geodataframe_equal(result, df)

    # reset fsspec registry
    fsspec.register_implementation(
        "memory", fsspec.implementations.memory.MemoryFileSystem, clobber=True
    )


def test_non_fsspec_url_with_storage_options_raises(naturalearth_lowres):
    with pytest.raises(ValueError, match="storage_options"):
        read_parquet(naturalearth_lowres, storage_options={"foo": "bar"})


def test_prefers_pyarrow_fs():
    filesystem, _ = _get_filesystem_path("file:///data.parquet")
    assert isinstance(filesystem, pyarrow.fs.LocalFileSystem)


def test_write_read_parquet_expand_user():
    gdf = geopandas.GeoDataFrame(geometry=[box(0, 0, 10, 10)], crs="epsg:4326")
    test_file = "~/test_file.parquet"
    gdf.to_parquet(test_file)
    pq_df = geopandas.read_parquet(test_file)
    assert_geodataframe_equal(gdf, pq_df, check_crs=True)
    os.remove(os.path.expanduser(test_file))


def test_write_read_feather_expand_user():
    gdf = geopandas.GeoDataFrame(geometry=[box(0, 0, 10, 10)], crs="epsg:4326")
    test_file = "~/test_file.feather"
    gdf.to_feather(test_file)
    f_df = geopandas.read_feather(test_file)
    assert_geodataframe_equal(gdf, f_df, check_crs=True)
    os.remove(os.path.expanduser(test_file))


@pytest.mark.parametrize("geometry", [[], [None]])
def test_write_empty_bbox(tmpdir, geometry):
    # empty dataframe or all missing geometries -> avoid bbox with NaNs
    gdf = geopandas.GeoDataFrame({"col": [1] * len(geometry)}, geometry=geometry)
    gdf.to_parquet(tmpdir / "test.parquet")

    from pyarrow.parquet import read_table

    table = read_table(tmpdir / "test.parquet")
    metadata = json.loads(table.schema.metadata[b"geo"])
    assert "encoding" in metadata["columns"]["geometry"]
    assert "bbox" not in metadata["columns"]["geometry"]


@pytest.mark.skipif(
    Version(pyarrow.__version__) < Version("19.0.0"),
    reason="This version of pyarrow does not support reading complex types",
)
@pytest.mark.parametrize("format", ["feather", "parquet"])
def test_write_read_to_pandas_kwargs(tmpdir, format):
    filename = os.path.join(str(tmpdir), f"test.{format}")

    # Use arrow types to ensure that we can assert the roundtrip was successful
    int_type = ArrowDtype(pyarrow.int64())
    str_type = ArrowDtype(pyarrow.string())
    complex_type = ArrowDtype(pyarrow.struct([pyarrow.field("foo", pyarrow.string())]))
    index = Index([0], dtype=ArrowDtype(pyarrow.int64()))

    gdf = geopandas.GeoDataFrame(
        {
            "geometry": [box(0, 0, 10, 10)],
            "i": Series([1], index=index, dtype=int_type),
            "s": Series(["a"], index=index, dtype=str_type),
            "c": Series([{"foo": "bar"}], index=index, dtype=complex_type),
        },
        index=index,
    )

    if format == "feather":
        gdf.to_feather(filename)
        read_func = read_feather
    else:
        gdf.to_parquet(filename)
        read_func = read_parquet

    # simulate the `dtype_backend="pyarrow"` option in `pandas.read_parquet`
    gdf_roundtrip = read_func(filename, to_pandas_kwargs={"types_mapper": ArrowDtype})
    assert isinstance(gdf_roundtrip, geopandas.GeoDataFrame)
    assert gdf_roundtrip.dtypes["i"] == int_type
    assert gdf_roundtrip.dtypes["s"] == str_type
    assert gdf_roundtrip.dtypes["c"] == complex_type
    assert_geodataframe_equal(gdf_roundtrip, gdf, check_dtype=True)


@pytest.mark.parametrize("format", ["feather", "parquet"])
def test_read_complex_type_with_numpy_backend_xfail(tmpdir, format):
    filename = os.path.join(str(tmpdir), f"test.{format}")
    complex_type = ArrowDtype(pyarrow.struct([pyarrow.field("foo", pyarrow.string())]))
    index = Index([0], dtype=ArrowDtype(pyarrow.int64()))
    gdf = geopandas.GeoDataFrame(
        {
            "geometry": [box(0, 0, 10, 10)],
            "c": Series([{"foo": "bar"}], index=index, dtype=complex_type),
        },
        index=index,
    )
    if format == "feather":
        gdf.to_feather(filename)
        read_func = read_feather
    else:
        gdf.to_parquet(filename)
        read_func = read_parquet
    # Note: due to bugs in pyarrow, we can't read complex types without using
    # the types mapper. This is a long standing pandas issue as noted here:
    # - https://github.com/pandas-dev/pandas/issues/53011
    # - https://github.com/apache/arrow/issues/39914
    match = re.escape("data type 'struct<foo: string>[pyarrow]' not understood")
    with pytest.raises(TypeError, match=match):
        read_func(filename)


@pytest.mark.parametrize("format", ["feather", "parquet"])
def test_write_read_default_crs(tmpdir, format):
    pyproj = pytest.importorskip("pyproj")
    if format == "feather":
        from pyarrow.feather import write_feather as write
    else:
        from pyarrow.parquet import write_table as write

    filename = os.path.join(str(tmpdir), f"test.{format}")
    gdf = geopandas.GeoDataFrame(geometry=[box(0, 0, 10, 10)])
    table = _geopandas_to_arrow(gdf)

    # update the geo metadata to strip 'crs' entry
    metadata = table.schema.metadata
    geo_metadata = _decode_metadata(metadata[b"geo"])
    del geo_metadata["columns"]["geometry"]["crs"]
    metadata.update({b"geo": _encode_metadata(geo_metadata)})
    table = table.replace_schema_metadata(metadata)

    write(table, filename)

    read = getattr(geopandas, f"read_{format}")
    df = read(filename)
    assert df.crs.equals(pyproj.CRS("OGC:CRS84"))


@pytest.mark.skipif(shapely.geos_version < (3, 10, 0), reason="requires GEOS>=3.10")
def test_write_iso_wkb(tmpdir):
    gdf = geopandas.GeoDataFrame(
        geometry=geopandas.GeoSeries.from_wkt(["POINT Z (1 2 3)"])
    )
    gdf.to_parquet(tmpdir / "test.parquet")

    from pyarrow.parquet import read_table

    table = read_table(tmpdir / "test.parquet")
    wkb = table["geometry"][0].as_py().hex()

    # correct ISO flavor
    assert wkb == "01e9030000000000000000f03f00000000000000400000000000000840"


@pytest.mark.skipif(shapely.geos_version >= (3, 10, 0), reason="tests GEOS<3.10")
def test_write_iso_wkb_old_geos(tmpdir):
    gdf = geopandas.GeoDataFrame(
        geometry=geopandas.GeoSeries.from_wkt(["POINT Z (1 2 3)"])
    )
    with pytest.raises(ValueError, match="Cannot write 3D"):
        gdf.to_parquet(tmpdir / "test.parquet")


@pytest.mark.parametrize(
    "format,schema_version",
    product(["feather", "parquet"], [None] + SUPPORTED_VERSIONS),
)
def test_write_spec_version(tmpdir, format, schema_version):
    if format == "feather":
        from pyarrow.feather import read_table
    else:
        from pyarrow.parquet import read_table

    filename = os.path.join(str(tmpdir), f"test.{format}")
    gdf = geopandas.GeoDataFrame(geometry=[box(0, 0, 10, 10)], crs="EPSG:4326")
    write = getattr(gdf, f"to_{format}")
    write(filename, schema_version=schema_version)

    # ensure that we can roundtrip data regardless of version
    read = getattr(geopandas, f"read_{format}")
    df = read(filename)
    assert_geodataframe_equal(df, gdf)

    # verify the correct version is written in the metadata
    schema_version = schema_version or METADATA_VERSION
    table = read_table(filename)
    metadata = json.loads(table.schema.metadata[b"geo"])
    assert metadata["version"] == schema_version

    # verify that CRS is correctly handled between versions
    if HAS_PYPROJ:
        if schema_version == "0.1.0":
            assert metadata["columns"]["geometry"]["crs"] == gdf.crs.to_wkt()

        else:
            crs_expected = gdf.crs.to_json_dict()
            _remove_id_from_member_of_ensembles(crs_expected)
            assert metadata["columns"]["geometry"]["crs"] == crs_expected

    # verify that geometry_type(s) is correctly handled between versions
    if Version(schema_version) <= Version("0.4.0"):
        assert "geometry_type" in metadata["columns"]["geometry"]
        assert metadata["columns"]["geometry"]["geometry_type"] == "Polygon"
    else:
        assert "geometry_types" in metadata["columns"]["geometry"]
        assert metadata["columns"]["geometry"]["geometry_types"] == ["Polygon"]


@pytest.mark.parametrize("version", ["0.1.0", "0.4.0", "1.0.0-beta.1"])
def test_read_versioned_file(version):
    """
    Verify that files for different metadata spec versions can be read
    created for each supported version:

    # small dummy test dataset (not naturalearth_lowres, as this can change over time)
    from shapely.geometry import box, MultiPolygon
    df = geopandas.GeoDataFrame(
        {"col_str": ["a", "b"], "col_int": [1, 2], "col_float": [0.1, 0.2]},
        geometry=[MultiPolygon([box(0, 0, 1, 1), box(2, 2, 3, 3)]), box(4, 4, 5,5)],
        crs="EPSG:4326",
    )
    df.to_feather(DATA_PATH / 'arrow' / f'test_data_v{METADATA_VERSION}.feather')
    df.to_parquet(DATA_PATH / 'arrow' / f'test_data_v{METADATA_VERSION}.parquet')
    """
    expected = geopandas.GeoDataFrame(
        {"col_str": ["a", "b"], "col_int": [1, 2], "col_float": [0.1, 0.2]},
        geometry=[MultiPolygon([box(0, 0, 1, 1), box(2, 2, 3, 3)]), box(4, 4, 5, 5)],
        crs="EPSG:4326",
    )

    df = geopandas.read_feather(DATA_PATH / "arrow" / f"test_data_v{version}.feather")
    assert_geodataframe_equal(df, expected, check_crs=True)

    df = geopandas.read_parquet(DATA_PATH / "arrow" / f"test_data_v{version}.parquet")
    assert_geodataframe_equal(df, expected, check_crs=True)


def test_read_gdal_files():
    """
    Verify that files written by GDAL can be read by geopandas.
    Since it is currently not yet straightforward to install GDAL with
    Parquet/Arrow enabled in our conda setup, we are testing with some
    generated files included in the repo (using GDAL 3.5.0):

    # small dummy test dataset (not naturalearth_lowres, as this can change over time)
    from shapely.geometry import box, MultiPolygon
    df = geopandas.GeoDataFrame(
        {"col_str": ["a", "b"], "col_int": [1, 2], "col_float": [0.1, 0.2]},
        geometry=[MultiPolygon([box(0, 0, 1, 1), box(2, 2, 3, 3)]), box(4, 4, 5,5)],
        crs="EPSG:4326",
    )
    df.to_file("test_data.gpkg", GEOMETRY_NAME="geometry")
    and then the gpkg file is converted to Parquet/Arrow with:
    $ ogr2ogr -f Parquet -lco FID= test_data_gdal350.parquet test_data.gpkg
    $ ogr2ogr -f Arrow -lco FID= -lco GEOMETRY_ENCODING=WKB test_data_gdal350.arrow test_data.gpkg

    Repeated for GDAL 3.9 which adds a bbox covering column:
    $ ogr2ogr -f Parquet -lco FID= test_data_gdal390.parquet test_data.gpkg
    """  # noqa: E501
    pytest.importorskip("pyproj")
    expected = geopandas.GeoDataFrame(
        {"col_str": ["a", "b"], "col_int": [1, 2], "col_float": [0.1, 0.2]},
        geometry=[MultiPolygon([box(0, 0, 1, 1), box(2, 2, 3, 3)]), box(4, 4, 5, 5)],
        crs="EPSG:4326",
    )

    df = geopandas.read_parquet(DATA_PATH / "arrow" / "test_data_gdal350.parquet")
    assert_geodataframe_equal(df, expected, check_crs=True)

    df = geopandas.read_feather(DATA_PATH / "arrow" / "test_data_gdal350.arrow")
    assert_geodataframe_equal(df, expected, check_crs=True)

    df = geopandas.read_parquet(DATA_PATH / "arrow" / "test_data_gdal390.parquet")
    # recent GDAL no longer writes CRS in metadata in case of EPSG:4326, so comes back
    # as default OGC:CRS84
    expected = expected.to_crs("OGC:CRS84")
    assert_geodataframe_equal(df, expected, check_crs=True)

    df = geopandas.read_parquet(
        DATA_PATH / "arrow" / "test_data_gdal390.parquet", bbox=(0, 0, 2, 2)
    )
    assert len(df) == 1


def test_parquet_read_partitioned_dataset(tmpdir, naturalearth_lowres):
    # we don't yet explicitly support this (in writing), but for Parquet it
    # works for reading (by relying on pyarrow.read_table)
    df = read_file(naturalearth_lowres)

    # manually create partitioned dataset
    basedir = tmpdir / "partitioned_dataset"
    basedir.mkdir()
    df[:100].to_parquet(basedir / "data1.parquet")
    df[100:].to_parquet(basedir / "data2.parquet")

    result = read_parquet(basedir)
    assert_geodataframe_equal(result, df)


def test_parquet_read_partitioned_dataset_fsspec(tmpdir, naturalearth_lowres):
    fsspec = pytest.importorskip("fsspec")

    df = read_file(naturalearth_lowres)

    # manually create partitioned dataset
    memfs = fsspec.filesystem("memory")
    memfs.mkdir("partitioned_dataset")
    with memfs.open("partitioned_dataset/data1.parquet", "wb") as f:
        df[:100].to_parquet(f)
    with memfs.open("partitioned_dataset/data2.parquet", "wb") as f:
        df[100:].to_parquet(f)

    result = read_parquet("memory://partitioned_dataset")
    assert_geodataframe_equal(result, df)


@pytest.mark.parametrize(
    "geometry_type",
    ["point", "linestring", "polygon", "multipoint", "multilinestring", "multipolygon"],
)
def test_read_parquet_geoarrow(geometry_type):
    result = geopandas.read_parquet(
        DATA_PATH
        / "arrow"
        / "geoparquet"
        / f"data-{geometry_type}-encoding_native.parquet"
    )
    expected = geopandas.read_parquet(
        DATA_PATH
        / "arrow"
        / "geoparquet"
        / f"data-{geometry_type}-encoding_wkb.parquet"
    )
    assert_geodataframe_equal(result, expected, check_crs=True)


@pytest.mark.parametrize(
    "geometry_type",
    ["point", "linestring", "polygon", "multipoint", "multilinestring", "multipolygon"],
)
def test_geoarrow_roundtrip(tmp_path, geometry_type):
    df = geopandas.read_parquet(
        DATA_PATH
        / "arrow"
        / "geoparquet"
        / f"data-{geometry_type}-encoding_wkb.parquet"
    )

    df.to_parquet(tmp_path / "test.parquet", geometry_encoding="geoarrow")
    result = geopandas.read_parquet(tmp_path / "test.parquet")
    assert_geodataframe_equal(result, df, check_crs=True)


def test_to_parquet_bbox_structure_and_metadata(tmpdir, naturalearth_lowres):
    # check metadata being written for covering.
    from pyarrow import parquet

    df = read_file(naturalearth_lowres)
    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, write_covering_bbox=True)

    table = parquet.read_table(filename)
    metadata = json.loads(table.schema.metadata[b"geo"].decode("utf-8"))
    assert metadata["columns"]["geometry"]["covering"] == {
        "bbox": {
            "xmin": ["bbox", "xmin"],
            "ymin": ["bbox", "ymin"],
            "xmax": ["bbox", "xmax"],
            "ymax": ["bbox", "ymax"],
        }
    }
    assert "bbox" in table.schema.names
    assert [field.name for field in table.schema.field("bbox").type] == [
        "xmin",
        "ymin",
        "xmax",
        "ymax",
    ]


@pytest.mark.parametrize(
    "geometry, expected_bbox",
    [
        (Point(1, 3), {"xmin": 1.0, "ymin": 3.0, "xmax": 1.0, "ymax": 3.0}),
        (
            LineString([(1, 1), (3, 3)]),
            {"xmin": 1.0, "ymin": 1.0, "xmax": 3.0, "ymax": 3.0},
        ),
        (
            Polygon([(2, 1), (1, 2), (2, 3), (3, 2)]),
            {"xmin": 1.0, "ymin": 1.0, "xmax": 3.0, "ymax": 3.0},
        ),
        (
            MultiPolygon([box(0, 0, 1, 1), box(2, 2, 3, 3), box(4, 4, 5, 5)]),
            {"xmin": 0.0, "ymin": 0.0, "xmax": 5.0, "ymax": 5.0},
        ),
    ],
    ids=["Point", "LineString", "Polygon", "Multipolygon"],
)
def test_to_parquet_bbox_values(tmpdir, geometry, expected_bbox):
    # check bbox bounds being written for different geometry types.
    import pyarrow.parquet as pq

    df = GeoDataFrame(data=[[1, 2]], columns=["a", "b"], geometry=[geometry])
    filename = os.path.join(str(tmpdir), "test.pq")

    df.to_parquet(filename, write_covering_bbox=True)

    result = pq.read_table(filename).to_pandas()
    assert result["bbox"][0] == expected_bbox


def test_read_parquet_bbox_single_point(tmpdir):
    # confirm that on a single point, bbox will pick it up.
    df = GeoDataFrame(data=[[1, 2]], columns=["a", "b"], geometry=[Point(1, 1)])
    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, write_covering_bbox=True)
    pq_df = read_parquet(filename, bbox=(1, 1, 1, 1))
    assert len(pq_df) == 1
    assert pq_df.geometry[0] == Point(1, 1)


@pytest.mark.parametrize("geometry_name", ["geometry", "custum_geom_col"])
def test_read_parquet_bbox(tmpdir, naturalearth_lowres, geometry_name):
    # check bbox is being used to filter results.
    df = read_file(naturalearth_lowres)
    if geometry_name != "geometry":
        df = df.rename_geometry(geometry_name)

    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, write_covering_bbox=True)

    pq_df = read_parquet(filename, bbox=(0, 0, 10, 10))

    assert pq_df["name"].values.tolist() == [
        "France",
        "Benin",
        "Nigeria",
        "Cameroon",
        "Togo",
        "Ghana",
        "Burkina Faso",
        "Gabon",
        "Eq. Guinea",
    ]


@pytest.mark.parametrize("geometry_name", ["geometry", "custum_geom_col"])
def test_read_parquet_bbox_partitioned(tmpdir, naturalearth_lowres, geometry_name):
    # check bbox is being used to filter results on partitioned data.
    df = read_file(naturalearth_lowres)
    if geometry_name != "geometry":
        df = df.rename_geometry(geometry_name)

    # manually create partitioned dataset
    basedir = tmpdir / "partitioned_dataset"
    basedir.mkdir()
    df[:100].to_parquet(basedir / "data1.parquet", write_covering_bbox=True)
    df[100:].to_parquet(basedir / "data2.parquet", write_covering_bbox=True)

    pq_df = read_parquet(basedir, bbox=(0, 0, 10, 10))

    assert pq_df["name"].values.tolist() == [
        "France",
        "Benin",
        "Nigeria",
        "Cameroon",
        "Togo",
        "Ghana",
        "Burkina Faso",
        "Gabon",
        "Eq. Guinea",
    ]


@pytest.mark.parametrize(
    "geometry, bbox",
    [
        (LineString([(1, 1), (3, 3)]), (1.5, 1.5, 3.5, 3.5)),
        (LineString([(1, 1), (3, 3)]), (3, 3, 3, 3)),
        (LineString([(1, 1), (3, 3)]), (1.5, 1.5, 2.5, 2.5)),
        (Polygon([(0, 0), (4, 0), (4, 4), (0, 4)]), (1, 1, 3, 3)),
        (Polygon([(0, 0), (4, 0), (4, 4), (0, 4)]), (1, 1, 5, 5)),
        (Polygon([(0, 0), (4, 0), (4, 4), (0, 4)]), (2, 2, 4, 4)),
        (Polygon([(0, 0), (4, 0), (4, 4), (0, 4)]), (4, 4, 4, 4)),
        (Polygon([(0, 0), (4, 0), (4, 4), (0, 4)]), (1, 1, 5, 3)),
    ],
)
def test_read_parquet_bbox_partial_overlap_of_geometry(tmpdir, geometry, bbox):
    df = GeoDataFrame(data=[[1, 2]], columns=["a", "b"], geometry=[geometry])
    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, write_covering_bbox=True)

    pq_df = read_parquet(filename, bbox=bbox)
    assert len(pq_df) == 1


def test_read_parquet_no_bbox(tmpdir, naturalearth_lowres):
    # check error message when parquet lacks a bbox column but
    # want to use bbox kwarg in read_parquet.
    df = read_file(naturalearth_lowres)
    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename)
    with pytest.raises(ValueError, match="Specifying 'bbox' not supported"):
        read_parquet(filename, bbox=(0, 0, 20, 20))


def test_read_parquet_no_bbox_partitioned(tmpdir, naturalearth_lowres):
    # check error message when partitioned parquet data does not have
    # a bbox column but want to use kwarg to read_parquet.
    df = read_file(naturalearth_lowres)

    # manually create partitioned dataset
    basedir = tmpdir / "partitioned_dataset"
    basedir.mkdir()
    df[:100].to_parquet(basedir / "data1.parquet")
    df[100:].to_parquet(basedir / "data2.parquet")

    with pytest.raises(ValueError, match="Specifying 'bbox' not supported"):
        read_parquet(basedir, bbox=(0, 0, 20, 20))


def test_convert_bbox_to_parquet_filter():
    # check conversion of bbox to parquet filter expression
    import pyarrow.compute as pc

    bbox = (0, 0, 25, 35)
    expected = ~(
        (pc.field(("bbox", "xmin")) > 25)
        | (pc.field(("bbox", "ymin")) > 35)
        | (pc.field(("bbox", "xmax")) < 0)
        | (pc.field(("bbox", "ymax")) < 0)
    )
    assert expected.equals(_convert_bbox_to_parquet_filter(bbox, "bbox"))


def test_read_parquet_bbox_column_default_behaviour(tmpdir, naturalearth_lowres):
    # check that bbox column is not read in by default

    df = read_file(naturalearth_lowres)
    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, write_covering_bbox=True)
    result1 = read_parquet(filename)
    assert "bbox" not in result1

    result2 = read_parquet(filename, columns=["name", "geometry"])
    assert "bbox" not in result2
    assert list(result2.columns) == ["name", "geometry"]


@pytest.mark.parametrize(
    "filters",
    [
        [("gdp_md_est", ">", 20000)],
        pc.field("gdp_md_est") > 20000,
    ],
)
def test_read_parquet_filters_and_bbox(tmpdir, naturalearth_lowres, filters):
    df = read_file(naturalearth_lowres)
    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, write_covering_bbox=True)

    result = read_parquet(filename, filters=filters, bbox=(0, 0, 20, 20))
    assert result["name"].values.tolist() == [
        "Dem. Rep. Congo",
        "France",
        "Nigeria",
        "Cameroon",
        "Ghana",
        "Algeria",
        "Libya",
    ]


@pytest.mark.parametrize(
    "filters",
    [
        ([("gdp_md_est", ">", 15000), ("gdp_md_est", "<", 16000)]),
        ((pc.field("gdp_md_est") > 15000) & (pc.field("gdp_md_est") < 16000)),
    ],
)
def test_read_parquet_filters_without_bbox(tmpdir, naturalearth_lowres, filters):
    df = read_file(naturalearth_lowres)
    filename = os.path.join(str(tmpdir), "test.pq")
    df.to_parquet(filename, write_covering_bbox=True)

    result = read_parquet(filename, filters=filters)
    assert result["name"].values.tolist() == ["Burkina Faso", "Mozambique", "Albania"]


def test_read_parquet_file_with_custom_bbox_encoding_fieldname(tmpdir):
    import pyarrow.parquet as pq

    data = {
        "name": ["point1", "point2", "point3"],
        "geometry": [Point(1, 1), Point(2, 2), Point(3, 3)],
    }
    df = GeoDataFrame(data)
    filename = os.path.join(str(tmpdir), "test.pq")

    table = _geopandas_to_arrow(
        df,
        schema_version="1.1.0",
        write_covering_bbox=True,
    )
    metadata = table.schema.metadata  # rename_columns results in wiping of metadata

    table = table.rename_columns(["name", "geometry", "custom_bbox_name"])

    geo_metadata = json.loads(metadata[b"geo"])
    geo_metadata["columns"]["geometry"]["covering"]["bbox"] = {
        "xmin": ["custom_bbox_name", "xmin"],
        "ymin": ["custom_bbox_name", "ymin"],
        "xmax": ["custom_bbox_name", "xmax"],
        "ymax": ["custom_bbox_name", "ymax"],
    }
    metadata.update({b"geo": _encode_metadata(geo_metadata)})

    table = table.replace_schema_metadata(metadata)
    pq.write_table(table, filename)

    pq_table = pq.read_table(filename)
    assert "custom_bbox_name" in pq_table.schema.names

    pq_df = read_parquet(filename, bbox=(1.5, 1.5, 2.5, 2.5))
    assert pq_df["name"].values.tolist() == ["point2"]


def test_to_parquet_with_existing_bbox_column(tmpdir, naturalearth_lowres):
    df = read_file(naturalearth_lowres)
    df = df.assign(bbox=[0] * len(df))
    filename = os.path.join(str(tmpdir), "test.pq")

    with pytest.raises(
        ValueError, match="An existing column 'bbox' already exists in the dataframe"
    ):
        df.to_parquet(filename, write_covering_bbox=True)


def test_read_parquet_bbox_points(tmp_path):
    # check bbox filtering on point geometries
    df = geopandas.GeoDataFrame(
        {"col": range(10)}, geometry=[Point(i, i) for i in range(10)]
    )
    df.to_parquet(tmp_path / "test.parquet", geometry_encoding="geoarrow")

    result = geopandas.read_parquet(tmp_path / "test.parquet", bbox=(0, 0, 10, 10))
    assert len(result) == 10
    result = geopandas.read_parquet(tmp_path / "test.parquet", bbox=(3, 3, 5, 5))
    assert len(result) == 3


def test_non_geo_parquet_read_with_proper_error(tmp_path):
    # https://github.com/geopandas/geopandas/issues/3556

    gdf = GeoDataFrame(
        {"col": [1, 2, 3]},
        geometry=geopandas.points_from_xy([1, 2, 3], [1, 2, 3]),
        crs="EPSG:4326",
    )
    del gdf["geometry"]
    # for the purposes of testing, force a situation where gdf will be written without
    # a geometry column (but with the other metadata)
    gdf.__class__ = GeoDataFrame

    gdf.to_parquet(tmp_path / "test_no_geometry.parquet")
    with pytest.raises(
        ValueError, match="No geometry columns are included in the columns read"
    ):
        geopandas.read_parquet(tmp_path / "test_no_geometry.parquet")