fix(assets): batch prefix filters so scans work with many model folders

app/assets/helpers.py

@@ -37,14 +37,51 @@ def sql_path_under_prefix(
     and normalizes there. Normalizing the column in SQL is not an option anyway:
     it would need a per-row Python call and would defeat the index.
     """
-    base = os.path.abspath(prefix)
-    stem = base if base.endswith(os.sep) else base + os.sep
+    base, stem = _base_and_stem(prefix)
     return sa.or_(
         column == base,
         sa.func.substr(column, 1, len(stem)) == stem,
     )
 
 
+def _base_and_stem(prefix: str) -> tuple[str, str]:
+    base = os.path.abspath(prefix)
+    return base, base if base.endswith(os.sep) else base + os.sep
+
+
+def stored_path_under_prefixes(prefixes: list[str]) -> Callable[[str], bool]:
+    """The Python twin of sql_path_under_prefix OR'd over ``prefixes``: the same
+    case-sensitive string test on a stored path, for filtering rows already fetched.
+
+    Unlike path_prefix_matcher, it neither normalizes nor normcases the path.
+    """
+    pairs = [_base_and_stem(prefix) for prefix in prefixes]
+    bases = frozenset(base for base, _ in pairs)
+    stems = tuple(stem for _, stem in pairs)
+    return lambda path: path in bases or path.startswith(stems)
+
+
+# Each prefix adds two terms to one flat OR, and SQLite rejects an expression deeper than
+# 1000, so about 500 prefixes in one statement fail with "Expression tree is too large".
+# SQLAlchemy flattens nested ORs, so more prefixes than this are split across statements,
+# or filtered in Python with stored_path_under_prefixes.
+PREFIX_BATCH_SIZE = 200
+
+
+def sql_path_under_prefix_batches(
+    column: ColumnElement[str], prefixes: list[str]
+) -> list[ColumnElement[bool]]:
+    """sql_path_under_prefix OR'd over each run of at most PREFIX_BATCH_SIZE prefixes.
+
+    Run one statement per predicate and merge. Nested or overlapping prefixes can put a
+    row in more than one batch, so the caller dedupes.
+    """
+    return [
+        sa.or_(*(sql_path_under_prefix(column, p) for p in prefixes[i:i + PREFIX_BATCH_SIZE]))
+        for i in range(0, len(prefixes), PREFIX_BATCH_SIZE)
+    ]
+
+
 def path_prefix_matcher(prefixes: Iterable[str]) -> Callable[[str], bool]:
     """Return ``path -> Path(path).is_relative_to(<any prefix>)``, with the prefixes
     normalized once.

app/assets/scanner.py

@@ -12,6 +12,7 @@
 import logging
 import os
 from dataclasses import dataclass
+from itertools import islice
 from pathlib import Path
 from typing import Callable, Literal, NamedTuple, Protocol, TypedDict
 
@@ -28,7 +29,14 @@
     create_record,
 )
 from app.assets.database.models import Asset, AssetContent
