Singularity Data Dominion

FTS5 Safety Protocol (Rule 15 — MANDATORY)

Every FTS5 query must follow this exact sequence. No exceptions.

import re

def safe_fts5_search(conn, table, fts_table, query, limit=25):
    """FTS5 search with sanitization and LIKE fallback."""
    # Step 1: Sanitize — strip everything except word chars, spaces, wildcards, quotes
    clean = re.sub(r'[^\w\s*"]', ' ', query).strip()
    if not clean:
        return []

    # Step 2: Try FTS5 MATCH
    try:
        rows = conn.execute(f"""
            SELECT *, snippet({fts_table}, 0, '>>>', '<<<', '...', 32) AS snip
            FROM {fts_table}
            WHERE {fts_table} MATCH ?
            ORDER BY rank LIMIT ?
        """, (clean, limit)).fetchall()
        if rows:
            return rows
    except Exception:
        pass  # FTS5 failed — fall through to LIKE

    # Step 3: LIKE fallback with parameterized bind
    like_param = f"%{clean}%"
    return conn.execute(f"""
        SELECT * FROM {table}
        WHERE quote_text LIKE ? OR category LIKE ?
        LIMIT ?
    """, (like_param, like_param, limit)).fetchall()

Schema Introspection (Rule 16 — Before ANY Unfamiliar Table)

def get_columns(conn, table_name):
    """Introspect table schema before querying. MANDATORY for unfamiliar tables."""
    cols = conn.execute(f"PRAGMA table_info({table_name})").fetchall()
    if not cols:
        raise ValueError(f"Table '{table_name}' does not exist")
    return {row[1]: row[2] for row in cols}  # {name: type}

def adaptive_select(conn, table, required_cols, optional_cols=None):
    """Build SELECT dynamically based on actual schema."""
    actual = set(get_columns(conn, table).keys())
    select_cols = [c for c in required_cols if c in actual]
    if optional_cols:
        select_cols += [c for c in optional_cols if c in actual]
    if not select_cols:
        raise ValueError(f"No required columns found in {table}")
    return f"SELECT {', '.join(select_cols)} FROM {table}"

Batch Operations (10-100× Faster)

# CORRECT — executemany batch insert
rows = [(r["text"], r["source"], r["lane"]) for r in evidence_data]
conn.executemany(
    "INSERT OR IGNORE INTO evidence_quotes (quote_text, source_file, lane) VALUES (?, ?, ?)",
    rows
)
conn.commit()

# WRONG — row-by-row (10-100× slower)
for r in evidence_data:
    conn.execute("INSERT INTO evidence_quotes ...", (r["text"],))
    conn.commit()  # commit per row = disaster

Consolidate COUNT(*) Calls

-- CORRECT — single round-trip with subqueries
SELECT
    (SELECT COUNT(*) FROM evidence_quotes) AS evidence_count,
    (SELECT COUNT(*) FROM timeline_events) AS timeline_count,
    (SELECT COUNT(*) FROM authority_chains_v2) AS authority_count,
    (SELECT COUNT(*) FROM impeachment_matrix) AS impeachment_count;

-- WRONG — 4 separate queries (4 round-trips)
SELECT COUNT(*) FROM evidence_quotes;
SELECT COUNT(*) FROM timeline_events;
-- ... etc

Composite Indexes for Hot Queries

-- Evidence by lane + category (hot query in MEEK routing)
CREATE INDEX IF NOT EXISTS idx_eq_lane_cat ON evidence_quotes(lane, category);

-- Authority chains by citation + lane
CREATE INDEX IF NOT EXISTS idx_ac_cite_lane ON authority_chains_v2(primary_citation, lane);

-- Timeline by date range (chronology builder)
CREATE INDEX IF NOT EXISTS idx_te_date ON timeline_events(event_date);

-- Impeachment by target + severity (cross-exam prep)
CREATE INDEX IF NOT EXISTS idx_imp_target_sev ON impeachment_matrix(target, impeachment_value DESC);

Cursor-Based Pagination (NOT OFFSET)

-- CORRECT — cursor-based (fast at any depth, O(log n))
SELECT id, quote_text, source_file
FROM evidence_quotes
WHERE id > :last_seen_id
ORDER BY id LIMIT 50;

-- WRONG — OFFSET (degrades linearly, scans skipped rows)
SELECT id, quote_text, source_file
FROM evidence_quotes
ORDER BY id LIMIT 50 OFFSET 150000;  -- scans 150K rows to skip them

Cross-Database Federation with ATTACH

# Federation across 70+ databases on 6 drives
conn.execute("ATTACH DATABASE ? AS authority", (str(authority_db_path),))
conn.execute("ATTACH DATABASE ? AS brain", (str(brain_db_path),))

# Cross-DB join (prefix ALL table refs with schema alias)
results = conn.execute("""
    SELECT e.quote_text, a.citation_text, b.interpretation
    FROM main.evidence_quotes e
    JOIN authority.authority_chains_v2 a ON e.authority_id = a.id
    JOIN brain.interpretations b ON a.citation_id = b.citation_id
    WHERE e.lane = ?
""", ("A",)).fetchall()

# exFAT drives (J:\) — NO WAL MODE
# conn for J:\ databases:
conn_j = sqlite3.connect("file:///J:/path/db.sqlite?immutable=1", uri=True)
# Or if writes needed:
conn_j.execute("PRAGMA journal_mode = DELETE")
conn_j.execute("PRAGMA synchronous = FULL")

Layer 2: DuckDB Analytics (10-100× Faster Than SQLite OLAP)

When to Use DuckDB vs SQLite

