Solr + Postgres JSONB: Faceted, Full-Text, and Geo Search at Speed¶

# seed/seed_and_index.py
import os
import json
import time
import psycopg2
import requests
from datetime import datetime, timedelta
from typing import List, Dict, Any

DB_DSN = os.getenv("DB_DSN", "postgresql://postgres:postgres@postgres:5432/app")
SOLR_URL = os.getenv("SOLR_URL", "http://solr:8983/solr/appcore")

def rows_since(conn: psycopg2.extensions.connection, ts: datetime) -> List[Dict[str, Any]]:
    """Fetch rows changed since timestamp."""
    with conn.cursor() as cur:
        cur.execute("""
          SELECT id::text, org, published_at, author,
                 ST_X(ST_AsText(geom::geometry)) AS lon,
                 ST_Y(ST_AsText(geom::geometry)) AS lat,
                 doc
          FROM articles
          WHERE updated_at > %s
          ORDER BY updated_at
        """, (ts,))

        docs = []
        for row in cur:
            doc = {
                "id": row[0],
                "org_s": row[1],
                "published_dt": row[2].isoformat() if row[2] else None,
                "author_s": row[3],
                "loc_p": f"{row[5]},{row[4]}" if row[4] and row[5] else None,
                "title_t": row[6].get("title"),
                "body_t": row[6].get("body"),
                "tags_ss": row[6].get("tags", []),
                "lang_s": row[6].get("lang")
            }
            # Remove None values
            doc = {k: v for k, v in doc.items() if v is not None}
            docs.append(doc)

        return docs

def solr_upsert(docs: List[Dict[str, Any]]) -> None:
    """Upsert documents to Solr."""
    if not docs:
        return

    payload = {
        "add": [
            {"doc": doc, "commitWithin": 2000} 
            for doc in docs
        ]
    }

    try:
        r = requests.post(f"{SOLR_URL}/update", json=payload, timeout=30)
        r.raise_for_status()

        # Commit the changes
        requests.get(f"{SOLR_URL}/update?commit=true", timeout=10)
        print(f"✅ Indexed {len(docs)} documents")

    except requests.exceptions.RequestException as e:
        print(f"❌ Solr update failed: {e}")
        raise

def setup_solr_schema() -> None:
    """Setup Solr schema via Schema API."""
    schema_config = {
        "add-field": [
            {"name":"id","type":"string","stored":True,"indexed":True},
            {"name":"org_s","type":"string","stored":True,"indexed":True},
            {"name":"published_dt","type":"pdate","stored":True,"indexed":True},
            {"name":"author_s","type":"string","stored":True,"indexed":True},
            {"name":"loc_p","type":"location","stored":True,"indexed":True},
            {"name":"title_t","type":"text_en","stored":True,"indexed":True},
            {"name":"body_t","type":"text_en","stored":False,"indexed":True},
            {"name":"tags_ss","type":"strings","stored":True,"indexed":True},
            {"name":"lang_s","type":"string","stored":True,"indexed":True}
        ],
        "add-copy-field":[
            {"source":"title_t","dest":"_text_"},
            {"source":"body_t","dest":"_text_"}
        ]
    }

    try:
        r = requests.post(f"{SOLR_URL}/schema", json=schema_config, timeout=30)
        r.raise_for_status()
        print("✅ Solr schema configured")
    except requests.exceptions.RequestException as e:
        print(f"⚠️  Schema setup failed (may already exist): {e}")

def main():
    """Main indexing loop."""
    print("🚀 Starting Postgres → Solr indexer")

    # Setup Solr schema
    setup_solr_schema()

    # Connect to Postgres
    conn = psycopg2.connect(DB_DSN)
    checkpoint = datetime.utcnow() - timedelta(days=365)

    print(f"📊 Starting from checkpoint: {checkpoint.isoformat()}")

    try:
        while True:
            docs = rows_since(conn, checkpoint)
            if docs:
                solr_upsert(docs)
                checkpoint = datetime.utcnow()
                print(f"🔄 Updated checkpoint: {checkpoint.isoformat()}")
            else:
                print("⏳ No new documents, waiting...")

            time.sleep(5)

    except KeyboardInterrupt:
        print("🛑 Indexer stopped by user")
    finally:
        conn.close()

