Add Trino integration tests and update dependencies

- Introduced new tests for registering tables and schema existence in Trino.
- Added Trino connection fixtures to support integration testing.
- Updated `pyproject.toml` to include the Trino dependency.
- Enhanced the Makefile with a new command to run Trino integration tests.

Author: dingo4dev

Makefile

@@ -103,6 +103,10 @@ test-integration-rebuild: ## Rebuild integration Docker services from scratch
 	docker compose -f dev/docker-compose-integration.yml rm -f
 	docker compose -f dev/docker-compose-integration.yml build --no-cache
 
+test-integration-trino:
+	sh ./dev/run-trino.sh
+	$(TEST_RUNNER) pytest tests/ -m integration_trino $(PYTEST_ARGS)
+
 test-s3: ## Run tests marked with @pytest.mark.s3
 	sh ./dev/run-minio.sh
 	$(TEST_RUNNER) pytest tests/ -m s3 $(PYTEST_ARGS)

dev/run-trino.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+#
+# 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.
+#
+
+set -ex
+
+if [ $(docker ps -q --filter "name=pyiceberg-trino" --filter "status=running" ) ]; then
+    echo "Trino service running"
+else
+    docker compose -f dev/docker-compose-trino.yml kill
+    docker compose -f dev/docker-compose-trino.yml up -d
+    while [ -z $(docker ps -q --filter "name=pyiceberg-trino" --filter "status=running" ) ]
+    do
+      echo "Waiting for Trino"
+      sleep 1
+    done
+fi

poetry.lock

@@ -104,6 +104,7 @@ deptry = ">=0.14,<0.24"
 docutils = "!=0.21.post1"   # https://github.com/python-poetry/poetry/issues/9248#issuecomment-2026240520
 mypy-boto3-glue = ">=1.28.18"
 mypy-boto3-dynamodb = ">=1.28.18"
+trino = {extras = ["sqlalchemy"], version = "^0.335.0"}
 
 [tool.poetry.group.docs.dependencies]
 # for mkdocs
@@ -326,6 +327,7 @@ markers = [
   "s3: marks a test as requiring access to s3 compliant storage (use with --aws-access-key-id, --aws-secret-access-key, and --endpoint args)",
   "adls: marks a test as requiring access to adls compliant storage (use with --adls.account-name, --adls.account-key, and --adls.endpoint args)",
   "integration: marks integration tests against Apache Spark",
+  "integration_trino: marks integration tests against Trino",
   "gcs: marks a test as requiring access to gcs compliant storage (use with --gs.token, --gs.project, and --gs.endpoint)",
   "benchmark: collection of tests to validate read/write performance before and after a change"
 ]

pyproject.toml

@@ -35,18 +35,12 @@
 from pathlib import Path
 from random import choice, randint
 from tempfile import TemporaryDirectory
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Dict,
-    Generator,
-    List,
-    Optional,
-)
+from typing import TYPE_CHECKING, Any, Dict, Generator, List, Optional
 
 import boto3
 import pytest
 from moto import mock_aws
+from sqlalchemy import Connection
 
 from pyiceberg.catalog import Catalog, load_catalog
 from pyiceberg.catalog.noop import NoopCatalog
@@ -141,6 +135,18 @@ def pytest_addoption(parser: pytest.Parser) -> None:
         "--gcs.oauth2.token", action="store", default="anon", help="The GCS authentication method for tests marked gcs"
     )
     parser.addoption("--gcs.project-id", action="store", default="test", help="The GCP project for tests marked gcs")
+    parser.addoption(
+        "--trino.rest.endpoint",
+        action="store",
+        default="trino://test@localhost:8082/warehouse_rest",
+        help="The Trino REST endpoint URL for tests marked as integration_trino",
+    )
+    parser.addoption(
+        "--trino.hive.endpoint",
+        action="store",
+        default="trino://test@localhost:8082/warehouse_hive",
+        help="The Trino Hive endpoint URL for tests marked as integration_trino",
+    )
 
 
 @pytest.fixture(scope="session")
@@ -2436,6 +2442,28 @@ def bound_reference_uuid() -> BoundReference[str]:
     return BoundReference(field=NestedField(1, "field", UUIDType(), required=False), accessor=Accessor(position=0, inner=None))
 
 
