test_sql.py

# 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 os
from pathlib import Path
from typing import Any, Generator, cast

import pyarrow as pa
import pytest
from pydantic_core import ValidationError
from pytest_lazyfixture import lazy_fixture
from sqlalchemy import Engine, create_engine, inspect
from sqlalchemy.exc import ArgumentError, IntegrityError

from pyiceberg.catalog import (
    Catalog,
    load_catalog,
)
from pyiceberg.catalog.sql import (
    DEFAULT_ECHO_VALUE,
    DEFAULT_POOL_PRE_PING_VALUE,
    IcebergTables,
    SqlCatalog,
    SqlCatalogBaseTable,
)
from pyiceberg.exceptions import (
    CommitFailedException,
    NamespaceAlreadyExistsError,
    NamespaceNotEmptyError,
    NoSuchNamespaceError,
    NoSuchPropertyException,
    NoSuchTableError,
    TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import _dataframe_to_data_files, schema_to_pyarrow
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC
from pyiceberg.schema import Schema
from pyiceberg.table import TableProperties
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
    NullOrder,
    SortDirection,
    SortField,
    SortOrder,
)
from pyiceberg.transforms import IdentityTransform
from pyiceberg.typedef import Identifier
from pyiceberg.types import IntegerType, NestedField, StringType, strtobool

CATALOG_TABLES = [c.__tablename__ for c in SqlCatalogBaseTable.__subclasses__()]


@pytest.fixture(scope="module")
def catalog_name() -> str:
    return "test_sql_catalog"


@pytest.fixture(name="random_table_identifier")
def fixture_random_table_identifier(warehouse: Path, database_name: str, table_name: str) -> Identifier:
    os.makedirs(f"{warehouse}/{database_name}/{table_name}/metadata/", exist_ok=True)
    return database_name, table_name


@pytest.fixture(name="another_random_table_identifier")
def fixture_another_random_table_identifier(warehouse: Path, database_name: str, table_name: str) -> Identifier:
    database_name = database_name + "_new"
    table_name = table_name + "_new"
    os.makedirs(f"{warehouse}/{database_name}/{table_name}/metadata/", exist_ok=True)
    return database_name, table_name


@pytest.fixture(name="random_hierarchical_identifier")
def fixture_random_hierarchical_identifier(warehouse: Path, hierarchical_namespace_name: str, table_name: str) -> Identifier:
    os.makedirs(f"{warehouse}/{hierarchical_namespace_name}/{table_name}/metadata/", exist_ok=True)
    return Catalog.identifier_to_tuple(".".join((hierarchical_namespace_name, table_name)))


@pytest.fixture(name="another_random_hierarchical_identifier")
def fixture_another_random_hierarchical_identifier(
    warehouse: Path, hierarchical_namespace_name: str, table_name: str
) -> Identifier:
    hierarchical_namespace_name = hierarchical_namespace_name + "_new"
    table_name = table_name + "_new"
    os.makedirs(f"{warehouse}/{hierarchical_namespace_name}/{table_name}/metadata/", exist_ok=True)
    return Catalog.identifier_to_tuple(".".join((hierarchical_namespace_name, table_name)))