-from app.assets.helpers import path_prefix_matcher, sql_path_under_prefix, to_stored_hash
+from app.assets.helpers import (
+    PREFIX_BATCH_SIZE,
+    path_prefix_matcher,
+    sql_path_under_prefix,
+    sql_path_under_prefix_batches,
+    stored_path_under_prefixes,
+    to_stored_hash,
+)
 from app.assets.lifecycle import get_excluded_scan_roots
 from app.assets.scanner_changes import (
     clear_pending_verifications,
@@ -379,19 +387,21 @@ def live_references_safely(root: RootType) -> dict[str, list[_ReferenceObservati
     live: dict[str, list[_ReferenceObservation]] = {}
     if not prefixes:
         return live
-    stmt = sa.select(
-        AssetContent.id, AssetContent.path, AssetContent.size_bytes, AssetContent.mtime_ns
-    ).where(
-        AssetContent.is_missing.is_(False),
-        sa.or_(*(sql_path_under_prefix(AssetContent.path, p) for p in prefixes)),
-    )
+    seen: set[str] = set()
     try:
         with create_session() as session:
-            for content_id, path, size_bytes, mtime_ns in session.execute(stmt):
-                yield_gil(run=RESCAN_YIELD_RUN)
-                live.setdefault(os.path.abspath(path), []).append(
-                    _ReferenceObservation(content_id, size_bytes, mtime_ns, None)
-                )
+            for under_prefixes in sql_path_under_prefix_batches(AssetContent.path, prefixes):
+                stmt = sa.select(
+                    AssetContent.id, AssetContent.path, AssetContent.size_bytes, AssetContent.mtime_ns
+                ).where(AssetContent.is_missing.is_(False), under_prefixes)
+                for content_id, path, size_bytes, mtime_ns in session.execute(stmt):
+                    yield_gil(run=RESCAN_YIELD_RUN)
+                    if content_id in seen:
+                        continue
+                    seen.add(content_id)
+                    live.setdefault(os.path.abspath(path), []).append(
+                        _ReferenceObservation(content_id, size_bytes, mtime_ns, None)
+                    )
     except Exception as exc:
         logging.exception("fast DB scan failed for %s: %s", root, exc)
         emit("scanner.fast_scan_failed", root=root, error_type=error_type(exc))
@@ -745,12 +755,10 @@ def insert_asset_specs(
         return created, first_error
 
 
-def build_unenriched_candidates_statement(
-    prefixes: list[str],
-    compute_hashes: bool,
-    last_seen_id: str | None,
-    limit: int = 1000,
+def unenriched_candidates_query(
+    compute_hashes: bool, last_seen_id: str | None
 ) -> sa.Select[tuple[str, str, str]]:
+    """Every unenriched live candidate after ``last_seen_id``, in id order."""
     query = (
         sa.select(AssetContent.id, Asset.id, AssetContent.path)
         .join(Asset, Asset.content_id == AssetContent.id)
@@ -767,11 +775,19 @@ def build_unenriched_candidates_statement(
         query = query.where(Asset.system_metadata.is_(None))
     if last_seen_id is not None:
         query = query.where(Asset.id > last_seen_id)
+    return query.order_by(Asset.id.asc())
+
+
+def build_unenriched_candidates_statement(
+    prefixes: list[str],
+    compute_hashes: bool,
+    last_seen_id: str | None,
+    limit: int = 1000,
+) -> sa.Select[tuple[str, str, str]]:
+    """The next page of candidates under at most PREFIX_BATCH_SIZE prefixes."""
     return (
-        query.where(
-            sa.or_(*(sql_path_under_prefix(AssetContent.path, p) for p in prefixes))
-        )
-        .order_by(Asset.id.asc())
+        unenriched_candidates_query(compute_hashes, last_seen_id)
+        .where(sa.or_(*(sql_path_under_prefix(AssetContent.path, p) for p in prefixes)))
         .limit(limit)
     )
 
@@ -789,14 +805,24 @@ def get_unenriched_assets_for_roots(
     if not prefixes:
         return []
 
-    query = build_unenriched_candidates_statement(
-        prefixes,
-        compute_hashes,
-        last_seen_id,
-        limit,
-    )
     with create_session() as sess:
-        rows = sess.execute(query).all()
+        if len(prefixes) <= PREFIX_BATCH_SIZE:
+            statement = build_unenriched_candidates_statement(
+                prefixes,
+                compute_hashes,
+                last_seen_id,
+                limit,
+            )
+            rows = sess.execute(statement).all()
+        else:
+            # Too many prefixes for one SQL predicate. Paging each batch separately
+            # would rescan to the end of the table on every page for any batch with
+            # few matches, so filter a single id-ordered pass here instead.
+            is_under = stored_path_under_prefixes(prefixes)
+            candidates = sess.execute(
+                unenriched_candidates_query(compute_hashes, last_seen_id).execution_options(yield_per=500)
+            )
+            rows = list(islice((row for row in candidates if is_under(row[2])), limit))
 
     return [
         UnenrichedContent(content_id, record_id, file_path)

app/assets/scanner_changes.py

@@ -9,7 +9,7 @@
 from __future__ import annotations
 
 import os
-from collections.abc import Iterable
+from collections.abc import Iterator
 from typing import Literal
 
 import sqlalchemy as sa
@@ -22,7 +22,7 @@
     mark_content_missing,
     unset_content_missing,
 )
-from app.assets.helpers import path_prefix_matcher, sql_path_under_prefix, to_stored_hash
+from app.assets.helpers import path_prefix_matcher, sql_path_under_prefix_batches, to_stored_hash
 from app.assets.services.file_utils import get_mtime_ns
 from app.assets.services.path_utils import compute_loader_path, get_name_and_tags_from_asset_path
 from app.assets.services.snapshot_hash import snapshot_hash
@@ -259,12 +259,12 @@ def drain_pending_verifications(session: Session, limit: int | None = None) -> i
     return processed
 
 
-def live_contents_under_prefixes(session: Session, prefixes: list[str]) -> Iterable[AssetContent]:
+def live_contents_under_prefixes(session: Session, prefixes: list[str]) -> Iterator[AssetContent]:
     """Stream the live contents under the prefixes in batches; consume it inside the session."""
-    if not prefixes:
-        return []
-    stmt = sa.select(AssetContent).where(
-        AssetContent.is_missing.is_(False),
-        sa.or_(*(sql_path_under_prefix(AssetContent.path, prefix) for prefix in prefixes)),
-    )
-    return session.scalars(stmt.execution_options(yield_per=500))
+    seen: set[str] = set()
+    for under_prefixes in sql_path_under_prefix_batches(AssetContent.path, prefixes):
+        stmt = sa.select(AssetContent).where(AssetContent.is_missing.is_(False), under_prefixes)
+        for content in session.scalars(stmt.execution_options(yield_per=500)):
+            if content.id not in seen:
+                seen.add(content.id)
+                yield content