feast-dev
@@ -28,6 +28,7 @@ Commands:
   materialize              Run a (non-incremental) materialization job to...
   materialize-incremental  Run an incremental materialization job to ingest...
   permissions              Access permissions
+  registry                 Manage the feature registry
   registry-dump            Print contents of the metadata registry
   teardown                 Tear down deployed feature store infrastructure
   version                  Display Feast SDK version
@@ -483,6 +484,18 @@ reader             driver_hourly_stats_fresh   FeatureView      DESCRIBE
 ```
 
 
+## Registry
+
+### create-schema
+
+Pre-create the SQL registry schema so the application does not need DDL privileges at runtime. Use this with `schema_mode: verify` or `schema_mode: skip` in your `feature_store.yaml`.
+
+```text
+feast registry create-schema
+```
+
+This command only applies to SQL-based registries (`registry_type: sql`). It is safe to run multiple times — existing tables are not modified.
+
 ## Teardown
 
 Tear down deployed feature store infrastructure

@@ -66,6 +66,7 @@ registry:
 |-------|------|---------|-------------|
 | `registry_type` | string | `file` | Registry backend (`file`, `sql`, etc.) |
 | `path` | string | — | Connection string or file path |
+| `schema_mode` | string | `auto` | SQL registry only. `auto`: create tables on startup; `verify`: check tables exist, error if missing; `skip`: no DDL or verification. See [SQL Registry docs](registries/sql.md#schema-management-schema_mode). |
 | `mcp.enabled` | bool | `false` | Enable MCP (Model Context Protocol) on the REST registry server |
 
 When `registry.mcp.enabled` is `true`, the REST registry server exposes registry

@@ -80,9 +80,36 @@ docker build \
 If you are running Feast in Kubernetes, set the `image.repository` and
 `imagePullSecrets` Helm values accordingly to utilize your custom image.
 
+## Schema management (`schema_mode`)
+
+By default, the SQL registry creates its tables on every startup (`schema_mode: auto`). In production environments where the application should not have DDL privileges, you can pre-create the schema and configure the registry to only verify it:
+
+```yaml
+registry:
+    registry_type: sql
+    path: postgresql://db:5432/feast
+    schema_mode: verify   # or "skip"
+```
+
+| Value | Behavior |
+|---|---|
+| `auto` (default) | Creates tables if they don't exist. Current behavior, no breaking change. |
+| `verify` | Skips DDL. Checks that all expected tables exist on startup; raises an error listing missing tables if any are absent. When a separate `read_path` is configured, the read replica is also verified — a lagging replica (e.g. mid-migration) will block startup. Note: this is a table-level check only — it does not verify individual columns. A schema created by an older Feast version (missing newer columns) will pass verification but may fail at query time. |
+| `skip` | Skips both creation and verification. Use when schema is managed entirely outside Feast (e.g. by a migration tool). |
+
+### Pre-creating the schema
+
+When using `verify` or `skip` mode, run the following CLI command with a user that has DDL privileges to create the schema before starting the application:
+
+```shell
+feast registry create-schema
+```
+
+This reads `feature_store.yaml`, connects to the configured database, and creates all required tables. It is safe to run multiple times — existing tables are not modified.
+
 There are some things to note about how the SQL registry works:
-- Once instantiated, the Registry ensures the tables needed to store data exist, and creates them if they do not.
-- Upon tearing down the feast project, the registry ensures that the tables are dropped from the database.
+- When `schema_mode` is `auto` (the default), the Registry ensures the tables needed to store data exist, and creates them if they do not.
+- Upon tearing down the feast project, the registry deletes all rows from the registry tables (it does not drop the tables themselves). This runs regardless of `schema_mode` and requires only DML (`DELETE`) privileges, not DDL.
 - The schema for how data is laid out in tables can be found in the table definitions in [`sdk/python/feast/infra/registry/sql.py`](https://github.com/feast-dev/feast/blob/master/sdk/python/feast/infra/registry/sql.py). It is intentionally simple, storing the serialized protobuf versions of each Feast object keyed by its name.
 
 ## MySQL: serialized-proto columns use `LONGBLOB`

@@ -40,6 +40,7 @@
 from feast.cli.on_demand_feature_views import on_demand_feature_views_cmd
 from feast.cli.permissions import feast_permissions_cmd
 from feast.cli.projects import projects_cmd
+from feast.cli.registry import registry_cmd
 from feast.cli.saved_datasets import saved_datasets_cmd
 from feast.cli.serve import (
     serve_command,
@@ -644,6 +645,7 @@ def demo_notebooks_command(ctx: click.Context, output_dir: str, overwrite: bool)
 cli.add_command(on_demand_feature_views_cmd)
 cli.add_command(feast_permissions_cmd)
 cli.add_command(projects_cmd)
+cli.add_command(registry_cmd)
 cli.add_command(saved_datasets_cmd)
 cli.add_command(stream_feature_views_cmd)
 cli.add_command(label_views_cmd)

@@ -0,0 +1,46 @@
+import click
+from sqlalchemy import create_engine
+
+from feast.infra.registry.sql import SqlRegistryConfig, metadata
+from feast.repo_config import load_repo_config
+from feast.repo_operations import cli_check_repo
+
+
+@click.group(name="registry")
+def registry_cmd() -> None:
+    """
+    Manage the feature registry
+    """
+    pass
+
+
+@registry_cmd.command("create-schema")
+@click.pass_context
+def registry_create_schema(ctx: click.Context) -> None:
+    """
+    Pre-create the SQL registry schema.
+
+    Use this when schema_mode is set to 'verify' or 'skip' so the application
+    does not need DDL privileges at runtime.
+    """
+    repo = ctx.obj["CHDIR"]
+    fs_yaml_file = ctx.obj["FS_YAML_FILE"]
+    cli_check_repo(repo, fs_yaml_file)
+    repo_config = load_repo_config(repo, fs_yaml_file)
+
+    if repo_config is None:
+        raise click.ClickException("Could not load feature_store.yaml")
+
+    registry_config = repo_config.registry
+    if not isinstance(registry_config, SqlRegistryConfig):
+        raise click.ClickException(
+            "This command only applies to SQL-based registries "
+            f"(registry_type='sql'). Current type: '{registry_config.registry_type}'"
+        )
+
+    engine = create_engine(
+        registry_config.path, **registry_config.sqlalchemy_config_kwargs
+    )
+    metadata.create_all(engine)
+    engine.dispose()
+    click.echo("SQL registry schema created successfully.")
@@ -4,7 +4,7 @@
 from datetime import datetime, timezone
 from enum import Enum
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Union, cast
+from typing import Any, Callable, Dict, List, Literal, Optional, Union, cast
 
 from pydantic import StrictInt, StrictStr, field_validator
 from sqlalchemy import (  # type: ignore
@@ -26,6 +26,9 @@
     text,
     update,
 )
+from sqlalchemy import (
+    inspect as sa_inspect,
+)
 from sqlalchemy.dialects import mysql
 from sqlalchemy.engine import Engine
 from sqlalchemy.exc import IntegrityError
@@ -306,6 +309,12 @@ class SqlRegistryConfig(RegistryConfig):
     thread_pool_executor_worker_count: StrictInt = 0
     """ int: Number of worker threads to use for asynchronous caching in SQL Registry. If set to 0, it doesn't use ThreadPoolExecutor. """
 
+    schema_mode: Literal["auto", "verify", "skip"] = "auto"
+    """ str: Controls schema creation on startup.
+    'auto' (default) — creates tables if they don't exist (current behavior).
+    'verify' — skips DDL; checks that all expected tables exist and raises an error if any are missing.
+    'skip' — skips both creation and verification. """
+
     @field_validator("read_path")
     def validate_read_path(cls, read_path: Optional[str]) -> Optional[str]:
         # Mirror `RegistryConfig.validate_path`: a bare `postgresql://` read_path