if __name__ == "__main__":
    main()

# fastapi/app.py
import os
import json
from typing import List, Optional
from fastapi import FastAPI, Query, HTTPException
import psycopg2
import requests
from urllib.parse import urlencode

DB_DSN = os.getenv("DB_DSN")
SOLR_URL = os.getenv("SOLR_URL")

app = FastAPI(title="Hybrid Search API", version="1.0.0")

def pg_candidates(org: Optional[str], date_from: Optional[str], 
                 date_to: Optional[str], tags: Optional[List[str]]) -> List[str]:
    """Phase 1: Get candidate IDs from Postgres filters."""
    sql = """
      SELECT id::text FROM articles
      WHERE (%s::text IS NULL OR org = %s)
        AND (%s::timestamptz IS NULL OR published_at >= %s)
        AND (%s::timestamptz IS NULL OR published_at <= %s)
        AND (%s::text[] IS NULL OR doc->'tags' ?| %s)
      LIMIT 5000
    """

    with psycopg2.connect(DB_DSN) as conn, conn.cursor() as cur:
        cur.execute(sql, (org, org, date_from, date_from, date_to, date_to, tags, tags))
        return [row[0] for row in cur.fetchall()]

@app.get("/search")
def search(
    q: str = "",
    org: Optional[str] = None,
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    tags: Optional[List[str]] = Query(default=None),
    near: Optional[str] = None,  # "lat,lon,radius"
    facets: bool = True,
    page: int = 1,
    size: int = 10
):
    """
    Hybrid search endpoint combining Postgres filters with Solr search.

    - q: Full-text search query
    - org: Organization filter
    - date_from/date_to: Date range filter
    - tags: Tag filters
    - near: Geo proximity (lat,lon,radius_km)
    - facets: Enable faceting
    - page/size: Pagination
    """

    # Phase 1: Postgres filters → candidate IDs
    candidate_ids = pg_candidates(org, date_from, date_to, tags)
    if not candidate_ids:
        return {"hits": [], "facets": {}, "total": 0}

    # Phase 2: Solr search with ID intersection
    solr_params = {
        "q": q or "*:*",
        "rows": size,
        "start": (page - 1) * size,
        "hl": "true",
        "hl.fl": "title_t,body_t",
        "defType": "edismax",
        "qf": "title_t^3 body_t^1",
        "fq": f"id:({' '.join(candidate_ids[:1000])})"  # Cap fq size
    }

    # Add geo search if specified
    if near:
        try:
            lat, lon, radius = near.split(",")
            solr_params.update({
                "sfield": "loc_p",
                "pt": f"{lat},{lon}",
                "d": radius,
                "sort": "geodist() asc"
            })
        except ValueError:
            raise HTTPException(status_code=400, detail="Invalid near format. Use: lat,lon,radius")

    # Add faceting
    if facets:
        solr_params.update({
            "facet": "true",
            "facet.field": ["org_s", "tags_ss"],
            "facet.range": "published_dt",
            "facet.range.gap": "+1DAY",
            "facet.range.start": "NOW-30DAYS/DAY",
            "facet.range.end": "NOW/DAY"
        })

    # Fire Solr query
    try:
        r = requests.get(f"{SOLR_URL}/select", params=solr_params, timeout=10)
        r.raise_for_status()
        data = r.json()
    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"Solr query failed: {e}")

    # Process results
    hits = []
    highlighting = data.get("highlighting", {})

    for doc in data["response"]["docs"]:
        hit = {
            "id": doc["id"],
            "org": doc.get("org_s"),
            "title": doc.get("title_t"),
            "author": doc.get("author_s"),
            "tags": doc.get("tags_ss", []),
            "score": doc.get("score"),
            "snippet": " ".join(highlighting.get(doc["id"], {}).get("body_t", [])[:1]) if highlighting else None
        }
        hits.append(hit)

    # Extract facets
    facets_out = data.get("facet_counts", {}) if facets else {}

    return {
        "hits": hits,
        "facets": facets_out,
        "total": data["response"]["numFound"],
        "page": page,
        "size": size
    }

