GraphDB (Storage)

DuckDB-backed storage and query layer. Owns the database schema, handles inserts, resolves edges, manages phantom nodes, and provides all query methods.

GraphDB

GraphDB(db_path=None)

DuckDB-backed knowledge graph storage.

The sole storage layer for the SQL indexer. Manages a DuckDB database containing repos, files, nodes, edges, column usage, and column lineage tables. Provides insert/upsert methods for the indexer and query methods consumed by the MCP tool layer.

Thread-safety (read/write separation): Write operations are serialised through _write_lock (a threading.RLock). Read operations use a fresh cursor and require no lock -- DuckDB MVCC ensures snapshot isolation.

The ``write_transaction()`` context manager holds the lock for the
full ``BEGIN .. COMMIT`` scope so no other thread can interleave
write statements.  ``asyncio.to_thread()`` callers are safe because
reads are lock-free and writes acquire the lock internally.

Initialise the database.

Parameters:

Name	Type	Description	Default
db_path	str | Path | None	Path to DuckDB file. None for in-memory (testing).	None

Source code in src/sqlprism/core/graph.py

def __init__(self, db_path: str | Path | None = None):
    """Initialise the database.

    Args:
        db_path: Path to DuckDB file. None for in-memory (testing).
    """
    self.db_path = str(db_path) if db_path else ":memory:"
    self.conn = duckdb.connect(self.db_path)
    self._write_lock = threading.RLock()
    # Thread-local flag: only the thread holding _write_lock inside
    # write_transaction() sets this to True.  _execute_read checks it
    # to decide whether to use the main connection (to see uncommitted
    # writes) or a fresh cursor (snapshot isolation).
    self._tlocal = threading.local()
    self._init_schema()
    self._has_pgq = False
    self._init_pgq()

has_pgq property

has_pgq

Whether DuckPGQ is available for graph queries.

close

close()

Close the underlying DuckDB connection.

Source code in src/sqlprism/core/graph.py

def close(self) -> None:
    """Close the underlying DuckDB connection."""
    with self._write_lock:
        self.conn.close()

refresh_property_graph

refresh_property_graph()

Refresh the property graph after data changes (e.g., reindex).

Source code in src/sqlprism/core/graph.py

def refresh_property_graph(self) -> None:
    """Refresh the property graph after data changes (e.g., reindex)."""
    self._create_property_graph()

clear_snippet_cache staticmethod

clear_snippet_cache()

Clear the cached file contents used for snippet extraction.

Should be called after reindex to avoid serving stale content.

Source code in src/sqlprism/core/graph.py

@staticmethod
def clear_snippet_cache() -> None:
    """Clear the cached file contents used for snippet extraction.

    Should be called after reindex to avoid serving stale content.
    """
    _read_file_lines.cache_clear()

write_transaction

write_transaction()

Context manager that holds _write_lock for a full transaction.

Acquires the write lock, issues BEGIN TRANSACTION, yields, then COMMIT on success or ROLLBACK on exception.

Re-entrant: if the current thread already holds the lock and is inside a transaction, yields without starting a nested one (DuckDB does not support nested transactions).

Source code in src/sqlprism/core/graph.py

@contextmanager
def write_transaction(self):
    """Context manager that holds ``_write_lock`` for a full transaction.

    Acquires the write lock, issues ``BEGIN TRANSACTION``, yields, then
    ``COMMIT`` on success or ``ROLLBACK`` on exception.

    Re-entrant: if the current thread already holds the lock and is
    inside a transaction, yields without starting a nested one (DuckDB
    does not support nested transactions).
    """
    if getattr(self._tlocal, "in_transaction", False):
        yield
        return
    with self._write_lock:
        self.conn.execute("BEGIN TRANSACTION")
        self._tlocal.in_transaction = True
        try:
            yield
            self.conn.execute("COMMIT")
        except Exception:
            self.conn.execute("ROLLBACK")
            raise
        finally:
            self._tlocal.in_transaction = False

transaction

transaction()

Backward-compatible alias for :meth:write_transaction.

.. deprecated:: 0.6 Use :meth:write_transaction instead.

Source code in src/sqlprism/core/graph.py

@contextmanager
def transaction(self):
    """Backward-compatible alias for :meth:`write_transaction`.

    .. deprecated:: 0.6
        Use :meth:`write_transaction` instead.
    """
    warnings.warn(
        "transaction() is deprecated, use write_transaction() instead",
        DeprecationWarning,
        stacklevel=2,
    )
    with self.write_transaction():
        yield

upsert_repo

upsert_repo(name, path, repo_type='sql')

Create or update a repo entry.

Updates the stored path and repo_type if changed.

Parameters:

Name	Type	Description	Default
name	str	Unique repo name used as the identifier across the index.	required
path	str	Absolute filesystem path to the repo root.	required
repo_type	str	One of 'sql', 'dbt', 'sqlmesh'.	'sql'

Returns:

Type	Description
int	The repo_id (existing or newly created).

Source code in src/sqlprism/core/graph.py

def upsert_repo(self, name: str, path: str, repo_type: str = "sql") -> int:
    """Create or update a repo entry.

    Updates the stored path and repo_type if changed.

    Args:
        name: Unique repo name used as the identifier across the index.
        path: Absolute filesystem path to the repo root.
        repo_type: One of ``'sql'``, ``'dbt'``, ``'sqlmesh'``.

    Returns:
        The ``repo_id`` (existing or newly created).
    """
    with self._write_lock:
        existing = self._execute_write(
            "SELECT repo_id, path, repo_type FROM repos WHERE name = ?", [name],
        ).fetchone()
        if existing:
            if existing[1] != str(path) or existing[2] != repo_type:
                self._execute_write(
                    "UPDATE repos SET path = ?, repo_type = ? WHERE repo_id = ?",
                    [str(path), repo_type, existing[0]],
                )
            return existing[0]
        result = self._execute_write(
            "INSERT INTO repos (name, path, repo_type) VALUES (?, ?, ?) RETURNING repo_id",
            [name, str(path), repo_type],
        ).fetchone()
        return result[0]

update_repo_metadata

update_repo_metadata(repo_id, commit=None, branch=None)

Update the last indexed commit/branch for a repo.

Source code in src/sqlprism/core/graph.py

def update_repo_metadata(self, repo_id: int, commit: str | None = None, branch: str | None = None) -> None:
    """Update the last indexed commit/branch for a repo."""
    with self._write_lock:
        self._execute_write(
            "UPDATE repos SET last_commit = ?, last_branch = ?, indexed_at = now() WHERE repo_id = ?",
            [commit, branch, repo_id],
        )

get_source_fingerprint

get_source_fingerprint(repo_id)

Get the stored source fingerprint for a repo.

Source code in src/sqlprism/core/graph.py

def get_source_fingerprint(self, repo_id: int) -> str | None:
    """Get the stored source fingerprint for a repo."""
    row = self._execute_read(
        "SELECT source_fingerprint FROM repos WHERE repo_id = ?", [repo_id],
    ).fetchone()
    return row[0] if row else None

update_source_fingerprint

update_source_fingerprint(repo_id, fingerprint)

Store the source fingerprint for a repo.

Source code in src/sqlprism/core/graph.py

def update_source_fingerprint(self, repo_id: int, fingerprint: str) -> None:
    """Store the source fingerprint for a repo."""
    with self._write_lock:
        self._execute_write(
            "UPDATE repos SET source_fingerprint = ? WHERE repo_id = ?",
            [fingerprint, repo_id],
        )

get_rendered_cache

get_rendered_cache(repo_id)

Load cached rendered SQL for all models in a repo.

Returns:

Type	Description
dict[str, tuple[str, dict]]	Dict mapping model_name -> (rendered_sql, column_schemas_dict).

Source code in src/sqlprism/core/graph.py

def get_rendered_cache(self, repo_id: int) -> dict[str, tuple[str, dict]]:
    """Load cached rendered SQL for all models in a repo.

    Returns:
        Dict mapping model_name -> (rendered_sql, column_schemas_dict).
    """
    rows = self._execute_read(
        "SELECT model_name, rendered_sql, column_schemas "
        "FROM rendered_cache WHERE repo_id = ?",
        [repo_id],
    ).fetchall()
    result = {}
    for name, sql, schemas_json in rows:
        schemas = json.loads(schemas_json) if schemas_json else {}
        result[name] = (sql, schemas)
    return result

update_rendered_cache

update_rendered_cache(repo_id, models, column_schemas)

Replace the rendered SQL cache for a repo.

Source code in src/sqlprism/core/graph.py

def update_rendered_cache(
    self, repo_id: int, models: dict[str, str], column_schemas: dict[str, dict[str, str]],
) -> None:
    """Replace the rendered SQL cache for a repo."""
    with self._write_lock:
        self._execute_write("DELETE FROM rendered_cache WHERE repo_id = ?", [repo_id])
        for model_name, sql in models.items():
            schemas = column_schemas.get(model_name, {})
            self._execute_write(
                "INSERT INTO rendered_cache (repo_id, model_name, rendered_sql, column_schemas) "
                "VALUES (?, ?, ?, ?)",
                [repo_id, model_name, sql, json.dumps(schemas)],
            )

delete_repo

delete_repo(repo_id)

Delete a repo and all associated data (manual cascade).

DuckDB does not support ON DELETE CASCADE, so child rows are deleted in dependency order: lineage, column_usage, edges, nodes, files, then the repo itself.

Parameters:

Name	Type	Description	Default
repo_id	int	ID of the repo to delete.	required

Source code in src/sqlprism/core/graph.py

def delete_repo(self, repo_id: int) -> None:
    """Delete a repo and all associated data (manual cascade).

    DuckDB does not support ``ON DELETE CASCADE``, so child rows are
    deleted in dependency order: lineage, column_usage, edges, nodes,
    files, then the repo itself.

    Args:
        repo_id: ID of the repo to delete.
    """
    with self.write_transaction():
        self._delete_repo_impl(repo_id)

get_all_repos

get_all_repos()

Return all repos as (repo_id, name, path, repo_type) tuples.

Source code in src/sqlprism/core/graph.py

def get_all_repos(self) -> list[tuple[int, str, str, str]]:
    """Return all repos as (repo_id, name, path, repo_type) tuples."""
    return self._execute_read(
        "SELECT repo_id, name, path, repo_type FROM repos"
    ).fetchall()

get_file_checksum

get_file_checksum(repo_id, path)

Get the stored checksum for a single file in a repo.

Source code in src/sqlprism/core/graph.py

def get_file_checksum(self, repo_id: int, path: str) -> str | None:
    """Get the stored checksum for a single file in a repo."""
    row = self._execute_read(
        "SELECT checksum FROM files WHERE repo_id = ? AND path = ?",
        [repo_id, path],
    ).fetchone()
    return row[0] if row else None

find_node_name_by_file

find_node_name_by_file(repo_id, rel_path)

Find the primary node name for a file path in a repo.

Used by sqlmesh reindex to resolve file paths to model names. Returns the first table/view node name found, or None.

Source code in src/sqlprism/core/graph.py

def find_node_name_by_file(self, repo_id: int, rel_path: str) -> str | None:
    """Find the primary node name for a file path in a repo.

    Used by sqlmesh reindex to resolve file paths to model names.
    Returns the first table/view node name found, or ``None``.
    """
    row = self._execute_read(
        "SELECT n.name FROM nodes n "
        "JOIN files f ON n.file_id = f.file_id "
        "WHERE f.repo_id = ? AND f.path = ? AND n.kind IN ('table', 'view') "
        "ORDER BY n.kind, n.name LIMIT 1",
        [repo_id, rel_path],
    ).fetchone()
    return row[0] if row else None

find_file_paths_by_stem

find_file_paths_by_stem(repo_id, stem)

Find stored file paths whose filename stem matches.

Used by dbt/sqlmesh on-save reindex to map filesystem paths (e.g. models/orders.sql) to stored paths that may differ (e.g. staging/orders.sql for dbt, catalog/schema/orders.sql for sqlmesh).

Parameters:

Name	Type	Description	Default
repo_id	int	Repo to search within.	required
stem	str	Filename stem without extension (e.g. "orders").	required

Returns:

Type	Description
list[str]	List of matching stored file paths.

Source code in src/sqlprism/core/graph.py

def find_file_paths_by_stem(self, repo_id: int, stem: str) -> list[str]:
    """Find stored file paths whose filename stem matches.

    Used by dbt/sqlmesh on-save reindex to map filesystem paths
    (e.g. ``models/orders.sql``) to stored paths that may differ
    (e.g. ``staging/orders.sql`` for dbt, ``catalog/schema/orders.sql``
    for sqlmesh).

    Args:
        repo_id: Repo to search within.
        stem: Filename stem without extension (e.g. ``"orders"``).

    Returns:
        List of matching stored file paths.
    """
    rows = self._execute_read(
        "SELECT path FROM files WHERE repo_id = ? "
        "AND (path = ? OR path LIKE ?)",
        [repo_id, stem + ".sql", "%/" + stem + ".sql"],
    ).fetchall()
    return [r[0] for r in rows]

get_file_checksums

get_file_checksums(repo_id)

Get {path: checksum} for all files in a repo.

Source code in src/sqlprism/core/graph.py

def get_file_checksums(self, repo_id: int) -> dict[str, str]:
    """Get {path: checksum} for all files in a repo."""
    rows = self._execute_read("SELECT path, checksum FROM files WHERE repo_id = ?", [repo_id]).fetchall()
    return {path: checksum for path, checksum in rows}

delete_file_data

delete_file_data(repo_id, path)

Delete all data for a file (nodes, edges, column_usage, file record).

Nodes that have inbound edges from OTHER files are converted to phantom nodes (file_id=NULL) instead of being deleted, so that cross-file edges survive incremental reindex. cleanup_phantoms() will later merge these phantoms with the newly-inserted real nodes.

Wraps in a write_transaction if not already inside one.

Source code in src/sqlprism/core/graph.py

def delete_file_data(self, repo_id: int, path: str) -> None:
    """Delete all data for a file (nodes, edges, column_usage, file record).

    Nodes that have inbound edges from OTHER files are converted to phantom
    nodes (file_id=NULL) instead of being deleted, so that cross-file edges
    survive incremental reindex. cleanup_phantoms() will later merge these
    phantoms with the newly-inserted real nodes.

    Wraps in a write_transaction if not already inside one.
    """
    with self.write_transaction():
        self._delete_file_data_impl(repo_id, path)

insert_file

insert_file(repo_id, path, language, checksum)

Insert a file record.

Parameters:

Name	Type	Description	Default
repo_id	int	Owning repo.	required
path	str	Relative file path within the repo.	required
language	str	Language identifier (e.g. "sql").	required
checksum	str	SHA-256 hex digest of the file content.	required

Returns:

Type	Description
int	The newly assigned file_id.

Source code in src/sqlprism/core/graph.py

def insert_file(self, repo_id: int, path: str, language: str, checksum: str) -> int:
    """Insert a file record.

    Args:
        repo_id: Owning repo.
        path: Relative file path within the repo.
        language: Language identifier (e.g. ``"sql"``).
        checksum: SHA-256 hex digest of the file content.

    Returns:
        The newly assigned ``file_id``.
    """
    with self._write_lock:
        result = self._execute_write(
            "INSERT INTO files (repo_id, path, language, checksum) VALUES (?, ?, ?, ?) RETURNING file_id",
            [repo_id, path, language, checksum],
        ).fetchone()
        return result[0]

insert_node

insert_node(
    file_id,
    kind,
    name,
    language,
    line_start=None,
    line_end=None,
    metadata=None,
    schema=None,
)

Insert a single node.

Parameters:

Name	Type	Description	Default
file_id	int | None	Owning file, or None for phantom nodes.	required
kind	str	Node kind (e.g. "table", "view", "cte").	required
name	str	Unqualified entity name.	required
language	str	Language identifier.	required
line_start	int | None	First source line, if known.	None
line_end	int | None	Last source line, if known.	None
metadata	dict | None	Arbitrary JSON-serialisable metadata.	None
schema	str | None	Database schema qualifier (e.g. "staging").	None

Returns:

Type	Description
int	The newly assigned node_id.

Source code in src/sqlprism/core/graph.py

def insert_node(
    self,
    file_id: int | None,
    kind: str,
    name: str,
    language: str,
    line_start: int | None = None,
    line_end: int | None = None,
    metadata: dict | None = None,
    schema: str | None = None,
) -> int:
    """Insert a single node.

    Args:
        file_id: Owning file, or ``None`` for phantom nodes.
        kind: Node kind (e.g. ``"table"``, ``"view"``, ``"cte"``).
        name: Unqualified entity name.
        language: Language identifier.
        line_start: First source line, if known.
        line_end: Last source line, if known.
        metadata: Arbitrary JSON-serialisable metadata.
        schema: Database schema qualifier (e.g. ``"staging"``).

    Returns:
        The newly assigned ``node_id``.
    """
    with self._write_lock:
        result = self._execute_write(
            "INSERT INTO nodes (file_id, kind, name, language, "
            "line_start, line_end, metadata, schema) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?) RETURNING node_id",
            [
                file_id,
                kind,
                name,
                language,
                line_start,
                line_end,
                json.dumps(metadata) if metadata else None,
                schema,
            ],
        ).fetchone()
        return result[0]

resolve_node

resolve_node(
    name,
    kind,
    repo_id=None,
    schema=None,
    same_repo_only=False,
)

Find a node by name and kind.

Matches on short name (e.g. "orders") which covers both unqualified references and qualified ones (stored as short name plus schema column). Search order:

1. Exact (name + kind) in same repo
2. Exact (name + kind) cross-repo
3. Kind-relaxed (name only, real nodes with a file) in same repo
4. Kind-relaxed cross-repo

