Fix BM25 partial predicate autogenerate drift

Author: ankitml

paradedb/sqlalchemy/alembic.py

@@ -273,18 +273,76 @@ def _normalized_expression_list(expressions: list[str]) -> list[str]:
     return [_normalize_bm25_expression(expr) for expr in expressions]
 
 
+def _normalize_sql_for_compare(expr: str) -> str:
+    """Normalize SQL text outside string literals for stable comparisons.
+
+    This collapses whitespace, removes identifier quotes, and lowercases
+    non-literal SQL text while preserving the exact contents of single-quoted
+    string literals.
+    """
+    out: list[str] = []
+    in_single = False
+    in_double = False
+    pending_space = False
+    i = 0
+
+    while i < len(expr):
+        ch = expr[i]
+
+        if ch == "'" and not in_double:
+            if pending_space and out and out[-1] != " ":
+                out.append(" ")
+            pending_space = False
+            out.append(ch)
+            if in_single and i + 1 < len(expr) and expr[i + 1] == "'":
+                out.append("'")
+                i += 2
+                continue
+            in_single = not in_single
+            i += 1
+            continue
+
+        if ch == '"' and not in_single:
+            pending_space = False
+            in_double = not in_double
+            i += 1
+            continue
+
+        if in_single:
+            out.append(ch)
+            i += 1
+            continue
+
+        if ch.isspace():
+            pending_space = True
+            i += 1
+            continue
+
+        if pending_space and out and out[-1] != " ":
+            out.append(" ")
+        pending_space = False
+
+        if in_double:
+            out.append(ch)
+        else:
+            out.append(ch.lower())
+        i += 1
+
+    return "".join(out).strip()
+
+
 def _normalize_where(clause: str | None) -> str | None:
     """Normalize a WHERE clause string for comparison.
 
-    Strips whitespace, removes double quotes, and lowercases to reduce
-    false-positive drift detection between PostgreSQL's normalized form
-    and the SQLAlchemy-compiled form.
+    Reduces false-positive drift between PostgreSQL's normalized form and the
+    SQLAlchemy-compiled form while preserving the exact contents of
+    single-quoted string literals.
     """
     if clause is None:
         return None
-    normalized = " ".join(clause.split())
-    normalized = normalized.replace('"', "")
-    return normalized.lower()
+    normalized = _normalize_sql_for_compare(clause)
+    normalized = normalized.replace("::text", "")
+    return _strip_non_pdb_qualifiers(normalized)
 
 
 def _render_where_from_index(index) -> str | None:
@@ -293,13 +351,16 @@ def _render_where_from_index(index) -> str | None:
     if where_clause is None:
         return None
     if isinstance(where_clause, ClauseElement):