DuckDB + SQLite Scanner Integration

import duckdb

con = duckdb.connect()
con.install_extension("sqlite_scanner")
con.load_extension("sqlite_scanner")

# ATTACH litigation_context.db as read-only
con.execute("ATTACH 'litigation_context.db' AS lit (TYPE sqlite, READ_ONLY)")

# 10-100× faster analytical query
result = con.execute("""
    SELECT lane, category,
           COUNT(*) AS cnt,
           COUNT(DISTINCT source_file) AS unique_sources
    FROM lit.evidence_quotes
    WHERE lane IN ('A', 'D', 'E')
    GROUP BY lane, category
    ORDER BY cnt DESC
""").fetchdf()

Litigation Analytics Patterns

-- Judicial violation trends by month (DuckDB)
SELECT strftime(event_date, '%Y-%m') AS month,
       violation_type,
       COUNT(*) AS violations
FROM lit.judicial_violations
GROUP BY month, violation_type
ORDER BY month, violations DESC;

-- Evidence density heatmap per lane + category
SELECT lane, category,
       COUNT(*) AS evidence_count,
       AVG(CAST(impeachment_value AS FLOAT)) AS avg_severity
FROM lit.impeachment_matrix
GROUP BY lane, category
HAVING evidence_count > 5
ORDER BY avg_severity DESC;

-- Filing readiness aggregation with window functions
SELECT lane, vehicle_name,
       evidence_count, authority_count,
       RANK() OVER (ORDER BY evidence_count + authority_count DESC) AS readiness_rank
FROM lit.filing_readiness;

Layer 3: LanceDB Vector Search (75K Vectors, 384-dim)

Embedding Generation

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")  # 80 MB, CPU-fast

# Generate embeddings for evidence text
texts = ["parental alienation documented", "ex parte order issued"]
embeddings = model.encode(texts, normalize_embeddings=True)
# Shape: (2, 384) — 384-dimensional unit vectors

Vector Search via NEXUS Tool

# Preferred: use the vector_search extension tool
# Returns semantically similar content from 75K vectors
result = vector_search(query="judicial bias ex parte orders", top_k=10)

Hybrid Search: BM25 + Vector Fusion

def hybrid_search(query, keyword_weight=0.4, vector_weight=0.6, top_k=20):
    """Combine BM25 keyword search with vector semantic search."""
    # Stage 1: BM25 keyword results (tantivy/FTS5)
    keyword_results = safe_fts5_search(conn, "evidence_quotes", "evidence_fts", query)

    # Stage 2: Vector similarity results (LanceDB)
    vector_results = vector_search(query=query, top_k=top_k)

    # Stage 3: Reciprocal Rank Fusion (RRF)
    scores = {}
    for rank, r in enumerate(keyword_results):
        scores[r["id"]] = keyword_weight / (60 + rank)
    for rank, r in enumerate(vector_results):
        rid = r.get("id", rank)
        scores[rid] = scores.get(rid, 0) + vector_weight / (60 + rank)

    return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]

When to Use Each Search Mode

Cross-Encoder Reranking (25-35% MRR Boost)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

# Rerank top-50 candidates from bi-encoder retrieval
pairs = [(query, doc["text"]) for doc in candidates[:50]]
scores = reranker.predict(pairs)

# Sort by cross-encoder score for precision
reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)

Layer 4: Polars DataFrames (2-10× Faster Than pandas)

Lazy Evaluation for Large Datasets

import polars as pl

# Lazy frame — computation deferred until .collect()
lf = pl.scan_csv("evidence_export.csv")
result = (
    lf.filter(pl.col("lane") == "A")
      .group_by("category")
      .agg([
          pl.col("quote_text").count().alias("count"),
          pl.col("impeachment_value").mean().alias("avg_severity"),
      ])
      .sort("count", descending=True)
      .collect()
)

DuckDB → Polars Integration

import duckdb, polars as pl

con = duckdb.connect()
# DuckDB query → Polars DataFrame (zero-copy via Arrow)
df = con.execute("""
    SELECT lane, category, COUNT(*) AS cnt
    FROM lit.evidence_quotes GROUP BY lane, category
""").pl()  # .pl() returns Polars DataFrame directly

Layer 5: RAG Pipeline Architecture

End-to-End Pipeline

Document Ingestion (pypdfium2, python-docx)
    ↓
Chunking (512 tokens, 64-token overlap)
    ↓
Embedding (all-MiniLM-L6-v2, 384-dim, CPU)
    ↓
Storage (LanceDB — 75K vectors on disk)
    ↓
Retrieval (hybrid: BM25 keyword + vector cosine)
    ↓
Reranking (cross-encoder/ms-marco-MiniLM)
    ↓
Generation (Ollama llama3.2:3b local OR context assembly)

PDF Extraction (pypdfium2 — 5× Faster Than PyMuPDF)

import pypdfium2 as pdfium

def extract_pdf_text(path, max_pages=30):
    """Extract text from PDF using pypdfium2 (5× faster than PyMuPDF)."""
    pdf = pdfium.PdfDocument(path)
    pages = []
    for i in range(min(len(pdf), max_pages)):
        page = pdf[i]
        text = page.get_textpage().get_text_range()
        pages.append({"page_number": i + 1, "text": text})
    pdf.close()
    return pages

Anti-Patterns (VIOLATIONS = IMMEDIATE FAILURE)

Performance Budgets

Decision Matrix: Which Tool for Which Data Task

Key NEXUS Extension Tools

END OF SINGULARITY-data-dominion v2.0.0 — DuckDB · LanceDB · Polars · FTS5 · RAG · ZERO API