The kind-relaxed fallback handles the common case where a SQL reference uses target_kind="table" but the actual node was indexed as "query" or "view" (e.g. sqlmesh/dbt models). Within the fallback, non-matching kinds are ranked table > view > query and query-local aliases (cte, subquery) are placed last so a real node always beats an alias sharing the same name.

Parameters:

Name	Type	Description	Default
name	str	Unqualified entity name.	required
kind	str	Node kind to match.	required
repo_id	int | None	Prefer nodes from this repo. Falls back to cross-repo search if not found (unless same_repo_only is set).	None
schema	str | None	Optional schema qualifier. When provided, only nodes with a matching schema column are returned.	None
same_repo_only	bool	Skip cross-repo steps 2 and 4. Callers that persist data against the resolved node (e.g. column-def inserts) must use this to avoid attaching rows to a different repo's node.	False

Returns:

Type	Description
int | None	The node_id if found, otherwise None.

Source code in src/sqlprism/core/graph.py

def resolve_node(
    self,
    name: str,
    kind: str,
    repo_id: int | None = None,
    schema: str | None = None,
    same_repo_only: bool = False,
) -> int | None:
    """Find a node by name and kind.

    Matches on short name (e.g. ``"orders"``) which covers both
    unqualified references and qualified ones (stored as short name
    plus schema column). Search order:

    1. Exact (name + kind) in same repo
    2. Exact (name + kind) cross-repo
    3. Kind-relaxed (name only, real nodes with a file) in same repo
    4. Kind-relaxed cross-repo

    The kind-relaxed fallback handles the common case where a SQL
    reference uses ``target_kind="table"`` but the actual node was
    indexed as ``"query"`` or ``"view"`` (e.g. sqlmesh/dbt models).
    Within the fallback, non-matching kinds are ranked
    ``table > view > query`` and query-local aliases (``cte``,
    ``subquery``) are placed last so a real node always beats an alias
    sharing the same name.

    Args:
        name: Unqualified entity name.
        kind: Node kind to match.
        repo_id: Prefer nodes from this repo. Falls back to cross-repo
            search if not found (unless ``same_repo_only`` is set).
        schema: Optional schema qualifier. When provided, only nodes
            with a matching ``schema`` column are returned.
        same_repo_only: Skip cross-repo steps 2 and 4. Callers that
            persist data against the resolved node (e.g. column-def
            inserts) must use this to avoid attaching rows to a
            different repo's node.

    Returns:
        The ``node_id`` if found, otherwise ``None``.
    """
    schema_clause = ""
    schema_params: list = []
    if schema is not None:
        schema_clause = " AND n.schema = ?"
        schema_params = [schema]

    if repo_id:
        # 1. Exact kind match in same repo
        row = self._execute_read(
            "SELECT n.node_id FROM nodes n "
            "JOIN files f ON n.file_id = f.file_id "
            "WHERE n.name = ? AND n.kind = ? AND f.repo_id = ?" + schema_clause + " LIMIT 1",
            [name, kind, repo_id, *schema_params],
        ).fetchone()
        if row:
            return row[0]

    if not same_repo_only:
        # 2. Exact kind match cross-repo
        row = self._execute_read(
            "SELECT n.node_id FROM nodes n WHERE n.name = ? AND n.kind = ?" + schema_clause + " LIMIT 1",
            [name, kind, *schema_params],
        ).fetchone()
        if row:
            return row[0]

    # 3. Kind-relaxed: match by name only, prefer real nodes (file_id IS NOT NULL).
    # Secondary rank (when requested kind doesn't match): table > view > query;
    # CTEs and subqueries are query-local aliases and must never win over a
    # real node when a user looks up a name without a kind filter. A final
    # node_id tiebreaker keeps ordering deterministic across runs.
    kind_rank = _kind_rank_sql("n.kind")
    if repo_id:
        row = self._execute_read(
            "SELECT n.node_id FROM nodes n "
            "JOIN files f ON n.file_id = f.file_id "
            "WHERE n.name = ? AND f.repo_id = ?" + schema_clause
            + f" ORDER BY (n.kind = ?) DESC, {kind_rank} ASC, n.node_id ASC LIMIT 1",
            [name, repo_id, *schema_params, kind],
        ).fetchone()
        if row:
            return row[0]

    if same_repo_only:
        return None

    # 4. Kind-relaxed cross-repo
    row = self._execute_read(
        "SELECT n.node_id FROM nodes n "
        "WHERE n.name = ? AND n.file_id IS NOT NULL" + schema_clause
        + f" ORDER BY (n.kind = ?) DESC, {kind_rank} ASC, n.node_id ASC LIMIT 1",
        [name, *schema_params, kind],
    ).fetchone()
    return row[0] if row else None

get_or_create_phantom

get_or_create_phantom(name, kind, language)

Get an existing phantom node or create one. Returns node_id.

Source code in src/sqlprism/core/graph.py

def get_or_create_phantom(self, name: str, kind: str, language: str) -> int:
    """Get an existing phantom node or create one. Returns node_id."""
    with self._write_lock:
        row = self._execute_write(
            "SELECT node_id FROM nodes WHERE name = ? AND kind = ? AND file_id IS NULL LIMIT 1",
            [name, kind],
        ).fetchone()
        if row:
            return row[0]
        # insert_node acquires _write_lock (RLock is re-entrant)
        return self.insert_node(file_id=None, kind=kind, name=name, language=language)

cleanup_phantoms

cleanup_phantoms()

Repoint edges from phantom nodes to real counterparts, then delete phantoms.

A phantom node (file_id IS NULL) can be replaced when a real node with the same name+kind exists. Edges pointing to/from the phantom are updated to reference the real node, then the phantom is deleted.

Returns the number of phantom nodes cleaned up.

Source code in src/sqlprism/core/graph.py

def cleanup_phantoms(self) -> int:
    """Repoint edges from phantom nodes to real counterparts, then delete phantoms.

    A phantom node (file_id IS NULL) can be replaced when a real node with
    the same name+kind exists. Edges pointing to/from the phantom are updated
    to reference the real node, then the phantom is deleted.

    Returns the number of phantom nodes cleaned up.
    """
    with self._write_lock:
        # Find phantoms that have a real counterpart
        phantoms = self._execute_write(
            "SELECT p.node_id AS phantom_id, r.node_id AS real_id "
            "FROM nodes p "
            "JOIN nodes r ON p.name = r.name AND p.kind = r.kind "
            "AND COALESCE(p.schema, '') = COALESCE(r.schema, '') "
            "WHERE p.file_id IS NULL AND r.file_id IS NOT NULL"
        ).fetchall()

        if not phantoms:
            # Still check for orphaned phantoms (no edges at all)
            orphaned = self._execute_write(
                "SELECT node_id FROM nodes "
                "WHERE file_id IS NULL "
                "AND node_id NOT IN (SELECT source_id FROM edges) "
                "AND node_id NOT IN (SELECT target_id FROM edges)"
            ).fetchall()
            # Also find stale phantoms: phantoms whose only inbound edges
            # come from other phantoms (no real node references them).
            stale = self._execute_write(
                "SELECT p.node_id FROM nodes p "
                "WHERE p.file_id IS NULL "
                "AND p.node_id IN (SELECT target_id FROM edges) "
                "AND NOT EXISTS ("
                "  SELECT 1 FROM edges e "
                "  JOIN nodes src ON e.source_id = src.node_id "
                "  WHERE e.target_id = p.node_id AND src.file_id IS NOT NULL"
                ")"
            ).fetchall()
            to_delete = {row[0] for row in orphaned} | {row[0] for row in stale}
            if to_delete:
                delete_ids = list(to_delete)
                placeholders = ",".join(["?"] * len(delete_ids))
                # Remove edges referencing stale phantoms before deleting nodes
                self._execute_write(
                    f"DELETE FROM edges WHERE source_id IN ({placeholders}) OR target_id IN ({placeholders})",
                    delete_ids + delete_ids,
                )
                self._execute_write(
                    f"DELETE FROM nodes WHERE node_id IN ({placeholders})",
                    delete_ids,
                )
                return len(to_delete)
            return 0

        # Batch repoint edges: single UPDATE per direction using a mapping table
        # instead of O(phantoms) individual UPDATEs.
        mapping_values = ", ".join([f"({phantom_id}, {real_id})" for phantom_id, real_id in phantoms])
        self._execute_write(
            f"UPDATE edges SET source_id = m.real_id "
            f"FROM (VALUES {mapping_values}) AS m(phantom_id, real_id) "
            f"WHERE edges.source_id = m.phantom_id"
        )
        self._execute_write(
            f"UPDATE edges SET target_id = m.real_id "
            f"FROM (VALUES {mapping_values}) AS m(phantom_id, real_id) "
            f"WHERE edges.target_id = m.phantom_id"
        )

        # Delete all phantoms that had real counterparts
        phantom_ids = [p[0] for p in phantoms]
        placeholders = ",".join(["?"] * len(phantom_ids))
        self._execute_write(
            f"DELETE FROM nodes WHERE node_id IN ({placeholders})",
            phantom_ids,
        )

        # Clean up orphaned phantoms: phantom nodes with no edges at all
        orphaned = self._execute_write(
            "SELECT node_id FROM nodes "
            "WHERE file_id IS NULL "
            "AND node_id NOT IN (SELECT source_id FROM edges) "
            "AND node_id NOT IN (SELECT target_id FROM edges)"
        ).fetchall()

        if orphaned:
            orphan_ids = [row[0] for row in orphaned]
            placeholders = ",".join(["?"] * len(orphan_ids))
            self._execute_write(
                f"DELETE FROM nodes WHERE node_id IN ({placeholders})",
                orphan_ids,
            )

        return len(phantoms) + len(orphaned)

merge_duplicate_nodes

merge_duplicate_nodes()

Merge stub nodes into their defining counterparts.

When file A references model X (creating a "table" stub node attached to A's file_id) and file B defines model X (creating a "query" or "view" node), edges targeting the stub should be repointed to the defining node. The stub is then deleted.

A "defining" node is one whose kind is NOT "table" (i.e. it was parsed from its own SQL file as a query/view/cte). A "stub" is a "table" node that merely records a reference in another file.

Only merges nodes within the same repo to avoid cross-repo name collisions (e.g. two repos both defining orders).

Returns the number of stub nodes merged.

Source code in src/sqlprism/core/graph.py

def merge_duplicate_nodes(self) -> int:
    """Merge stub nodes into their defining counterparts.

    When file A references model X (creating a ``"table"`` stub node
    attached to A's file_id) and file B defines model X (creating a
    ``"query"`` or ``"view"`` node), edges targeting the stub should be
    repointed to the defining node. The stub is then deleted.

    A "defining" node is one whose kind is NOT ``"table"`` (i.e. it was
    parsed from its own SQL file as a query/view/cte). A "stub" is a
    ``"table"`` node that merely records a reference in another file.

    Only merges nodes within the **same repo** to avoid cross-repo
    name collisions (e.g. two repos both defining ``orders``).

    Returns the number of stub nodes merged.
    """
    with self._write_lock:
        # Find stub → defining pairs in the same repo.
        # Schema match is relaxed: a stub with schema "sushi" (from a
        # qualified reference like sushi.orders) matches a defining node
        # with schema NULL (common for sqlmesh/dbt models whose own file
        # doesn't set a schema). Exact matches are preferred via ORDER BY.
        dupes = self._execute_write(
            "SELECT stub.node_id AS stub_id, def.node_id AS def_id "
            "FROM nodes stub "
            "JOIN files fs ON stub.file_id = fs.file_id "
            "JOIN nodes def ON stub.name = def.name "
            "JOIN files fd ON def.file_id = fd.file_id "
            "WHERE stub.kind = 'table' "
            "  AND def.kind != 'table' "
            "  AND fs.repo_id = fd.repo_id "
            "  AND stub.node_id != def.node_id "
            "  AND (COALESCE(stub.schema, '') = COALESCE(def.schema, '') "
            "       OR def.schema IS NULL OR stub.schema IS NULL)"
        ).fetchall()

        if not dupes:
            return 0

        # Repoint edges from stubs to defining nodes
        mapping_values = ", ".join(
            [f"({stub_id}, {def_id})" for stub_id, def_id in dupes]
        )
        self._execute_write(
            f"UPDATE edges SET source_id = m.def_id "
            f"FROM (VALUES {mapping_values}) AS m(stub_id, def_id) "
            f"WHERE edges.source_id = m.stub_id"
        )
        self._execute_write(
            f"UPDATE edges SET target_id = m.def_id "
            f"FROM (VALUES {mapping_values}) AS m(stub_id, def_id) "
            f"WHERE edges.target_id = m.stub_id"
        )

        # Delete duplicate edges that may have been created by the merge
        self._execute_write(
            "DELETE FROM edges WHERE rowid NOT IN ("
            "  SELECT MIN(rowid) FROM edges "
            "  GROUP BY source_id, target_id, relationship"
            ")"
        )

        # Delete the stub nodes
        stub_ids = [s[0] for s in dupes]
        placeholders = ",".join(["?"] * len(stub_ids))
        self._execute_write(
            f"DELETE FROM nodes WHERE node_id IN ({placeholders})",
            stub_ids,
        )

        return len(dupes)

insert_edge

insert_edge(
    source_id,
    target_id,
    relationship,
    context=None,
    metadata=None,
)

Insert an edge. Returns edge_id.

Source code in src/sqlprism/core/graph.py

def insert_edge(
    self,
    source_id: int,
    target_id: int,
    relationship: str,
    context: str | None = None,
    metadata: dict | None = None,
) -> int:
    """Insert an edge. Returns edge_id."""
    with self._write_lock:
        result = self._execute_write(
            "INSERT INTO edges (source_id, target_id, relationship, context, metadata) "
            "VALUES (?, ?, ?, ?, ?) RETURNING edge_id",
            [
                source_id,
                target_id,
                relationship,
                context,
                json.dumps(metadata) if metadata else None,
            ],
        ).fetchone()
        return result[0]

insert_nodes_batch

insert_nodes_batch(rows)

Batch insert nodes.

Parameters:

Name	Type	Description	Default
rows	list[tuple]	List of tuples, each containing (file_id, kind, name, language, line_start, line_end, metadata_json, schema).	required

Returns:

Type	Description
list[int]	node_id values in insertion order.

Source code in src/sqlprism/core/graph.py

def insert_nodes_batch(
    self,
    rows: list[tuple],
) -> list[int]:
    """Batch insert nodes.

    Args:
        rows: List of tuples, each containing
            ``(file_id, kind, name, language, line_start, line_end, metadata_json, schema)``.

    Returns:
        ``node_id`` values in insertion order.
    """
    if not rows:
        return []
    chunk_size = 200
    all_ids = []
    with self._write_lock:
        for i in range(0, len(rows), chunk_size):
            chunk = rows[i : i + chunk_size]
            placeholders = ", ".join(["(?, ?, ?, ?, ?, ?, ?, ?)"] * len(chunk))
            flat = [v for row in chunk for v in row]
            result = self.conn.execute(
                "INSERT INTO nodes (file_id, kind, name, language, "
                "line_start, line_end, metadata, schema) "
                f"VALUES {placeholders} RETURNING node_id",
                flat,
            ).fetchall()
            all_ids.extend(r[0] for r in result)
    return all_ids

insert_edges_batch

insert_edges_batch(rows)

Batch insert edges.

Parameters:

Name	Type	Description	Default
rows	list[tuple]	List of tuples, each containing (source_id, target_id, relationship, context, metadata_json).	required

Source code in src/sqlprism/core/graph.py

def insert_edges_batch(self, rows: list[tuple]) -> None:
    """Batch insert edges.

    Args:
        rows: List of tuples, each containing
            ``(source_id, target_id, relationship, context, metadata_json)``.
    """
    if not rows:
        return
    with self._write_lock:
        self.conn.executemany(
            "INSERT INTO edges (source_id, target_id, relationship, context, metadata) VALUES (?, ?, ?, ?, ?)",
            rows,
        )

insert_column_usage_batch

insert_column_usage_batch(rows)

Batch insert column usage records.

Parameters:

Name	Type	Description	Default
rows	list[tuple]	List of tuples, each containing (node_id, table_name, column_name, usage_type, file_id, alias, transform).	required

Source code in src/sqlprism/core/graph.py

def insert_column_usage_batch(self, rows: list[tuple]) -> None:
    """Batch insert column usage records.

    Args:
        rows: List of tuples, each containing
            ``(node_id, table_name, column_name, usage_type, file_id, alias, transform)``.
    """
    if not rows:
        return
    with self._write_lock:
        self.conn.executemany(
            "INSERT INTO column_usage (node_id, table_name, column_name, "
            "usage_type, file_id, alias, transform) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            rows,
        )

insert_column_lineage_batch

insert_column_lineage_batch(rows)

Batch insert column lineage hops.

Parameters:

Name	Type	Description	Default
rows	list[tuple]	List of tuples, each containing (file_id, output_node, output_column, chain_index, hop_index, hop_column, hop_table, hop_expression).	required

Source code in src/sqlprism/core/graph.py

def insert_column_lineage_batch(self, rows: list[tuple]) -> None:
    """Batch insert column lineage hops.

    Args:
        rows: List of tuples, each containing
            ``(file_id, output_node, output_column, chain_index,
            hop_index, hop_column, hop_table, hop_expression)``.
    """
    if not rows:
        return
    with self._write_lock:
        self.conn.executemany(
            "INSERT INTO column_lineage "
            "(file_id, output_node, output_column, chain_index, "
            "hop_index, hop_column, hop_table, hop_expression) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            rows,
        )

insert_columns_batch

insert_columns_batch(rows)

Batch insert/upsert column definitions.

Parameters:

Name	Type	Description	Default
rows	list[tuple]	List of tuples, each containing (node_id, column_name, data_type, position, source, description).	required

Returns:

Type	Description
int	Number of rows processed (inserts + upserts).

Source code in src/sqlprism/core/graph.py

def insert_columns_batch(self, rows: list[tuple]) -> int:
    """Batch insert/upsert column definitions.

    Args:
        rows: List of tuples, each containing
            ``(node_id, column_name, data_type, position, source,
            description)``.

    Returns:
        Number of rows processed (inserts + upserts).
    """
    if not rows:
        return 0
    with self._write_lock:
        self.conn.executemany(
            "INSERT INTO columns "
            "(node_id, column_name, data_type, position, source, description) "
            "VALUES (?, ?, ?, ?, ?, ?) "
            "ON CONFLICT (node_id, column_name) DO UPDATE SET "
            "data_type = COALESCE(EXCLUDED.data_type, columns.data_type), "
            "position = COALESCE(EXCLUDED.position, columns.position), "
            "source = EXCLUDED.source, "
            "description = COALESCE(EXCLUDED.description, columns.description)",
            rows,
        )
    return len(rows)

insert_column_usage

insert_column_usage(
    node_id,
    table_name,
    column_name,
    usage_type,
    file_id,
    alias=None,
    transform=None,
)

Insert a column usage record.

Source code in src/sqlprism/core/graph.py

def insert_column_usage(
    self,
    node_id: int,
    table_name: str,
    column_name: str,
    usage_type: str,
    file_id: int,
    alias: str | None = None,
    transform: str | None = None,
) -> None:
    """Insert a column usage record."""
    with self._write_lock:
        self._execute_write(
            "INSERT INTO column_usage (node_id, table_name, column_name, "
            "usage_type, file_id, alias, transform) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            [node_id, table_name, column_name, usage_type, file_id, alias, transform],
        )

insert_column_lineage

insert_column_lineage(
    file_id,
    output_node,
    output_column,
    hop_index,
    hop_column,
    hop_table,
    hop_expression=None,
    chain_index=0,
)

Insert a single hop in a column lineage chain.

Source code in src/sqlprism/core/graph.py

def insert_column_lineage(
    self,
    file_id: int,
    output_node: str,
    output_column: str,
    hop_index: int,
    hop_column: str,
    hop_table: str,
    hop_expression: str | None = None,
    chain_index: int = 0,
) -> None:
    """Insert a single hop in a column lineage chain."""
    with self._write_lock:
        self._execute_write(
            "INSERT INTO column_lineage "
            "(file_id, output_node, output_column, chain_index, "
            "hop_index, hop_column, hop_table, hop_expression) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            [
                file_id,
                output_node,
                output_column,
                chain_index,
                hop_index,
                hop_column,
                hop_table,
                hop_expression,
            ],
        )