@@ -316,6 +325,16 @@ def validate_read_path(cls, read_path: Optional[str]) -> Optional[str]:
         return read_path
 
 
+class FeastRegistrySchemaError(Exception):
+    def __init__(self, missing_tables: List[str]) -> None:
+        tables = ", ".join(missing_tables)
+        super().__init__(
+            f"SQL registry schema is incomplete — missing tables: {tables}. "
+            "Run 'feast registry create-schema' to create them, "
+            "or set schema_mode='auto' to create tables on startup."
+        )
+
+
 class SqlRegistry(CachingRegistry):
     def __init__(
         self,
@@ -339,7 +358,12 @@ def __init__(
             )
         else:
             self.read_engine = self.write_engine
-        metadata.create_all(self.write_engine)
+        if registry_config.schema_mode == "auto":
+            metadata.create_all(self.write_engine)
+        elif registry_config.schema_mode == "verify":
+            self._verify_schema(self.write_engine)
+            if self.read_engine is not self.write_engine:
+                self._verify_schema(self.read_engine)
         self._warn_if_narrow_blob_columns(self.write_engine)
         if self.read_engine is not self.write_engine:
             # A read replica can be on a different schema version (e.g. mid
@@ -357,13 +381,25 @@ def __init__(
             cache_ttl_seconds=registry_config.cache_ttl_seconds,
             cache_mode=registry_config.cache_mode,
         )
-        # Sync feast_metadata to projects table
-        # when purge_feast_metadata is set to True, Delete data from
-        # feast_metadata table and list_project_metadata will not return any data
         self._sync_feast_metadata_to_projects_table()
         if not self.purge_feast_metadata:
             self._maybe_init_project_metadata(project)
 
+    @staticmethod
+    def _verify_schema(engine: Engine, registry_metadata: MetaData = metadata) -> None:
+        """Verify that all expected registry tables exist in the database.
+
+        Raises ``FeastRegistrySchemaError`` listing missing tables and
+        suggesting ``feast registry create-schema``.
+        """
+        expected = set(registry_metadata.tables.keys())
+        actual = set(
+            sa_inspect(engine).get_table_names(schema=registry_metadata.schema)
+        )
+        missing = expected - actual
+        if missing:
+            raise FeastRegistrySchemaError(sorted(missing))
+
     @staticmethod
     def _warn_if_narrow_blob_columns(
         engine: Engine, registry_metadata: MetaData = metadata

@@ -29,11 +29,15 @@
 from feast.feature_view import FeatureView
 from feast.infra.offline_stores.file_source import FileSource
 from feast.infra.registry.sql import (
+    FeastRegistrySchemaError,
     ProtoBytes,
     SqlRegistry,
     SqlRegistryConfig,
     feature_views,
 )
+from feast.infra.registry.sql import (
+    metadata as registry_metadata,
+)
 from feast.protos.feast.core.Transformation_pb2 import (
     FeatureTransformationV2,
     UserDefinedFunctionV2,
@@ -585,3 +589,88 @@ def test_list_feature_views_updated_since_naive_treated_as_utc(sqlite_registry):
         "test_project", tags=None, updated_since=past_aware
     )
     assert len(result) == len(result_aware)
+
+
+class TestSchemaMode:
+    def test_schema_mode_auto_creates_tables(self, tmp_path):
+        """Default schema_mode='auto' creates tables on init (existing behavior)."""
+        db_file = tmp_path / "auto.db"
+        config = SqlRegistryConfig(
+            registry_type="sql",
+            path=f"sqlite:///{db_file}",
+            schema_mode="auto",
+        )
+        registry = SqlRegistry(config, "test_project", None)
+        from sqlalchemy import create_engine, inspect
+
+        engine = create_engine(f"sqlite:///{db_file}")
+        tables = set(inspect(engine).get_table_names())
+        expected = set(registry_metadata.tables.keys())
+        assert expected.issubset(tables)
+        engine.dispose()
+        registry.teardown()
+
+    def test_schema_mode_verify_raises_when_tables_missing(self, tmp_path):
+        """schema_mode='verify' raises FeastRegistrySchemaError on empty database."""
+        db_file = tmp_path / "verify_empty.db"
+        config = SqlRegistryConfig(
+            registry_type="sql",
+            path=f"sqlite:///{db_file}",
+            schema_mode="verify",
+        )
+        with pytest.raises(FeastRegistrySchemaError, match="missing tables"):
+            SqlRegistry(config, "test_project", None)
+
+    def test_schema_mode_verify_passes_when_tables_exist(self, tmp_path):
+        """schema_mode='verify' succeeds when schema was pre-created."""
+        db_file = tmp_path / "verify_ok.db"
+        db_url = f"sqlite:///{db_file}"
+        from sqlalchemy import create_engine
+
+        engine = create_engine(db_url)
+        registry_metadata.create_all(engine)
+        engine.dispose()
+
+        config = SqlRegistryConfig(
+            registry_type="sql",
+            path=db_url,
+            schema_mode="verify",
+        )
+        registry = SqlRegistry(config, "test_project", None)
+        registry.teardown()
+
+    def test_schema_mode_skip_does_not_run_ddl(self, tmp_path):
+        """schema_mode='skip' calls neither create_all nor _verify_schema."""
+        from unittest.mock import patch
+
+        db_file = tmp_path / "skip.db"
+        db_url = f"sqlite:///{db_file}"
+        from sqlalchemy import create_engine
+
+        engine = create_engine(db_url)
+        registry_metadata.create_all(engine)
+        engine.dispose()
+
+        with (
+            patch.object(registry_metadata, "create_all") as mock_create,
+            patch.object(SqlRegistry, "_verify_schema") as mock_verify,
+        ):
+            config = SqlRegistryConfig(
+                registry_type="sql",
+                path=db_url,
+                schema_mode="skip",
+            )
+            SqlRegistry(config, "test_project", None)
+            mock_create.assert_not_called()
+            mock_verify.assert_not_called()
+
+    def test_schema_mode_invalid_value_rejected(self):
+        """schema_mode only accepts 'auto', 'verify', 'skip'."""
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError):
+            SqlRegistryConfig(
+                registry_type="sql",
+                path="sqlite:///dummy.db",
+                schema_mode="bogus",
+            )