From 3c2392cae52ac3f39c19d8c2f968baeccd14c788 Mon Sep 17 00:00:00 2001 From: Chaitany Patel Date: Wed, 2 Sep 2026 15:24:00 +0530 Subject: [PATCH 1/2] feat: Add SQL schema discovery library for PostgreSQL Introduces feast.schema_discovery, which introspects a PostgreSQL database from an ad-hoc connection URL. It requires neither a feature_store.yaml nor a registered data source, so a user can explore a database before defining anything in Feast. SchemaDiscoveryService exposes list_schemas(), list_tables() and describe_table(). Column discovery deliberately combines two sources: - Feast's own DataSource supplies column types. Its get_table_column_names_and_types() reports canonical catalog spellings such as "character varying", which is what pg_type_to_feast_value_type keys on. SQLAlchemy reports "VARCHAR(255)" for the same column, which maps to UNKNOWN. - SQLAlchemy's Inspector supplies schema and table navigation, plus the nullability and primary-key metadata that a cursor description omits. Table and schema names reach the database through string interpolation in DataSource.get_table_query_string(), so identifiers are validated against a strict unquoted-identifier pattern before any connection is opened. That pattern is anchored with \Z rather than $, because $ also matches immediately before a trailing newline and would admit "users\n". Errors subclass FeastError and carry http_status_code(), so the REST layer added later can map them without a translation table. v1 accepts PostgreSQL only; every other dialect is refused by allowlist. Signed-off-by: Chaitany Patel --- sdk/python/feast/schema_discovery/__init__.py | 39 +++ sdk/python/feast/schema_discovery/dialects.py | 188 +++++++++++ sdk/python/feast/schema_discovery/errors.py | 84 +++++ .../feast/schema_discovery/introspector.py | 150 +++++++++ sdk/python/feast/schema_discovery/models.py | 85 +++++ sdk/python/feast/schema_discovery/service.py | 167 ++++++++++ .../tests/unit/schema_discovery/__init__.py | 0 .../unit/schema_discovery/test_dialects.py | 213 ++++++++++++ .../unit/schema_discovery/test_service.py | 311 ++++++++++++++++++ 9 files changed, 1237 insertions(+) create mode 100644 sdk/python/feast/schema_discovery/__init__.py create mode 100644 sdk/python/feast/schema_discovery/dialects.py create mode 100644 sdk/python/feast/schema_discovery/errors.py create mode 100644 sdk/python/feast/schema_discovery/introspector.py create mode 100644 sdk/python/feast/schema_discovery/models.py create mode 100644 sdk/python/feast/schema_discovery/service.py create mode 100644 sdk/python/tests/unit/schema_discovery/__init__.py create mode 100644 sdk/python/tests/unit/schema_discovery/test_dialects.py create mode 100644 sdk/python/tests/unit/schema_discovery/test_service.py diff --git a/sdk/python/feast/schema_discovery/__init__.py b/sdk/python/feast/schema_discovery/__init__.py new file mode 100644 index 00000000000..c88100f60d5 --- /dev/null +++ b/sdk/python/feast/schema_discovery/__init__.py @@ -0,0 +1,39 @@ +from feast.schema_discovery.errors import ( + DatabaseConnectionError, + DatabaseConnectionTimeout, + InvalidConnectionUrlError, + InvalidIdentifierError, + MissingDialectDependencyError, + SchemaDiscoveryError, + SchemaNotFoundError, + TableNotFoundError, + UnsupportedDialectError, +) +from feast.schema_discovery.models import ( + ColumnSchema, + DatabaseSchemas, + SchemaDiscoveryRequest, + SchemaInfo, + TableInfo, + TableSchema, +) +from feast.schema_discovery.service import SchemaDiscoveryService + +__all__ = [ + "ColumnSchema", + "DatabaseConnectionError", + "DatabaseConnectionTimeout", + "DatabaseSchemas", + "InvalidConnectionUrlError", + "InvalidIdentifierError", + "MissingDialectDependencyError", + "SchemaDiscoveryError", + "SchemaDiscoveryRequest", + "SchemaDiscoveryService", + "SchemaInfo", + "SchemaNotFoundError", + "TableInfo", + "TableNotFoundError", + "TableSchema", + "UnsupportedDialectError", +] diff --git a/sdk/python/feast/schema_discovery/dialects.py b/sdk/python/feast/schema_discovery/dialects.py new file mode 100644 index 00000000000..f864c9d7d4e --- /dev/null +++ b/sdk/python/feast/schema_discovery/dialects.py @@ -0,0 +1,188 @@ +"""Dialect allowlist, connection-URL parsing, and identifier validation. + +v1 supports PostgreSQL only. Adding a dialect means registering its backend +name here plus the ``DataSource``/offline-store config pair used to read +column types. +""" + +import re +from typing import Optional, Tuple + +from sqlalchemy.engine import URL, make_url +from sqlalchemy.exc import ArgumentError + +from feast.repo_config import RepoConfig +from feast.schema_discovery.errors import ( + InvalidConnectionUrlError, + InvalidIdentifierError, + MissingDialectDependencyError, + UnsupportedDialectError, +) + +POSTGRES = "postgresql" + +#: Backend names accepted by schema discovery. +SUPPORTED_DIALECTS = frozenset({POSTGRES}) + +#: Driver used when the URL names a backend without one (``postgresql://``). +_DEFAULT_DRIVERS = {POSTGRES: "psycopg"} + +#: Packaging extra that provides each dialect's driver. +_DIALECT_EXTRAS = {POSTGRES: "postgres"} + +# Unquoted SQL identifier. Deliberately strict: table names reach the database +# through string interpolation in DataSource.get_table_query_string(). +# Anchored with \Z, not $: "$" also matches before a trailing newline, which +# would let "users\n" through. +_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_$]*\Z") + +# RepoConfig requires a registry, but schema discovery only ever reads +# config.offline_store. Nothing resolves this path. +_UNUSED_REGISTRY = "schema_discovery_unused_registry.db" + + +def parse_connection_url(connection_url: str) -> URL: + """Parse and validate a connection URL, returning a SQLAlchemy ``URL``. + + Raises: + InvalidConnectionUrlError: URL is malformed or missing host/database. + UnsupportedDialectError: the backend is not in the v1 allowlist. + """ + if not connection_url or not connection_url.strip(): + raise InvalidConnectionUrlError("connection_url must not be empty") + + try: + url = make_url(connection_url.strip()) + except ArgumentError as e: + raise InvalidConnectionUrlError(str(e)) from e + + backend = url.get_backend_name() + if backend not in SUPPORTED_DIALECTS: + raise UnsupportedDialectError(backend, SUPPORTED_DIALECTS) + + if not url.host: + raise InvalidConnectionUrlError("no host specified") + if not url.database: + raise InvalidConnectionUrlError("no database specified") + if not url.username: + raise InvalidConnectionUrlError("no username specified") + + return url + + +def normalize_engine_url(url: URL) -> URL: + """Pin a driver on a bare backend URL so the right DBAPI is loaded.""" + if "+" not in url.drivername: + driver = _DEFAULT_DRIVERS[url.get_backend_name()] + return url.set(drivername=f"{url.drivername}+{driver}") + return url + + +def validate_identifier(identifier: str) -> str: + """Validate a single unquoted SQL identifier.""" + if not identifier or not _IDENTIFIER_RE.fullmatch(identifier): + raise InvalidIdentifierError(identifier) + return identifier + + +def split_table_name( + table: str, schema: Optional[str] = None +) -> Tuple[Optional[str], str]: + """Split ``table`` into ``(schema, table)``, validating both parts. + + A schema qualifier on ``table`` wins over the ``schema`` argument. + + Raises: + InvalidIdentifierError: either part is not a valid unquoted identifier, + or the name has more than two parts. + """ + if not table: + raise InvalidIdentifierError(table) + + parts = table.split(".") + if len(parts) > 2: + raise InvalidIdentifierError(table) + + if len(parts) == 2: + return validate_identifier(parts[0]), validate_identifier(parts[1]) + + validated = validate_identifier(parts[0]) + return (validate_identifier(schema) if schema else None), validated + + +def build_repo_config(url: URL, db_schema: str) -> RepoConfig: + """Build an in-memory ``RepoConfig`` from a connection URL. + + Produces the config that ``DataSource.get_table_column_names_and_types()`` + needs, without a ``feature_store.yaml`` or a registered data source. + """ + backend = url.get_backend_name() + if backend != POSTGRES: + raise UnsupportedDialectError(backend, SUPPORTED_DIALECTS) + + try: + from feast.infra.offline_stores.contrib.postgres_offline_store.postgres import ( + PostgreSQLOfflineStoreConfig, + ) + except ImportError as e: + raise MissingDialectDependencyError(backend, _DIALECT_EXTRAS[backend]) from e + + # parse_connection_url() guarantees these, but build_repo_config() is also + # reachable directly. + if not url.host or not url.database or not url.username: + raise InvalidConnectionUrlError("host, database and username are required") + + query = {k: v for k, v in url.query.items() if isinstance(v, str)} + + offline_store = PostgreSQLOfflineStoreConfig( + host=url.host, + port=url.port or 5432, + database=url.database, + db_schema=db_schema, + user=url.username, + password=url.password or "", + sslmode=query.get("sslmode", "require"), + sslkey_path=query.get("sslkey"), + sslcert_path=query.get("sslcert"), + sslrootcert_path=query.get("sslrootcert"), + ) + + return RepoConfig( + project="schema_discovery", + provider="local", + registry=_UNUSED_REGISTRY, + offline_store=offline_store, + entity_key_serialization_version=3, + ) + + +def build_data_source(url: URL, schema: str, table: str): + """Build an unregistered ``DataSource`` pointing at ``schema.table``. + + The source is never applied to a registry — it exists only so that + ``get_table_column_names_and_types()`` and + ``source_datatype_to_feast_value_type()`` can be reused for type discovery. + """ + backend = url.get_backend_name() + if backend != POSTGRES: + raise UnsupportedDialectError(backend, SUPPORTED_DIALECTS) + + try: + from feast.infra.offline_stores.contrib.postgres_offline_store.postgres_source import ( + PostgreSQLSource, + ) + except ImportError as e: + raise MissingDialectDependencyError(backend, _DIALECT_EXTRAS[backend]) from e + + return PostgreSQLSource( + name=f"schema_discovery_{schema}_{table}", + table=f"{schema}.{table}", + ) + + +def default_schema_for(url: URL) -> str: + """The schema searched when a table name carries no qualifier.""" + backend = url.get_backend_name() + if backend != POSTGRES: + raise UnsupportedDialectError(backend, SUPPORTED_DIALECTS) + return "public" diff --git a/sdk/python/feast/schema_discovery/errors.py b/sdk/python/feast/schema_discovery/errors.py new file mode 100644 index 00000000000..5e3e1ac044b --- /dev/null +++ b/sdk/python/feast/schema_discovery/errors.py @@ -0,0 +1,84 @@ +from typing import Iterable + +from fastapi import status as HttpStatusCode + +from feast.errors import FeastError + + +class SchemaDiscoveryError(FeastError): + """Base class for all schema discovery failures.""" + + +class InvalidConnectionUrlError(SchemaDiscoveryError): + def __init__(self, reason: str): + super().__init__(f"Invalid connection URL: {reason}") + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_400_BAD_REQUEST + + +class UnsupportedDialectError(SchemaDiscoveryError): + def __init__(self, dialect: str, supported: Iterable[str]): + super().__init__( + f"Unsupported database dialect '{dialect}'. " + f"Supported dialects: {', '.join(sorted(supported))}." + ) + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_400_BAD_REQUEST + + +class InvalidIdentifierError(SchemaDiscoveryError): + def __init__(self, identifier: str): + super().__init__( + f"Invalid SQL identifier '{identifier}'. Expected an unquoted name, " + f"optionally schema-qualified (e.g. 'users' or 'public.users')." + ) + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_400_BAD_REQUEST + + +class SchemaNotFoundError(SchemaDiscoveryError): + def __init__(self, schema: str): + super().__init__(f"Schema '{schema}' was not found in the database.") + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_404_NOT_FOUND + + +class TableNotFoundError(SchemaDiscoveryError): + def __init__(self, table: str): + super().__init__(f"Table '{table}' was not found in the database.") + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_404_NOT_FOUND + + +class DatabaseConnectionError(SchemaDiscoveryError): + def __init__(self, reason: str): + super().__init__(f"Failed to connect to or query the database: {reason}") + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_502_BAD_GATEWAY + + +class DatabaseConnectionTimeout(SchemaDiscoveryError): + def __init__(self, timeout_seconds: int): + super().__init__( + f"Timed out connecting to the database after {timeout_seconds}s." + ) + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_408_REQUEST_TIMEOUT + + +class MissingDialectDependencyError(SchemaDiscoveryError): + def __init__(self, dialect: str, extra: str): + super().__init__( + f"Schema discovery for '{dialect}' requires optional dependencies. " + f"Install them with: pip install 'feast[{extra}]'" + ) + + def http_status_code(self) -> int: + return HttpStatusCode.HTTP_400_BAD_REQUEST diff --git a/sdk/python/feast/schema_discovery/introspector.py b/sdk/python/feast/schema_discovery/introspector.py new file mode 100644 index 00000000000..e262da02fc9 --- /dev/null +++ b/sdk/python/feast/schema_discovery/introspector.py @@ -0,0 +1,150 @@ +"""SQLAlchemy-backed introspection for navigating a database. + +``DataSource.get_table_column_names_and_types()`` can only describe a table you +already know the name of, so schema/table *navigation* — and the nullability and +primary-key metadata that the cursor description does not carry — comes from +SQLAlchemy's ``Inspector``. +""" + +import logging +from contextlib import contextmanager +from typing import Any, Dict, Iterator, List + +from sqlalchemy import create_engine +from sqlalchemy import inspect as sa_inspect +from sqlalchemy.engine import URL, Engine +from sqlalchemy.exc import DBAPIError, NoSuchTableError, SQLAlchemyError +from sqlalchemy.pool import NullPool + +from feast.schema_discovery.dialects import normalize_engine_url +from feast.schema_discovery.errors import ( + DatabaseConnectionError, + DatabaseConnectionTimeout, + SchemaNotFoundError, + TableNotFoundError, +) +from feast.schema_discovery.models import TableInfo + +logger = logging.getLogger(__name__) + +# Namespaces that belong to the database itself, not the user's data. +_SYSTEM_SCHEMAS = frozenset({"information_schema", "pg_catalog", "pg_toast"}) + + +def _is_system_schema(name: str) -> bool: + return name in _SYSTEM_SCHEMAS or name.startswith("pg_") + + +@contextmanager +def engine_for(url: URL, timeout: int) -> Iterator[Engine]: + """Yield a short-lived, unpooled engine for one introspection call.""" + engine = create_engine( + normalize_engine_url(url), + poolclass=NullPool, + connect_args={"connect_timeout": timeout}, + ) + try: + yield engine + finally: + engine.dispose() + + +@contextmanager +def _translate_errors(timeout: int) -> Iterator[None]: + """Map SQLAlchemy/DBAPI failures onto schema discovery errors.""" + try: + yield + except DBAPIError as e: + if "timeout" in str(e).lower(): + raise DatabaseConnectionTimeout(timeout) from e + raise DatabaseConnectionError(str(e.orig or e)) from e + except SQLAlchemyError as e: + raise DatabaseConnectionError(str(e)) from e + + +def list_tables(url: URL, schema: str, timeout: int) -> List[TableInfo]: + """List tables and views within ``schema``. + + The existence check shares this connection rather than opening its own. + + Raises: + SchemaNotFoundError: ``schema`` does not exist in the database. + """ + with _translate_errors(timeout), engine_for(url, timeout) as engine: + inspector = sa_inspect(engine) + if schema not in inspector.get_schema_names(): + raise SchemaNotFoundError(schema) + tables = inspector.get_table_names(schema=schema) + views = inspector.get_view_names(schema=schema) + + return [TableInfo(name=n, type="table") for n in sorted(tables)] + [ + TableInfo(name=n, type="view") for n in sorted(views) + ] + + +def list_schemas_with_tables( + url: URL, timeout: int, include_tables: bool +) -> Dict[str, List[TableInfo]]: + """List schemas and, when requested, the tables in each. + + Reuses a single engine across every schema rather than reconnecting per + schema. + """ + result: Dict[str, List[TableInfo]] = {} + + with _translate_errors(timeout), engine_for(url, timeout) as engine: + inspector = sa_inspect(engine) + schemas = sorted( + n for n in inspector.get_schema_names() if not _is_system_schema(n) + ) + + for schema in schemas: + if not include_tables: + result[schema] = [] + continue + tables = inspector.get_table_names(schema=schema) + views = inspector.get_view_names(schema=schema) + result[schema] = [ + TableInfo(name=n, type="table") for n in sorted(tables) + ] + [TableInfo(name=n, type="view") for n in sorted(views)] + + return result + + +def get_column_metadata( + url: URL, schema: str, table: str, timeout: int +) -> Dict[str, Dict[str, Any]]: + """Return ``{column_name: {"nullable": bool, "primary_key": bool}}``. + + Raises: + TableNotFoundError: the table does not exist in ``schema``. + """ + with _translate_errors(timeout), engine_for(url, timeout) as engine: + inspector = sa_inspect(engine) + try: + columns = inspector.get_columns(table, schema=schema) + except NoSuchTableError as e: + raise TableNotFoundError(f"{schema}.{table}") from e + + if not columns: + raise TableNotFoundError(f"{schema}.{table}") + + try: + pk_columns = set( + inspector.get_pk_constraint(table, schema=schema).get( + "constrained_columns" + ) + or [] + ) + except SQLAlchemyError: + # Views and some permission setups expose no PK constraint. + logger.debug("No primary key constraint readable for %s.%s", schema, table) + pk_columns = set() + + return { + c["name"]: { + "nullable": bool(c.get("nullable", True)), + "primary_key": c["name"] in pk_columns, + } + for c in columns + } diff --git a/sdk/python/feast/schema_discovery/models.py b/sdk/python/feast/schema_discovery/models.py new file mode 100644 index 00000000000..3a3ab42e51c --- /dev/null +++ b/sdk/python/feast/schema_discovery/models.py @@ -0,0 +1,85 @@ +from typing import List, Literal, Optional + +from pydantic import BaseModel, Field + + +class SchemaDiscoveryRequest(BaseModel): + """Request payload for the schema discovery endpoint.""" + + connection_url: str = Field( + ..., + description=( + "SQLAlchemy-compatible connection URL including credentials, e.g. " + "'postgresql://user:password@host:5432/mydb'." # pragma: allowlist secret + ), + ) + table: Optional[str] = Field( + None, + description=( + "Table to describe. Either simple ('users') or schema-qualified " + "('public.users'). When omitted, schemas are listed instead." + ), + ) + schema_name: Optional[str] = Field( + None, + alias="schema", + description=( + "Database schema/namespace. Qualifies `table` when `table` is " + "unqualified; scopes the listing when `table` is omitted." + ), + ) + include_tables: bool = Field( + True, + description="When `table` is omitted, also list the tables in each schema.", + ) + + model_config = {"populate_by_name": True} + + +class ColumnSchema(BaseModel): + """A single column of a database table.""" + + name: str + sql_type: str = Field( + ..., description="Native SQL type as reported by the database." + ) + nullable: bool + primary_key: bool + feast_type: Optional[str] = Field( + None, + description=( + "Name of the mapped Feast ValueType, or 'UNKNOWN' when the native " + "type has no Feast equivalent." + ), + ) + + +class TableInfo(BaseModel): + """A table or view within a schema.""" + + name: str + type: Literal["table", "view"] = "table" + + +class SchemaInfo(BaseModel): + """A database schema (namespace), optionally with its tables.""" + + name: str + tables: Optional[List[TableInfo]] = None + + +class DatabaseSchemas(BaseModel): + """Response returned when no table is requested.""" + + database: str + dialect: str + schemas: List[SchemaInfo] + + +class TableSchema(BaseModel): + """Response returned when a table is requested.""" + + database: str + dialect: str + table: str + columns: List[ColumnSchema] diff --git a/sdk/python/feast/schema_discovery/service.py b/sdk/python/feast/schema_discovery/service.py new file mode 100644 index 00000000000..f0d1b9120ca --- /dev/null +++ b/sdk/python/feast/schema_discovery/service.py @@ -0,0 +1,167 @@ +"""Orchestration for SQL schema discovery. + +Discovery combines two sources of truth: + +* **SQLAlchemy** (``introspector``) navigates the database — which schemas and + tables exist — and supplies nullability and primary-key metadata. +* **Feast's own ``DataSource``** supplies column types. This matters: Feast's + type mappers key on canonical catalog spellings such as + ``character varying``, which is what the DBAPI cursor reports, whereas + SQLAlchemy reports ``VARCHAR(255)`` and would map to ``UNKNOWN``. +""" + +import logging +from typing import List, Optional, Union + +from feast.errors import ZeroColumnQueryResult +from feast.schema_discovery import introspector +from feast.schema_discovery.dialects import ( + build_data_source, + build_repo_config, + default_schema_for, + parse_connection_url, + split_table_name, + validate_identifier, +) +from feast.schema_discovery.errors import ( + DatabaseConnectionError, + DatabaseConnectionTimeout, + SchemaDiscoveryError, + TableNotFoundError, +) +from feast.schema_discovery.models import ( + ColumnSchema, + DatabaseSchemas, + SchemaDiscoveryRequest, + SchemaInfo, + TableInfo, + TableSchema, +) + +logger = logging.getLogger(__name__) + +DEFAULT_CONNECT_TIMEOUT_SECONDS = 10 + + +class SchemaDiscoveryService: + """Introspects a SQL database from an ad-hoc connection URL. + + Requires neither a ``feature_store.yaml`` nor a registered data source. + """ + + def __init__(self, connect_timeout: int = DEFAULT_CONNECT_TIMEOUT_SECONDS): + self.connect_timeout = connect_timeout + + def discover( + self, request: SchemaDiscoveryRequest + ) -> Union[DatabaseSchemas, TableSchema]: + """Dispatch on the request: a table describes columns, otherwise schemas.""" + if request.table: + return self.describe_table( + request.connection_url, request.table, request.schema_name + ) + return self.list_schemas( + request.connection_url, + include_tables=request.include_tables, + schema=request.schema_name, + ) + + def list_schemas( + self, + connection_url: str, + include_tables: bool = True, + schema: Optional[str] = None, + ) -> DatabaseSchemas: + """List schemas in the database, optionally with their tables. + + When ``schema`` is given the listing is scoped to that one schema. + """ + url = parse_connection_url(connection_url) + + if schema: + validate_identifier(schema) + # Always listed, so a missing schema is a 404 even when the caller + # does not want the tables themselves. + found_tables = introspector.list_tables(url, schema, self.connect_timeout) + schemas = [ + SchemaInfo(name=schema, tables=found_tables if include_tables else None) + ] + else: + found = introspector.list_schemas_with_tables( + url, self.connect_timeout, include_tables + ) + schemas = [ + SchemaInfo(name=name, tables=tables if include_tables else None) + for name, tables in found.items() + ] + + return DatabaseSchemas( + database=url.database or "", + dialect=url.get_backend_name(), + schemas=schemas, + ) + + def list_tables(self, connection_url: str, schema: str) -> List[TableInfo]: + """List the tables and views in one schema.""" + url = parse_connection_url(connection_url) + validate_identifier(schema) + return introspector.list_tables(url, schema, self.connect_timeout) + + def describe_table( + self, connection_url: str, table: str, schema: Optional[str] = None + ) -> TableSchema: + """Describe one table's columns, with Feast type mappings.""" + url = parse_connection_url(connection_url) + parsed_schema, table_name = split_table_name(table, schema) + resolved_schema = parsed_schema or default_schema_for(url) + qualified = f"{resolved_schema}.{table_name}" + + # SQLAlchemy first: it validates existence and carries the nullability + # and primary-key metadata the cursor description omits. + metadata = introspector.get_column_metadata( + url, resolved_schema, table_name, self.connect_timeout + ) + + source = build_data_source(url, resolved_schema, table_name) + columns_and_types = self._read_column_types( + source, url, resolved_schema, table_name + ) + to_feast_type = type(source).source_datatype_to_feast_value_type() + + columns = [] + for name, sql_type in columns_and_types: + column_meta = metadata.get(name, {}) + columns.append( + ColumnSchema( + name=name, + sql_type=sql_type, + nullable=bool(column_meta.get("nullable", True)), + primary_key=bool(column_meta.get("primary_key", False)), + feast_type=to_feast_type(sql_type).name, + ) + ) + + return TableSchema( + database=url.database or "", + dialect=url.get_backend_name(), + table=qualified, + columns=columns, + ) + + def _read_column_types(self, source, url, schema: str, table: str): + """Read ``(column_name, native_sql_type)`` pairs via Feast's DataSource.""" + qualified = f"{schema}.{table}" + config = build_repo_config(url, db_schema=schema) + + try: + return list(source.get_table_column_names_and_types(config)) + except ZeroColumnQueryResult as e: + raise TableNotFoundError(qualified) from e + except SchemaDiscoveryError: + raise + except Exception as e: + if type(e).__name__ in ("UndefinedTable", "InvalidSchemaName"): + raise TableNotFoundError(qualified) from e + if "timeout" in str(e).lower(): + raise DatabaseConnectionTimeout(self.connect_timeout) from e + raise DatabaseConnectionError(str(e)) from e diff --git a/sdk/python/tests/unit/schema_discovery/__init__.py b/sdk/python/tests/unit/schema_discovery/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/sdk/python/tests/unit/schema_discovery/test_dialects.py b/sdk/python/tests/unit/schema_discovery/test_dialects.py new file mode 100644 index 00000000000..ab26b504f2f --- /dev/null +++ b/sdk/python/tests/unit/schema_discovery/test_dialects.py @@ -0,0 +1,213 @@ +import pytest + +from feast.schema_discovery.dialects import ( + build_data_source, + build_repo_config, + default_schema_for, + normalize_engine_url, + parse_connection_url, + split_table_name, + validate_identifier, +) +from feast.schema_discovery.errors import ( + InvalidConnectionUrlError, + InvalidIdentifierError, + UnsupportedDialectError, +) + +VALID_URL = ( + "postgresql://alice:s3cret@db.internal:5432/mydb" # pragma: allowlist secret +) + + +class TestParseConnectionUrl: + def test_parses_postgres_url(self): + url = parse_connection_url(VALID_URL) + assert url.get_backend_name() == "postgresql" + assert url.host == "db.internal" + assert url.port == 5432 + assert url.database == "mydb" + assert url.username == "alice" + assert url.password == "s3cret" # pragma: allowlist secret + + def test_accepts_explicit_driver(self): + url = parse_connection_url("postgresql+psycopg://alice:p@h:5432/db") + assert url.get_backend_name() == "postgresql" + + def test_strips_surrounding_whitespace(self): + assert parse_connection_url(f" {VALID_URL} ").database == "mydb" + + @pytest.mark.parametrize("url", ["", " ", None]) + def test_rejects_empty(self, url): + with pytest.raises(InvalidConnectionUrlError): + parse_connection_url(url) + + def test_rejects_malformed(self): + with pytest.raises(InvalidConnectionUrlError): + parse_connection_url("not a url at all") + + @pytest.mark.parametrize( + "url,missing", + [ + ("postgresql://alice@/mydb", "host"), + ("postgresql://alice@host:5432/", "database"), + ("postgresql://host:5432/mydb", "username"), + ], + ) + def test_rejects_incomplete_url(self, url, missing): + with pytest.raises(InvalidConnectionUrlError): + parse_connection_url(url) + + @pytest.mark.parametrize( + "url", + [ + "mysql://u:p@h:3306/db", + "snowflake://u:p@account/db", + "bigquery://project/dataset", + "sqlite:///local.db", + "trino://u@h:8080/catalog", + ], + ) + def test_rejects_unsupported_dialects(self, url): + """v1 is PostgreSQL only.""" + with pytest.raises(UnsupportedDialectError): + parse_connection_url(url) + + def test_unsupported_dialect_error_names_supported_set(self): + with pytest.raises(UnsupportedDialectError, match="postgresql"): + parse_connection_url("mysql://u:p@h:3306/db") + + +class TestNormalizeEngineUrl: + def test_pins_driver_on_bare_backend(self): + url = normalize_engine_url(parse_connection_url(VALID_URL)) + assert url.drivername == "postgresql+psycopg" + + def test_preserves_explicit_driver(self): + url = normalize_engine_url( + parse_connection_url("postgresql+psycopg2://a:p@h:5432/db") + ) + assert url.drivername == "postgresql+psycopg2" + + +class TestValidateIdentifier: + @pytest.mark.parametrize( + "ident", ["users", "_private", "Table1", "a$b", "user_events_v2", "T"] + ) + def test_accepts_valid(self, ident): + assert validate_identifier(ident) == ident + + @pytest.mark.parametrize( + "ident", + [ + "", + "1users", + "user events", + "users;", + "users--", + '"users"', + "users'", + "us*ers", + "users\n", + "sch.tbl", + ], + ) + def test_rejects_invalid(self, ident): + with pytest.raises(InvalidIdentifierError): + validate_identifier(ident) + + +class TestSplitTableName: + def test_simple_name_without_schema(self): + assert split_table_name("users") == (None, "users") + + def test_simple_name_with_schema_arg(self): + assert split_table_name("users", "analytics") == ("analytics", "users") + + def test_qualified_name(self): + assert split_table_name("public.users") == ("public", "users") + + def test_qualifier_beats_schema_arg(self): + assert split_table_name("public.users", "analytics") == ("public", "users") + + def test_rejects_three_part_name(self): + with pytest.raises(InvalidIdentifierError): + split_table_name("db.public.users") + + def test_rejects_empty(self): + with pytest.raises(InvalidIdentifierError): + split_table_name("") + + @pytest.mark.parametrize( + "table", + [ + "users; DROP TABLE accounts", + "users WHERE 1=1", + "users) UNION SELECT * FROM secrets --", + "pg_shadow--", + "users/*", + ], + ) + def test_rejects_sql_injection_attempts(self, table): + """Table names are interpolated into SQL, so they must be strictly validated.""" + with pytest.raises(InvalidIdentifierError): + split_table_name(table) + + def test_rejects_injection_in_schema_arg(self): + with pytest.raises(InvalidIdentifierError): + split_table_name("users", "public; DROP SCHEMA x") + + +class TestBuildRepoConfig: + def test_maps_url_parts_onto_offline_store(self): + config = build_repo_config(parse_connection_url(VALID_URL), db_schema="public") + store = config.offline_store + assert store.host == "db.internal" + assert store.port == 5432 + assert store.database == "mydb" + assert store.user == "alice" + assert store.password == "s3cret" # pragma: allowlist secret + assert store.db_schema == "public" + + def test_defaults_port_when_absent(self): + config = build_repo_config( + parse_connection_url("postgresql://a:p@h/db"), db_schema="public" + ) + assert config.offline_store.port == 5432 + + def test_honours_non_default_schema(self): + config = build_repo_config( + parse_connection_url(VALID_URL), db_schema="analytics" + ) + assert config.offline_store.db_schema == "analytics" + + def test_sslmode_defaults_to_require(self): + config = build_repo_config(parse_connection_url(VALID_URL), db_schema="public") + assert config.offline_store.sslmode == "require" + + def test_sslmode_read_from_query_string(self): + config = build_repo_config( + parse_connection_url(f"{VALID_URL}?sslmode=disable"), db_schema="public" + ) + assert config.offline_store.sslmode == "disable" + + def test_does_not_require_feature_store_yaml(self): + """The whole point: an ad-hoc config with no repo on disk.""" + config = build_repo_config(parse_connection_url(VALID_URL), db_schema="public") + assert config.project == "schema_discovery" + + +class TestBuildDataSource: + def test_builds_qualified_source(self): + source = build_data_source(parse_connection_url(VALID_URL), "public", "users") + assert source.get_table_query_string() == "public.users" + + def test_exposes_feast_type_mapper(self): + source = build_data_source(parse_connection_url(VALID_URL), "public", "users") + mapper = type(source).source_datatype_to_feast_value_type() + assert mapper.__name__ == "pg_type_to_feast_value_type" + + +class TestDefaultSchema: + def test_postgres_defaults_to_public(self): + assert default_schema_for(parse_connection_url(VALID_URL)) == "public" diff --git a/sdk/python/tests/unit/schema_discovery/test_service.py b/sdk/python/tests/unit/schema_discovery/test_service.py new file mode 100644 index 00000000000..31172ded93e --- /dev/null +++ b/sdk/python/tests/unit/schema_discovery/test_service.py @@ -0,0 +1,311 @@ +import pytest + +from feast.schema_discovery import introspector +from feast.schema_discovery.errors import ( + DatabaseConnectionError, + DatabaseConnectionTimeout, + InvalidConnectionUrlError, + InvalidIdentifierError, + SchemaDiscoveryError, + SchemaNotFoundError, + TableNotFoundError, + UnsupportedDialectError, +) +from feast.schema_discovery.models import SchemaDiscoveryRequest, TableInfo +from feast.schema_discovery.service import SchemaDiscoveryService + +VALID_URL = ( + "postgresql://alice:s3cret@db.internal:5432/mydb" # pragma: allowlist secret +) + + +def _raise_schema_not_found(*args, **kwargs): + raise SchemaNotFoundError("nope") + + +@pytest.fixture +def service(): + return SchemaDiscoveryService(connect_timeout=3) + + +@pytest.fixture +def no_network(monkeypatch): + """Make every introspection call explode, proving it was never reached.""" + + def boom(*args, **kwargs): + raise AssertionError("attempted to connect to the database") + + for name in ( + "list_tables", + "list_schemas_with_tables", + "get_column_metadata", + ): + monkeypatch.setattr(introspector, name, boom) + + +class TestValidationHappensBeforeConnecting: + """Bad input must be rejected without opening a connection.""" + + def test_invalid_url_rejected(self, service, no_network): + with pytest.raises(InvalidConnectionUrlError): + service.list_schemas("not a url") + + def test_unsupported_dialect_rejected(self, service, no_network): + with pytest.raises(UnsupportedDialectError): + service.list_schemas("mysql://u:p@h:3306/db") + + def test_injection_in_table_rejected(self, service, no_network): + with pytest.raises(InvalidIdentifierError): + service.describe_table(VALID_URL, "users; DROP TABLE accounts") + + def test_injection_in_schema_rejected(self, service, no_network): + with pytest.raises(InvalidIdentifierError): + service.list_tables(VALID_URL, "public; DROP SCHEMA x") + + def test_three_part_table_rejected(self, service, no_network): + with pytest.raises(InvalidIdentifierError): + service.describe_table(VALID_URL, "db.public.users") + + def test_unsupported_dialect_rejected_on_describe(self, service, no_network): + with pytest.raises(UnsupportedDialectError): + service.describe_table("bigquery://project/dataset", "users") + + +class TestDescribeTableMerge: + """Column types come from Feast's DataSource; nullability/PK from SQLAlchemy.""" + + @pytest.fixture + def stubbed(self, service, monkeypatch): + monkeypatch.setattr( + introspector, + "get_column_metadata", + lambda *a, **k: { + "user_id": {"nullable": False, "primary_key": True}, + "email": {"nullable": True, "primary_key": False}, + "created_at": {"nullable": False, "primary_key": False}, + }, + ) + monkeypatch.setattr( + SchemaDiscoveryService, + "_read_column_types", + lambda self, source, url, schema, table: [ + ("user_id", "bigint"), + ("email", "character varying"), + ("created_at", "timestamp without time zone"), + ], + ) + return service + + def test_maps_postgres_types_to_feast_types(self, stubbed): + result = stubbed.describe_table(VALID_URL, "public.users") + assert [(c.name, c.feast_type) for c in result.columns] == [ + ("user_id", "INT64"), + ("email", "STRING"), + ("created_at", "UNIX_TIMESTAMP"), + ] + + def test_merges_nullability_and_primary_key(self, stubbed): + by_name = { + c.name: c for c in stubbed.describe_table(VALID_URL, "users").columns + } + assert by_name["user_id"].primary_key is True + assert by_name["user_id"].nullable is False + assert by_name["email"].primary_key is False + assert by_name["email"].nullable is True + + def test_reports_native_sql_type(self, stubbed): + by_name = { + c.name: c for c in stubbed.describe_table(VALID_URL, "users").columns + } + assert by_name["email"].sql_type == "character varying" + + def test_preserves_column_order(self, stubbed): + result = stubbed.describe_table(VALID_URL, "users") + assert [c.name for c in result.columns] == ["user_id", "email", "created_at"] + + def test_qualifies_unqualified_table_with_default_schema(self, stubbed): + assert stubbed.describe_table(VALID_URL, "users").table == "public.users" + + def test_honours_explicit_schema(self, stubbed): + result = stubbed.describe_table(VALID_URL, "users", schema="analytics") + assert result.table == "analytics.users" + + def test_reports_database_and_dialect(self, stubbed): + result = stubbed.describe_table(VALID_URL, "users") + assert result.database == "mydb" + assert result.dialect == "postgresql" + + def test_unmapped_type_becomes_unknown(self, service, monkeypatch): + monkeypatch.setattr( + introspector, + "get_column_metadata", + lambda *a, **k: {"weird": {"nullable": True, "primary_key": False}}, + ) + monkeypatch.setattr( + SchemaDiscoveryService, + "_read_column_types", + lambda self, source, url, schema, table: [("weird", "some_custom_type")], + ) + column = service.describe_table(VALID_URL, "users").columns[0] + assert column.feast_type == "UNKNOWN" + assert column.sql_type == "some_custom_type" + + def test_column_missing_from_metadata_defaults_safely(self, service, monkeypatch): + monkeypatch.setattr(introspector, "get_column_metadata", lambda *a, **k: {}) + monkeypatch.setattr( + SchemaDiscoveryService, + "_read_column_types", + lambda self, source, url, schema, table: [("orphan", "integer")], + ) + column = service.describe_table(VALID_URL, "users").columns[0] + assert column.nullable is True + assert column.primary_key is False + + +class TestListSchemas: + def test_lists_schemas_with_tables(self, service, monkeypatch): + monkeypatch.setattr( + introspector, + "list_schemas_with_tables", + lambda url, timeout, include_tables: { + "public": [ + TableInfo(name="users", type="table"), + TableInfo(name="active_users", type="view"), + ], + "analytics": [TableInfo(name="daily_metrics", type="table")], + }, + ) + result = service.list_schemas(VALID_URL) + assert result.database == "mydb" + assert result.dialect == "postgresql" + assert [s.name for s in result.schemas] == ["public", "analytics"] + assert [t.type for t in result.schemas[0].tables] == ["table", "view"] + + def test_omits_tables_when_not_requested(self, service, monkeypatch): + monkeypatch.setattr( + introspector, + "list_schemas_with_tables", + lambda url, timeout, include_tables: {"public": []}, + ) + result = service.list_schemas(VALID_URL, include_tables=False) + assert result.schemas[0].tables is None + + def test_scopes_to_single_schema(self, service, monkeypatch): + monkeypatch.setattr( + introspector, + "list_tables", + lambda *a, **k: [TableInfo(name="daily_metrics", type="table")], + ) + result = service.list_schemas(VALID_URL, schema="analytics") + assert [s.name for s in result.schemas] == ["analytics"] + + def test_missing_schema_raises_not_found(self, service, monkeypatch): + monkeypatch.setattr(introspector, "list_tables", _raise_schema_not_found) + with pytest.raises(SchemaNotFoundError): + service.list_schemas(VALID_URL, schema="nope") + + def test_missing_schema_raises_even_without_tables(self, service, monkeypatch): + """The 404 must survive include_tables=False.""" + monkeypatch.setattr(introspector, "list_tables", _raise_schema_not_found) + with pytest.raises(SchemaNotFoundError): + service.list_schemas(VALID_URL, schema="nope", include_tables=False) + + def test_list_tables_missing_schema_raises_not_found(self, service, monkeypatch): + monkeypatch.setattr(introspector, "list_tables", _raise_schema_not_found) + with pytest.raises(SchemaNotFoundError): + service.list_tables(VALID_URL, "nope") + + +class TestDiscoverDispatch: + def test_table_present_describes_columns(self, service, monkeypatch): + monkeypatch.setattr( + introspector, "get_column_metadata", lambda *a, **k: {"id": {}} + ) + monkeypatch.setattr( + SchemaDiscoveryService, + "_read_column_types", + lambda self, source, url, schema, table: [("id", "integer")], + ) + result = service.discover( + SchemaDiscoveryRequest(connection_url=VALID_URL, table="public.users") + ) + assert result.table == "public.users" + + def test_table_absent_lists_schemas(self, service, monkeypatch): + monkeypatch.setattr( + introspector, + "list_schemas_with_tables", + lambda url, timeout, include_tables: {"public": []}, + ) + result = service.discover(SchemaDiscoveryRequest(connection_url=VALID_URL)) + assert [s.name for s in result.schemas] == ["public"] + + +class TestRequestModel: + def test_schema_alias_is_accepted(self): + request = SchemaDiscoveryRequest( + connection_url=VALID_URL, table="users", schema="analytics" + ) + assert request.schema_name == "analytics" + + def test_field_name_also_accepted(self): + request = SchemaDiscoveryRequest( + connection_url=VALID_URL, schema_name="analytics" + ) + assert request.schema_name == "analytics" + + def test_include_tables_defaults_true(self): + assert SchemaDiscoveryRequest(connection_url=VALID_URL).include_tables is True + + +class TestErrorHttpStatusCodes: + @pytest.mark.parametrize( + "error,expected", + [ + (InvalidConnectionUrlError("bad"), 400), + (UnsupportedDialectError("mysql", ["postgresql"]), 400), + (InvalidIdentifierError("x;"), 400), + (SchemaNotFoundError("nope"), 404), + (TableNotFoundError("public.nope"), 404), + (DatabaseConnectionTimeout(10), 408), + (DatabaseConnectionError("refused"), 502), + ], + ) + def test_maps_to_http_status(self, error, expected): + assert error.http_status_code() == expected + + def test_all_errors_share_a_base(self): + assert isinstance(SchemaNotFoundError("x"), SchemaDiscoveryError) + + +class TestRealConnectionFailures: + """No server needed: a closed port exercises driver load and error translation.""" + + CLOSED_PORT_URL = "postgresql://u:p@127.0.0.1:1/db?sslmode=disable" + + def test_engine_loads_psycopg_driver(self): + from feast.schema_discovery.dialects import parse_connection_url + from feast.schema_discovery.introspector import engine_for + + with engine_for(parse_connection_url(self.CLOSED_PORT_URL), 2) as engine: + assert engine.dialect.name == "postgresql" + assert engine.dialect.driver == "psycopg" + + def test_list_schemas_translates_refused_connection(self): + service = SchemaDiscoveryService(connect_timeout=2) + with pytest.raises(DatabaseConnectionError) as excinfo: + service.list_schemas(self.CLOSED_PORT_URL) + assert excinfo.value.http_status_code() == 502 + + def test_describe_table_translates_refused_connection(self): + service = SchemaDiscoveryService(connect_timeout=2) + with pytest.raises(DatabaseConnectionError): + service.describe_table(self.CLOSED_PORT_URL, "public.users") + + def test_password_is_not_leaked_in_error_message(self): + service = SchemaDiscoveryService(connect_timeout=2) + with pytest.raises(DatabaseConnectionError) as excinfo: + service.list_schemas( + "postgresql://u:sup3rs3cret@127.0.0.1:1/db?sslmode=disable" # pragma: allowlist secret + ) + assert "sup3rs3cret" not in str(excinfo.value) From b57482c9767c83e561fcb8b62b3e46a42179eb32 Mon Sep 17 00:00:00 2001 From: Chaitany Patel Date: Fri, 4 Sep 2026 15:54:39 +0530 Subject: [PATCH 2/2] test: Add integration tests for PostgreSQL schema discovery Exercises SchemaDiscoveryService against a real PostgreSQL 16 instance via testcontainers, following the PostgresContainer pattern already used elsewhere in the integration suite. Covers schema listing with system namespaces excluded, table and view enumeration, column type mapping across the common PostgreSQL types, primary-key and nullability reporting, describing a view, and the not-found and connection-failure paths. Includes a regression guard asserting that no column in the fixture table maps to UNKNOWN. That assertion fails if column types ever start coming from SQLAlchemy's type spellings ("VARCHAR(255)") rather than the catalog spellings the Feast type mappers expect. These tests require Docker and are marked with pytest.mark.integration. Signed-off-by: Chaitany Patel --- .../integration/schema_discovery/__init__.py | 0 .../schema_discovery/test_schema_discovery.py | 178 ++++++++++++++++++ 2 files changed, 178 insertions(+) create mode 100644 sdk/python/tests/integration/schema_discovery/__init__.py create mode 100644 sdk/python/tests/integration/schema_discovery/test_schema_discovery.py diff --git a/sdk/python/tests/integration/schema_discovery/__init__.py b/sdk/python/tests/integration/schema_discovery/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/sdk/python/tests/integration/schema_discovery/test_schema_discovery.py b/sdk/python/tests/integration/schema_discovery/test_schema_discovery.py new file mode 100644 index 00000000000..671806dbc4c --- /dev/null +++ b/sdk/python/tests/integration/schema_discovery/test_schema_discovery.py @@ -0,0 +1,178 @@ +"""End-to-end schema discovery against a real PostgreSQL instance.""" + +import pytest +from testcontainers.postgres import PostgresContainer + +from feast.schema_discovery import SchemaDiscoveryService +from feast.schema_discovery.errors import ( + DatabaseConnectionError, + SchemaNotFoundError, + TableNotFoundError, +) + +DDL = """ +CREATE TABLE public.users ( + user_id BIGINT PRIMARY KEY, + email VARCHAR(255), + age INTEGER NOT NULL, + balance DOUBLE PRECISION, + is_active BOOLEAN NOT NULL, + created_at TIMESTAMP NOT NULL, + metadata JSONB +); + +CREATE VIEW public.active_users AS + SELECT user_id, email FROM public.users WHERE is_active; + +CREATE SCHEMA analytics; +CREATE TABLE analytics.daily_metrics ( + metric_date DATE PRIMARY KEY, + value NUMERIC +); +""" + + +@pytest.fixture(scope="module") +def postgres_url(): + with PostgresContainer( + "postgres:16", + username="root", + password="test!@#$%", # pragma: allowlist secret + dbname="test", + ).with_exposed_ports(5432) as container: + import psycopg + + host = container.get_container_host_ip() + port = container.get_exposed_port(5432) + url = ( # pragma: allowlist secret + f"postgresql://root:test!%40%23%24%25@{host}:{port}/test?sslmode=disable" + ) + + with psycopg.connect( + host=host, + port=port, + user="root", + password="test!@#$%", # pragma: allowlist secret + dbname="test", + sslmode="disable", + ) as conn: + conn.execute(DDL) + conn.commit() + + yield url + + +@pytest.fixture +def service(): + return SchemaDiscoveryService(connect_timeout=10) + + +@pytest.mark.integration +class TestListSchemas: + def test_lists_user_schemas_only(self, service, postgres_url): + result = service.list_schemas(postgres_url, include_tables=False) + names = {s.name for s in result.schemas} + assert {"public", "analytics"} <= names + assert not any(n.startswith("pg_") for n in names) + assert "information_schema" not in names + + def test_reports_database_and_dialect(self, service, postgres_url): + result = service.list_schemas(postgres_url, include_tables=False) + assert result.database == "test" + assert result.dialect == "postgresql" + + def test_includes_tables_and_views(self, service, postgres_url): + result = service.list_schemas(postgres_url, include_tables=True) + public = next(s for s in result.schemas if s.name == "public") + by_name = {t.name: t.type for t in public.tables} + assert by_name["users"] == "table" + assert by_name["active_users"] == "view" + + def test_scopes_to_one_schema(self, service, postgres_url): + result = service.list_schemas(postgres_url, schema="analytics") + assert [s.name for s in result.schemas] == ["analytics"] + assert [t.name for t in result.schemas[0].tables] == ["daily_metrics"] + + def test_unknown_schema_raises(self, service, postgres_url): + with pytest.raises(SchemaNotFoundError): + service.list_schemas(postgres_url, schema="does_not_exist") + + +@pytest.mark.integration +class TestListTables: + def test_lists_tables_in_schema(self, service, postgres_url): + tables = service.list_tables(postgres_url, "analytics") + assert [t.name for t in tables] == ["daily_metrics"] + + def test_unknown_schema_raises(self, service, postgres_url): + with pytest.raises(SchemaNotFoundError): + service.list_tables(postgres_url, "does_not_exist") + + +@pytest.mark.integration +class TestDescribeTable: + def test_maps_every_column_to_a_feast_type(self, service, postgres_url): + result = service.describe_table(postgres_url, "public.users") + by_name = {c.name: c for c in result.columns} + + assert by_name["user_id"].feast_type == "INT64" + assert by_name["email"].feast_type == "STRING" + assert by_name["age"].feast_type == "INT32" + assert by_name["balance"].feast_type == "DOUBLE" + assert by_name["is_active"].feast_type == "BOOL" + assert by_name["created_at"].feast_type == "UNIX_TIMESTAMP" + assert by_name["metadata"].feast_type == "MAP" + + def test_no_column_falls_back_to_unknown(self, service, postgres_url): + """Regression guard: SQLAlchemy type spellings would map to UNKNOWN.""" + result = service.describe_table(postgres_url, "public.users") + assert all(c.feast_type != "UNKNOWN" for c in result.columns) + + def test_reports_primary_key(self, service, postgres_url): + result = service.describe_table(postgres_url, "public.users") + pks = {c.name for c in result.columns if c.primary_key} + assert pks == {"user_id"} + + def test_reports_nullability(self, service, postgres_url): + by_name = { + c.name: c + for c in service.describe_table(postgres_url, "public.users").columns + } + assert by_name["email"].nullable is True + assert by_name["age"].nullable is False + assert by_name["user_id"].nullable is False + + def test_unqualified_table_uses_public(self, service, postgres_url): + result = service.describe_table(postgres_url, "users") + assert result.table == "public.users" + + def test_describes_table_in_non_default_schema(self, service, postgres_url): + result = service.describe_table(postgres_url, "analytics.daily_metrics") + assert {c.name for c in result.columns} == {"metric_date", "value"} + + def test_schema_argument_qualifies_table(self, service, postgres_url): + result = service.describe_table( + postgres_url, "daily_metrics", schema="analytics" + ) + assert result.table == "analytics.daily_metrics" + + def test_describes_a_view(self, service, postgres_url): + result = service.describe_table(postgres_url, "public.active_users") + assert [c.name for c in result.columns] == ["user_id", "email"] + assert all(c.primary_key is False for c in result.columns) + + def test_unknown_table_raises(self, service, postgres_url): + with pytest.raises(TableNotFoundError): + service.describe_table(postgres_url, "public.does_not_exist") + + def test_unknown_schema_raises(self, service, postgres_url): + with pytest.raises(TableNotFoundError): + service.describe_table(postgres_url, "nosuchschema.users") + + +@pytest.mark.integration +class TestConnectionFailures: + def test_bad_credentials_raise_connection_error(self, service, postgres_url): + broken = postgres_url.replace("root:", "wronguser:") + with pytest.raises(DatabaseConnectionError): + service.list_schemas(broken)