Upload 4 files

Dockerfile

@@ -6,7 +6,7 @@ COPY requirements.txt .
 
 RUN pip install --no-cache-dir -r requirements.txt
 
-COPY app.py build_fulltext_db.py ./
+COPY app.py build_fulltext_db.py start.sh ./
 
 COPY static/ static/
 
@@ -16,10 +16,6 @@ COPY CCRD/ CCRD/
 
 COPY CW/ CW/
 
-# Full-text indexes are derived artifacts. Build them inside the Space image so
-# the repository never needs to store SQLite files larger than the LFS limit.
-RUN python build_fulltext_db.py
-
 EXPOSE 7860
 
-CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860", "--log-level", "info"]
+CMD ["sh", "/app/start.sh"]

app.py

@@ -5,6 +5,7 @@ import json
 import random
 import re
 import sqlite3
+import threading
 import time
 import unicodedata
 from functools import lru_cache
@@ -91,39 +92,63 @@ def decode_doc_numbers(payload: bytes, candidates: set[int] | None = None) -> se
 class FulltextDatabases:
 
     def __init__(self, directory: Path):
+        self.directory = directory
+        self.lock = threading.RLock()
         self.connections = {}
         self.has_doc_counts = {}
         self.tokenizer_versions = {}
         self.has_content_fts = {}
-        for path in directory.glob("*.sqlite3"):
-            uri = f"file:{path.resolve()}?mode=ro&immutable=1"
-            connection = sqlite3.connect(uri, uri=True, check_same_thread=False)
-            self.connections[path.stem] = connection
+        self.refresh()
+
+    def refresh(self) -> None:
+        with self.lock:
+            for path in self.directory.glob("*.sqlite3"):
+                if path.stem in self.connections:
+                    continue
+                try:
+                    self._open(path)
+                except sqlite3.Error as exc:
+                    print(f"fulltext_open_failed={path.name}:{exc}")
+
+    def _open(self, path: Path) -> None:
+        uri = f"file:{path.resolve()}?mode=ro&immutable=1"
+        connection = sqlite3.connect(uri, uri=True, check_same_thread=False)
+        try:
             columns = {row[1] for row in connection.execute("PRAGMA table_info(postings)")}
-            self.has_doc_counts[path.stem] = "doc_count" in columns
             has_metadata = connection.execute(
                 "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'metadata'"
             ).fetchone()
             version_row = connection.execute(
                 "SELECT value FROM metadata WHERE key = 'tokenizer_version'"
             ).fetchone() if has_metadata else None