@pytest.fixture(scope="module")
def catalog_memory(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
    props = {
        "uri": "sqlite:///:memory:",
        "warehouse": f"file://{warehouse}",
    }
    catalog = SqlCatalog(catalog_name, **props)
    catalog.create_tables()
    yield catalog
    catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
    props = {
        "uri": f"sqlite:////{warehouse}/sql-catalog",
        "warehouse": f"file://{warehouse}",
    }
    catalog = SqlCatalog(catalog_name, **props)
    catalog.create_tables()
    yield catalog
    catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_uri(warehouse: Path) -> str:
    return f"sqlite:////{warehouse}/sql-catalog"


@pytest.fixture(scope="module")
def alchemy_engine(catalog_uri: str) -> Engine:
    return create_engine(catalog_uri)


@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
    props = {
        "uri": f"sqlite:////{warehouse}/sql-catalog",
        "warehouse": f"file://{warehouse}",
    }
    catalog = SqlCatalog(catalog_name, **props)
    catalog.engine.dialect.supports_sane_rowcount = False
    catalog.create_tables()
    yield catalog
    catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(catalog_name: str, warehouse: Path) -> Generator[SqlCatalog, None, None]:
    props = {
        "uri": f"sqlite:////{warehouse}/sql-catalog",
        "warehouse": f"file://{warehouse}",
        PY_IO_IMPL: FSSPEC_FILE_IO,
    }
    catalog = SqlCatalog(catalog_name, **props)
    catalog.create_tables()
    yield catalog
    catalog.destroy_tables()


def test_creation_with_no_uri(catalog_name: str) -> None:
    with pytest.raises(NoSuchPropertyException):
        SqlCatalog(catalog_name, not_uri="unused")


def test_creation_with_unsupported_uri(catalog_name: str) -> None:
    with pytest.raises(ArgumentError):
        SqlCatalog(catalog_name, uri="unsupported:xxx")


def test_creation_with_echo_parameter(catalog_name: str, warehouse: Path) -> None:
    # echo_param, expected_echo_value
    test_cases = [(None, strtobool(DEFAULT_ECHO_VALUE)), ("debug", "debug"), ("true", True), ("false", False)]

    for echo_param, expected_echo_value in test_cases:
        props = {
            "uri": f"sqlite:////{warehouse}/sql-catalog",
            "warehouse": f"file://{warehouse}",
        }
        # None is for default value
        if echo_param is not None:
            props["echo"] = echo_param
        catalog = SqlCatalog(catalog_name, **props)
        assert catalog.engine._echo == expected_echo_value, (
            f"Assertion failed: expected echo value {expected_echo_value}, "
            f"but got {catalog.engine._echo}. For echo_param={echo_param}"
        )


def test_creation_with_pool_pre_ping_parameter(catalog_name: str, warehouse: Path) -> None:
    # pool_pre_ping_param, expected_pool_pre_ping_value
    test_cases = [
        (None, strtobool(DEFAULT_POOL_PRE_PING_VALUE)),
        ("true", True),
        ("false", False),
    ]

    for pool_pre_ping_param, expected_pool_pre_ping_value in test_cases:
        props = {
            "uri": f"sqlite:////{warehouse}/sql-catalog",
            "warehouse": f"file://{warehouse}",
        }
        # None is for default value
        if pool_pre_ping_param is not None:
            props["pool_pre_ping"] = pool_pre_ping_param

        catalog = SqlCatalog(catalog_name, **props)
        assert catalog.engine.pool._pre_ping == expected_pool_pre_ping_value, (
            f"Assertion failed: expected pool_pre_ping value {expected_pool_pre_ping_value}, "
            f"but got {catalog.engine.pool._pre_ping}. For pool_pre_ping_param={pool_pre_ping_param}"
        )


def test_creation_from_impl(catalog_name: str, warehouse: Path) -> None:
    assert isinstance(
        load_catalog(
            catalog_name,
            **{
                "py-catalog-impl": "pyiceberg.catalog.sql.SqlCatalog",
                "uri": f"sqlite:////{warehouse}/sql-catalog",
                "warehouse": f"file://{warehouse}",
            },
        ),
        SqlCatalog,
    )


def confirm_no_tables_exist(alchemy_engine: Engine) -> None:
    inspector = inspect(alchemy_engine)
    for c in SqlCatalogBaseTable.__subclasses__():
        if inspector.has_table(c.__tablename__):
            c.__table__.drop(alchemy_engine)

    any_table_exists = any(t for t in inspector.get_table_names() if t in CATALOG_TABLES)
    if any_table_exists:
        pytest.raises(TableAlreadyExistsError, "Tables exist, but should not have been created yet")


def confirm_all_tables_exist(catalog: SqlCatalog) -> None:
    all_tables_exists = True
    for t in CATALOG_TABLES:
        if t not in inspect(catalog.engine).get_table_names():
            all_tables_exists = False

    assert isinstance(catalog, SqlCatalog), "Catalog should be a SQLCatalog"
    assert all_tables_exists, "Tables should have been created"


def load_catalog_for_catalog_table_creation(catalog_name: str, catalog_uri: str) -> SqlCatalog:
    catalog = load_catalog(
        catalog_name,
        type="sql",
        uri=catalog_uri,
        init_catalog_tables="true",
    )

    return cast(SqlCatalog, catalog)


def test_creation_when_no_tables_exist(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None:
    confirm_no_tables_exist(alchemy_engine)
    catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri)
    confirm_all_tables_exist(catalog)


def test_creation_when_one_tables_exists(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None:
    confirm_no_tables_exist(alchemy_engine)

    # Create one table
    inspector = inspect(alchemy_engine)
    IcebergTables.__table__.create(bind=alchemy_engine)
    assert IcebergTables.__tablename__ in [t for t in inspector.get_table_names() if t in CATALOG_TABLES]

    catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri)
    confirm_all_tables_exist(catalog)


def test_creation_when_all_tables_exists(alchemy_engine: Engine, catalog_name: str, catalog_uri: str) -> None:
    confirm_no_tables_exist(alchemy_engine)

    # Create all tables
    inspector = inspect(alchemy_engine)
    SqlCatalogBaseTable.metadata.create_all(bind=alchemy_engine)
    for c in CATALOG_TABLES:
        assert c in [t for t in inspector.get_table_names() if t in CATALOG_TABLES]

    catalog = load_catalog_for_catalog_table_creation(catalog_name=catalog_name, catalog_uri=catalog_uri)
    confirm_all_tables_exist(catalog)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_create_tables_idempotency(catalog: SqlCatalog) -> None:
    # Second initialization should not fail even if tables are already created
    catalog.create_tables()
    catalog.create_tables()


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_table_default_sort_order(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_nested)
    assert table.sort_order().order_id == 0, "Order ID must match"
    assert table.sort_order().is_unsorted is True, "Order must be unsorted"
    catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_v1_table(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_nested, properties={"format-version": "1"})
    assert table.sort_order().order_id == 0, "Order ID must match"
    assert table.sort_order().is_unsorted is True, "Order must be unsorted"
    assert table.format_version == 1
    assert table.spec() == UNPARTITIONED_PARTITION_SPEC
    catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_table_with_pyarrow_schema(
    catalog: SqlCatalog,
    pyarrow_schema_simple_without_ids: pa.Schema,
    iceberg_table_schema_simple: Schema,
    table_identifier: Identifier,
) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, pyarrow_schema_simple_without_ids)
    assert table.schema() == iceberg_table_schema_simple
    catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_write_pyarrow_schema(catalog: SqlCatalog, table_identifier: Identifier) -> None:
    import pyarrow as pa

    pyarrow_table = pa.Table.from_arrays(
        [
            pa.array([None, "A", "B", "C"]),  # 'foo' column
            pa.array([1, 2, 3, 4]),  # 'bar' column
            pa.array([True, None, False, True]),  # 'baz' column
            pa.array([None, "A", "B", "C"]),  # 'large' column
        ],
        schema=pa.schema(
            [
                pa.field("foo", pa.large_string(), nullable=True),
                pa.field("bar", pa.int32(), nullable=False),
                pa.field("baz", pa.bool_(), nullable=True),
                pa.field("large", pa.large_string(), nullable=True),
            ]
        ),
    )
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, pyarrow_table.schema)
    table.append(pyarrow_table)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_table_custom_sort_order(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    order = SortOrder(SortField(source_id=2, transform=IdentityTransform(), null_order=NullOrder.NULLS_FIRST))
    table = catalog.create_table(table_identifier, table_schema_nested, sort_order=order)
    given_sort_order = table.sort_order()
    assert given_sort_order.order_id == 1, "Order ID must match"
    assert len(given_sort_order.fields) == 1, "Order must have 1 field"
    assert given_sort_order.fields[0].direction == SortDirection.ASC, "Direction must match"
    assert given_sort_order.fields[0].null_order == NullOrder.NULLS_FIRST, "Null order must match"
    assert isinstance(given_sort_order.fields[0].transform, IdentityTransform), "Transform must match"
    catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_table_with_default_warehouse_location(
    warehouse: Path, catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier
) -> None:
    identifier_tuple = Catalog.identifier_to_tuple(table_identifier)
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    catalog.create_table(table_identifier, table_schema_nested)
    table = catalog.load_table(table_identifier)
    assert table.name() == identifier_tuple
    assert table.metadata_location.startswith(f"file://{warehouse}")
    assert os.path.exists(table.metadata_location[len("file://") :])
    catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_table_with_given_location_removes_trailing_slash(
    warehouse: Path, catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier
) -> None:
    identifier_tuple = Catalog.identifier_to_tuple(table_identifier)
    namespace = Catalog.namespace_from(table_identifier)
    table_name = Catalog.table_name_from(identifier_tuple)
    location = f"file://{warehouse}/{catalog.name}/{table_name}-given"
    catalog.create_namespace(namespace)
    catalog.create_table(table_identifier, table_schema_nested, location=f"{location}/")
    table = catalog.load_table(table_identifier)
    assert table.name() == identifier_tuple
    assert table.metadata_location.startswith(f"file://{warehouse}")
    assert os.path.exists(table.metadata_location[len("file://") :])
    assert table.location() == location
    catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_duplicated_table(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    catalog.create_table(table_identifier, table_schema_nested)
    with pytest.raises(TableAlreadyExistsError):
        catalog.create_table(table_identifier, table_schema_nested)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_create_table_if_not_exists_duplicated_table(
    catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier
) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table1 = catalog.create_table(table_identifier, table_schema_nested)
    table2 = catalog.create_table_if_not_exists(table_identifier, table_schema_nested)
    assert table1.name() == table2.name()


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_create_table_with_non_existing_namespace(catalog: SqlCatalog, table_schema_nested: Schema, table_name: str) -> None:
    identifier = ("invalid", table_name)
    with pytest.raises(NoSuchNamespaceError):
        catalog.create_table(identifier, table_schema_nested)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_create_table_without_namespace(catalog: SqlCatalog, table_schema_nested: Schema, table_name: str) -> None:
    with pytest.raises(NoSuchNamespaceError):
        catalog.create_table(table_name, table_schema_nested)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_register_table(catalog: SqlCatalog, table_identifier: Identifier, metadata_location: str) -> None:
    identifier_tuple = Catalog.identifier_to_tuple(table_identifier)
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.register_table(table_identifier, metadata_location)
    assert table.name() == identifier_tuple
    assert table.metadata_location == metadata_location
    assert os.path.exists(metadata_location)
    catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_register_existing_table(catalog: SqlCatalog, table_identifier: Identifier, metadata_location: str) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    catalog.register_table(table_identifier, metadata_location)
    with pytest.raises(TableAlreadyExistsError):
        catalog.register_table(table_identifier, metadata_location)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_register_table_with_non_existing_namespace(catalog: SqlCatalog, metadata_location: str, table_name: str) -> None:
    identifier = ("invalid", table_name)
    with pytest.raises(NoSuchNamespaceError):
        catalog.register_table(identifier, metadata_location)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_register_table_without_namespace(catalog: SqlCatalog, metadata_location: str, table_name: str) -> None:
    with pytest.raises(ValueError):
        catalog.register_table(table_name, metadata_location)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_load_table(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_nested)
    loaded_table = catalog.load_table(table_identifier)
    assert table.name() == loaded_table.name()
    assert table.metadata_location == loaded_table.metadata_location
    assert table.metadata == loaded_table.metadata


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_load_table_from_self_identifier(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    identifier_tuple = Catalog.identifier_to_tuple(table_identifier)
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_nested)
    intermediate = catalog.load_table(table_identifier)
    assert intermediate.name() == identifier_tuple
    loaded_table = catalog.load_table(intermediate.name())
    assert table.name() == loaded_table.name()
    assert table.metadata_location == loaded_table.metadata_location
    assert table.metadata == loaded_table.metadata


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_drop_table(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    identifier_tuple = Catalog.identifier_to_tuple(table_identifier)
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_nested)
    assert table.name() == identifier_tuple
    catalog.drop_table(table_identifier)
    with pytest.raises(NoSuchTableError):
        catalog.load_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_drop_table_from_self_identifier(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    identifier_tuple = Catalog.identifier_to_tuple(table_identifier)
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_nested)
    assert table.name() == identifier_tuple
    catalog.drop_table(table.name())
    with pytest.raises(NoSuchTableError):
        catalog.load_table(table.name())
    with pytest.raises(NoSuchTableError):
        catalog.load_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_drop_table_that_does_not_exist(catalog: SqlCatalog, table_identifier: Identifier) -> None:
    with pytest.raises(NoSuchTableError):
        catalog.drop_table(table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "from_table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
@pytest.mark.parametrize(
    "to_table_identifier",
    [
        lazy_fixture("another_random_table_identifier"),
        lazy_fixture("another_random_hierarchical_identifier"),
    ],
)
def test_rename_table(
    catalog: SqlCatalog, table_schema_nested: Schema, from_table_identifier: Identifier, to_table_identifier: Identifier
) -> None:
    from_namespace = Catalog.namespace_from(from_table_identifier)
    to_namespace = Catalog.namespace_from(to_table_identifier)
    catalog.create_namespace(from_namespace)
    catalog.create_namespace(to_namespace)
    table = catalog.create_table(from_table_identifier, table_schema_nested)
    assert table.name() == from_table_identifier
    catalog.rename_table(from_table_identifier, to_table_identifier)
    new_table = catalog.load_table(to_table_identifier)
    assert new_table.name() == to_table_identifier
    assert new_table.metadata_location == table.metadata_location
    with pytest.raises(NoSuchTableError):
        catalog.load_table(from_table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "from_table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
@pytest.mark.parametrize(
    "to_table_identifier",
    [
        lazy_fixture("another_random_table_identifier"),
        lazy_fixture("another_random_hierarchical_identifier"),
    ],
)
def test_rename_table_from_self_identifier(
    catalog: SqlCatalog, table_schema_nested: Schema, from_table_identifier: Identifier, to_table_identifier: Identifier
) -> None:
    from_namespace = Catalog.namespace_from(from_table_identifier)
    to_namespace = Catalog.namespace_from(to_table_identifier)
    catalog.create_namespace(from_namespace)
    catalog.create_namespace(to_namespace)
    table = catalog.create_table(from_table_identifier, table_schema_nested)
    assert table.name() == from_table_identifier
    catalog.rename_table(table.name(), to_table_identifier)
    new_table = catalog.load_table(to_table_identifier)
    assert new_table.name() == to_table_identifier
    assert new_table.metadata_location == table.metadata_location
    with pytest.raises(NoSuchTableError):
        catalog.load_table(table.name())
    with pytest.raises(NoSuchTableError):
        catalog.load_table(from_table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "from_table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
@pytest.mark.parametrize(
    "to_table_identifier",
    [
        lazy_fixture("another_random_table_identifier"),
        lazy_fixture("another_random_hierarchical_identifier"),
    ],
)
def test_rename_table_to_existing_one(
    catalog: SqlCatalog, table_schema_nested: Schema, from_table_identifier: Identifier, to_table_identifier: Identifier
) -> None:
    from_namespace = Catalog.namespace_from(from_table_identifier)
    to_namespace = Catalog.namespace_from(to_table_identifier)
    catalog.create_namespace(from_namespace)
    catalog.create_namespace(to_namespace)
    table = catalog.create_table(from_table_identifier, table_schema_nested)
    assert table.name() == from_table_identifier
    new_table = catalog.create_table(to_table_identifier, table_schema_nested)
    assert new_table.name() == to_table_identifier
    with pytest.raises(TableAlreadyExistsError):
        catalog.rename_table(from_table_identifier, to_table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "from_table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
@pytest.mark.parametrize(
    "to_table_identifier",
    [
        lazy_fixture("another_random_table_identifier"),
        lazy_fixture("another_random_hierarchical_identifier"),
    ],
)
def test_rename_missing_table(catalog: SqlCatalog, from_table_identifier: Identifier, to_table_identifier: Identifier) -> None:
    to_namespace = Catalog.namespace_from(to_table_identifier)
    catalog.create_namespace(to_namespace)
    with pytest.raises(NoSuchTableError):
        catalog.rename_table(from_table_identifier, to_table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "from_table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
@pytest.mark.parametrize(
    "to_table_identifier",
    [
        lazy_fixture("another_random_table_identifier"),
        lazy_fixture("another_random_hierarchical_identifier"),
    ],
)
def test_rename_table_to_missing_namespace(
    catalog: SqlCatalog, table_schema_nested: Schema, from_table_identifier: Identifier, to_table_identifier: Identifier
) -> None:
    from_namespace = Catalog.namespace_from(from_table_identifier)
    catalog.create_namespace(from_namespace)
    table = catalog.create_table(from_table_identifier, table_schema_nested)
    assert table.name() == from_table_identifier
    with pytest.raises(NoSuchNamespaceError):
        catalog.rename_table(from_table_identifier, to_table_identifier)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier_1",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier_2",
    [
        lazy_fixture("another_random_table_identifier"),
        lazy_fixture("another_random_hierarchical_identifier"),
    ],
)
def test_list_tables(
    catalog: SqlCatalog, table_schema_nested: Schema, table_identifier_1: Identifier, table_identifier_2: Identifier
) -> None:
    namespace_1 = Catalog.namespace_from(table_identifier_1)
    namespace_2 = Catalog.namespace_from(table_identifier_2)
    catalog.create_namespace(namespace_1)
    catalog.create_namespace(namespace_2)
    catalog.create_table(table_identifier_1, table_schema_nested)
    catalog.create_table(table_identifier_2, table_schema_nested)
    identifier_list = catalog.list_tables(namespace_1)
    assert len(identifier_list) == 1
    assert table_identifier_1 in identifier_list

    identifier_list = catalog.list_tables(namespace_2)
    assert len(identifier_list) == 1
    assert table_identifier_2 in identifier_list


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_list_tables_when_missing_namespace(catalog: SqlCatalog, namespace: str) -> None:
    with pytest.raises(NoSuchNamespaceError):
        catalog.list_tables(namespace)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_create_namespace_if_not_exists(catalog: SqlCatalog, database_name: str) -> None:
    catalog.create_namespace(database_name)
    assert (database_name,) in catalog.list_namespaces()
    catalog.create_namespace_if_not_exists(database_name)
    assert (database_name,) in catalog.list_namespaces()


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_create_namespace(catalog: SqlCatalog, namespace: str) -> None:
    catalog.create_namespace(namespace)
    assert (Catalog.identifier_to_tuple(namespace)[:1]) in catalog.list_namespaces()


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_create_duplicate_namespace(catalog: SqlCatalog, namespace: str) -> None:
    catalog.create_namespace(namespace)
    with pytest.raises(NamespaceAlreadyExistsError):
        catalog.create_namespace(namespace)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_create_namespaces_sharing_same_prefix(catalog: SqlCatalog, namespace: str) -> None:
    catalog.create_namespace(namespace + "_1")
    # Second namespace is a prefix of the first one, make sure it can be added.
    catalog.create_namespace(namespace)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_create_namespace_with_comment_and_location(catalog: SqlCatalog, namespace: str) -> None:
    test_location = "/test/location"
    test_properties = {
        "comment": "this is a test description",
        "location": test_location,
    }
    catalog.create_namespace(namespace=namespace, properties=test_properties)
    loaded_database_list = catalog.list_namespaces()
    assert Catalog.identifier_to_tuple(namespace)[:1] in loaded_database_list
    properties = catalog.load_namespace_properties(namespace)
    assert properties["comment"] == "this is a test description"
    assert properties["location"] == test_location


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
@pytest.mark.filterwarnings("ignore")
def test_create_namespace_with_null_properties(catalog: SqlCatalog, namespace: str) -> None:
    with pytest.raises(IntegrityError):
        catalog.create_namespace(namespace=namespace, properties={None: "value"})  # type: ignore

    with pytest.raises(IntegrityError):
        catalog.create_namespace(namespace=namespace, properties={"key": None})


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("empty_namespace", ["", (), (""), ("", ""), " ", (" ")])
def test_create_namespace_with_empty_identifier(catalog: SqlCatalog, empty_namespace: Any) -> None:
    with pytest.raises(NoSuchNamespaceError):
        catalog.create_namespace(empty_namespace)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_namespace_exists(catalog: SqlCatalog) -> None:
    for ns in [("db1",), ("db1", "ns1"), ("db2", "ns1"), ("db3", "ns1", "ns2")]:
        catalog.create_namespace(ns)
        assert catalog._namespace_exists(ns)

    assert catalog._namespace_exists("db2")  # `db2` exists because `db2.ns1` exists
    assert catalog._namespace_exists("db3.ns1")  # `db3.ns1` exists because `db3.ns1.ns2` exists
    assert not catalog._namespace_exists("db_")  # make sure '_' is escaped in the query
    assert not catalog._namespace_exists("db%")  # make sure '%' is escaped in the query


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_list_namespaces(catalog: SqlCatalog) -> None:
    namespace_list = ["db", "db.ns1", "db.ns1.ns2", "db.ns2", "db2", "db2.ns1", "db%"]
    for namespace in namespace_list:
        if not catalog._namespace_exists(namespace):
            catalog.create_namespace(namespace)

    ns_list = catalog.list_namespaces()
    for ns in [("db",), ("db%",), ("db2",)]:
        assert ns in ns_list

    ns_list = catalog.list_namespaces("db")
    assert sorted(ns_list) == [("db", "ns1"), ("db", "ns2")]

    ns_list = catalog.list_namespaces("db.ns1")
    assert sorted(ns_list) == [("db", "ns1", "ns2")]

    ns_list = catalog.list_namespaces("db.ns1.ns2")
    assert len(ns_list) == 0


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_list_namespaces_fuzzy_match(catalog: SqlCatalog) -> None:
    namespace_list = ["db.ns1", "db.ns1.ns2", "db.ns2", "db.ns1X.ns3", "db_.ns1.ns2", "db2.ns1.ns2"]
    for namespace in namespace_list:
        if not catalog._namespace_exists(namespace):
            catalog.create_namespace(namespace)

    assert catalog.list_namespaces("db.ns1") == [("db", "ns1", "ns2")]

    assert catalog.list_namespaces("db_.ns1") == [("db_", "ns1", "ns2")]


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_list_non_existing_namespaces(catalog: SqlCatalog) -> None:
    with pytest.raises(NoSuchNamespaceError):
        catalog.list_namespaces("does_not_exist")


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_drop_namespace(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    assert catalog._namespace_exists(namespace)
    catalog.create_table(table_identifier, table_schema_nested)
    with pytest.raises(NamespaceNotEmptyError):
        catalog.drop_namespace(namespace)
    catalog.drop_table(table_identifier)
    catalog.drop_namespace(namespace)
    assert not catalog._namespace_exists(namespace)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_drop_non_existing_namespaces(catalog: SqlCatalog) -> None:
    with pytest.raises(NoSuchNamespaceError):
        catalog.drop_namespace("does_not_exist")


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_load_namespace_properties(catalog: SqlCatalog, namespace: str) -> None:
    warehouse_location = "/test/location"
    test_properties = {
        "comment": "this is a test description",
        "location": f"{warehouse_location}/{namespace}",
        "test_property1": "1",
        "test_property2": "2",
        "test_property3": "3",
    }

    catalog.create_namespace(namespace, test_properties)
    listed_properties = catalog.load_namespace_properties(namespace)
    for k, v in listed_properties.items():
        assert k in test_properties
        assert v == test_properties[k]


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_load_empty_namespace_properties(catalog: SqlCatalog, namespace: str) -> None:
    catalog.create_namespace(namespace)
    listed_properties = catalog.load_namespace_properties(namespace)
    assert listed_properties == {"exists": "true"}


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
def test_load_namespace_properties_non_existing_namespace(catalog: SqlCatalog) -> None:
    with pytest.raises(NoSuchNamespaceError):
        catalog.load_namespace_properties("does_not_exist")


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("namespace", [lazy_fixture("database_name"), lazy_fixture("hierarchical_namespace_name")])
def test_update_namespace_properties(catalog: SqlCatalog, namespace: str) -> None:
    warehouse_location = "/test/location"
    test_properties = {
        "comment": "this is a test description",
        "location": f"{warehouse_location}/{namespace}",
        "test_property1": "1",
        "test_property2": "2",
        "test_property3": "3",
    }
    removals = {"test_property1", "test_property2", "test_property3", "should_not_removed"}
    updates = {"test_property4": "4", "test_property5": "5", "comment": "updated test description"}
    catalog.create_namespace(namespace, test_properties)
    update_report = catalog.update_namespace_properties(namespace, removals, updates)
    for k in updates.keys():
        assert k in update_report.updated
    for k in removals:
        if k == "should_not_removed":
            assert k in update_report.missing
        else:
            assert k in update_report.removed
    assert catalog.load_namespace_properties(namespace) == {
        "comment": "updated test description",
        "test_property4": "4",
        "test_property5": "5",
        "location": f"{warehouse_location}/{namespace}",
    }


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_nested)
    last_updated_ms = table.metadata.last_updated_ms
    original_table_metadata_location = table.metadata_location
    original_table_last_updated_ms = table.metadata.last_updated_ms

    assert catalog._parse_metadata_version(table.metadata_location) == 0
    assert table.metadata.current_schema_id == 0

    transaction = table.transaction()
    update = transaction.update_schema()
    update.add_column(path="b", field_type=IntegerType())
    update.commit()
    transaction.commit_transaction()

    updated_table_metadata = table.metadata

    assert catalog._parse_metadata_version(table.metadata_location) == 1
    assert updated_table_metadata.current_schema_id == 1
    assert len(updated_table_metadata.schemas) == 2
    new_schema = next(schema for schema in updated_table_metadata.schemas if schema.schema_id == 1)
    assert new_schema
    assert new_schema == update._apply()
    assert new_schema.find_field("b").field_type == IntegerType()
    assert updated_table_metadata.last_updated_ms > last_updated_ms
    assert len(updated_table_metadata.metadata_log) == 1
    assert updated_table_metadata.metadata_log[0].metadata_file == original_table_metadata_location
    assert updated_table_metadata.metadata_log[0].timestamp_ms == original_table_last_updated_ms


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
        lazy_fixture("catalog_sqlite_fsspec"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(table_identifier, table_schema_simple)

    df = pa.Table.from_pydict(
        {
            "foo": ["a"],
            "bar": [1],
            "baz": [True],
        },
        schema=schema_to_pyarrow(table_schema_simple),
    )

    table.append(df)

    # new snapshot is written in APPEND mode
    assert len(table.metadata.snapshots) == 1
    assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
    assert table.metadata.snapshots[0].parent_snapshot_id is None
    assert table.metadata.snapshots[0].sequence_number == 1
    assert table.metadata.snapshots[0].summary is not None
    assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
    assert table.metadata.snapshots[0].summary["added-data-files"] == "1"
    assert table.metadata.snapshots[0].summary["added-records"] == "1"
    assert table.metadata.snapshots[0].summary["total-data-files"] == "1"
    assert table.metadata.snapshots[0].summary["total-records"] == "1"
    assert len(table.metadata.metadata_log) == 1

    # read back the data
    assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_concurrent_commit_table(catalog: SqlCatalog, table_schema_simple: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    table_a = catalog.create_table(table_identifier, table_schema_simple)
    table_b = catalog.load_table(table_identifier)

    with table_a.update_schema() as update:
        update.add_column(path="b", field_type=IntegerType())

    with pytest.raises(CommitFailedException, match="Requirement failed: current schema id has changed: expected 0, found 1"):
        # This one should fail since it already has been updated
        with table_b.update_schema() as update:
            update.add_column(path="c", field_type=IntegerType())


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize("format_version", [1, 2])
def test_write_and_evolve(catalog: SqlCatalog, format_version: int) -> None:
    identifier = f"default.arrow_write_data_and_evolve_schema_v{format_version}"

    try:
        catalog.create_namespace("default")
    except NamespaceAlreadyExistsError:
        pass

    try:
        catalog.drop_table(identifier=identifier)
    except NoSuchTableError:
        pass

    pa_table = pa.Table.from_pydict(
        {
            "foo": ["a", None, "z"],
        },
        schema=pa.schema([pa.field("foo", pa.large_string(), nullable=True)]),
    )

    tbl = catalog.create_table(identifier=identifier, schema=pa_table.schema, properties={"format-version": str(format_version)})

    pa_table_with_column = pa.Table.from_pydict(
        {
            "foo": ["a", None, "z"],
            "bar": [19, None, 25],
        },
        schema=pa.schema(
            [
                pa.field("foo", pa.large_string(), nullable=True),
                pa.field("bar", pa.int32(), nullable=True),
            ]
        ),
    )

    with tbl.transaction() as txn:
        with txn.update_schema() as schema_txn:
            schema_txn.union_by_name(pa_table_with_column.schema)

        txn.append(pa_table_with_column)
        txn.overwrite(pa_table_with_column)
        txn.delete("foo = 'a'")


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize("format_version", [1, 2])
def test_create_table_transaction(catalog: SqlCatalog, format_version: int) -> None:
    identifier = f"default.arrow_create_table_transaction_{catalog.name}_{format_version}"
    try:
        catalog.create_namespace("default")
    except NamespaceAlreadyExistsError:
        pass

    try:
        catalog.drop_table(identifier=identifier)
    except NoSuchTableError:
        pass

    pa_table = pa.Table.from_pydict(
        {
            "foo": ["a", None, "z"],
        },
        schema=pa.schema([pa.field("foo", pa.large_string(), nullable=True)]),
    )

    pa_table_with_column = pa.Table.from_pydict(
        {
            "foo": ["a", None, "z"],
            "bar": [19, None, 25],
        },
        schema=pa.schema(
            [
                pa.field("foo", pa.large_string(), nullable=True),
                pa.field("bar", pa.int32(), nullable=True),
            ]
        ),
    )

    with catalog.create_table_transaction(
        identifier=identifier, schema=pa_table.schema, properties={"format-version": str(format_version)}
    ) as txn:
        with txn.update_snapshot().fast_append() as snapshot_update:
            for data_file in _dataframe_to_data_files(table_metadata=txn.table_metadata, df=pa_table, io=txn._table.io):
                snapshot_update.append_data_file(data_file)

        with txn.update_schema() as schema_txn:
            schema_txn.union_by_name(pa_table_with_column.schema)

        with txn.update_snapshot().fast_append() as snapshot_update:
            for data_file in _dataframe_to_data_files(
                table_metadata=txn.table_metadata, df=pa_table_with_column, io=txn._table.io
            ):
                snapshot_update.append_data_file(data_file)

    tbl = catalog.load_table(identifier=identifier)
    assert tbl.format_version == format_version
    assert len(tbl.scan().to_arrow()) == 6


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_table_properties_int_value(catalog: SqlCatalog, table_schema_simple: Schema, table_identifier: Identifier) -> None:
    # table properties can be set to int, but still serialized to string
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    property_with_int = {"property_name": 42}
    table = catalog.create_table(table_identifier, table_schema_simple, properties=property_with_int)
    assert isinstance(table.properties["property_name"], str)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_table_properties_raise_for_none_value(
    catalog: SqlCatalog, table_schema_simple: Schema, table_identifier: Identifier
) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    property_with_none = {"property_name": None}
    with pytest.raises(ValidationError) as exc_info:
        _ = catalog.create_table(table_identifier, table_schema_simple, properties=property_with_none)
    assert "None type is not a supported value in properties: property_name" in str(exc_info.value)


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize(
    "table_identifier",
    [
        lazy_fixture("random_table_identifier"),
        lazy_fixture("random_hierarchical_identifier"),
    ],
)
def test_table_exists(catalog: SqlCatalog, table_schema_simple: Schema, table_identifier: Identifier) -> None:
    namespace = Catalog.namespace_from(table_identifier)
    catalog.create_namespace(namespace)
    catalog.create_table(table_identifier, table_schema_simple, properties={"format-version": "2"})
    existing_table = table_identifier
    # Act and Assert for an existing table
    assert catalog.table_exists(existing_table) is True

    # Act and Assert for a non-existing table
    assert catalog.table_exists(("non", "exist")) is False


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
    ],
)
@pytest.mark.parametrize("format_version", [1, 2])
def test_merge_manifests_local_file_system(catalog: SqlCatalog, arrow_table_with_null: pa.Table, format_version: int) -> None:
    # To catch manifest file name collision bug during merge:
    # https://github.com/apache/iceberg-python/pull/363#discussion_r1660691918
    catalog.create_namespace_if_not_exists("default")
    try:
        catalog.drop_table("default.test_merge_manifest")
    except NoSuchTableError:
        pass
    tbl = catalog.create_table(
        "default.test_merge_manifest",
        arrow_table_with_null.schema,
        properties={
            "commit.manifest-merge.enabled": "true",
            "commit.manifest.min-count-to-merge": "2",
            "format-version": format_version,
        },
    )

    for _ in range(5):
        tbl.append(arrow_table_with_null)

    assert len(tbl.scan().to_arrow()) == 5 * len(arrow_table_with_null)
    current_snapshot = tbl.current_snapshot()
    assert current_snapshot
    manifests = current_snapshot.manifests(tbl.io)
    assert len(manifests) == 1


@pytest.mark.parametrize(
    "catalog",
    [
        lazy_fixture("catalog_memory"),
        lazy_fixture("catalog_sqlite"),
        lazy_fixture("catalog_sqlite_without_rowcount"),
    ],
)
def test_delete_metadata_multiple(catalog: SqlCatalog, table_schema_nested: Schema, random_table_identifier: str) -> None:
    namespace = Catalog.namespace_from(random_table_identifier)
    catalog.create_namespace(namespace)
    table = catalog.create_table(random_table_identifier, table_schema_nested)

    original_metadata_location = table.metadata_location

    for i in range(5):
        with table.transaction() as transaction:
            with transaction.update_schema() as update:
                update.add_column(path=f"new_column_{i}", field_type=IntegerType())

    assert len(table.metadata.metadata_log) == 5
    assert os.path.exists(original_metadata_location[len("file://") :])

    # Set the max versions property to 2, and delete after commit
    new_property = {
        TableProperties.METADATA_PREVIOUS_VERSIONS_MAX: "2",
        TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED: "true",
    }

    with table.transaction() as transaction:
        transaction.set_properties(properties=new_property)

    # Verify that only the most recent metadata files are kept
    assert len(table.metadata.metadata_log) == 2
    updated_metadata_1, updated_metadata_2 = table.metadata.metadata_log

    # new metadata log was added, so earlier metadata logs are removed.
    with table.transaction() as transaction:
        with transaction.update_schema() as update:
            update.add_column(path="new_column_x", field_type=IntegerType())

    assert len(table.metadata.metadata_log) == 2
    assert not os.path.exists(original_metadata_location[len("file://") :])
    assert not os.path.exists(updated_metadata_1.metadata_file[len("file://") :])
    assert os.path.exists(updated_metadata_2.metadata_file[len("file://") :])


class TestSqlCatalogClose:
    """Test SqlCatalog close functionality."""

    def test_sql_catalog_close(self, catalog_sqlite: SqlCatalog) -> None:
        """Test that SqlCatalog close method properly disposes the engine."""
        # Verify engine exists
        assert hasattr(catalog_sqlite, "engine")

        # Close the catalog
        catalog_sqlite.close()

        # Verify engine is disposed by checking that the engine still exists
        assert hasattr(catalog_sqlite, "engine")

    def test_sql_catalog_context_manager(self, warehouse: Path) -> None:
        """Test that SqlCatalog works as a context manager."""
        with SqlCatalog("test", uri="sqlite:///:memory:", warehouse=str(warehouse)) as catalog:
            # Verify engine exists
            assert hasattr(catalog, "engine")

            # Create a namespace and table to test functionality
            catalog.create_namespace("test_db")
            schema = Schema(NestedField(1, "name", StringType(), required=True))
            catalog.create_table(("test_db", "test_table"), schema)

        # Verify engine is disposed after exiting context
        assert hasattr(catalog, "engine")

    def test_sql_catalog_context_manager_with_exception(self) -> None:
        """Test that SqlCatalog context manager properly closes even with exceptions."""
        catalog = None
        try:
            with SqlCatalog("test", uri="sqlite:///:memory:") as cat:
                catalog = cat
                # Verify engine exists
                assert hasattr(catalog, "engine")
                raise ValueError("Test exception")
        except ValueError:
            pass

        # Verify engine is disposed even after exception
        assert catalog is not None
        assert hasattr(catalog, "engine")

    def test_sql_catalog_multiple_close_calls(self, catalog_sqlite: SqlCatalog) -> None:
        """Test that multiple close calls on SqlCatalog are safe."""
        # First close
        catalog_sqlite.close()

        # Second close should not raise an exception
        catalog_sqlite.close()