@app.get("/suggest")
def suggest(q: str = Query(..., min_length=1)):
    """Get search suggestions from Solr."""
    try:
        r = requests.get(f"{SOLR_URL}/suggest", params={
            "suggest": "true",
            "suggest.dictionary": "titleSuggester",
            "suggest.q": q
        }, timeout=5)
        r.raise_for_status()
        return r.json()
    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"Suggestion failed: {e}")

@app.get("/health")
def health():
    """Health check endpoint."""
    return {"status": "healthy", "service": "hybrid-search"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# cdc/solr_sink.py
import json
import requests
from kafka import KafkaConsumer
from typing import Dict, Any

SOLR_URL = "http://solr:8983/solr/appcore"

def process_kafka_message(message: Dict[str, Any]) -> None:
    """Process CDC message and update Solr."""
    op = message.get("op")
    after = message.get("after", {})
    before = message.get("before", {})

    if op == "c":
        # Create - index new document
        doc = transform_to_solr_doc(after)
        solr_upsert([doc])
    elif op == "u":
        # Update - reindex document
        doc = transform_to_solr_doc(after)
        solr_upsert([doc])
    elif op == "d":
        # Delete - remove from Solr
        doc_id = before.get("id")
        solr_delete(doc_id)

def transform_to_solr_doc(row: Dict[str, Any]) -> Dict[str, Any]:
    """Transform Postgres row to Solr document."""
    # Implementation similar to indexer
    pass

def solr_upsert(docs: List[Dict[str, Any]]) -> None:
    """Upsert documents to Solr."""
    payload = {"add": [{"doc": doc} for doc in docs]}
    requests.post(f"{SOLR_URL}/update", json=payload)

def solr_delete(doc_id: str) -> None:
    """Delete document from Solr."""
    payload = {"delete": {"id": doc_id}}
    requests.post(f"{SOLR_URL}/update", json=payload)

# Kafka consumer
consumer = KafkaConsumer('app.public.articles', 
                        bootstrap_servers=['kafka:9092'],
                        value_deserializer=lambda m: json.loads(m.decode('utf-8')))

for message in consumer:
    process_kafka_message(message.value)

# Batch size limits
MAX_BATCH_SIZE = 1000
COMMIT_WITHIN_MS = 2000

# Rate limiting
import time
time.sleep(0.1)  # 100ms between batches

# Dynamic field mapping for new JSONB keys
def map_jsonb_to_solr(doc: Dict[str, Any]) -> Dict[str, Any]:
    """Map JSONB fields to Solr dynamic fields."""
    solr_doc = {}

    for key, value in doc.items():
        if isinstance(value, str):
            solr_doc[f"{key}_s"] = value
        elif isinstance(value, list):
            solr_doc[f"{key}_ss"] = value
        elif isinstance(value, int):
            solr_doc[f"{key}_i"] = value
        elif isinstance(value, datetime):
            solr_doc[f"{key}_dt"] = value.isoformat()

    return solr_doc

# monitoring/solr_metrics.py
import requests
import time
from prometheus_client import Gauge, Counter, start_http_server

# Metrics
solr_docs_total = Gauge('solr_documents_total', 'Total documents in Solr')
solr_query_duration = Gauge('solr_query_duration_seconds', 'Query duration')
solr_errors_total = Counter('solr_errors_total', 'Total Solr errors')

def collect_solr_metrics():
    """Collect Solr metrics for Prometheus."""
    try:
        r = requests.get("http://solr:8983/solr/appcore/admin/mbeans")
        data = r.json()

        # Extract document count
        docs = data['solr-mbeans']['CORE']['appcore']['searcher']['numDocs']
        solr_docs_total.set(docs)

    except Exception as e:
        solr_errors_total.inc()
        print(f"Metrics collection failed: {e}")

if __name__ == "__main__":
    start_http_server(8001)
    while True:
        collect_solr_metrics()
        time.sleep(30)

# Complete hybrid search setup
def setup_hybrid_search():
    # 1. Postgres JSONB source
    # 2. Solr schema with dynamic fields
    # 3. Python indexer (incremental)
    # 4. FastAPI hybrid endpoint
    # 5. Solr features (facets, suggest, geo)
    # 6. Optional CDC pipeline
    # 7. Monitoring & ops

    print("Hybrid search setup complete!")