-        return str(
-            where_clause.compile(
-                dialect=postgresql.dialect(),
-                compile_kwargs={"literal_binds": True},
-            )
+        return _strip_relation_qualifiers(
+            str(
+                where_clause.compile(
+                    dialect=postgresql.dialect(),
+                    compile_kwargs={"literal_binds": True},
+                )
+            ),
+            index.table.name,
         )
-    return str(where_clause)
+    return _strip_relation_qualifiers(str(where_clause), index.table.name)
 
 
 def _suppress_standard_bm25_ops(upgrade_ops, bm25_names: set[str]) -> None:

tests/integration/test_alembic_integration.py

@@ -534,7 +534,85 @@ def test_autogenerate_detects_changed_partial_predicate(engine):
 
 
 # ---------------------------------------------------------------------------
-# 2d. Autogenerate round-trip converges (no diff after applying ops)
+# 2d. Matching partial indexes should not emit drift
+# ---------------------------------------------------------------------------
+
+def test_autogenerate_no_op_when_partial_indexes_match(engine):
+    _setup_autogen_table(engine, with_index=False)
+    try:
+        with engine.begin() as conn:
+            conn.execute(text(
+                f'CREATE INDEX "{_AG_IDX}" ON "{_AG_TABLE}" '
+                f"USING bm25 (id, description) WITH (key_field='id') WHERE (id > 2)"
+            ))
+
+        m = MetaData()
+        t = Table(_AG_TABLE, m, Column("id", Integer, primary_key=True), Column("description", Text))
+        from sqlalchemy.schema import Index
+
+        Index(
+            _AG_IDX,
+            BM25Field(t.c.id),
+            BM25Field(t.c.description),
+            postgresql_using="bm25",
+            postgresql_with={"key_field": "id"},
+            postgresql_where=t.c.id > 2,
+        )
+
+        upgrade_ops = _run_comparator(engine, m)
+        our_ops = [
+            op for op in upgrade_ops.ops
+            if getattr(op, "index_name", None) == _AG_IDX
+        ]
+        assert not our_ops, f"Expected no ops for matching partial index, got: {our_ops}"
+    finally:
+        _teardown_autogen_table(engine)
+
+
+# ---------------------------------------------------------------------------
+# 2e. String-literal case drift in partial predicates should be detected
+# ---------------------------------------------------------------------------
+
+def test_autogenerate_detects_changed_partial_string_literal_case(engine):
+    _setup_autogen_table(engine, with_index=False)
+    try:
+        with engine.begin() as conn:
+            conn.execute(text(
+                f'CREATE INDEX "{_AG_IDX}" ON "{_AG_TABLE}" '
+                f"USING bm25 (id, description) WITH (key_field='id') "
+                f"WHERE (description = 'ACTIVE')"
+            ))
+
+        m = MetaData()
+        t = Table(_AG_TABLE, m, Column("id", Integer, primary_key=True), Column("description", Text))
+        from sqlalchemy.schema import Index
+
+        Index(
+            _AG_IDX,
+            BM25Field(t.c.id),
+            BM25Field(t.c.description),
+            postgresql_using="bm25",
+            postgresql_with={"key_field": "id"},
+            postgresql_where="description = 'active'::text",
+        )
+
+        upgrade_ops = _run_comparator(engine, m)
+        drop_ops = [
+            op for op in upgrade_ops.ops
+            if isinstance(op, pdb_alembic.DropBM25IndexOp) and op.index_name == _AG_IDX
+        ]
+        create_ops = [
+            op for op in upgrade_ops.ops
+            if isinstance(op, pdb_alembic.CreateBM25IndexOp) and op.index_name == _AG_IDX
+        ]
+        assert len(drop_ops) == 1, "Expected DropBM25IndexOp for string-literal case change"
+        assert len(create_ops) == 1, "Expected CreateBM25IndexOp for string-literal case change"
+    finally:
+        _teardown_autogen_table(engine)
+
+
+# ---------------------------------------------------------------------------
+# 2f. Autogenerate round-trip converges (no diff after applying ops)
 # ---------------------------------------------------------------------------
 
 def test_autogenerate_round_trip_converges(engine):
@@ -553,7 +631,7 @@ def test_autogenerate_round_trip_converges(engine):
 
 
 # ---------------------------------------------------------------------------
-# 2e. Full lifecycle: create → query → reindex → query → drop
+# 2g. Full lifecycle: create → query → reindex → query → drop
 # ---------------------------------------------------------------------------
 
 _LIFECYCLE_TABLE = "alembic_lifecycle_test"
@@ -597,7 +675,7 @@ def test_alembic_create_reindex_drop_is_queryable(engine):
 
 
 # ---------------------------------------------------------------------------
-# 2f. Reindex concurrently with AUTOCOMMIT
+# 2h. Reindex concurrently with AUTOCOMMIT
 # ---------------------------------------------------------------------------
 
 _CONC_TABLE = "alembic_conc_test"
@@ -638,7 +716,7 @@ def test_alembic_reindex_concurrently_autocommit(engine):
 
 
 # ---------------------------------------------------------------------------
-# 2g. Autogenerate detects changed key_field
+# 2i. Autogenerate detects changed key_field
 # ---------------------------------------------------------------------------
 
 def test_autogenerate_detects_changed_key_field(engine):
@@ -651,14 +729,15 @@ def test_autogenerate_detects_changed_key_field(engine):
                 f"USING bm25 (id, description) WITH (key_field='id')"
             ))
 
-        # MetaData declares key_field='description' (different)
+        # MetaData declares key_field='description' (different) but keeps the
+        # expression list identical so only key_field drift is under test.
         m = MetaData()
         t = Table(_AG_TABLE, m, Column("id", Integer, primary_key=True), Column("description", Text))
         from sqlalchemy.schema import Index
         Index(
             _AG_IDX,
-            BM25Field(t.c.description),
             BM25Field(t.c.id),
+            BM25Field(t.c.description),
             postgresql_using="bm25",
             postgresql_with={"key_field": "description"},
         )
@@ -674,7 +753,7 @@ def test_autogenerate_detects_changed_key_field(engine):
 
 
 # ---------------------------------------------------------------------------
-# 2h. Expression index lifecycle (with tokenizer)
+# 2j. Expression index lifecycle (with tokenizer)
 # ---------------------------------------------------------------------------
 
 _EXPR_TABLE = "alembic_expr_test"
@@ -719,7 +798,7 @@ def test_alembic_expression_index_lifecycle(engine):
 
 
 # ---------------------------------------------------------------------------
-# 2i. Multi-tokenizer expression lifecycle
+# 2k. Multi-tokenizer expression lifecycle
 # ---------------------------------------------------------------------------
 
 _MULTI_TABLE = "alembic_multi_tok_test"