feat: diagnostics helpers, partial/concurrent indexes, agg fix, top_hits from_

Author: ankitml

CLAUDE.md

@@ -0,0 +1,27 @@
+# Testing
+
+## Unit Tests
+
+Run unit tests (no database required) before every push — this is what CI checks first:
+
+```bash
+.venv/bin/python -m pytest -m "not integration"
+```
+
+## Integration Tests
+
+Integration tests require a running ParadeDB/Postgres instance. Use the provided script to start the container and set up the correct DSN:
+
+```bash
+PARADEDB_PASSWORD=postgres scripts/run_integration_tests.sh
+```
+
+To run a subset of tests, pass pytest selectors:
+
+```bash
+PARADEDB_PASSWORD=postgres scripts/run_integration_tests.sh tests/integration/test_indexing_integration.py::test_bm25_partial_index_generates_where_clause
+```
+
+The script sets `PARADEDB_TEST_DSN` and `DATABASE_URL` automatically. The default container name is `paradedb-sqlalchemy-integration` on port `5443`.
+
+Some integration tests require newer pg_search versions and will be automatically skipped if the feature is not available (e.g. diagnostics functions like `pdb.indexes()`).

paradedb/__init__.py

@@ -1,3 +1,64 @@
 from . import sqlalchemy
+from .sqlalchemy.diagnostics import (
+    paradedb_index_segments,
+    paradedb_indexes,
+    paradedb_verify_all_indexes,
+    paradedb_verify_index,
+)
+from .sqlalchemy.facets import with_rows
+from .sqlalchemy.indexing import BM25Field, assert_indexed, describe, tokenize
+from .sqlalchemy.pdb import agg, score, snippet, snippet_positions, snippets
+from .sqlalchemy.search import (
+    ProximityExpr,
+    all,
+    fuzzy,
+    match_all,
+    match_any,
+    more_like_this,
+    near,
+    parse,
+    phrase,
+    phrase_prefix,
+    prox_array,
+    prox_regex,
+    proximity,
+    range_term,
+    regex,
+    regex_phrase,
+    term,
+)
 
-__all__ = ["sqlalchemy"]
+__all__ = [
+    "BM25Field",
+    "ProximityExpr",
+    "agg",
+    "all",
+    "assert_indexed",
+    "describe",
+    "fuzzy",
+    "match_all",
+    "match_any",
+    "more_like_this",
+    "near",
+    "parse",
+    "paradedb_index_segments",
+    "paradedb_indexes",
+    "paradedb_verify_all_indexes",
+    "paradedb_verify_index",
+    "phrase",
+    "phrase_prefix",
+    "prox_array",
+    "prox_regex",
+    "proximity",
+    "range_term",
+    "regex",
+    "regex_phrase",
+    "score",
+    "snippet",
+    "snippet_positions",
+    "snippets",
+    "sqlalchemy",
+    "term",
+    "tokenize",
+    "with_rows",
+]

paradedb/sqlalchemy/__init__.py

@@ -1,3 +1,3 @@
-from . import expr, facets, indexing, inspect, pdb, search, select_with
+from . import diagnostics, expr, facets, indexing, inspect, pdb, search, select_with
 
-__all__ = ["expr", "facets", "indexing", "inspect", "pdb", "search", "select_with"]
+__all__ = ["diagnostics", "expr", "facets", "indexing", "inspect", "pdb", "search", "select_with"]