+@pytest.fixture(scope="session")
+def trino_hive_conn(request: pytest.FixtureRequest) -> Generator[Connection, None, None]:
+    from sqlalchemy import create_engine
+
+    trino_endpoint = request.config.getoption("--trino.hive.endpoint")
+    engine = create_engine(trino_endpoint)
+    connection = engine.connect()
+    yield connection
+    connection.close()
+
+
+@pytest.fixture(scope="session")
+def trino_rest_conn(request: pytest.FixtureRequest) -> Generator[Connection, None, None]:
+    from sqlalchemy import create_engine
+
+    trino_endpoint = request.config.getoption("--trino.rest.endpoint")
+    engine = create_engine(trino_endpoint)
+    connection = engine.connect()
+    yield connection
+    connection.close()
+
+
 @pytest.fixture(scope="session")
 def session_catalog() -> Catalog:
     return load_catalog(

tests/conftest.py

@@ -15,19 +15,14 @@
 # specific language governing permissions and limitations
 # under the License.
 import pytest
+from sqlalchemy import Connection, inspect
 
 from pyiceberg.catalog import Catalog
 from pyiceberg.exceptions import NoSuchTableError, TableAlreadyExistsError
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.table import Table
-from pyiceberg.types import (
-    BooleanType,
-    DateType,
-    IntegerType,
-    NestedField,
-    StringType,
-)
+from pyiceberg.types import BooleanType, DateType, IntegerType, NestedField, StringType
 
 TABLE_SCHEMA = Schema(
     NestedField(field_id=1, name="foo", field_type=BooleanType(), required=False),
@@ -86,3 +81,35 @@ def test_register_table_existing(
     # Assert that registering the table again raises TableAlreadyExistsError
     with pytest.raises(TableAlreadyExistsError):
         catalog.register_table(("default", "register_table_existing"), metadata_location=tbl.metadata_location)
+
+
+@pytest.mark.integration_trino
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    "catalog, trino_conn",
+    [
+        (pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("trino_hive_conn")),
+        (pytest.lazy_fixture("session_catalog"), pytest.lazy_fixture("trino_rest_conn")),
+    ],
+)
+def test_register_table_existing_in_trino(
+    catalog: Catalog,
+    trino_conn: Connection,
+) -> None:
+    """Test the registration of a table in the catalog that already exists in Trino.
+    This test verifies that a table can be registered in the catalog with an existing
+    metadata location and properly reflected in Trino.
+    """
+    namespace = "default"
+    table_name = "register_table_trino"
+    identifier = f"{namespace}.{table_name}"
+    location = f"s3a://warehouse/{namespace}/{table_name}"
+    tbl = _create_table(catalog, identifier, 2, location)
+    assert catalog.table_exists(identifier=identifier)
+    assert table_name in inspect(trino_conn).get_table_names(schema=namespace)
+    catalog.drop_table(identifier=identifier)
+    assert not catalog.table_exists(identifier=identifier)
+    assert table_name not in inspect(trino_conn).get_table_names(schema=namespace)
+    catalog.register_table((namespace, table_name), metadata_location=tbl.metadata_location)
+    assert catalog.table_exists(identifier=identifier)
+    assert table_name in inspect(trino_conn).get_table_names(schema=namespace)

tests/integration/test_register_table.py

@@ -17,6 +17,7 @@
 # pylint:disable=redefined-outer-name
 
 import pytest
+from sqlalchemy import Connection, inspect
 
 from pyiceberg.catalog.rest import RestCatalog
 
@@ -61,3 +62,22 @@ def test_create_namespace_if_already_existing(catalog: RestCatalog) -> None:
     catalog.create_namespace_if_not_exists(TEST_NAMESPACE_IDENTIFIER)
 
     assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
+
+
+@pytest.mark.integration
+@pytest.mark.integration_trino
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog")])
+def test_schema_exists_in_trino(trino_rest_conn: Connection, catalog: RestCatalog) -> None:
+    """Verifies that an Iceberg namespace correctly appears as a schema in Trino.
+
+    This test ensures the synchronization between Iceberg's namespace concept and
+    Trino's schema concept, confirming that after creating a namespace in the Iceberg
+    catalog, it becomes visible as a schema in the Trino environment.
+    """
+
+    if not catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER):
+        catalog.drop_namespace(TEST_NAMESPACE_IDENTIFIER)
+    catalog.create_namespace_if_not_exists(TEST_NAMESPACE_IDENTIFIER)
+
+    assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
+    assert TEST_NAMESPACE_IDENTIFIER.lower() in inspect(trino_rest_conn).get_schema_names()