[Feature][DataLoader] Add catalog for loading tables

.github/workflows/build-run-tests.yml

@@ -43,6 +43,10 @@ jobs:
         working-directory: integrations/python/dataloader
         run: make sync verify
 
+      - name: Run Data Loader Integration Tests
+        working-directory: integrations/python/dataloader
+        run: make integration-tests TOKEN_FILE=../../../tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/resources/dummy.token
+
       - name: Install dependencies
         run: pip install -r scripts/python/requirements.txt
 

integrations/python/dataloader/Makefile

@@ -1,4 +1,4 @@
-.PHONY: help sync clean lint format format-check check test verify build package-check
+.PHONY: help sync clean lint format format-check check test verify build package-check integration-tests
 
 help:
 	@echo "Available commands:"
@@ -37,6 +37,9 @@ build:
 package-check:
 	uv run twine check dist/*
 
+integration-tests:
+	uv run python tests/integration_test_catalog.py $(TOKEN_FILE)
+
 clean:
 	rm -rf build/
 	rm -rf dist/

integrations/python/dataloader/README.md

@@ -54,9 +54,28 @@ filters = (col("age") >= 18) & (col("country").is_in(["US", "CA"])) & ~col("emai
 # Set up environment
 make sync
 
-# Run tests
-make test
+# Run all checks and unit tests
+make verify
 
 # See all available commands
 make help
+```
+
+### Integration Tests
+
+Integration tests run the data loader end to end against an instance of OpenHouse running in Docker.
+
+**Prerequisites:** Docker and a Gradle build of OpenHouse.
+
+```bash
+# Create docker OpenHouse instance from repo root
+./gradlew clean build
+docker compose -f infra/recipes/docker-compose/oh-only/docker-compose.yml up -d --build
+
+# Run integration tests
+make -C integrations/python/dataloader sync
+make -C integrations/python/dataloader integration-tests TOKEN_FILE=../../../tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/resources/dummy.token
+
+# Stop docker
+docker compose -f infra/recipes/docker-compose/oh-only/docker-compose.yml down
 ```

integrations/python/dataloader/pyproject.toml

@@ -10,10 +10,10 @@ readme = "README.md"
 requires-python = ">=3.12"
 license = {text = "BSD-2-Clause"}
 keywords = ["openhouse", "data-loader", "lakehouse", "iceberg", "datafusion"]
-dependencies = ["datafusion==51.0.0", "pyiceberg~=0.11.0"]
+dependencies = ["datafusion==51.0.0", "pyiceberg~=0.11.0", "requests>=2.31.0"]
 
 [project.optional-dependencies]
-dev = ["ruff>=0.9.0", "pytest>=8.0.0", "twine>=6.0.0"]
+dev = ["responses>=0.25.0", "ruff>=0.9.0", "pytest>=8.0.0", "twine>=6.0.0"]
 
 [tool.hatch.version]
 source = "vcs"

integrations/python/dataloader/src/openhouse/dataloader/__init__.py

@@ -1,7 +1,15 @@
 from importlib.metadata import version
 
+from openhouse.dataloader.catalog import OpenHouseCatalog, OpenHouseCatalogError
 from openhouse.dataloader.data_loader import DataLoaderContext, OpenHouseDataLoader
 from openhouse.dataloader.filters import always_true, col
 
 __version__ = version("openhouse.dataloader")
-__all__ = ["OpenHouseDataLoader", "DataLoaderContext", "always_true", "col"]
+__all__ = [
+    "OpenHouseDataLoader",
+    "DataLoaderContext",
+    "OpenHouseCatalog",
+    "OpenHouseCatalogError",
+    "always_true",
+    "col",
+]

integrations/python/dataloader/src/openhouse/dataloader/catalog.py

@@ -0,0 +1,155 @@
+import logging
+from typing import Any
+
+import requests
+from pyiceberg.catalog import Catalog
+from pyiceberg.exceptions import NoSuchTableError
+from pyiceberg.io import load_file_io
+from pyiceberg.serializers import FromInputFile
+from pyiceberg.table import Table
+from pyiceberg.typedef import Identifier
+
+logger = logging.getLogger(__name__)
+
+_TABLE_LOCATION = "tableLocation"
+
+
+class OpenHouseCatalogError(Exception):
+    """Error raised when the OpenHouse catalog fails to load a table."""
+
+
+class OpenHouseCatalog(Catalog):
+    """Client-side catalog implementation for Iceberg tables in OpenHouse.
+
+    Leverages the OpenHouse Tables Service REST API to load table metadata.
+
+    Args:
+        name: Catalog name
+        uri: OpenHouse Tables Service base URL
+        auth_token: JWT Bearer token for authentication
+        ssl_ca_cert: Path to CA cert bundle for SSL verification
+        timeout_seconds: HTTP request timeout in seconds
+        properties: Additional properties forwarded to the PyIceberg Catalog base class
+            (e.g., ``py-io-impl`` for custom FileIO)
+    """
+
+    def __init__(
+        self,
+        name: str,
+        uri: str,
+        auth_token: str | None = None,
+        ssl_ca_cert: str | None = None,
+        timeout_seconds: float = 30,
+        properties: dict[str, str] | None = None,
+    ):
+        super().__init__(name, uri=uri, **(properties or {}))
+        self._uri = uri.rstrip("/")
+        self._timeout = timeout_seconds
+        logger.info("Initializing OpenHouseCatalog for service at %s", self._uri)
+        self._session = requests.Session()
+        self._session.headers["Content-Type"] = "application/json"
+
+        if auth_token is not None:
+            self._session.headers["Authorization"] = f"Bearer {auth_token}"
+
+        if ssl_ca_cert is not None:
+            self._session.verify = ssl_ca_cert
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_: Any):
+        self.close()
+
+    def close(self):
+        self._session.close()
+
+    def load_table(self, identifier: str | Identifier) -> Table:
+        database, table = self.identifier_to_database_and_table(identifier)
+        url = f"{self._uri}/v1/databases/{database}/tables/{table}"
+        logger.info("Calling load_table for table: %s.%s", database, table)
+
+        response = self._session.get(url, timeout=self._timeout)
+        if not response.ok:
+            if response.status_code == 404:
+                raise NoSuchTableError(f"Table {database}.{table} does not exist")
+            raise OSError(
+                f"Failed to load table {database}.{table}: HTTP {response.status_code}. Response: {response.text}"
+            )
+
+        table_response = response.json()
+        metadata_location = table_response.get(_TABLE_LOCATION)
+        if not metadata_location:
+            raise OpenHouseCatalogError(
+                f"Response for table {database}.{table} is missing '{_TABLE_LOCATION}'. Response: {table_response}"
+            )
+
+        file_io = load_file_io(properties=self.properties, location=metadata_location)
+        metadata_file = file_io.new_input(metadata_location)
+        metadata = FromInputFile.table_metadata(metadata_file)
+
+        logger.debug("Calling load_table succeeded")
+        return Table(
+            identifier=(database, table),
+            metadata=metadata,
+            metadata_location=metadata_location,
+            io=file_io,
+            catalog=self,
+        )
+
+    # -- Unsupported operations --
+    # Required by the Catalog ABC but not needed for read-only table loading.
+
+    def drop_table(self, *_: Any, **__: Any) -> None:
+        raise NotImplementedError
+
+    def purge_table(self, *_: Any, **__: Any) -> None:
+        raise NotImplementedError
+
+    def rename_table(self, *_: Any, **__: Any) -> Table:
+        raise NotImplementedError
+
+    def create_table(self, *_: Any, **__: Any) -> Table:
+        raise NotImplementedError
+
+    def create_table_transaction(self, *_: Any, **__: Any) -> Any:
+        raise NotImplementedError
+
+    def register_table(self, *_: Any, **__: Any) -> Table:
+        raise NotImplementedError
+
+    def commit_table(self, *_: Any, **__: Any) -> Any:
+        raise NotImplementedError
+
+    def list_tables(self, *_: Any, **__: Any) -> list[Identifier]:
+        raise NotImplementedError
+
+    def list_namespaces(self, *_: Any, **__: Any) -> list[Identifier]:
+        raise NotImplementedError
+
+    def create_namespace(self, *_: Any, **__: Any) -> None:
+        raise NotImplementedError
+
+    def drop_namespace(self, *_: Any, **__: Any) -> None:
+        raise NotImplementedError
+
+    def load_namespace_properties(self, *_: Any, **__: Any) -> dict[str, str]:
+        raise NotImplementedError
+
+    def update_namespace_properties(self, *_: Any, **__: Any) -> Any:
+        raise NotImplementedError
+
+    def list_views(self, *_: Any, **__: Any) -> list[Identifier]:
+        raise NotImplementedError
+
+    def drop_view(self, *_: Any, **__: Any) -> None:
+        raise NotImplementedError
+
+    def table_exists(self, *_: Any, **__: Any) -> bool:
+        raise NotImplementedError
+
+    def view_exists(self, *_: Any, **__: Any) -> bool:
+        raise NotImplementedError
+
+    def namespace_exists(self, *_: Any, **__: Any) -> bool:
+        raise NotImplementedError

integrations/python/dataloader/src/openhouse/dataloader/data_loader.py

@@ -1,6 +1,8 @@
 from collections.abc import Iterable, Mapping, Sequence
 from dataclasses import dataclass
 
+from pyiceberg.catalog import Catalog
+
 from openhouse.dataloader.data_loader_split import DataLoaderSplit
 from openhouse.dataloader.filters import Filter, always_true
 from openhouse.dataloader.table_identifier import TableIdentifier
@@ -31,6 +33,7 @@ class OpenHouseDataLoader:
 
     def __init__(
         self,
+        catalog: Catalog,
         database: str,
         table: str,
         branch: str | None = None,
@@ -40,13 +43,15 @@ def __init__(
     ):
         """
         Args:
+            catalog: Catalog for loading table metadata
             database: Database name
             table: Table name
             branch: Optional branch name
             columns: Column names to load, or None to load all columns
             filters: Row filter expression, defaults to always_true() (all rows)
             context: Data loader context
         """
+        self._catalog = catalog
         self._table = TableIdentifier(database, table, branch)
         self._columns = columns
         self._filters = filters if filters is not None else always_true()