Adds Docker Compose setup for Trino

Adds a Docker Compose configuration for setting up Trino with Iceberg, including support for Hive Metastore and REST catalog types. This allows for easier testing and development with Trino and Iceberg.

closes apache#2219

Author: dingo4dev

Makefile

@@ -117,6 +117,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/docker-compose-integration.yml

@@ -57,6 +57,21 @@ services:
       - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
       - CATALOG_S3_ENDPOINT=http://minio:9000
       - CATALOG_JDBC_STRICT__MODE=true
+
+  trino:
+    image: trinodb/trino:478
+    container_name: pyiceberg-trino
+    networks:
+      iceberg_net:
+    ports:
+      - 8082:8080
+    environment:
+      - CATALOG_MANAGEMENT=dynamic
+    depends_on:
+      - rest
+      - hive
+    volumes:
+      - ./trino/catalog:/etc/trino/catalog
   minio:
     image: minio/minio
     container_name: pyiceberg-minio

dev/docker-compose-trino.yml

@@ -0,0 +1,80 @@
+services:
+  rest:
+    image: apache/iceberg-rest-fixture
+    container_name: pyiceberg-rest
+    networks:
+      iceberg_net:
+    ports:
+      - 8181:8181
+    environment:
+      - AWS_ACCESS_KEY_ID=admin
+      - AWS_SECRET_ACCESS_KEY=password
+      - AWS_REGION=us-east-1
+      - CATALOG_WAREHOUSE=s3://warehouse/
+      - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
+      - CATALOG_S3_ENDPOINT=http://minio:9000
+
+  trino:
+    image: trinodb/trino:478
+    container_name: pyiceberg-trino
+    networks:
+      iceberg_net:
+    ports:
+      - 8082:8080
+    environment:
+      - CATALOG_MANAGEMENT=dynamic
+    depends_on:
+      - rest
+      - hive
+    volumes:
+      - ./trino/catalog:/etc/trino/catalog
+      - ./trino/config.properties:/etc/trino/config.properties
+
+  minio:
+    image: minio/minio
+    container_name: pyiceberg-minio
+    environment:
+      - MINIO_ROOT_USER=admin
+      - MINIO_ROOT_PASSWORD=password
+      - MINIO_DOMAIN=minio
+    networks:
+      iceberg_net:
+        aliases:
+          - warehouse.minio
+    ports:
+      - 9001:9001
+      - 9000:9000
+    command: ["server", "/data", "--console-address", ":9001"]
+  mc:
+    depends_on:
+      - minio
+    image: minio/mc
+    container_name: pyiceberg-mc
+    networks:
+      iceberg_net:
+    environment:
+      - AWS_ACCESS_KEY_ID=admin
+      - AWS_SECRET_ACCESS_KEY=password
+      - AWS_REGION=us-east-1
+    entrypoint: >
+      /bin/sh -c "
+      until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
+      /usr/bin/mc mb minio/warehouse;
+      /usr/bin/mc policy set public minio/warehouse;
+      tail -f /dev/null
+      "
+
+  hive:
+    build: hive/
+    container_name: hive
+    hostname: hive
+    networks:
+      iceberg_net:
+    ports:
+      - 9083:9083
+    environment:
+      SERVICE_NAME: "metastore"
+      SERVICE_OPTS: "-Dmetastore.warehouse.dir=s3a://warehouse/hive/"
+
+networks:
+  iceberg_net:

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

dev/trino/catalog/warehouse_hive.properties

@@ -0,0 +1,13 @@
+connector.name=iceberg
+iceberg.catalog.type=hive_metastore
+iceberg.expire-snapshots.min-retention=0d
+iceberg.remove-orphan-files.min-retention=0d
+iceberg.register-table-procedure.enabled=true
+hive.metastore.uri=thrift://hive:9083
+iceberg.hive-catalog-name=hive
+fs.native-s3.enabled=true
+s3.region=us-east-1
+s3.aws-access-key=admin
+s3.aws-secret-key=password
+s3.endpoint=http://minio:9000
+s3.path-style-access=false

dev/trino/catalog/warehouse_rest.properties

@@ -0,0 +1,15 @@
+connector.name=iceberg
+iceberg.catalog.type=rest
+iceberg.rest-catalog.uri=http://rest:8181
+iceberg.rest-catalog.warehouse=s3://warehouse/default
+iceberg.rest-catalog.nested-namespace-enabled=true
+iceberg.rest-catalog.case-insensitive-name-matching=true
+iceberg.expire-snapshots.min-retention=0d
+iceberg.remove-orphan-files.min-retention=0d
+iceberg.register-table-procedure.enabled=true
+fs.native-s3.enabled=true
+s3.region=us-east-1
+s3.aws-access-key=admin
+s3.aws-secret-key=password
+s3.endpoint=http://minio:9000
+s3.path-style-access=false

dev/trino/config.properties

@@ -0,0 +1,8 @@
+#single node install config
+coordinator=true
+node-scheduler.include-coordinator=true
+http-server.http.port=8080
+discovery.uri=http://localhost:8080
+http-server.process-forwarded=true
+http-server.https.enabled=false
+catalog.management=${ENV:CATALOG_MANAGEMENT}

pyproject.toml

@@ -119,6 +119,8 @@ dev = [
     "docutils!=0.21.post1",
     "mypy-boto3-glue>=1.28.18",
     "mypy-boto3-dynamodb>=1.28.18",
+    "trino[sqlalchemy]>=0.336.0",
+    "pre-commit>=4.4.0",
 ]
 # for mkdocs
 docs = [
@@ -153,6 +155,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",
 ]

tests/conftest.py

@@ -47,6 +47,7 @@
 import pytest
 from moto import mock_aws
 from pydantic_core import to_json
+from sqlalchemy import Connection
 
 from pyiceberg.catalog import Catalog, load_catalog
 from pyiceberg.catalog.noop import NoopCatalog
@@ -143,6 +144,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")
@@ -2574,6 +2587,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/integration/test_rest_catalog.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()