-            self.tokenizer_versions[path.stem] = version_row[0] if version_row else "legacy"
-            self.has_content_fts[path.stem] = bool(connection.execute(
+            has_content_fts = bool(connection.execute(
                 "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'content_fts'"
             ).fetchone())
+        except Exception:
+            connection.close()
+            raise
+        self.connections[path.stem] = connection
+        self.has_doc_counts[path.stem] = "doc_count" in columns
+        self.tokenizer_versions[path.stem] = version_row[0] if version_row else "legacy"
+        self.has_content_fts[path.stem] = has_content_fts
 
     def close(self) -> None:
-        for connection in self.connections.values():
-            connection.close()
+        with self.lock:
+            for connection in self.connections.values():
+                connection.close()
 
     def search_source(self, source: str, query: str, exact: bool = False) -> set[int]:
-        connection = self.connections.get(source)
+        with self.lock:
+            connection = self.connections.get(source)
+            if connection is None:
+                self.refresh()
+                connection = self.connections.get(source)
+            if connection is None:
+                return set()
         version = self.tokenizer_versions.get(source)
         if exact and self.has_content_fts.get(source) and "*" not in query and "?" not in query and len(normalize_text(query)) >= 3:
-            phrase = normalize_text(query)
-            grams = [phrase[index:index + 3] for index in range(len(phrase) - 2)]
+            phrase = normalize_text(query).replace('"', '""')
             try:
-                match_query = " AND ".join('"' + gram.replace('"', '""') + '"' for gram in grams)
                 rows = connection.execute(
                     "SELECT rowid FROM content_fts WHERE content_fts MATCH ?",
                     (f'"{phrase}"',),
@@ -135,7 +160,7 @@ class FulltextDatabases:
             tokens = [*query_tokens(query), *literal_query_tokens(query)]
         else:
             tokens = query_tokens(query) if version == TOKENIZER_VERSION else query_terms(query)
-        if connection is None or not tokens:
+        if not tokens:
             return set()
         matched = None
         token_hashes = [hashlib.sha256(token.encode("utf-8")).digest()[:16] for token in tokens]
@@ -173,6 +198,7 @@ class FulltextDatabases:
         )
 
     def summaries(self, doc_ids: list[str]) -> dict[str, str]:
+        self.refresh()
         grouped = {}
         for doc_id in doc_ids:
             grouped.setdefault(doc_id.split(":", 1)[0], []).append(doc_id)

build_fulltext_db.py

@@ -3,8 +3,9 @@ import gzip
 import hashlib
 import json
 import re
+import shutil
 import sqlite3
-import unicodedata
+import struct
 import unicodedata
 from collections import defaultdict
 from pathlib import Path
@@ -15,6 +16,7 @@ TOKEN_RE = re.compile(r"[a-z0-9]+|[\u4e00-\u9fff\u3400-\u4dbf]+")
 SUMMARY_CHARS = 700
 TOKENIZER_VERSION = "cjk-bigram-boundary-fts5-v5"
 LITERAL_PREFIX = "\0literal:"
+POSTING_BUCKETS = 256
 
 def load_json_gz(path: Path):
     return json.loads(gzip.decompress(path.read_bytes()).decode("utf-8"))
@@ -68,55 +70,91 @@ def build_summary(text: str) -> str:
 def fts_content(text: str) -> str:
     return unicodedata.normalize("NFKC", text or "").casefold().replace("\0", " ")
 
+def database_ready(path: Path, expected_documents: int) -> bool:
+    if not path.exists():
+        return False
+    try:
+        uri = f"file:{path.resolve()}?mode=ro&immutable=1"
+        with sqlite3.connect(uri, uri=True) as connection:
+            version = connection.execute("SELECT value FROM metadata WHERE key = 'tokenizer_version'").fetchone()
+            documents = int(connection.execute("SELECT COUNT(*) FROM documents").fetchone()[0])
+            fts_rows = int(connection.execute("SELECT COUNT(*) FROM content_fts").fetchone()[0])
+        return bool(version and version[0] == TOKENIZER_VERSION and documents == expected_documents and fts_rows == expected_documents)
+    except Exception:
+        return False
+
 def build_source_db(source: str, source_records: list[dict]) -> None:
     db_path = FULLTEXT_DIR / f"{source}.sqlite3"
+    if database_ready(db_path, len(source_records)):
+        print(f"{source}: current index already ready", flush=True)
+        return
     tmp_path = db_path.with_suffix(".sqlite3.tmp")
-    postings: dict[bytes, set[int]] = defaultdict(set)
-    documents = []
-    content_rows = []
+    bucket_dir = FULLTEXT_DIR / f".{source}-postings"
     missing = 0
     if tmp_path.exists():
         tmp_path.unlink()
-    for index, record in enumerate(source_records, 1):
-        doc_id = record["doc_id"]
-        file_path = get_doc_storage_path(record)
-        if not file_path.exists():
-            missing += 1
-            continue
-        text = file_path.read_text(encoding="utf-8", errors="ignore")
-        documents.append((doc_id, build_summary(text)))
-        number = doc_number(doc_id)
-        # FTS5 tokenization stops at embedded NUL bytes even though selecting the
-        # virtual column returns the remainder. Replace them only in derived index data.
-        content_rows.append((number, fts_content(text)))
-        for token in index_tokens(text) | literal_tokens(text):
-            token_hash = hashlib.sha256(token.encode("utf-8")).digest()[:16]
-            postings[token_hash].add(number)
-        if index % 1000 == 0:
-            print(f"{source}: indexed {index}/{len(source_records)} records")
-    with sqlite3.connect(tmp_path) as connection:
-        connection.execute("PRAGMA journal_mode = OFF")
-        connection.execute("PRAGMA synchronous = OFF")
-        connection.execute("PRAGMA temp_store = MEMORY")
-        connection.execute("CREATE TABLE documents (doc_id TEXT PRIMARY KEY, summary TEXT NOT NULL)")
-        connection.execute("CREATE TABLE postings (token_hash BLOB PRIMARY KEY, docs BLOB NOT NULL, doc_count INTEGER NOT NULL)")
-        connection.execute("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
-        connection.execute("INSERT INTO metadata (key, value) VALUES ('tokenizer_version', ?)", (TOKENIZER_VERSION,))
-        connection.execute("CREATE VIRTUAL TABLE content_fts USING fts5(content, tokenize='trigram', detail='full')")
-        connection.executemany("INSERT INTO documents (doc_id, summary) VALUES (?, ?)", documents)
-        connection.executemany("INSERT INTO content_fts (rowid, content) VALUES (?, ?)", content_rows)
-        connection.executemany(
-            "INSERT INTO postings (token_hash, docs, doc_count) VALUES (?, ?, ?)",
-            (
-                (token_hash, encode_doc_ids(list(doc_numbers)), len(doc_numbers))
-                for token_hash, doc_numbers in postings.items()
-            ),
-        )
-        connection.execute("CREATE INDEX idx_documents_doc_id ON documents(doc_id)")
+    shutil.rmtree(bucket_dir, ignore_errors=True)
+    bucket_dir.mkdir(parents=True)
+    bucket_files = [(bucket_dir / f"{index:02x}.bin").open("wb") for index in range(POSTING_BUCKETS)]
+    document_count = 0
+    posting_count = 0
+    try:
+        with sqlite3.connect(tmp_path) as connection:
+            connection.execute("PRAGMA journal_mode = OFF")
+            connection.execute("PRAGMA synchronous = OFF")
+            connection.execute("PRAGMA temp_store = FILE")
+            connection.execute("CREATE TABLE documents (doc_id TEXT PRIMARY KEY, summary TEXT NOT NULL)")
+            connection.execute("CREATE TABLE postings (token_hash BLOB PRIMARY KEY, docs BLOB NOT NULL, doc_count INTEGER NOT NULL)")
+            connection.execute("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
+            connection.execute("INSERT INTO metadata (key, value) VALUES ('tokenizer_version', ?)", (TOKENIZER_VERSION,))
+            connection.execute("CREATE VIRTUAL TABLE content_fts USING fts5(content, tokenize='trigram', detail='full')")
+            for index, record in enumerate(source_records, 1):
+                doc_id = record["doc_id"]
+                file_path = get_doc_storage_path(record)
+                if not file_path.exists():
+                    missing += 1
+                    continue
+                text = file_path.read_text(encoding="utf-8", errors="ignore")
+                number = doc_number(doc_id)
+                connection.execute("INSERT INTO documents (doc_id, summary) VALUES (?, ?)", (doc_id, build_summary(text)))
+                connection.execute("INSERT INTO content_fts (rowid, content) VALUES (?, ?)", (number, fts_content(text)))
+                document_count += 1
+                for token in index_tokens(text) | literal_tokens(text):
+                    token_hash = hashlib.sha256(token.encode("utf-8")).digest()[:16]
+                    bucket_files[token_hash[0]].write(token_hash + struct.pack(">I", number))
+                if index % 100 == 0:
+                    connection.commit()
+                if index % 1000 == 0:
+                    print(f"{source}: indexed {index}/{len(source_records)} records", flush=True)
+            connection.commit()
+            for bucket_file in bucket_files:
+                bucket_file.close()
+            for bucket_index in range(POSTING_BUCKETS):
+                grouped: dict[bytes, list[int]] = defaultdict(list)
+                with (bucket_dir / f"{bucket_index:02x}.bin").open("rb") as source_file:
+                    while chunk := source_file.read(20):
+                        if len(chunk) != 20:
+                            raise RuntimeError("corrupt posting bucket")
+                        grouped[chunk[:16]].append(struct.unpack(">I", chunk[16:])[0])
+                connection.executemany(
+                    "INSERT INTO postings (token_hash, docs, doc_count) VALUES (?, ?, ?)",
+                    ((token_hash, encode_doc_ids(numbers), len(numbers)) for token_hash, numbers in grouped.items()),
+                )
+                posting_count += len(grouped)
+                if bucket_index % 16 == 15:
+                    connection.commit()
+                    print(f"{source}: merged posting buckets {bucket_index + 1}/{POSTING_BUCKETS}", flush=True)
+            connection.execute("CREATE INDEX idx_documents_doc_id ON documents(doc_id)")
+    finally:
+        for bucket_file in bucket_files:
+            if not bucket_file.closed:
+                bucket_file.close()
+        shutil.rmtree(bucket_dir, ignore_errors=True)
     tmp_path.replace(db_path)
     print(
         f"{source}: wrote {db_path.relative_to(BASE_DIR)} "
-        f"with {len(documents)} documents, {len(postings)} tokens, {missing} missing files"
+        f"with {document_count} documents, {posting_count} tokens, {missing} missing files",
+        flush=True,
     )
 
 def main() -> None:

start.sh

@@ -0,0 +1,7 @@
+#!/bin/sh
+set -eu
+
+mkdir -p /app/data/fulltext
+python /app/build_fulltext_db.py &
+
+exec uvicorn app:app --host 0.0.0.0 --port 7860 --log-level info