query_column_lineage

query_column_lineage(
    table=None,
    column=None,
    output_node=None,
    repo=None,
    limit=100,
    offset=0,
)

Query column lineage chains.

Can search by:

• output_node + column: "where does this output column come from?"
• table + column at any hop: "where does this source column flow to?"

limit applies to chain count (distinct output_node/output_column/chain_index combinations), not raw hop rows.

Parameters:

Name	Type	Description	Default
table	str | None	Filter by hop table name.	None
column	str | None	Filter by column name (output or any hop, depending on whether output_node is also set).	None
output_node	str | None	Filter by the node that produces the output column.	None
repo	str | None	Filter by repo name.	None
limit	int	Maximum number of lineage chains to return.	100
offset	int	Pagination offset (in chains).	0

Returns:

Type	Description
dict	Dict with keys "chains" (list of chain dicts, each with
dict	output_node, output_column, chain_index, hops,
dict	file, and repo) and "total_count" (int).

Source code in src/sqlprism/core/graph.py

def query_column_lineage(
    self,
    table: str | None = None,
    column: str | None = None,
    output_node: str | None = None,
    repo: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> dict:
    """Query column lineage chains.

    Can search by:

    - ``output_node`` + ``column``: "where does this output column come from?"
    - ``table`` + ``column`` at any hop: "where does this source column flow to?"

    ``limit`` applies to chain count (distinct
    ``output_node``/``output_column``/``chain_index`` combinations),
    not raw hop rows.

    Args:
        table: Filter by hop table name.
        column: Filter by column name (output or any hop, depending
            on whether ``output_node`` is also set).
        output_node: Filter by the node that produces the output column.
        repo: Filter by repo name.
        limit: Maximum number of lineage chains to return.
        offset: Pagination offset (in chains).

    Returns:
        Dict with keys ``"chains"`` (list of chain dicts, each with
        ``output_node``, ``output_column``, ``chain_index``, ``hops``,
        ``file``, and ``repo``) and ``"total_count"`` (int).
    """
    # Build WHERE clauses for both outer (cl) and inner (cl2) aliases
    outer_where: list[str] = []
    inner_where: list[str] = []
    params: list = []

    if output_node:
        outer_where.append("cl.output_node = ?")
        inner_where.append("cl2.output_node = ?")
        params.append(output_node)
    if column:
        if output_node:
            outer_where.append("cl.output_column = ?")
            inner_where.append("cl2.output_column = ?")
            params.append(column)
        else:
            outer_where.append("(cl.output_column = ? OR cl.hop_column = ?)")
            inner_where.append("(cl2.output_column = ? OR cl2.hop_column = ?)")
            params.extend([column, column])
    if table:
        outer_where.append("cl.hop_table = ?")
        inner_where.append("cl2.hop_table = ?")
        params.append(table)
    if repo:
        outer_where.append("r.name = ?")
        inner_where.append("r2.name = ?")
        params.append(repo)

    if not outer_where:
        return {"chains": [], "total_count": 0}

    # True total count of matching chains (before pagination)
    count_sql = (
        "SELECT COUNT(*) FROM ("
        "  SELECT DISTINCT cl2.output_node, cl2.output_column, cl2.chain_index "
        "  FROM column_lineage cl2 "
        "  JOIN files f2 ON cl2.file_id = f2.file_id "
        "  JOIN repos r2 ON f2.repo_id = r2.repo_id "
        f"  WHERE {' AND '.join(inner_where)} "
        ")"
    )
    total_count = self._execute_read(count_sql, params).fetchone()[0]

    # Subquery selects distinct chains with LIMIT, then outer query
    # fetches all hops for those chains. This ensures LIMIT counts chains,
    # not individual hop rows.
    sql = (
        "SELECT cl.output_node, cl.output_column, cl.chain_index, cl.hop_index, "
        "cl.hop_column, cl.hop_table, cl.hop_expression, "
        "f.path, r.name as repo_name "
        "FROM column_lineage cl "
        "JOIN files f ON cl.file_id = f.file_id "
        "JOIN repos r ON f.repo_id = r.repo_id "
        "WHERE (cl.output_node, cl.output_column, cl.chain_index) IN ("
        "  SELECT DISTINCT cl2.output_node, cl2.output_column, cl2.chain_index "
        "  FROM column_lineage cl2 "
        "  JOIN files f2 ON cl2.file_id = f2.file_id "
        "  JOIN repos r2 ON f2.repo_id = r2.repo_id "
        f"  WHERE {' AND '.join(inner_where)} "
        "  ORDER BY cl2.output_node, cl2.output_column, cl2.chain_index "
        "  LIMIT ? OFFSET ?"
        ") "
        "ORDER BY cl.output_node, cl.output_column, cl.chain_index, cl.hop_index"
    )

    # params duplicated: once for inner subquery, once not needed for outer
    # (outer filters via the IN subquery)
    rows = self._execute_read(sql, [*params, limit, offset]).fetchall()

    # Group by (output_node, output_column, chain_index) into chains
    chains: dict[tuple[str, str, int], dict] = {}
    for r in rows:
        key = (r[0], r[1], r[2])
        if key not in chains:
            chains[key] = {
                "output_node": r[0],
                "output_column": r[1],
                "chain_index": r[2],
                "hops": [],
                "file": r[7],
                "repo": r[8],
            }
        chains[key]["hops"].append(
            {
                "index": r[3],
                "column": r[4],
                "table": r[5],
                "expression": r[6],
            }
        )

    return {"chains": list(chains.values()), "total_count": total_count}

get_table_columns

get_table_columns(repo_id=None)

Build a schema catalog from indexed columns and column usage.

Prefers the authoritative columns table (joined with nodes) which carries real data_type information. Falls back to column_usage for any additional columns not present in the columns table, assigning them the default type "TEXT".

When the columns table is empty the behaviour is identical to the previous column-usage-only implementation.

Suitable for passing to sqlglot.optimizer.qualify_columns or sqlglot.lineage.

Parameters:

Name	Type	Description	Default
repo_id	int | None	Restrict to columns from this repo. None returns columns across all repos.	None

Returns:

Type	Description
dict[str, dict[str, str]]	{table_name: {column_name: data_type, ...}} mapping.

Source code in src/sqlprism/core/graph.py

def get_table_columns(self, repo_id: int | None = None) -> dict[str, dict[str, str]]:
    """Build a schema catalog from indexed columns and column usage.

    Prefers the authoritative ``columns`` table (joined with ``nodes``)
    which carries real ``data_type`` information.  Falls back to
    ``column_usage`` for any additional columns not present in the
    ``columns`` table, assigning them the default type ``"TEXT"``.

    When the ``columns`` table is empty the behaviour is identical to
    the previous column-usage-only implementation.

    Suitable for passing to ``sqlglot.optimizer.qualify_columns`` or
    ``sqlglot.lineage``.

    Args:
        repo_id: Restrict to columns from this repo. ``None`` returns
            columns across all repos.

    Returns:
        ``{table_name: {column_name: data_type, ...}}`` mapping.
    """
    schema: dict[str, dict[str, str]] = {}

    # 1. Authoritative columns from columns table (with real types).
    # Note: phantom nodes (file_id IS NULL) are intentionally excluded
    # since they lack a verified repo association.
    if repo_id is not None:
        col_rows = self._execute_read(
            "SELECT n.name, c.column_name, c.data_type "
            "FROM columns c "
            "JOIN nodes n ON c.node_id = n.node_id "
            "JOIN files f ON n.file_id = f.file_id "
            "WHERE f.repo_id = ?",
            [repo_id],
        ).fetchall()
    else:
        col_rows = self._execute_read(
            "SELECT n.name, c.column_name, c.data_type "
            "FROM columns c "
            "JOIN nodes n ON c.node_id = n.node_id"
        ).fetchall()

    for table, col, dtype in col_rows:
        if table not in schema:
            schema[table] = {}
        schema[table][col] = dtype or "TEXT"

    # 2. Fallback: fill gaps from column_usage
    if repo_id is not None:
        usage_rows = self._execute_read(
            "SELECT DISTINCT cu.table_name, cu.column_name "
            "FROM column_usage cu "
            "JOIN files f ON cu.file_id = f.file_id "
            "WHERE f.repo_id = ? AND cu.column_name != '*'",
            [repo_id],
        ).fetchall()
    else:
        usage_rows = self._execute_read(
            "SELECT DISTINCT table_name, column_name FROM column_usage WHERE column_name != '*'"
        ).fetchall()

    for table, col in usage_rows:
        if table not in schema:
            schema[table] = {}
        if col not in schema[table]:  # Don't overwrite columns table entries
            schema[table][col] = "TEXT"

    return schema

get_cross_repo_columns

get_cross_repo_columns(current_repo_id)

Build a schema catalog that spans all repos, with current repo winning.

Upstream mesh refs (e.g. dbt cross-project ref() or sqlmesh multi-project indexing) need column schemas from sibling repos so SELECT * through CTEs can expand. Two passes:

1. Aggregate columns from every other repo (repo_id != current) into a flat catalog. Sibling repos with same-named tables collapse into one entry by design — the catalog is name-addressed, not repo-scoped, so later sibling rows naturally layer over earlier ones.
2. Overlay the current repo's columns so local definitions win deterministically over any sibling collision.

The two-query form avoids the non-deterministic ordering that get_table_columns(None) would produce for same-named tables across repos.

Source code in src/sqlprism/core/graph.py

def get_cross_repo_columns(self, current_repo_id: int) -> dict[str, dict[str, str]]:
    """Build a schema catalog that spans all repos, with current repo winning.

    Upstream mesh refs (e.g. dbt cross-project ``ref()`` or sqlmesh
    multi-project indexing) need column schemas from sibling repos so
    ``SELECT *`` through CTEs can expand. Two passes:

    1. Aggregate columns from every *other* repo (``repo_id != current``)
       into a flat catalog.  Sibling repos with same-named tables collapse
       into one entry by design — the catalog is name-addressed, not
       repo-scoped, so later sibling rows naturally layer over earlier ones.
    2. Overlay the current repo's columns so local definitions win
       deterministically over any sibling collision.

    The two-query form avoids the non-deterministic ordering that
    ``get_table_columns(None)`` would produce for same-named tables across
    repos.
    """
    schema: dict[str, dict[str, str]] = {}

    sibling_rows = self._execute_read(
        "SELECT n.name, c.column_name, c.data_type "
        "FROM columns c "
        "JOIN nodes n ON c.node_id = n.node_id "
        "JOIN files f ON n.file_id = f.file_id "
        "WHERE f.repo_id != ?",
        [current_repo_id],
    ).fetchall()
    for table, col, dtype in sibling_rows:
        schema.setdefault(table, {})[col] = dtype or "TEXT"

    sibling_usage = self._execute_read(
        "SELECT DISTINCT cu.table_name, cu.column_name "
        "FROM column_usage cu "
        "JOIN files f ON cu.file_id = f.file_id "
        "WHERE f.repo_id != ? AND cu.column_name != '*'",
        [current_repo_id],
    ).fetchall()
    for table, col in sibling_usage:
        schema.setdefault(table, {}).setdefault(col, "TEXT")

    # Overlay current repo — local definitions always win.
    for table, cols in self.get_table_columns(current_repo_id).items():
        schema[table] = {**schema.get(table, {}), **cols}
    return schema

query_schema

query_schema(name, repo=None)

Return the full schema for a named table or model.

Includes column definitions with types, descriptions, and the node's upstream/downstream dependencies.

Parameters:

Name	Type	Description	Default
name	str	Entity name to look up.	required
repo	str | None	Optional repo name filter for disambiguation.	None

Returns:

Type	Description
dict	Dict with name, kind, file, repo, columns,
dict	upstream, and downstream keys. Returns an error
dict	key when the entity is not found.

Source code in src/sqlprism/core/graph.py

def query_schema(
    self,
    name: str,
    repo: str | None = None,
) -> dict:
    """Return the full schema for a named table or model.

    Includes column definitions with types, descriptions, and the
    node's upstream/downstream dependencies.

    Args:
        name: Entity name to look up.
        repo: Optional repo name filter for disambiguation.

    Returns:
        Dict with ``name``, ``kind``, ``file``, ``repo``, ``columns``,
        ``upstream``, and ``downstream`` keys.  Returns an ``error``
        key when the entity is not found.
    """
    # 1. Find the node — ORDER BY prefers real nodes over phantoms
    if repo:
        node_rows = self._execute_read(
            "SELECT n.node_id, n.kind, f.path, r.name "
            "FROM nodes n "
            "LEFT JOIN files f ON n.file_id = f.file_id "
            "LEFT JOIN repos r ON f.repo_id = r.repo_id "
            "WHERE n.name = ? AND r.name = ? "
            "ORDER BY (n.file_id IS NULL), n.node_id DESC",
            [name, repo],
        ).fetchall()
    else:
        node_rows = self._execute_read(
            "SELECT n.node_id, n.kind, f.path, r.name "
            "FROM nodes n "
            "LEFT JOIN files f ON n.file_id = f.file_id "
            "LEFT JOIN repos r ON f.repo_id = r.repo_id "
            "WHERE n.name = ? AND n.file_id IS NOT NULL "
            "ORDER BY n.node_id DESC",
            [name],
        ).fetchall()

    if not node_rows:
        return {"error": f"Model '{name}' not found in the index."}

    # Use the first match (real node preferred) for all queries
    first = node_rows[0]
    node_id = first[0]
    node_kind = first[1]
    file_path = first[2]
    repo_name = first[3]

    # 2. Get columns
    col_rows = self._execute_read(
        "SELECT column_name, data_type, position, source, description "
        "FROM columns WHERE node_id = ? ORDER BY position",
        [node_id],
    ).fetchall()

    columns = [
        {
            "name": r[0],
            "type": r[1] or "UNKNOWN",
            "position": r[2],
            "source": r[3],
            "description": r[4],
        }
        for r in col_rows
    ]

    # 3. Get upstream (outbound edges — what this node references)
    upstream_rows = self._execute_read(
        "SELECT DISTINCT n2.name, n2.kind "
        "FROM edges e "
        "JOIN nodes n2 ON e.target_id = n2.node_id "
        "WHERE e.source_id = ?",
        [node_id],
    ).fetchall()

    upstream = [{"name": r[0], "kind": r[1]} for r in upstream_rows]

    # 4. Get downstream (inbound edges — what depends on this node)
    downstream_rows = self._execute_read(
        "SELECT DISTINCT n2.name, n2.kind "
        "FROM edges e "
        "JOIN nodes n2 ON e.source_id = n2.node_id "
        "WHERE e.target_id = ?",
        [node_id],
    ).fetchall()

    downstream = [{"name": r[0], "kind": r[1]} for r in downstream_rows]

    result = {
        "name": name,
        "kind": node_kind,
        "file": file_path,
        "repo": repo_name,
        "columns": columns,
        "upstream": upstream,
        "downstream": downstream,
    }
    if len(node_rows) > 1:
        result["matches"] = len(node_rows)
    return result

query_check_impact

query_check_impact(model, changes, repo=None)

Analyze downstream impact of column changes on a model.

For each proposed change (column removal, rename, or addition), queries the column_usage table to classify downstream models as breaking, warning, or safe.

Note: add_column does not account for SELECT * usage — downstream models using wildcard selects may still be affected.

The repo filter restricts the producer lookup (which model_found is based on) — it answers "does this repo define the named model?". Downstream consumer discovery deliberately spans all repos: consumer files in other repos index each cross-project ref() as a local shadow of the producer's name, and omitting those would hide the headline use case of cross-repo impact analysis (#131).

Name-collision caveat: downstream discovery matches on node name, so two repos that independently define unrelated models with the same name will be conflated — consumers of repo B's users are reported as impacts on repo A's users. This is the same tradeoff made by the trace traversal; naming should be globally unique within a federated project or impact results will include false positives.

Parameters:

Name	Type	Description	Default
model	str	The model/table name whose columns are changing.	required
changes	list[dict]	List of change dicts. Supported actions: - {"action": "remove_column", "column": "col"} - {"action": "rename_column", "old": "old", "new": "new"} - {"action": "add_column", "column": "col"}	required
repo	str | None	Optional repo name filter for the producer lookup.	None

Returns:

Type	Description
dict	Dict with model, model_found, changes_analyzed,
dict	impacts (one entry per change with breaking,
dict	warnings, and safe lists), and a summary with
dict	totals.

Source code in src/sqlprism/core/graph.py

def query_check_impact(
    self,
    model: str,
    changes: list[dict],
    repo: str | None = None,
) -> dict:
    """Analyze downstream impact of column changes on a model.

    For each proposed change (column removal, rename, or addition),
    queries the ``column_usage`` table to classify downstream models
    as **breaking**, **warning**, or **safe**.

    Note: ``add_column`` does not account for ``SELECT *`` usage —
    downstream models using wildcard selects may still be affected.

    The ``repo`` filter restricts the **producer lookup** (which
    ``model_found`` is based on) — it answers "does this repo define
    the named model?". Downstream consumer discovery deliberately
    spans all repos: consumer files in other repos index each
    cross-project ``ref()`` as a local shadow of the producer's
    name, and omitting those would hide the headline use case of
    cross-repo impact analysis (#131).

    **Name-collision caveat**: downstream discovery matches on node
    *name*, so two repos that independently define unrelated models
    with the same name will be conflated — consumers of repo B's
    ``users`` are reported as impacts on repo A's ``users``. This is
    the same tradeoff made by the trace traversal; naming should be
    globally unique within a federated project or impact results
    will include false positives.

    Args:
        model: The model/table name whose columns are changing.
        changes: List of change dicts.  Supported actions:
            - ``{"action": "remove_column", "column": "col"}``
            - ``{"action": "rename_column", "old": "old", "new": "new"}``
            - ``{"action": "add_column", "column": "col"}``
        repo: Optional repo name filter for the producer lookup.

    Returns:
        Dict with ``model``, ``model_found``, ``changes_analyzed``,
        ``impacts`` (one entry per change with ``breaking``,
        ``warnings``, and ``safe`` lists), and a ``summary`` with
        totals.
    """
    breaking_types = {"select", "join_on", "insert", "update"}
    warning_types = {
        "where", "group_by", "order_by", "having",
        "partition_by", "window_order", "qualify",
    }

    # Producer lookup — scoped by repo when provided — establishes
    # whether the named model exists. Phantoms excluded to avoid
    # false positives for unresolved refs.
    if repo:
        producer_rows = self._execute_read(
            "SELECT n.node_id FROM nodes n "
            "JOIN files f ON n.file_id = f.file_id "
            "JOIN repos r ON f.repo_id = r.repo_id "
            "WHERE n.name = ? AND r.name = ?",
            [model, repo],
        ).fetchall()
    else:
        producer_rows = self._execute_read(
            "SELECT n.node_id FROM nodes n "
            "WHERE n.name = ? AND n.file_id IS NOT NULL",
            [model],
        ).fetchall()

    if not producer_rows:
        return {
            "model": model,
            "model_found": False,
            "changes_analyzed": len(changes),
            "impacts": [],
            "summary": {"total_breaking": 0, "total_warnings": 0, "total_safe": 0},
        }

    # Downstream discovery spans every node with this name — real or
    # shadow, any repo. Consumer files in other repos point their
    # ref() edges at a local shadow (same name), so restricting the
    # target set to the producer repo's node would miss them (#131).
    same_name_rows = self._execute_read(
        "SELECT node_id FROM nodes WHERE name = ?",
        [model],
    ).fetchall()
    target_ids = [r[0] for r in same_name_rows]
    if not target_ids:
        # Producer existed at the first lookup but no same-name nodes
        # remain — nothing reachable, return an empty shape rather
        # than building `IN ()` which DuckDB rejects.
        return {
            "model": model,
            "model_found": True,
            "changes_analyzed": len(changes),
            "impacts": [
                {"change": c, "breaking": [], "warnings": [], "safe": []}
                for c in changes
            ],
            "summary": {"total_breaking": 0, "total_warnings": 0, "total_safe": 0},
        }
    placeholders = ", ".join("?" for _ in target_ids)

    # Fetch downstream models via edges, excluding the model itself
    # (by name, so shadow self-loops in other repos are also dropped).
    # The explicit non-dataflow filter keeps defines / inserts_into
    # file-stem self-loops out of the downstream set (#127).
    ds_rows = self._execute_read(
        "SELECT DISTINCT n2.name, n2.kind "
        "FROM edges e "
        "JOIN nodes n2 ON e.source_id = n2.node_id "
        "JOIN nodes nt ON nt.node_id = e.target_id "
        f"WHERE e.target_id IN ({placeholders}) "
        "AND n2.name <> ? "
        f"AND {self._non_dataflow_edge_filter('e', 'n2', 'nt')}",
        [*target_ids, model],
    ).fetchall()
    all_downstream = [{"name": r[0], "kind": r[1]} for r in ds_rows]

    impacts: list[dict] = []
    total_breaking = 0
    total_warnings = 0
    total_safe = 0

    for change in changes:
        action = change.get("action", "")

        # Determine the column name to check
        if action == "add_column":
            # Always safe — no downstream references exist yet
            # (does not account for SELECT * usage)
            impacts.append({
                "change": change,
                "breaking": [],
                "warnings": [],
                "safe": [{"model": d["name"], "kind": d["kind"]} for d in all_downstream],
            })
            total_safe += len(all_downstream)
            continue
        elif action == "rename_column":
            col_name = change.get("old", "")
        elif action == "remove_column":
            col_name = change.get("column", "")
        else:
            # Unknown action — record as skipped so len(impacts) == changes_analyzed
            impacts.append({
                "change": change,
                "skipped": True,
                "reason": f"unknown action '{action}'",
                "breaking": [],
                "warnings": [],
                "safe": [],
            })
            continue

        # Query column_usage for downstream references to this column.
        # The consumer side is intentionally unscoped — repo filter only
        # gates the producer lookup above, not which consumers are
        # considered (#131).
        usage_sql = (
            "SELECT DISTINCT n.name AS node_name, n.kind AS node_kind, cu.usage_type "
            "FROM column_usage cu "
            "JOIN nodes n ON cu.node_id = n.node_id "
            "WHERE cu.table_name = ? AND cu.column_name = ?"
        )
        usage_rows = self._execute_read(usage_sql, [model, col_name]).fetchall()

        # Group usage_types per downstream model
        model_usage: dict[tuple[str, str], list[str]] = {}
        for r in usage_rows:
            key = (r[0], r[1])  # (node_name, node_kind)
            model_usage.setdefault(key, []).append(r[2])

        breaking: list[dict] = []
        warnings: list[dict] = []
        affected_names: set[str] = set()

        for (name, kind), usage_types in model_usage.items():
            affected_names.add(name)
            types_set = set(usage_types)
            if types_set & breaking_types:
                breaking.append({
                    "model": name,
                    "kind": kind,
                    "usage_types": sorted(types_set),
                })
            elif types_set & warning_types:
                warnings.append({
                    "model": name,
                    "kind": kind,
                    "usage_types": sorted(types_set),
                })

        # Safe = downstream models not in breaking or warning
        safe = [
            {"model": d["name"], "kind": d["kind"]}
            for d in all_downstream
            if d["name"] not in affected_names
        ]

        impacts.append({
            "change": change,
            "breaking": breaking,
            "warnings": warnings,
            "safe": safe,
        })
        total_breaking += len(breaking)
        total_warnings += len(warnings)
        total_safe += len(safe)

    return {
        "model": model,
        "model_found": True,
        "changes_analyzed": len(changes),
        "impacts": impacts,
        "summary": {
            "total_breaking": total_breaking,
            "total_warnings": total_warnings,
            "total_safe": total_safe,
        },
    }

query_find_path

query_find_path(from_model, to_model, max_hops=10)

Find the shortest path between two models using DuckPGQ.

Uses ANY SHORTEST for path-length discovery, then a BFS CTE to recover intermediate node names.

Parameters:

Name	Type	Description	Default
from_model	str	Source model name.	required
to_model	str	Target model name.	required
max_hops	int	Maximum edge traversals (clamped to 1..10).	10

Returns:

Type	Description
dict	Dict with path_found, path (list of node names),
dict	and length; or an error key when DuckPGQ is missing.

Source code in src/sqlprism/core/graph.py

def query_find_path(
    self,
    from_model: str,
    to_model: str,
    max_hops: int = 10,
) -> dict:
    """Find the shortest path between two models using DuckPGQ.

    Uses ``ANY SHORTEST`` for path-length discovery, then a BFS CTE
    to recover intermediate node names.

    Args:
        from_model: Source model name.
        to_model:   Target model name.
        max_hops:   Maximum edge traversals (clamped to 1..10).

    Returns:
        Dict with ``path_found``, ``path`` (list of node names),
        and ``length``; or an ``error`` key when DuckPGQ is missing.
    """
    if not self.has_pgq:
        return {
            "error": (
                "DuckPGQ not installed. "
                "Install with: INSTALL duckpgq FROM community"
            ),
        }

    max_hops = max(min(max_hops, 10), 1)

    # Resolve model names to node_ids via parameterized query
    # (avoids string interpolation into GRAPH_TABLE SQL)
    from_row = self._execute_read(
        "SELECT node_id FROM nodes WHERE name = ? LIMIT 1",
        [from_model],
    ).fetchone()
    to_row = self._execute_read(
        "SELECT node_id FROM nodes WHERE name = ? LIMIT 1",
        [to_model],
    ).fetchone()
    if not from_row or not to_row:
        return {
            "from": from_model,
            "to": to_model,
            "path_found": False,
            "path": [],
            "length": 0,
        }

    from_id, to_id = from_row[0], to_row[0]

    # Step 1: Reachability probe via PGQ ANY SHORTEST.
    # DuckPGQ does not support bind parameters inside GRAPH_TABLE, so
    # integer node_ids are interpolated (safe, no escaping needed).
    # PGQ's quantified edge binding does not accept a per-edge WHERE
    # predicate — so this probe walks non-dataflow edges too. It is
    # used only as a "is there *any* path within max_hops?" hint; the
    # true shortest dataflow path is computed by the BFS below (#127).
    try:
        rows = self._execute_read(
            f"FROM GRAPH_TABLE (sqlprism_graph "
            f"MATCH p = ANY SHORTEST "
            f"(src:nodes WHERE src.node_id = {from_id})"
            f"-[e:edges]->{{1,{max_hops}}}"
            f"(dst:nodes WHERE dst.node_id = {to_id}) "
            f"COLUMNS (path_length(p) AS hops))",
        ).fetchall()
    except duckdb.Error as e:
        logger.warning("DuckPGQ find_path failed: %s", e)
        return {"error": f"Graph query failed: {e}"}

    if not rows:
        return {
            "from": from_model,
            "to": to_model,
            "path_found": False,
            "path": [],
            "length": 0,
        }

    # Step 2: BFS up to max_hops, filtered to dataflow edges only.
    # Returns the shortest dataflow path. If PGQ reported a path whose
    # shortest form used a defines/inserts_into self-loop, the BFS may
    # find a strictly longer dataflow path — or none. Either way the
    # answer here is correct and the PGQ-reported length is ignored.
    path_cte = f"""
    WITH RECURSIVE path_bfs AS (
        SELECT n.node_id, n.name, 0 as depth,
               ARRAY[n.name] as path_names
        FROM nodes n
        WHERE n.node_id = ?
        UNION ALL
        SELECT n2.node_id, n2.name, pb.depth + 1,
               array_append(pb.path_names, n2.name)
        FROM edges e
        JOIN nodes ns ON ns.node_id = e.source_id
        JOIN nodes n2 ON e.target_id = n2.node_id
        JOIN path_bfs pb ON e.source_id = pb.node_id
        WHERE pb.depth < {max_hops}
        AND NOT array_contains(pb.path_names, n2.name)
        AND {self._non_dataflow_edge_filter('e', 'ns', 'n2')}
    )
    SELECT path_names, depth FROM path_bfs
    WHERE node_id = ?
    ORDER BY depth
    LIMIT 1
    """
    path_rows = self._execute_read(
        path_cte, [from_id, to_id]
    ).fetchall()

    if path_rows:
        path_names = list(path_rows[0][0])
        length = int(path_rows[0][1])
    else:
        path_names = []
        length = 0

    return {
        "from": from_model,
        "to": to_model,
        "path_found": bool(path_names),
        "path": path_names,
        "length": length,
    }

query_find_critical_models

query_find_critical_models(top_n=20, repo=None)

Rank models by importance using PageRank and direct dependent count.

PageRank is computed over the full graph (all repos). The repo filter limits which models are returned, not the computation scope. This means a model's importance reflects cross-repo references.

Parameters:

Name	Type	Description	Default
top_n	int	Number of top models to return (default 20, max 100).	20
repo	str | None	Optional repo name filter.	None

Returns:

Type	Description
dict	Dict with models list and total_indexed_nodes; or
dict	error when DuckPGQ is not installed.

Source code in src/sqlprism/core/graph.py

def query_find_critical_models(
    self,
    top_n: int = 20,
    repo: str | None = None,
) -> dict:
    """Rank models by importance using PageRank and direct dependent count.

    PageRank is computed over the full graph (all repos). The ``repo``
    filter limits which models are returned, not the computation scope.
    This means a model's importance reflects cross-repo references.

    Args:
        top_n: Number of top models to return (default 20, max 100).
        repo: Optional repo name filter.

    Returns:
        Dict with ``models`` list and ``total_indexed_nodes``; or
        ``error`` when DuckPGQ is not installed.
    """
    if not self.has_pgq:
        return {"error": "DuckPGQ not installed. Install with: INSTALL duckpgq FROM community"}

    top_n = max(min(top_n, 100), 1)

    # PageRank scores — join back to nodes for names and node_ids
    # pagerank() returns (node_id, pagerank)
    try:
        if repo:
            pr_sql = (
                "SELECT n.node_id, n.name, n.kind, pr.pagerank "
                "FROM pagerank(sqlprism_graph, nodes, edges) pr "
                "JOIN nodes n ON n.node_id = pr.node_id "
                "JOIN files f ON n.file_id = f.file_id "
                "JOIN repos r ON f.repo_id = r.repo_id "
                "WHERE n.file_id IS NOT NULL AND r.name = ? "
                "ORDER BY pr.pagerank DESC LIMIT ?"
            )
            rows = self._execute_read(pr_sql, [repo, top_n]).fetchall()
        else:
            pr_sql = (
                "SELECT n.node_id, n.name, n.kind, pr.pagerank "
                "FROM pagerank(sqlprism_graph, nodes, edges) pr "
                "JOIN nodes n ON n.node_id = pr.node_id "
                "WHERE n.file_id IS NOT NULL "
                "ORDER BY pr.pagerank DESC LIMIT ?"
            )
            rows = self._execute_read(pr_sql, [top_n]).fetchall()
    except duckdb.Error as e:
        logger.warning("PageRank query failed: %s", e)
        return {"error": f"Graph query failed: {e}"}

    if not rows:
        if repo:
            total = self._execute_read(
                "SELECT COUNT(*) FROM nodes n JOIN files f ON n.file_id = f.file_id "
                "JOIN repos r ON f.repo_id = r.repo_id WHERE r.name = ?",
                [repo],
            ).fetchone()[0]
        else:
            total = self._execute_read(
                "SELECT COUNT(*) FROM nodes WHERE file_id IS NOT NULL"
            ).fetchone()[0]
        return {"models": [], "total_indexed_nodes": total}

    # Batch downstream count (direct dependents) using node_ids
    node_ids = [r[0] for r in rows]
    placeholders = ",".join("?" for _ in node_ids)
    ds_rows = self._execute_read(
        f"SELECT e.target_id, COUNT(DISTINCT e.source_id) "
        f"FROM edges e WHERE e.target_id IN ({placeholders}) "
        f"GROUP BY e.target_id",
        node_ids,
    ).fetchall()
    ds_map: dict[int, int] = {r[0]: r[1] for r in ds_rows}

    models = []
    for node_id, name, kind, pagerank_score in rows:
        models.append({
            "name": name,
            "kind": kind,
            "importance": round(pagerank_score, 6),
            "direct_dependents": ds_map.get(node_id, 0),
        })

    # Total indexed node count
    if repo:
        total = self._execute_read(
            "SELECT COUNT(*) FROM nodes n JOIN files f ON n.file_id = f.file_id "
            "JOIN repos r ON f.repo_id = r.repo_id WHERE r.name = ?",
            [repo],
        ).fetchone()[0]
    else:
        total = self._execute_read(
            "SELECT COUNT(*) FROM nodes WHERE file_id IS NOT NULL"
        ).fetchone()[0]

    return {
        "models": models,
        "total_indexed_nodes": total,
    }

query_detect_cycles

query_detect_cycles(repo=None, max_cycle_length=10)

Detect circular dependencies in the SQL dependency graph.

Uses a recursive CTE with revisit detection. No DuckPGQ required. Self-loops (a -> a) are not detected since they require depth < 1.

length in each cycle dict counts edges (equals number of distinct nodes in the cycle).

Parameters:

Name	Type	Description	Default
repo	str | None	Optional repo name filter (both edge endpoints must belong to the specified repo).	None
max_cycle_length	int	Maximum cycle length in edges (default 10, max 15, min 2).	10

Returns:

Type	Description
dict	Dict with has_cycles, cycles list, and
dict	total_nodes_in_scope.

Source code in src/sqlprism/core/graph.py

def query_detect_cycles(
    self,
    repo: str | None = None,
    max_cycle_length: int = 10,
) -> dict:
    """Detect circular dependencies in the SQL dependency graph.

    Uses a recursive CTE with revisit detection. No DuckPGQ required.
    Self-loops (a -> a) are not detected since they require depth < 1.

    ``length`` in each cycle dict counts edges (equals number of
    distinct nodes in the cycle).

    Args:
        repo: Optional repo name filter (both edge endpoints must
            belong to the specified repo).
        max_cycle_length: Maximum cycle length in edges (default 10,
            max 15, min 2).

    Returns:
        Dict with ``has_cycles``, ``cycles`` list, and
        ``total_nodes_in_scope``.
    """
    max_cycle_length = max(min(max_cycle_length, 15), 2)

    # Build a filtered edge CTE when repo is specified.
    # Both source and target must be in the repo to prevent
    # cross-repo edge leakage.
    if repo:
        edge_cte = (
            "edge_set AS ("
            "SELECT e.source_id, e.target_id FROM edges e "
            "JOIN nodes n1 ON e.source_id = n1.node_id "
            "JOIN files f1 ON n1.file_id = f1.file_id "
            "JOIN repos r1 ON f1.repo_id = r1.repo_id "
            "JOIN nodes n2 ON e.target_id = n2.node_id "
            "JOIN files f2 ON n2.file_id = f2.file_id "
            "JOIN repos r2 ON f2.repo_id = r2.repo_id "
            "WHERE r1.name = ? AND r2.name = ?), "
        )
        params: list = [repo, repo]
    else:
        edge_cte = "edge_set AS (SELECT source_id, target_id FROM edges), "
        params = []

    sql = f"""
    WITH RECURSIVE {edge_cte}
    cycle_detect AS (
        SELECT es.source_id AS start_node, es.target_id AS current_node,
               [es.source_id] AS path_ids, 1 AS depth, false AS is_cycle
        FROM edge_set es
        UNION ALL
        SELECT cd.start_node, es.target_id,
               list_append(cd.path_ids, cd.current_node),
               cd.depth + 1, es.target_id = cd.start_node
        FROM edge_set es
        JOIN cycle_detect cd ON es.source_id = cd.current_node
        WHERE cd.depth < ? AND NOT cd.is_cycle
        AND NOT list_contains(cd.path_ids, cd.current_node)
    )
    SELECT DISTINCT start_node, path_ids, depth
    FROM cycle_detect WHERE is_cycle
    ORDER BY depth, start_node
    LIMIT 100
    """
    params.append(max_cycle_length)

    rows = self._execute_read(sql, params).fetchall()

    # Deduplicate rotations: normalize each cycle by sorting and using min rotation
    seen_cycles: set[tuple] = set()
    cycles = []
    for start_node, path_ids, depth in rows:
        # Normalize: find the canonical rotation (start from smallest node_id)
        node_list = list(path_ids)
        min_idx = node_list.index(min(node_list))
        canonical = tuple(node_list[min_idx:] + node_list[:min_idx])
        if canonical in seen_cycles:
            continue
        seen_cycles.add(canonical)

        # Resolve names
        full_path_ids = [*node_list, start_node]
        placeholders = ",".join("?" for _ in full_path_ids)
        name_rows = self._execute_read(
            f"SELECT node_id, name FROM nodes WHERE node_id IN ({placeholders})",
            full_path_ids,
        ).fetchall()
        id_to_name = {r[0]: r[1] for r in name_rows}
        path_names = [id_to_name.get(nid, str(nid)) for nid in full_path_ids]

        cycles.append({
            "path": path_names,
            "length": depth,
        })

    # Total nodes checked
    if repo:
        total = self._execute_read(
            "SELECT COUNT(*) FROM nodes n JOIN files f ON n.file_id = f.file_id "
            "JOIN repos r ON f.repo_id = r.repo_id WHERE r.name = ?",
            [repo],
        ).fetchone()[0]
    else:
        total = self._execute_read(
            "SELECT COUNT(*) FROM nodes WHERE file_id IS NOT NULL"
        ).fetchone()[0]

    return {
        "has_cycles": len(cycles) > 0,
        "cycles": cycles,
        "total_nodes_in_scope": total,
    }

query_find_subgraphs

query_find_subgraphs(repo=None)

Find weakly connected components (subgraphs) in the dependency graph.

Uses DuckPGQ weakly_connected_component to partition the graph into disjoint subgraphs. Phantom nodes (file_id IS NULL) are excluded from the results.

Parameters:

Name	Type	Description	Default
repo	str | None	Optional repo name filter.	None

Returns:

Type	Description
dict	Dict with components, total_components,
dict	largest_component, orphaned_models, and
dict	total_nodes_in_scope; or error when DuckPGQ is not
dict	installed or the query fails.

Source code in src/sqlprism/core/graph.py

def query_find_subgraphs(self, repo: str | None = None) -> dict:
    """Find weakly connected components (subgraphs) in the dependency graph.

    Uses DuckPGQ ``weakly_connected_component`` to partition the graph
    into disjoint subgraphs.  Phantom nodes (file_id IS NULL) are
    excluded from the results.

    Args:
        repo: Optional repo name filter.

    Returns:
        Dict with ``components``, ``total_components``,
        ``largest_component``, ``orphaned_models``, and
        ``total_nodes_in_scope``; or ``error`` when DuckPGQ is not
        installed or the query fails.
    """
    if not self.has_pgq:
        return {"error": "DuckPGQ not installed. Install with: INSTALL duckpgq FROM community"}

    # Build WCC query — repo filter applied post-WCC via JOIN.
    # NOTE: WCC runs on the full property graph; repo filtering only
    # restricts which nodes appear in the result, not the component
    # assignments. Cross-repo edges may merge otherwise-disconnected
    # repo-local subgraphs into one component.
    repo_join = (
        "JOIN files f ON n.file_id = f.file_id "
        "JOIN repos r ON f.repo_id = r.repo_id "
    ) if repo else ""
    repo_where = "AND r.name = ? " if repo else ""
    params: list = [repo] if repo else []

    wcc_sql = (
        "SELECT n.name, wcc.componentId "
        "FROM weakly_connected_component(sqlprism_graph, nodes, edges) wcc "
        "JOIN nodes n ON n.node_id = wcc.node_id "
        f"{repo_join}"
        f"WHERE n.file_id IS NOT NULL {repo_where}"
    )
    fallback_sql = (
        "SELECT n.name FROM nodes n "
        f"{repo_join}"
        f"WHERE n.file_id IS NOT NULL {repo_where}"
    )

    try:
        rows = self._execute_read(wcc_sql, params).fetchall()
    except duckdb.Error as e:
        # WCC requires at least one edge; when the graph has no edges
        # (all nodes isolated) DuckPGQ raises a "CSR not found" error.
        # Fall back to treating each node as its own component.
        if "CSR" in str(e):
            fallback_rows = self._execute_read(fallback_sql, params).fetchall()
            rows = [(r[0], i) for i, r in enumerate(fallback_rows)]
        else:
            logger.warning("WCC query failed: %s", e)
            return {"error": f"Graph query failed: {e}"}

    # Group by component_id
    comp_map: dict[int, list[str]] = {}
    for name, component_id in rows:
        comp_map.setdefault(component_id, []).append(name)

    # Build components list sorted by size descending
    comp_list: list[dict[str, int | list[str]]] = [
        {
            "component_id": cid,
            "size": len(names),
            "models": sorted(names),
        }
        for cid, names in comp_map.items()
    ]
    comp_list.sort(key=lambda c: c["size"], reverse=True)

    largest_component: dict[str, object] | None = (
        {"name": comp_list[0]["models"][0], "size": comp_list[0]["size"]}  # type: ignore[index]
        if comp_list else None
    )
    orphaned_models: list[str] = sorted(
        name
        for c in comp_list
        if c["size"] == 1
        for name in c["models"]  # type: ignore[union-attr]
    )

    # Derive total from results to guarantee consistency with components
    total = sum(len(names) for names in comp_map.values())

    return {
        "components": comp_list,
        "total_components": len(comp_list),
        "largest_component": largest_component,
        "orphaned_models": orphaned_models,
        "total_nodes_in_scope": total,
    }

query_find_bottlenecks

query_find_bottlenecks(min_downstream=5, repo=None)

Identify bottleneck models with high fan-out and low clustering.

Uses plain SQL to count dependents and dependencies for each node. When DuckPGQ is available, enriches results with local clustering coefficient to better assess risk.

Terminology (edge convention: source REFERENCES target means source depends on target):

• downstream: models that depend on this node (edges pointing to it — high count = many things break if this node fails).
• upstream: models this node depends on (edges pointing from it — high count = many inputs).

Risk tiers (fixed thresholds):

• high: downstream > 20 and (no clustering data or clustering < 0.1)
• medium: downstream > 10
• low: everything else

When has_clustering is True, a clustering: null value means the node was absent from the LCC result set (e.g. isolated node).

Parameters:

Name	Type	Description	Default
min_downstream	int	Minimum downstream (dependents) count to include a node (default 5, clamped 1-100).	5
repo	str | None	Optional repo name filter.	None

Returns:

Type	Description
dict	Dict with bottlenecks list, total_analyzed, and
dict	has_clustering; each bottleneck includes name, kind,
dict	downstream, upstream, clustering, and risk.

Source code in src/sqlprism/core/graph.py

def query_find_bottlenecks(
    self,
    min_downstream: int = 5,
    repo: str | None = None,
) -> dict:
    """Identify bottleneck models with high fan-out and low clustering.

    Uses plain SQL to count dependents and dependencies for each node.
    When DuckPGQ is available, enriches results with local clustering
    coefficient to better assess risk.

    Terminology (edge convention: ``source REFERENCES target`` means
    source depends on target):

    - ``downstream``: models that depend on this node (edges pointing
      *to* it — high count = many things break if this node fails).
    - ``upstream``: models this node depends on (edges pointing *from*
      it — high count = many inputs).

    Risk tiers (fixed thresholds):

    - **high**: downstream > 20 and (no clustering data or clustering < 0.1)
    - **medium**: downstream > 10
    - **low**: everything else

    When ``has_clustering`` is True, a ``clustering: null`` value means
    the node was absent from the LCC result set (e.g. isolated node).

    Args:
        min_downstream: Minimum downstream (dependents) count to include
            a node (default 5, clamped 1-100).
        repo: Optional repo name filter.

    Returns:
        Dict with ``bottlenecks`` list, ``total_analyzed``, and
        ``has_clustering``; each bottleneck includes ``name``, ``kind``,
        ``downstream``, ``upstream``, ``clustering``, and ``risk``.
    """
    min_downstream = max(min(min_downstream, 100), 1)

    # Build repo join/where dynamically
    repo_join = (
        "JOIN files f ON n.file_id = f.file_id "
        "JOIN repos r ON f.repo_id = r.repo_id "
    ) if repo else ""
    repo_where = "AND r.name = ? " if repo else ""
    params: list = [repo] if repo else []

    # Fan-in / fan-out counts exclude non-dataflow edges (defines and
    # inserts_into self-loops) via a shared CTE — otherwise every
    # indexed model's fan-in is +1 too high via its own file-stem query
    # node (#127). A single `dataflow_edges` CTE centralises the filter
    # so the upstream and downstream counts can't drift.
    fan_sql = (
        "WITH dataflow_edges AS ("
        "SELECT e.source_id, e.target_id "
        "FROM edges e "
        "JOIN nodes ns ON ns.node_id = e.source_id "
        "JOIN nodes nt ON nt.node_id = e.target_id "
        f"WHERE {self._non_dataflow_edge_filter('e', 'ns', 'nt')}"
        "), "
        "upstream_counts AS ("
        "SELECT source_id, COUNT(DISTINCT target_id) AS upstream "
        "FROM dataflow_edges GROUP BY source_id"
        ") "
        "SELECT n.node_id, n.name, n.kind, "
        "COUNT(DISTINCT de.source_id) AS downstream, "
        "COALESCE(uc.upstream, 0) AS upstream "
        "FROM nodes n "
        "LEFT JOIN dataflow_edges de ON de.target_id = n.node_id "
        "LEFT JOIN upstream_counts uc ON uc.source_id = n.node_id "
        f"{repo_join}"
        f"WHERE n.file_id IS NOT NULL {repo_where}"
        "GROUP BY n.node_id, n.name, n.kind, uc.upstream "
        f"HAVING COUNT(DISTINCT de.source_id) >= ? "
        "ORDER BY downstream DESC"
    )
    params.append(min_downstream)

    rows = self._execute_read(fan_sql, params).fetchall()

    # Total nodes in scope — reuse repo filter
    total_sql = (
        f"SELECT COUNT(*) FROM nodes n {repo_join}"
        f"WHERE n.file_id IS NOT NULL {repo_where}"
    )
    total_params: list = [repo] if repo else []
    total = self._execute_read(total_sql, total_params).fetchone()[0]

    # Enrich with local clustering coefficient when DuckPGQ is available.
    # NOTE: LCC runs on the full property graph; when repo is set the
    # clustering values still reflect cross-repo neighbors.
    lcc_map: dict[int, float] = {}
    has_clustering = False
    if self.has_pgq and rows:
        try:
            lcc_rows = self._execute_read(
                "SELECT * FROM local_clustering_coefficient("
                "sqlprism_graph, nodes, edges)"
            ).fetchall()
            lcc_map = {r[0]: r[1] for r in lcc_rows}
            has_clustering = True
        except duckdb.Error:
            lcc_map = {}

    bottlenecks = []
    for node_id, name, kind, downstream, upstream in rows:
        clustering = lcc_map.get(node_id, None)

        if downstream > 20 and (clustering is None or clustering < 0.1):
            risk = "high"
        elif downstream > 10:
            risk = "medium"
        else:
            risk = "low"

        bottlenecks.append({
            "name": name,
            "kind": kind,
            "downstream": downstream,
            "upstream": upstream,
            "clustering": clustering,
            "risk": risk,
        })

    return {
        "bottlenecks": bottlenecks,
        "total_analyzed": total,
        "has_clustering": has_clustering,
    }

query_context

query_context(name, repo=None)

Return comprehensive context for a model.

Composes schema info, column usage summary, a code snippet, and optional graph metrics into a single response.

Parameters:

Name	Type	Description	Default
name	str	Entity name to look up.	required
repo	str | None	Optional repo name filter for disambiguation.	None

Returns:

Type	Description
dict	Dict with model, columns, upstream, downstream,
dict	column_usage_summary, snippet, and optionally
dict	graph_metrics keys.

Source code in src/sqlprism/core/graph.py

def query_context(self, name: str, repo: str | None = None) -> dict:
    """Return comprehensive context for a model.

    Composes schema info, column usage summary, a code snippet,
    and optional graph metrics into a single response.

    Args:
        name: Entity name to look up.
        repo: Optional repo name filter for disambiguation.

    Returns:
        Dict with ``model``, ``columns``, ``upstream``, ``downstream``,
        ``column_usage_summary``, ``snippet``, and optionally
        ``graph_metrics`` keys.
    """
    # 1. Schema lookup
    schema_result = self.query_schema(name, repo)
    if "error" in schema_result:
        return schema_result

    # 2. Column usage summary
    # Note: repo filter scopes to consumers *within* that repo.
    # Cross-repo usage (consumer in repo X referencing model in repo Y)
    # is excluded when a repo filter is applied.
    if repo:
        usage_sql = (
            "SELECT cu.column_name, cu.usage_type, COUNT(*) as cnt "
            "FROM column_usage cu "
            "JOIN files f ON cu.file_id = f.file_id "
            "JOIN repos r ON f.repo_id = r.repo_id "
            "WHERE cu.table_name = ? AND r.name = ? "
            "GROUP BY cu.column_name, cu.usage_type "
            "ORDER BY cnt DESC"
        )
        usage_rows = self._execute_read(usage_sql, [name, repo]).fetchall()
    else:
        usage_sql = (
            "SELECT cu.column_name, cu.usage_type, COUNT(*) as cnt "
            "FROM column_usage cu "
            "WHERE cu.table_name = ? "
            "GROUP BY cu.column_name, cu.usage_type "
            "ORDER BY cnt DESC"
        )
        usage_rows = self._execute_read(usage_sql, [name]).fetchall()

    # Aggregate total usage per column for top-10
    col_totals: dict[str, int] = {}
    join_keys: list[str] = []
    aggregations: list[str] = []
    seen_join: set[str] = set()
    seen_agg: set[str] = set()

    for col_name, usage_type, cnt in usage_rows:
        col_totals[col_name] = col_totals.get(col_name, 0) + cnt
        if usage_type == "join_on" and col_name not in seen_join:
            join_keys.append(col_name)
            seen_join.add(col_name)
        if usage_type in ("group_by", "partition_by") and col_name not in seen_agg:
            aggregations.append(col_name)
            seen_agg.add(col_name)

    most_used = sorted(col_totals, key=lambda c: col_totals[c], reverse=True)[:10]

    # 3. Code snippet (first 30 lines)
    snippet = self._read_snippet(
        schema_result["repo"],
        schema_result["file"],
        1,
        None,
        context_lines=0,
        max_lines=30,
    )

    # 4. Optional graph metrics (edge-based downstream_count, not usage-based)
    graph_metrics = None
    if self.has_pgq:
        try:
            repo_name = schema_result["repo"]
            if repo_name:
                pr_rows = self._execute_read(
                    "SELECT pr.pagerank FROM pagerank(sqlprism_graph, nodes, edges) pr "
                    "JOIN nodes n ON n.node_id = pr.node_id "
                    "JOIN files f ON n.file_id = f.file_id "
                    "JOIN repos r ON f.repo_id = r.repo_id "
                    "WHERE n.name = ? AND r.name = ?",
                    [name, repo_name],
                ).fetchall()
            else:
                pr_rows = self._execute_read(
                    "SELECT pr.pagerank FROM pagerank(sqlprism_graph, nodes, edges) pr "
                    "JOIN nodes n ON n.node_id = pr.node_id WHERE n.name = ?",
                    [name],
                ).fetchall()
            raw_score = pr_rows[0][0] if pr_rows else None
            graph_metrics = {
                "importance": round(raw_score, 6) if raw_score is not None else None,
                "downstream_count": len(schema_result.get("downstream", [])),
            }
        except (duckdb.Error, RuntimeError) as e:
            logger.debug("PageRank query failed: %s", e)
            graph_metrics = None

    # 5. Compose result
    result: dict = {
        "model": {
            "name": name,
            "kind": schema_result["kind"],
            "file": schema_result["file"],
            "repo": schema_result["repo"],
        },
        "columns": schema_result["columns"],
        "upstream": schema_result["upstream"],
        "downstream": schema_result["downstream"],
        "column_usage_summary": {
            "most_used_columns": most_used,
            "downstream_join_keys": join_keys,
            "downstream_aggregations": aggregations,
        },
        "snippet": snippet,
    }
    if graph_metrics is not None:
        result["graph_metrics"] = graph_metrics
    return result

query_references

query_references(
    name,
    kind=None,
    schema=None,
    repo=None,
    direction="both",
    include_snippets=True,
    limit=100,
    offset=0,
)

Find all references to/from a named entity.

defines edges are never surfaced in either direction — they represent CREATE-statement identity (the file-stem query node paired with the table/view it declares), not a real reference. Treating them as references would make a model appear to "reference itself" via its own CREATE wrapper (issue #122). Callers that need to locate the defining file should query the edges table directly.

Parameters:

Name	Type	Description	Default
name	str	Entity name to look up.	required
kind	str | None	Optional node kind filter.	None
schema	str | None	Optional database schema filter.	None
repo	str | None	Optional repo name filter.	None
direction	str	"both", "inbound", or "outbound".	'both'
include_snippets	bool	Attach source code snippets when True.	True
limit	int	Maximum edges per direction.	100
offset	int	Pagination offset.	0

Returns:

Type	Description
dict	Dict with keys "entity" (list of matched node dicts or
dict	None), "inbound" (list of referencing entities), and
dict	"outbound" (list of referenced entities). Each entry
dict	contains name, kind, relationship, context,
dict	file, repo, line, and optionally snippet.

Source code in src/sqlprism/core/graph.py

def query_references(
    self,
    name: str,
    kind: str | None = None,
    schema: str | None = None,
    repo: str | None = None,
    direction: str = "both",
    include_snippets: bool = True,
    limit: int = 100,
    offset: int = 0,
) -> dict:
    """Find all references to/from a named entity.

    ``defines`` edges are never surfaced in either direction — they
    represent CREATE-statement identity (the file-stem query node
    paired with the table/view it declares), not a real reference.
    Treating them as references would make a model appear to
    "reference itself" via its own CREATE wrapper (issue #122).
    Callers that need to locate the defining file should query the
    edges table directly.

    Args:
        name: Entity name to look up.
        kind: Optional node kind filter.
        schema: Optional database schema filter.
        repo: Optional repo name filter.
        direction: ``"both"``, ``"inbound"``, or ``"outbound"``.
        include_snippets: Attach source code snippets when ``True``.
        limit: Maximum edges per direction.
        offset: Pagination offset.

    Returns:
        Dict with keys ``"entity"`` (list of matched node dicts or
        ``None``), ``"inbound"`` (list of referencing entities), and
        ``"outbound"`` (list of referenced entities). Each entry
        contains ``name``, ``kind``, ``relationship``, ``context``,
        ``file``, ``repo``, ``line``, and optionally ``snippet``.
    """
    # Find the target node(s)
    where_clauses = ["n.name = ?"]
    params: list = [name]
    if kind:
        where_clauses.append("n.kind = ?")
        params.append(kind)
    if schema:
        where_clauses.append("n.schema = ?")
        params.append(schema)

    where_str = " AND ".join(where_clauses)
    node_query = f"SELECT n.node_id, n.kind, n.name FROM nodes n WHERE {where_str}"
    target_nodes = self._execute_read(node_query, params).fetchall()

    if not target_nodes:
        return {"entity": None, "inbound": [], "outbound": []}

    node_ids = [row[0] for row in target_nodes]
    placeholders = ",".join(["?"] * len(node_ids))

    result = {
        "entity": [{"node_id": r[0], "kind": r[1], "name": r[2]} for r in target_nodes],
        "inbound": [],
        "outbound": [],
    }

    if direction in ("both", "inbound"):
        inbound_sql = (
            f"SELECT n2.name, n2.kind, e.relationship, e.context, "
            f"f2.path, r2.name as repo_name, n2.line_start, n2.line_end "
            f"FROM edges e "
            f"JOIN nodes n2 ON e.source_id = n2.node_id "
            f"JOIN nodes nt ON nt.node_id = e.target_id "
            f"LEFT JOIN files f2 ON n2.file_id = f2.file_id "
            f"LEFT JOIN repos r2 ON f2.repo_id = r2.repo_id "
            f"WHERE e.target_id IN ({placeholders}) "
            f"AND {self._non_dataflow_edge_filter('e', 'n2', 'nt')} "
            f"LIMIT ? OFFSET ?"
        )
        for r in self._execute_read(inbound_sql, [*node_ids, limit, offset]).fetchall():
            entry = {
                "name": r[0],
                "kind": r[1],
                "relationship": r[2],
                "context": r[3],
                "file": r[4],
                "repo": r[5],
                "line": r[6],
            }
            if include_snippets:
                snippet = self._read_snippet(r[5], r[4], r[6], r[7])
                if snippet:
                    entry["snippet"] = snippet
            if not self._is_noise_node(entry):
                result["inbound"].append(entry)

    if direction in ("both", "outbound"):
        outbound_sql = (
            f"SELECT n2.name, n2.kind, e.relationship, e.context, "
            f"f2.path, r2.name as repo_name, n2.line_start, n2.line_end "
            f"FROM edges e "
            f"JOIN nodes ns ON ns.node_id = e.source_id "
            f"JOIN nodes n2 ON e.target_id = n2.node_id "
            f"LEFT JOIN files f2 ON n2.file_id = f2.file_id "
            f"LEFT JOIN repos r2 ON f2.repo_id = r2.repo_id "
            f"WHERE e.source_id IN ({placeholders}) "
            f"AND {self._non_dataflow_edge_filter('e', 'ns', 'n2')} "
            f"LIMIT ? OFFSET ?"
        )
        for r in self._execute_read(outbound_sql, [*node_ids, limit, offset]).fetchall():
            entry = {
                "name": r[0],
                "kind": r[1],
                "relationship": r[2],
                "context": r[3],
                "file": r[4],
                "repo": r[5],
                "line": r[6],
            }
            if include_snippets:
                snippet = self._read_snippet(r[5], r[4], r[6], r[7])
                if snippet:
                    entry["snippet"] = snippet
            if not self._is_noise_node(entry):
                result["outbound"].append(entry)

    return result

query_column_usage

query_column_usage(
    table,
    column=None,
    usage_type=None,
    repo=None,
    limit=100,
    offset=0,
)

Find column usage records for a table.

Parameters:

Name	Type	Description	Default
table	str	Table name to search for.	required
column	str | None	Optional column name filter.	None
usage_type	str | None	Optional usage type filter (e.g. "select", "where").	None
repo	str | None	Optional repo name filter.	None
limit	int	Maximum records to return.	100
offset	int	Pagination offset.	0

Returns:

Type	Description
dict	Dict with keys "usage" (list of usage dicts with table,
dict	column, usage_type, alias, node_name,
dict	node_kind, file, repo, line, and optionally
dict	transform), "summary" (dict mapping usage_type to count),
dict	and "total_count" (int).

Source code in src/sqlprism/core/graph.py

def query_column_usage(
    self,
    table: str,
    column: str | None = None,
    usage_type: str | None = None,
    repo: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> dict:
    """Find column usage records for a table.

    Args:
        table: Table name to search for.
        column: Optional column name filter.
        usage_type: Optional usage type filter (e.g. ``"select"``, ``"where"``).
        repo: Optional repo name filter.
        limit: Maximum records to return.
        offset: Pagination offset.

    Returns:
        Dict with keys ``"usage"`` (list of usage dicts with ``table``,
        ``column``, ``usage_type``, ``alias``, ``node_name``,
        ``node_kind``, ``file``, ``repo``, ``line``, and optionally
        ``transform``), ``"summary"`` (dict mapping usage_type to count),
        and ``"total_count"`` (int).
    """
    where = ["cu.table_name = ?"]
    params: list = [table]
    if column:
        where.append("cu.column_name = ?")
        params.append(column)
    if usage_type:
        where.append("cu.usage_type = ?")
        params.append(usage_type)

    joins = (
        "JOIN nodes n ON cu.node_id = n.node_id "
        "JOIN files f ON cu.file_id = f.file_id "
        "JOIN repos r ON f.repo_id = r.repo_id"
    )
    if repo:
        where.append("r.name = ?")
        params.append(repo)

    sql = (
        f"SELECT cu.table_name, cu.column_name, cu.usage_type, cu.alias, "
        f"n.name as node_name, n.kind as node_kind, f.path, r.name as repo_name, n.line_start, "
        f"cu.transform "
        f"FROM column_usage cu {joins} "
        f"WHERE {' AND '.join(where)} "
        f"ORDER BY cu.table_name, cu.column_name, cu.usage_type "
        f"LIMIT ? OFFSET ?"
    )

    rows = self._execute_read(sql, [*params, limit, offset]).fetchall()

    usage = []
    for r in rows:
        entry = {
            "table": r[0],
            "column": r[1],
            "usage_type": r[2],
            "alias": r[3],
            "node_name": r[4],
            "node_kind": r[5],
            "file": r[6],
            "repo": r[7],
            "line": r[8],
        }
        if r[9]:
            entry["transform"] = r[9]
        usage.append(entry)

    # True total count (before pagination)
    count_sql = f"SELECT COUNT(*) FROM column_usage cu {joins} WHERE {' AND '.join(where)}"
    total_count = self._execute_read(count_sql, params).fetchone()[0]

    # Summary by usage_type
    summary: dict[str, int] = {}
    for u in usage:
        summary[u["usage_type"]] = summary.get(u["usage_type"], 0) + 1

    return {"usage": usage, "summary": summary, "total_count": total_count}

query_search

query_search(
    pattern,
    kind=None,
    language=None,
    schema=None,
    repo=None,
    limit=20,
    offset=0,
    include_snippets=True,
)

Search nodes by name pattern (case-insensitive ILIKE).

Parameters:

Name	Type	Description	Default
pattern	str	Substring to match against node names.	required
kind	str | None	Filter by node kind (e.g. "table", "view").	None
language	str | None	Filter by language (e.g. "sql").	None
schema	str | None	Filter by database schema.	None
repo	str | None	Filter by repo name.	None
limit	int	Maximum number of matches to return.	20
offset	int	Number of matches to skip (for pagination).	0
include_snippets	bool	If True, attach source code snippets to results.	True

Returns:

Type	Description
dict	Dict with keys "matches" (list of match dicts with name,
dict	kind, language, file, repo, line_start,
dict	line_end, and optionally snippet) and "total_count"
dict	(int, total matching nodes before pagination).

Source code in src/sqlprism/core/graph.py

def query_search(
    self,
    pattern: str,
    kind: str | None = None,
    language: str | None = None,
    schema: str | None = None,
    repo: str | None = None,
    limit: int = 20,
    offset: int = 0,
    include_snippets: bool = True,
) -> dict:
    """Search nodes by name pattern (case-insensitive ``ILIKE``).

    Args:
        pattern: Substring to match against node names.
        kind: Filter by node kind (e.g. ``"table"``, ``"view"``).
        language: Filter by language (e.g. ``"sql"``).
        schema: Filter by database schema.
        repo: Filter by repo name.
        limit: Maximum number of matches to return.
        offset: Number of matches to skip (for pagination).
        include_snippets: If ``True``, attach source code snippets to results.

    Returns:
        Dict with keys ``"matches"`` (list of match dicts with ``name``,
        ``kind``, ``language``, ``file``, ``repo``, ``line_start``,
        ``line_end``, and optionally ``snippet``) and ``"total_count"``
        (int, total matching nodes before pagination).
    """
    escaped = pattern.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    where = ["n.name ILIKE ? ESCAPE '\\'"]
    params: list = [f"%{escaped}%"]
    if kind:
        where.append("n.kind = ?")
        params.append(kind)
    if language:
        where.append("n.language = ?")
        params.append(language)
    if schema:
        where.append("n.schema = ?")
        params.append(schema)

    joins = "LEFT JOIN files f ON n.file_id = f.file_id LEFT JOIN repos r ON f.repo_id = r.repo_id"
    if repo:
        where.append("r.name = ?")
        params.append(repo)

    count_sql = f"SELECT COUNT(*) FROM nodes n {joins} WHERE {' AND '.join(where)}"
    total = self._execute_read(count_sql, params).fetchone()[0]

    sql = (
        f"SELECT n.name, n.kind, n.language, f.path, r.name as repo_name, "
        f"n.line_start, n.line_end "
        f"FROM nodes n {joins} "
        f"WHERE {' AND '.join(where)} "
        f"ORDER BY n.name "
        f"LIMIT ? OFFSET ?"
    )
    rows = self._execute_read(sql, [*params, limit, offset]).fetchall()

    matches = []
    for r in rows:
        match = {
            "name": r[0],
            "kind": r[1],
            "language": r[2],
            "file": r[3],
            "repo": r[4],
            "line_start": r[5],
            "line_end": r[6],
        }
        if include_snippets:
            snippet = self._read_snippet(r[4], r[3], r[5], r[6])
            if snippet:
                match["snippet"] = snippet
        matches.append(match)

    return {"matches": matches, "total_count": total}

query_trace

query_trace(
    name,
    kind=None,
    direction="downstream",
    max_depth=3,
    repo=None,
    include_snippets=False,
    limit=100,
    exclude_edges=None,
)

Trace multi-hop dependency chains via DuckPGQ or recursive CTE.

When kind is unspecified, query-local aliases (cte, subquery) are excluded from start-node candidates so a within-query alias never becomes the reported root. If the name only resolves to such aliases (no real table/view/query exists), the call falls back to the alias match so the user still gets a trace. defines edges are never followed — they represent CREATE identity, not dataflow, and would otherwise pull a model's own file-stem query back into its trace (issue #122).

Parameters:

Name	Type	Description	Default
name	str	Starting entity name.	required
kind	str | None	Optional node kind filter for the starting node.	None
direction	str	"downstream", "upstream", or "both".	'downstream'
max_depth	int	Maximum hops to follow (capped at 10).	3
repo	str | None	Accepted for API parity with other query methods but not applied to traversal — the walk is over the name-quotient graph, which deliberately crosses repo boundaries so cross-project shadow refs resolve (#131). Results attribute each hop to its representative node's repo, and repos_affected reports every repo touched.	None
include_snippets	bool	Attach source code snippets when True.	False
limit	int	Maximum result rows.	100
exclude_edges	set[tuple[str, str]] | None	Optional set of (source_name, target_name) tuples. Any edge whose source and target names match a tuple in this set will be excluded from traversal. Used by PR-impact v2 to approximate a base-commit graph by removing newly-added edges from the HEAD graph.	None

Returns:

Type	Description
dict	Dict with keys "root" (starting node dict or None),
dict	"paths" (list of path-step dicts with name, kind,
dict	language, relationship, context, depth,
dict	file, repo, and optionally snippet),
dict	"depth_summary" ({depth: count}), and
dict	"repos_affected" (sorted list of repo names). When
dict	direction="both", paths are split into "downstream"
dict	and "upstream" keys instead of a single "paths".

Source code in src/sqlprism/core/graph.py

def query_trace(
    self,
    name: str,
    kind: str | None = None,
    direction: str = "downstream",
    max_depth: int = 3,
    repo: str | None = None,
    include_snippets: bool = False,
    limit: int = 100,
    exclude_edges: set[tuple[str, str]] | None = None,
) -> dict:
    """Trace multi-hop dependency chains via DuckPGQ or recursive CTE.

    When ``kind`` is unspecified, query-local aliases (``cte``,
    ``subquery``) are excluded from start-node candidates so a
    within-query alias never becomes the reported root. If the name
    only resolves to such aliases (no real table/view/query exists),
    the call falls back to the alias match so the user still gets a
    trace. ``defines`` edges are never followed — they represent
    CREATE identity, not dataflow, and would otherwise pull a
    model's own file-stem query back into its trace (issue #122).

    Args:
        name: Starting entity name.
        kind: Optional node kind filter for the starting node.
        direction: ``"downstream"``, ``"upstream"``, or ``"both"``.
        max_depth: Maximum hops to follow (capped at 10).
        repo: Accepted for API parity with other query methods but
            **not applied to traversal** — the walk is over the
            name-quotient graph, which deliberately crosses repo
            boundaries so cross-project shadow refs resolve (#131).
            Results attribute each hop to its representative node's
            repo, and ``repos_affected`` reports every repo touched.
        include_snippets: Attach source code snippets when ``True``.
        limit: Maximum result rows.
        exclude_edges: Optional set of ``(source_name, target_name)``
            tuples.  Any edge whose source and target names match a
            tuple in this set will be excluded from traversal.  Used
            by PR-impact v2 to approximate a base-commit graph by
            removing newly-added edges from the HEAD graph.

    Returns:
        Dict with keys ``"root"`` (starting node dict or ``None``),
        ``"paths"`` (list of path-step dicts with ``name``, ``kind``,
        ``language``, ``relationship``, ``context``, ``depth``,
        ``file``, ``repo``, and optionally ``snippet``),
        ``"depth_summary"`` (``{depth: count}``), and
        ``"repos_affected"`` (sorted list of repo names). When
        ``direction="both"``, paths are split into ``"downstream"``
        and ``"upstream"`` keys instead of a single ``"paths"``.
    """
    max_depth = max(min(max_depth, 10), 1)
    start_nodes = self._find_trace_start_nodes(name, kind)
    if not start_nodes:
        return {"root": None, "paths": [], "depth_summary": {}, "repos_affected": []}

    # Did _find_trace_start_nodes fall back to query-local aliases?
    # (it only returns them when nothing else matches the name)
    start_kinds = {r[2] for r in start_nodes}
    include_query_local_starts = bool(start_kinds) and start_kinds.issubset(_QUERY_LOCAL_KINDS)

    # "both" — run downstream and upstream separately and merge
    if direction == "both":
        down = self.query_trace(
            name,
            kind,
            "downstream",
            max_depth,
            repo,
            include_snippets,
            limit,
            exclude_edges,
        )
        up = self.query_trace(
            name,
            kind,
            "upstream",
            max_depth,
            repo,
            include_snippets,
            limit,
            exclude_edges,
        )
        return {
            "root": down["root"],
            "downstream": down["paths"],
            "upstream": up["paths"],
            "depth_summary": {
                depth: down["depth_summary"].get(depth, 0) + up["depth_summary"].get(depth, 0)
                for depth in set(down["depth_summary"]) | set(up["depth_summary"])
            },
            "repos_affected": list(set(down["repos_affected"] + up["repos_affected"])),
        }

    # Trace via the name-quotient graph (shadows with the same name
    # collapse into one logical hop). One CTE call covers all start
    # nodes with the given name — the recursive step naturally
    # unifies same-name shadows regardless of which repo they live in.
    #
    # Over-fetch so the post-SQL ``_is_noise_node`` filter can drop
    # dbt-test/CTE rows without truncating the final result below
    # ``limit`` when noisy rows happen to sort ahead of useful ones.
    raw_paths = self._trace_cte(
        name,
        kind,
        direction,
        max_depth,
        limit * 2,
        include_snippets,
        exclude_edges,
        include_query_local_starts=include_query_local_starts,
    )
    paths = [p for p in raw_paths if not self._is_noise_node(p)][:limit]

    depth_summary: dict[int, int] = {}
    repos_affected: set[str] = set()
    for p in paths:
        depth_summary[p["depth"]] = depth_summary.get(p["depth"], 0) + 1
        if p["repo"]:
            repos_affected.add(p["repo"])

    return {
        "root": {"name": start_nodes[0][1], "kind": start_nodes[0][2]},
        "paths": paths,
        "depth_summary": depth_summary,
        "repos_affected": sorted(repos_affected),
    }

get_index_status

get_index_status()

Return a summary of the current index state.

Returns:

Type	Description
dict	Dict with keys "repos" (list of repo summaries with
dict	name, path, last_commit, last_branch,
dict	indexed_at, file_count, node_count),
dict	"totals" (aggregate counts for files, nodes,
dict	edges, column_usage_records,
dict	column_lineage_chains), "phantom_nodes" (int),
dict	"cross_repo_edges" (int — edges where source and target
dict	are in different repos; excludes phantom nodes),
dict	"name_collisions" (list of dicts with name,
dict	kind, and repos for nodes defined in multiple repos),
dict	and "schema_version" (str).

Source code in src/sqlprism/core/graph.py

def get_index_status(self) -> dict:
    """Return a summary of the current index state.

    Returns:
        Dict with keys ``"repos"`` (list of repo summaries with
        ``name``, ``path``, ``last_commit``, ``last_branch``,
        ``indexed_at``, ``file_count``, ``node_count``),
        ``"totals"`` (aggregate counts for ``files``, ``nodes``,
        ``edges``, ``column_usage_records``,
        ``column_lineage_chains``), ``"phantom_nodes"`` (int),
        ``"cross_repo_edges"`` (int — edges where source and target
        are in different repos; excludes phantom nodes),
        ``"name_collisions"`` (list of dicts with ``name``,
        ``kind``, and ``repos`` for nodes defined in multiple repos),
        and ``"schema_version"`` (str).
    """
    repos = self._execute_read(
        "SELECT r.name, r.path, r.last_commit, r.last_branch, r.indexed_at, "
        "COUNT(DISTINCT f.file_id) as file_count, "
        "COUNT(DISTINCT n.node_id) as node_count "
        "FROM repos r "
        "LEFT JOIN files f ON r.repo_id = f.repo_id "
        "LEFT JOIN nodes n ON f.file_id = n.file_id "
        "GROUP BY r.repo_id, r.name, r.path, r.last_commit, r.last_branch, r.indexed_at"
    ).fetchall()

    totals = self._execute_read(
        "SELECT "
        "(SELECT COUNT(*) FROM files), "
        "(SELECT COUNT(*) FROM nodes), "
        "(SELECT COUNT(*) FROM edges), "
        "(SELECT COUNT(*) FROM column_usage), "
        "(SELECT COUNT(*) FROM nodes WHERE file_id IS NULL), "
        "(SELECT COUNT(DISTINCT output_node || '.' || output_column) FROM column_lineage)"
    ).fetchone()

    # Cross-repo edges: source and target belong to different repos.
    # INNER JOINs exclude phantom nodes (file_id IS NULL) which are
    # transient placeholders created during incremental reindex.
    cross_repo_edges = self._execute_read(
        "SELECT COUNT(*) FROM edges e "
        "JOIN nodes n1 ON e.source_id = n1.node_id "
        "JOIN nodes n2 ON e.target_id = n2.node_id "
        "JOIN files f1 ON n1.file_id = f1.file_id "
        "JOIN files f2 ON n2.file_id = f2.file_id "
        "WHERE f1.repo_id != f2.repo_id"
    ).fetchone()[0]

    # Name collisions: same (name, kind) defined in multiple repos.
    # Discriminates by kind to avoid false positives between e.g.
    # a table and a CTE sharing the same name.
    collision_rows = self._execute_read(
        "SELECT n.name, n.kind, "
        "LIST(DISTINCT r.name ORDER BY r.name) AS repos "  # DuckDB LIST agg
        "FROM nodes n "
        "JOIN files f ON n.file_id = f.file_id "
        "JOIN repos r ON f.repo_id = r.repo_id "
        "GROUP BY n.name, n.kind "
        "HAVING COUNT(DISTINCT r.repo_id) > 1 "
        "ORDER BY n.name, n.kind"
    ).fetchall()

    return {
        "repos": [
            {
                "name": r[0],
                "path": r[1],
                "last_commit": r[2],
                "last_branch": r[3],
                "indexed_at": str(r[4]) if r[4] else None,
                "file_count": r[5],
                "node_count": r[6],
            }
            for r in repos
        ],
        "totals": {
            "files": totals[0],
            "nodes": totals[1],
            "edges": totals[2],
            "column_usage_records": totals[3],
            "column_lineage_chains": totals[5],
        },
        "phantom_nodes": totals[4],
        "cross_repo_edges": cross_repo_edges,
        "name_collisions": [
            {"name": row[0], "kind": row[1], "repos": row[2]}
            for row in collision_rows
        ],
        "schema_version": "1.0",
    }

query_conventions

query_conventions(layer=None, repo=None)

Query stored conventions for a layer or all layers.

Parameters:

Name	Type	Description	Default
layer	str | None	Layer name (e.g. 'staging'). Omit for all layers.	None
repo	str | None	Repo name filter. Omit for all repos.	None

Returns:

Type	Description
dict	Dict with layers list, each containing convention data.
dict	Returns error key when no conventions found.

Source code in src/sqlprism/core/graph.py

def query_conventions(
    self,
    layer: str | None = None,
    repo: str | None = None,
) -> dict:
    """Query stored conventions for a layer or all layers.

    Args:
        layer: Layer name (e.g. 'staging'). Omit for all layers.
        repo: Repo name filter. Omit for all repos.

    Returns:
        Dict with ``layers`` list, each containing convention data.
        Returns ``error`` key when no conventions found.
    """
    # Resolve repo_id
    repo_id = None
    if repo:
        row = self._execute_read(
            "SELECT repo_id FROM repos WHERE name = ?", [repo]
        ).fetchone()
        if not row:
            return {"error": f"Repo '{repo}' not found"}
        repo_id = row[0]

    # Build query
    if layer and repo_id:
        rows = self._execute_read(
            "SELECT layer, convention_type, payload, confidence, "
            "source, model_count FROM conventions "
            "WHERE repo_id = ? AND layer = ? "
            "ORDER BY convention_type",
            [repo_id, layer],
        ).fetchall()
    elif repo_id:
        rows = self._execute_read(
            "SELECT layer, convention_type, payload, confidence, "
            "source, model_count FROM conventions "
            "WHERE repo_id = ? ORDER BY layer, convention_type",
            [repo_id],
        ).fetchall()
    elif layer:
        rows = self._execute_read(
            "SELECT layer, convention_type, payload, confidence, "
            "source, model_count FROM conventions "
            "WHERE layer = ? ORDER BY convention_type",
            [layer],
        ).fetchall()
    else:
        rows = self._execute_read(
            "SELECT layer, convention_type, payload, confidence, "
            "source, model_count FROM conventions "
            "ORDER BY layer, convention_type",
        ).fetchall()

    if not rows:
        if layer:
            # Check what layers exist — scoped to repo if specified
            if repo_id:
                available = self._execute_read(
                    "SELECT DISTINCT layer FROM conventions "
                    "WHERE repo_id = ?",
                    [repo_id],
                ).fetchall()
            else:
                available = self._execute_read(
                    "SELECT DISTINCT layer FROM conventions"
                ).fetchall()
            available_names = [r[0] for r in available]
            if available_names:
                return {
                    "error": f"No conventions for layer '{layer}'",
                    "available_layers": available_names,
                }
        return {
            "error": "No conventions found. Run reindex or "
            "`sqlprism conventions --refresh` first.",
        }

    # Group by layer
    layers_data: dict[str, dict] = {}
    for row_layer, conv_type, payload, conf, source, model_count in rows:
        if row_layer not in layers_data:
            layers_data[row_layer] = {
                "layer": row_layer,
                "model_count": model_count,
            }
        layer_data = layers_data[row_layer]

        try:
            parsed = json.loads(payload) if isinstance(payload, str) else payload
        except (json.JSONDecodeError, TypeError):
            logger.warning(
                "Malformed convention payload for %s/%s",
                row_layer,
                conv_type,
            )
            parsed = {}

        # Build convention sub-dict. Payload keys are spread first,
        # then confidence/source are set explicitly to avoid shadowing.
        conv_data = dict(parsed)
        conv_data["confidence"] = conf
        conv_data["source"] = source

        if conv_type == "naming":
            layer_data["naming"] = conv_data
        elif conv_type == "references":
            layer_data["allowed_references"] = conv_data
        elif conv_type == "required_columns":
            layer_data["required_columns"] = conv_data
        elif conv_type == "column_style":
            layer_data["column_style"] = conv_data

    result_layers = list(layers_data.values())

    # Small project advisory
    for ld in result_layers:
        if ld.get("model_count", 0) < 10:
            ld["note"] = (
                "Small project — conventions may be unreliable. "
                "Consider explicit overrides via "
                "`sqlprism conventions --init`."
            )

    if layer:
        return result_layers[0] if result_layers else {}

    return {"layers": result_layers}

upsert_tags

upsert_tags(repo_id, tags)

Bulk upsert semantic tag assignments.

Parameters:

Name	Type	Description	Default
repo_id	int	ID of the repo these tags belong to.	required
tags	list[dict]	List of dicts, each with keys tag_name, node_id, confidence, source.	required

Returns:

Type	Description
int	Number of rows processed (inserts + updates).

Source code in src/sqlprism/core/graph.py

def upsert_tags(self, repo_id: int, tags: list[dict]) -> int:
    """Bulk upsert semantic tag assignments.

    Args:
        repo_id: ID of the repo these tags belong to.
        tags: List of dicts, each with keys ``tag_name``, ``node_id``,
            ``confidence``, ``source``.

    Returns:
        Number of rows processed (inserts + updates).
    """
    if not tags:
        return 0
    rows = [
        (repo_id, t["tag_name"], t["node_id"], t["confidence"], t["source"])
        for t in tags
    ]
    with self._write_lock:
        self.conn.executemany(
            "INSERT INTO semantic_tags "
            "(repo_id, tag_name, node_id, confidence, source) "
            "VALUES (?, ?, ?, ?, ?) "
            "ON CONFLICT (repo_id, tag_name, node_id) DO UPDATE SET "
            "confidence = EXCLUDED.confidence, "
            "source = EXCLUDED.source",
            rows,
        )
    return len(rows)

get_tags

get_tags(repo_id, tag_name=None)

Retrieve tags for a repo, optionally filtered by tag name.

Parameters:

Name	Type	Description	Default
repo_id	int	ID of the repo.	required
tag_name	str | None	Optional tag name filter.	None

Returns:

Type	Description
list[dict]	List of dicts with tag_name, node_id, node_name,
list[dict]	confidence, source.

Source code in src/sqlprism/core/graph.py

def get_tags(
    self, repo_id: int, tag_name: str | None = None,
) -> list[dict]:
    """Retrieve tags for a repo, optionally filtered by tag name.

    Args:
        repo_id: ID of the repo.
        tag_name: Optional tag name filter.

    Returns:
        List of dicts with ``tag_name``, ``node_id``, ``node_name``,
        ``confidence``, ``source``.
    """
    sql = (
        "SELECT t.tag_name, t.node_id, n.name AS node_name, "
        "t.confidence, t.source "
        "FROM semantic_tags t "
        "JOIN nodes n ON t.node_id = n.node_id "
        "WHERE t.repo_id = ?"
    )
    params: list = [repo_id]
    if tag_name is not None:
        sql += " AND t.tag_name = ?"
        params.append(tag_name)
    sql += " ORDER BY t.tag_name, n.name"
    rows = self._execute_read(sql, params).fetchall()
    return [
        {
            "tag_name": r[0],
            "node_id": r[1],
            "node_name": r[2],
            "confidence": r[3],
            "source": r[4],
        }
        for r in rows
    ]

list_tag_names

list_tag_names(repo_id)

Return distinct tag names with model count and average confidence.

Parameters:

Name	Type	Description	Default
repo_id	int	ID of the repo.	required

Returns:

Type	Description
list[dict]	List of dicts with tag_name, model_count, avg_confidence.

Source code in src/sqlprism/core/graph.py

def list_tag_names(self, repo_id: int) -> list[dict]:
    """Return distinct tag names with model count and average confidence.

    Args:
        repo_id: ID of the repo.

    Returns:
        List of dicts with ``tag_name``, ``model_count``, ``avg_confidence``.
    """
    rows = self._execute_read(
        "SELECT tag_name, COUNT(*) AS model_count, "
        "AVG(confidence) AS avg_confidence "
        "FROM semantic_tags "
        "WHERE repo_id = ? "
        "GROUP BY tag_name "
        "ORDER BY tag_name",
        [repo_id],
    ).fetchall()
    return [
        {
            "tag_name": r[0],
            "model_count": r[1],
            "avg_confidence": r[2],
        }
        for r in rows
    ]

delete_repo_tags

delete_repo_tags(repo_id)

Delete all semantic tags for a repo (used before re-inference).

Parameters:

Name	Type	Description	Default
repo_id	int	ID of the repo whose tags should be removed.	required

Source code in src/sqlprism/core/graph.py

def delete_repo_tags(self, repo_id: int) -> None:
    """Delete all semantic tags for a repo (used before re-inference).

    Args:
        repo_id: ID of the repo whose tags should be removed.
    """
    with self._write_lock:
        self._execute_write(
            "DELETE FROM semantic_tags WHERE repo_id = ?", [repo_id],
        )

query_search_by_tag

query_search_by_tag(tag, repo=None, min_confidence=None)

Search for models matching a semantic tag.

Parameters:

Name	Type	Description	Default
tag	str	Tag name to search for.	required
repo	str | None	Optional repo name filter.	None
min_confidence	float | None	Optional minimum confidence threshold (0.0-1.0).	None

Returns:

Type	Description
dict	Dict with tag, total, and models list sorted by
dict	confidence descending. Returns suggestion when no matches.

Source code in src/sqlprism/core/graph.py

def query_search_by_tag(
    self,
    tag: str,
    repo: str | None = None,
    min_confidence: float | None = None,
) -> dict:
    """Search for models matching a semantic tag.

    Args:
        tag: Tag name to search for.
        repo: Optional repo name filter.
        min_confidence: Optional minimum confidence threshold (0.0-1.0).

    Returns:
        Dict with ``tag``, ``total``, and ``models`` list sorted by
        confidence descending. Returns ``suggestion`` when no matches.
    """
    repo_id = None
    if repo:
        row = self._execute_read(
            "SELECT repo_id FROM repos WHERE name = ?", [repo]
        ).fetchone()
        if not row:
            return {"error": f"Repo '{repo}' not found"}
        repo_id = row[0]

    sql = (
        "SELECT t.tag_name, t.node_id, n.name AS node_name, "
        "t.confidence, t.source "
        "FROM semantic_tags t "
        "JOIN nodes n ON t.node_id = n.node_id "
        "WHERE t.tag_name = ?"
    )
    params: list = [tag]

    if repo_id is not None:
        sql += " AND t.repo_id = ?"
        params.append(repo_id)
    if min_confidence is not None:
        sql += " AND t.confidence >= ?"
        params.append(min_confidence)

    sql += " ORDER BY t.confidence DESC"
    rows = self._execute_read(sql, params).fetchall()

    if not rows:
        return {
            "tag": tag,
            "total": 0,
            "models": [],
            "suggestion": "Run list_tags to see available tags.",
        }

    models = [
        {
            "node_name": r[2],
            "node_id": r[1],
            "confidence": r[3],
            "source": r[4],
        }
        for r in rows
    ]
    return {"tag": tag, "total": len(models), "models": models}

query_list_tags

query_list_tags(repo=None)

List all semantic tags with model counts and average confidence.

Parameters:

Name	Type	Description	Default
repo	str | None	Optional repo name filter.	None

Returns:

Type	Description
dict	Dict with tags list. Returns suggestion when empty.

Source code in src/sqlprism/core/graph.py

def query_list_tags(self, repo: str | None = None) -> dict:
    """List all semantic tags with model counts and average confidence.

    Args:
        repo: Optional repo name filter.

    Returns:
        Dict with ``tags`` list. Returns ``suggestion`` when empty.
    """
    repo_id = None
    if repo:
        row = self._execute_read(
            "SELECT repo_id FROM repos WHERE name = ?", [repo]
        ).fetchone()
        if not row:
            return {"error": f"Repo '{repo}' not found"}
        repo_id = row[0]

    sql = (
        "SELECT tag_name, COUNT(*) AS model_count, "
        "ROUND(AVG(confidence), 4) AS avg_confidence "
        "FROM semantic_tags"
    )
    params: list = []
    if repo_id is not None:
        sql += " WHERE repo_id = ?"
        params.append(repo_id)
    sql += " GROUP BY tag_name ORDER BY tag_name"

    rows = self._execute_read(sql, params).fetchall()

    if not rows:
        return {
            "tags": [],
            "suggestion": (
                "No semantic tags found. Run reindex or "
                "`sqlprism conventions --refresh` first."
            ),
        }

    return {
        "tags": [
            {
                "tag_name": r[0],
                "model_count": r[1],
                "avg_confidence": r[2],
            }
            for r in rows
        ],
    }

query_find_similar_models

query_find_similar_models(
    references=None,
    output_columns=None,
    model=None,
    limit=5,
    repo=None,
)

Find models similar to a given model or set of characteristics.

Parameters:

Name	Type	Description	Default
references	list[str] | None	List of reference/dependency names to match against.	None
output_columns	list[str] | None	List of output column names to match against.	None
model	str | None	Model name to use as the similarity target.	None
limit	int	Maximum number of results (default 5).	5
repo	str | None	Optional repo name filter.	None

Returns:

Type	Description
dict	Dict with similar list, count, and total_matches.

Source code in src/sqlprism/core/graph.py

def query_find_similar_models(
    self,
    references: list[str] | None = None,
    output_columns: list[str] | None = None,
    model: str | None = None,
    limit: int = 5,
    repo: str | None = None,
) -> dict:
    """Find models similar to a given model or set of characteristics.

    Args:
        references: List of reference/dependency names to match against.
        output_columns: List of output column names to match against.
        model: Model name to use as the similarity target.
        limit: Maximum number of results (default 5).
        repo: Optional repo name filter.

    Returns:
        Dict with ``similar`` list, ``count``, and ``total_matches``.
    """
    limit = max(1, min(limit, 50))

    def jaccard(a: set[str], b: set[str]) -> float:
        union = a | b
        if not union:
            return 0.0
        return len(a & b) / len(union)

    # --- Validate inputs ---
    if references is None and output_columns is None and model is None:
        return {
            "error": (
                "Provide at least one of: references, output_columns, model"
            )
        }

    # --- Repo resolution ---
    repo_id = None
    if repo:
        row = self._execute_read(
            "SELECT repo_id FROM repos WHERE name = ?", [repo]
        ).fetchone()
        if not row:
            return {"error": f"Repo '{repo}' not found"}
        repo_id = row[0]

    # --- Target sets ---
    target_refs: set[str] = set(references) if references is not None else set()
    target_cols: set[str] = set(output_columns) if output_columns is not None else set()
    target_model_node_id: int | None = None
    target_layer: str | None = None

    model = model or None  # normalize empty string
    if model:
        # Look up model node
        model_sql = (
            "SELECT n.node_id, f.path FROM nodes n "
            "JOIN files f ON n.file_id = f.file_id "
            "WHERE n.name = ? AND n.kind IN ('table', 'view')"
        )
        model_params: list = [model]
        if repo_id is not None:
            model_sql += " AND f.repo_id = ?"
            model_params.append(repo_id)
        model_rows = self._execute_read(model_sql, model_params).fetchmany(2)
        if not model_rows:
            return {"error": f"Model '{model}' not found"}
        if len(model_rows) > 1:
            return {
                "error": (
                    f"Multiple models named '{model}' found"
                    " — specify repo to disambiguate"
                )
            }
        model_row = model_rows[0]
        target_model_node_id = model_row[0]
        model_file_path: str = model_row[1]
        target_layer = self._extract_layer(model_file_path)

        # Extract model's references if not provided
        if references is None:
            ref_rows = self._execute_read(
                "SELECT n.name FROM edges e "
                "JOIN nodes n ON e.target_id = n.node_id "
                "WHERE e.source_id = ? AND e.relationship = 'references'",
                [target_model_node_id],
            ).fetchall()
            target_refs = {r[0] for r in ref_rows}

        # Extract model's output columns if not provided
        if output_columns is None:
            col_rows = self._execute_read(
                "SELECT column_name FROM columns WHERE node_id = ?",
                [target_model_node_id],
            ).fetchall()
            target_cols = {r[0] for r in col_rows}

    # If target has no refs and no columns, similarity is meaningless
    if not target_refs and not target_cols:
        return {
            "similar": [],
            "count": 0,
            "total_matches": 0,
            "suggestion": (
                "Model has no references or columns to compare against."
                if model else
                "No similar models found for the given criteria."
            ),
        }

    # --- Batch queries for candidates ---
    # Get all candidate nodes
    cand_sql = (
        "SELECT n.node_id, n.name, f.path FROM nodes n "
        "JOIN files f ON n.file_id = f.file_id "
        "WHERE n.kind IN ('table', 'view')"
    )
    cand_params: list = []
    if repo_id is not None:
        cand_sql += " AND f.repo_id = ?"
        cand_params.append(repo_id)
    if target_model_node_id is not None:
        cand_sql += " AND n.node_id != ?"
        cand_params.append(target_model_node_id)

    candidates = self._execute_read(cand_sql, cand_params).fetchall()
    if len(candidates) > 50_000:
        return {
            "error": (
                f"Too many candidate models ({len(candidates)})"
                " — specify repo to narrow the search"
            )
        }
    if not candidates:
        return {
            "similar": [],
            "count": 0,
            "total_matches": 0,
            "suggestion": "No similar models found for the given criteria.",
        }

    cand_node_ids = [c[0] for c in candidates]

    # Batch: all references and columns for candidate nodes (chunked)
    chunk_size = 5000
    refs_by_node: dict[int, set[str]] = {}
    cols_by_node: dict[int, set[str]] = {}

    for i in range(0, len(cand_node_ids), chunk_size):
        chunk = cand_node_ids[i : i + chunk_size]
        placeholders = ",".join("?" * len(chunk))

        ref_rows = self._execute_read(
            f"SELECT e.source_id, n.name FROM edges e "
            f"JOIN nodes n ON e.target_id = n.node_id "
            f"WHERE e.source_id IN ({placeholders}) "
            f"AND e.relationship = 'references'",
            chunk,
        ).fetchall()
        for source_id, name in ref_rows:
            refs_by_node.setdefault(source_id, set()).add(name)

        col_rows = self._execute_read(
            f"SELECT node_id, column_name FROM columns "
            f"WHERE node_id IN ({placeholders})",
            chunk,
        ).fetchall()
        for node_id, col_name in col_rows:
            cols_by_node.setdefault(node_id, set()).add(col_name)

    layer_by_node = {
        node_id: self._extract_layer(file_path)
        for node_id, _, file_path in candidates
    }

    # --- Score candidates ---
    scored: list[tuple[float, str, list[str], list[str], str]] = []
    for node_id, name, file_path in candidates:
        cand_refs = refs_by_node.get(node_id, set())
        cand_cols = cols_by_node.get(node_id, set())
        cand_layer = layer_by_node[node_id]

        layer_bonus = (
            0.1
            if target_layer is not None and cand_layer == target_layer
            else 0.0
        )
        similarity = (
            jaccard(target_refs, cand_refs) * 0.6
            + jaccard(target_cols, cand_cols) * 0.3
            + layer_bonus
        )

        shared_refs = sorted(target_refs & cand_refs)
        shared_cols = sorted(target_cols & cand_cols)

        if similarity >= 0.05:
            scored.append((similarity, name, shared_refs, shared_cols, file_path))

    if not scored:
        return {
            "similar": [],
            "count": 0,
            "total_matches": 0,
            "suggestion": "No similar models found for the given criteria.",
        }

    # Sort descending by similarity, apply limit
    scored.sort(key=lambda x: (-x[0], x[1]))
    total_matches = len(scored)
    scored = scored[:limit]

    # --- Build result ---
    similar: list[dict] = []
    for similarity, name, shared_refs, shared_cols, file_path in scored:
        entry: dict = {
            "name": name,
            "similarity": round(similarity, 4),
            "shared_refs": shared_refs,
            "shared_columns": shared_cols,
            "file": file_path,
        }
        if similarity >= _EXTEND_SUGGESTION_THRESHOLD:
            entry["suggestion"] = (
                f"Consider extending '{name}' instead of creating a new model"
            )
        similar.append(entry)

    return {
        "similar": similar,
        "count": len(similar),
        "total_matches": total_matches,
    }

query_suggest_placement

query_suggest_placement(references, name=None, repo=None)

Suggest where to place a new model based on its references.

Determines which layers the referenced models belong to, finds the convention rule that allows those layers as inputs, and recommends placement in that layer with its naming pattern.

Parameters:

Name	Type	Description	Default
references	list[str]	Tables this new model will reference.	required
name	str | None	Proposed model name (for naming validation).	None
repo	str | None	Optional repo name filter.	None

Returns:

Type	Description
dict	Dict with recommended_layer, recommended_path,
dict	naming_pattern, reason, similar_models, and
dict	optionally name_feedback.

Source code in src/sqlprism/core/graph.py

def query_suggest_placement(
    self,
    references: list[str],
    name: str | None = None,
    repo: str | None = None,
) -> dict:
    """Suggest where to place a new model based on its references.

    Determines which layers the referenced models belong to, finds the
    convention rule that allows those layers as inputs, and recommends
    placement in that layer with its naming pattern.

    Args:
        references: Tables this new model will reference.
        name: Proposed model name (for naming validation).
        repo: Optional repo name filter.

    Returns:
        Dict with ``recommended_layer``, ``recommended_path``,
        ``naming_pattern``, ``reason``, ``similar_models``, and
        optionally ``name_feedback``.
    """
    if not references:
        return {"error": "Provide at least one reference table."}

    # --- Repo resolution ---
    repo_id = None
    if repo:
        row = self._execute_read(
            "SELECT repo_id FROM repos WHERE name = ?", [repo]
        ).fetchone()
        if not row:
            return {"error": f"Repo '{repo}' not found"}
        repo_id = row[0]

    # --- Check conventions exist ---
    conv_sql = (
        "SELECT layer, convention_type, payload, confidence "
        "FROM conventions"
    )
    conv_params: list = []
    if repo_id is not None:
        conv_sql += " WHERE repo_id = ?"
        conv_params.append(repo_id)
    conv_rows = self._execute_read(conv_sql, conv_params).fetchall()

    if not conv_rows:
        return {
            "error": "No conventions found. Run reindex or "
            "`sqlprism conventions --refresh` first."
        }

    # --- Parse conventions into lookup structures ---
    # naming: layer -> {pattern, confidence}
    # references: layer -> {allowed_targets: [...], confidence}
    # path_patterns: layer -> path_pattern
    naming_by_layer: dict[str, dict] = {}
    refs_by_layer: dict[str, dict] = {}

    for layer, conv_type, payload, confidence in conv_rows:
        try:
            parsed = json.loads(payload) if isinstance(payload, str) else payload
        except (json.JSONDecodeError, TypeError):
            parsed = {}

        if conv_type == "naming":
            naming_by_layer[layer] = {
                "pattern": parsed.get("pattern", ""),
                "confidence": confidence,
            }
        elif conv_type == "references":
            refs_by_layer[layer] = {
                "allowed_targets": parsed.get("allowed_targets", []),
                "distribution": parsed.get("target_distribution", {}),
                "confidence": confidence,
            }

    # --- Determine layers of referenced models (batched) ---
    ref_layers: dict[str, str] = {}  # ref_name -> layer
    placeholders = ",".join("?" * len(references))
    ref_params: list = list(references)
    ref_sql = (
        "SELECT n.name, f.path FROM nodes n "
        "JOIN files f ON n.file_id = f.file_id "
        f"WHERE n.name IN ({placeholders}) "
        "AND n.kind IN ('table', 'view')"
    )
    if repo_id is not None:
        ref_sql += " AND f.repo_id = ?"
        ref_params.append(repo_id)
    for ref_name, fpath in self._execute_read(ref_sql, ref_params).fetchall():
        layer = self._extract_layer(fpath)
        if layer:  # skip models with no identifiable layer
            ref_layers[ref_name] = layer

    if not ref_layers:
        return {
            "error": "None of the referenced models were found in the index.",
            "references": references,
        }

    ref_layer_set = set(ref_layers.values())

    # --- Find the layer whose reference rule allows these ref layers ---
    # A candidate layer is one whose allowed_targets include ALL ref layers
    candidates: list[tuple[str, float]] = []  # (layer, confidence)
    for layer, rule in refs_by_layer.items():
        allowed = set(rule["allowed_targets"])
        if ref_layer_set <= allowed:
            candidates.append((layer, rule["confidence"]))

    # Also consider layers that partially match (for ambiguous case)
    partial_candidates: list[tuple[str, float, float]] = []
    if not candidates:
        for layer, rule in refs_by_layer.items():
            allowed = set(rule["allowed_targets"])
            overlap = ref_layer_set & allowed
            if overlap:
                coverage = len(overlap) / len(ref_layer_set)
                partial_candidates.append(
                    (layer, rule["confidence"], coverage)
                )

    # --- Build recommendation ---
    if candidates:
        # Pick highest confidence, break ties alphabetically
        candidates.sort(key=lambda x: (-x[1], x[0]))
        rec_layer, rec_confidence = candidates[0]
        ambiguous = False
    elif partial_candidates:
        # Pick highest coverage, then confidence, then alphabetical
        partial_candidates.sort(key=lambda x: (-x[2], -x[1], x[0]))
        rec_layer, rec_confidence, coverage = partial_candidates[0]
        ambiguous = True
    else:
        return {
            "error": "Could not determine placement — no convention "
            "rules match the referenced layers.",
            "ref_layers": sorted(ref_layer_set),
        }

    # --- Naming pattern ---
    naming = naming_by_layer.get(rec_layer, {})
    naming_pattern = naming.get("pattern", "")

    # --- Recommended path ---
    # Derive path from existing models in that layer (cursor iteration)
    path_cursor = self._execute_read(
        "SELECT f.path FROM nodes n "
        "JOIN files f ON n.file_id = f.file_id "
        "WHERE n.kind IN ('table', 'view')"
        + (" AND f.repo_id = ?" if repo_id is not None else ""),
        [repo_id] if repo_id is not None else [],
    )

    rec_path = ""
    while True:
        path_row = path_cursor.fetchone()
        if path_row is None:
            break
        fpath = path_row[0]
        layer_name = self._extract_layer(fpath)
        if layer_name == rec_layer:
            parts = fpath.replace("\\", "/").split("/")
            if len(parts) > 1:
                rec_path = "/".join(parts[:-1]) + "/"
            break

    # --- Name validation ---
    name_feedback: dict | None = None
    if name and naming_pattern:
        # Extract prefix: everything before the first placeholder {
        brace_idx = naming_pattern.find("{")
        prefix = naming_pattern[:brace_idx] if brace_idx > 0 else ""
        if prefix and not name.startswith(prefix):
            suggested_name = prefix + name
            name_feedback = {
                "matches_convention": False,
                "suggested_name": suggested_name,
                "reason": f"Convention for {rec_layer} is "
                f"'{naming_pattern}' — name should start "
                f"with '{prefix}'",
            }
        else:
            name_feedback = {
                "matches_convention": True,
                "reason": f"Name matches the {rec_layer} naming "
                f"convention '{naming_pattern}'",
            }

    # --- Similar models ---
    similar_result = self.query_find_similar_models(
        references=references, limit=3, repo=repo,
    )
    similar_models = [
        m["name"] for m in similar_result.get("similar", [])
    ]

    # --- Build reason ---
    ref_layer_str = ", ".join(sorted(ref_layer_set))
    reason = (
        f"References {ref_layer_str} models → {rec_layer} layer "
        f"per project conventions (confidence: {rec_confidence:.2f})"
    )
    if ambiguous:
        reason = (
            f"Mixed references from {ref_layer_str} — "
            f"most likely {rec_layer} layer based on partial "
            f"convention match (confidence: {rec_confidence:.2f})"
        )

    result: dict = {
        "recommended_layer": rec_layer,
        "recommended_path": rec_path,
        "naming_pattern": naming_pattern,
        "reason": reason,
        "similar_models": similar_models,
    }
    if name_feedback:
        result["name_feedback"] = name_feedback
    if ambiguous:
        result["ambiguous"] = True
        result["coverage"] = round(coverage, 2)

    return result