paradedb/sqlalchemy/diagnostics.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from sqlalchemy.engine import Engine
+
+
+def _exec_and_collect(conn, sql: str, params: list[Any]) -> list[dict[str, Any]]:
+    # exec_driver_sql requires a tuple for single-row execution, or None for no params.
+    result = conn.exec_driver_sql(sql, tuple(params) if params else None)
+    columns = list(result.keys())
+    return [dict(zip(columns, row, strict=False)) for row in result.fetchall()]
+
+
+def paradedb_indexes(engine: Engine) -> list[dict[str, Any]]:
+    """Return metadata for all BM25 indexes from ``pdb.indexes()``."""
+    with engine.connect() as conn:
+        return _exec_and_collect(conn, "SELECT * FROM pdb.indexes()", [])
+
+
+def paradedb_index_segments(engine: Engine, index: str) -> list[dict[str, Any]]:
+    """Return segment metadata for a BM25 index from ``pdb.index_segments()``."""
+    with engine.connect() as conn:
+        return _exec_and_collect(
+            conn, "SELECT * FROM pdb.index_segments(%s::regclass)", [index]
+        )
+
+
+def paradedb_verify_index(
+    engine: Engine,
+    index: str,
+    *,
+    heapallindexed: bool = False,
+    sample_rate: float | None = None,
+    report_progress: bool = False,
+    verbose: bool = False,
+    on_error_stop: bool = False,
+    segment_ids: Sequence[int] | None = None,
+) -> list[dict[str, Any]]:
+    """Run ``pdb.verify_index()`` for one BM25 index."""
+    sql_parts = ["SELECT * FROM pdb.verify_index(%s::regclass"]
+    params: list[Any] = [index]
+    if heapallindexed:
+        sql_parts.append(", heapallindexed => %s::boolean")
+        params.append(heapallindexed)
+    if sample_rate is not None:
+        sql_parts.append(", sample_rate => %s::double precision")
+        params.append(sample_rate)
+    if report_progress:
+        sql_parts.append(", report_progress => %s::boolean")
+        params.append(report_progress)
+    if verbose:
+        sql_parts.append(", verbose => %s::boolean")
+        params.append(verbose)
+    if on_error_stop:
+        sql_parts.append(", on_error_stop => %s::boolean")
+        params.append(on_error_stop)
+    if segment_ids is not None:
+        sql_parts.append(", segment_ids => %s::int[]")
+        params.append(list(segment_ids))
+    sql_parts.append(")")
+    with engine.connect() as conn:
+        return _exec_and_collect(conn, "".join(sql_parts), params)
+
+
+def paradedb_verify_all_indexes(
+    engine: Engine,
+    *,
+    schema_pattern: str | None = None,
+    index_pattern: str | None = None,
+    heapallindexed: bool = False,
+    sample_rate: float | None = None,
+    report_progress: bool = False,
+    on_error_stop: bool = False,
+) -> list[dict[str, Any]]:
+    """Run ``pdb.verify_all_indexes()`` across BM25 indexes."""
+    named_params: list[tuple[str, str, Any]] = []
+    if schema_pattern is not None:
+        named_params.append(("schema_pattern", "text", schema_pattern))
+    if index_pattern is not None:
+        named_params.append(("index_pattern", "text", index_pattern))
+    if heapallindexed:
+        named_params.append(("heapallindexed", "boolean", heapallindexed))
+    if sample_rate is not None:
+        named_params.append(("sample_rate", "double precision", sample_rate))
+    if report_progress:
+        named_params.append(("report_progress", "boolean", report_progress))
+    if on_error_stop:
+        named_params.append(("on_error_stop", "boolean", on_error_stop))
+
+    sql_parts = ["SELECT * FROM pdb.verify_all_indexes("]
+    params: list[Any] = []
+    if named_params:
+        sql_parts.append(
+            ", ".join(
+                f"{name} => %s::{pg_type}"
+                for name, pg_type, _ in named_params
+            )
+        )
+        params.extend(value for _, _, value in named_params)
+    sql_parts.append(")")
+
+    with engine.connect() as conn:
+        return _exec_and_collect(conn, "".join(sql_parts), params)
+
+
+__all__ = [
+    "paradedb_index_segments",
+    "paradedb_indexes",
+    "paradedb_verify_all_indexes",
+    "paradedb_verify_index",
+]

paradedb/sqlalchemy/facets.py

@@ -72,12 +72,15 @@ def range(*, field: str, ranges: list[dict[str, object]]) -> dict[str, dict[str,
 def top_hits(
     *,
     size: int | None = None,
-    sort: list[dict[str, str]] | None = None,
+    from_: int | None = None,
+    sort: list[dict[str, Any]] | None = None,
     docvalue_fields: list[str] | None = None,
 ) -> dict[str, dict[str, object]]:
     payload: dict[str, object] = {}
     if size is not None:
         payload["size"] = size
+    if from_ is not None:
+        payload["from"] = from_
     if sort is not None:
         payload["sort"] = sort
     if docvalue_fields is not None:

paradedb/sqlalchemy/pdb.py

@@ -77,4 +77,7 @@ def agg(spec: dict[str, Any], *, approximate: bool | None = None) -> ClauseEleme
     payload_expr = cast(literal(payload), JSONB)
     if approximate is None:
         return func.pdb.agg(payload_expr)
-    return PDBFunctionWithNamedArgs("agg", [payload_expr], [("approximate", approximate)])
+    # pdb.agg() takes an optional second positional boolean: true = exact (default),
+    # false = approximate (skip heap visibility checks, ~2-4x faster but may include
+    # stale rows). approximate=True → pass false; approximate=False → pass true.
+    return func.pdb.agg(payload_expr, literal(not approximate))

scripts/run_paradedb.sh

@@ -8,7 +8,7 @@ else
   RUNNING=0
 fi
 
-PARADEDB_IMAGE="${PARADEDB_IMAGE:-paradedb/paradedb:0.21.8-pg18}"
+PARADEDB_IMAGE="${PARADEDB_IMAGE:-paradedb/paradedb:0.21.9-pg18}"
 PARADEDB_CONTAINER_NAME="${PARADEDB_CONTAINER_NAME:-paradedb-sqlalchemy-integration}"
 PARADEDB_PORT="${PARADEDB_PORT:-5443}"
 PARADEDB_USER="${PARADEDB_USER:-postgres}"