fix(sql): canonicalize WHERE predicate operand order before planning

Normalize comparisons to column-first (`literal = col` -> `col = literal`,
flipping the operator) at the shared WHERE-to-filter conversion point used
by both regular SELECT/DELETE planning and recursive CTE join links. Without
this, a literal-first predicate bypassed the primary-key point-get rewrite
and fell back to a sequential scan, letting the same logical predicate route
to two different physical operators that could disagree after crash
recovery or DROP/recreate cycles.

Add regression coverage for the affected scenarios: point-get plan parity
for both operand orders, PK index vs. scan agreement after kill -9 and
after DROP/recreate on document_strict collections, and IS NULL polarity
on filtered reads/deletes. Reformat an unrelated assertion in the native
transaction visibility test to satisfy line-length formatting.

Author: farhan-syah

nodedb-sql/src/planner/cte/join_link.rs

@@ -76,10 +76,11 @@ pub(super) fn extract_recursive_info(expr: &SetExpr, cte_name: &str) -> Result<R
 
     let mut filters = Vec::new();
     if let Some(where_expr) = &select.selection {
-        let converted = crate::resolver::expr::convert_expr(where_expr)?;
-        filters.push(Filter {
-            expr: FilterExpr::Expr(converted),
-        });
+        // Route through the shared WHERE→filter converter so the predicate is
+        // operand-order canonicalized like every other WHERE clause.
+        filters.extend(crate::planner::select::convert_where_to_filters(
+            where_expr,
+        )?);
     }
 
     Ok((filters, join_link))

nodedb-sql/src/planner/select/helpers.rs

@@ -77,12 +77,79 @@ pub fn qualified_name(table: Option<&str>, name: &str) -> String {
 
 /// Convert a WHERE expression into a list of Filter.
 pub fn convert_where_to_filters(expr: &ast::Expr) -> Result<Vec<Filter>> {
-    let sql_expr = convert_expr(expr)?;
+    let sql_expr = canonicalize_predicate(convert_expr(expr)?);
     Ok(vec![Filter {
         expr: FilterExpr::Expr(sql_expr),
     }])
 }
 
+/// Canonicalize comparison operand order so a bare column always sits on the
+/// left of a comparison (`column <op> literal`), flipping the operator when a
+/// swap happens (`5 < id` → `id > 5`).
+///
+/// Written literal-first (`'x' = id`), a predicate is logically identical to
+/// its column-first spelling but reaches the planner in a shape only some
+/// extractors recognize: the primary-key point-get rewrite accepts
+/// `column = literal` but not `literal = column`, so the two spellings would
+/// route to different physical operators (index point-lookup vs sequential
+/// scan) that disagree whenever the index and the stored tuples diverge.
+/// Normalizing at this single WHERE→filter choke point guarantees every
+/// downstream extractor — point-get, index lookup, scan-filter fast path —
+/// sees one canonical shape.
+fn canonicalize_predicate(expr: SqlExpr) -> SqlExpr {
+    match expr {
+        SqlExpr::BinaryOp { left, op, right } => {
+            let left = canonicalize_predicate(*left);
+            let right = canonicalize_predicate(*right);
+            // Only swap when the column sits on the right and the left side is
+            // not itself a bare column — the exact shape (`literal = column`)
+            // the point-get / fast-path extractors fail to recognize.
+            if let Some(flipped) = flip_comparison(op)
+                && !matches!(left, SqlExpr::Column { .. })
+                && matches!(right, SqlExpr::Column { .. })
+            {
+                return SqlExpr::BinaryOp {
+                    left: Box::new(right),
+                    op: flipped,
+                    right: Box::new(left),
+                };
+            }
+            SqlExpr::BinaryOp {
+                left: Box::new(left),
+                op,
+                right: Box::new(right),
+            }
+        }
+        SqlExpr::UnaryOp { op, expr } => SqlExpr::UnaryOp {
+            op,
+            expr: Box::new(canonicalize_predicate(*expr)),
+        },
+        other => other,
+    }
+}
+
+/// Map a comparison operator to the operator that preserves meaning when its
+/// two operands are swapped. Returns `None` for non-comparison operators
+/// (arithmetic, logical, string), which are never operand-swapped here.
+fn flip_comparison(op: BinaryOp) -> Option<BinaryOp> {
+    match op {
+        BinaryOp::Eq => Some(BinaryOp::Eq),
+        BinaryOp::Ne => Some(BinaryOp::Ne),
+        BinaryOp::Gt => Some(BinaryOp::Lt),
+        BinaryOp::Ge => Some(BinaryOp::Le),
+        BinaryOp::Lt => Some(BinaryOp::Gt),
+        BinaryOp::Le => Some(BinaryOp::Ge),
+        BinaryOp::Add
+        | BinaryOp::Sub
+        | BinaryOp::Mul
+        | BinaryOp::Div
+        | BinaryOp::Mod
+        | BinaryOp::And
+        | BinaryOp::Or
+        | BinaryOp::Concat => None,
+    }
+}
+
 pub fn extract_func_args(func: &ast::Function) -> Result<Vec<ast::Expr>> {
     match &func.args {
         ast::FunctionArguments::List(args) => Ok(args

nodedb-sql/tests/point_get_operand_order.rs

@@ -0,0 +1,81 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Primary-key equality must be rewritten to a point-lookup regardless of
+//! operand order.
+//!
+//! `WHERE id = 'x'` and `WHERE 'x' = id` are logically identical. The
+//! point-get optimizer must produce the same `SqlPlan::PointGet` for both.
+//! If it rewrites only the column-first spelling and leaves the literal-first
+//! spelling as a `Scan`, the same predicate is answered by two different
+//! physical operators — an index point-lookup vs a sequential scan — which
+//! disagree whenever the primary-key index and the stored tuples diverge
+//! (the operand-direction asymmetry the ghost-tuple report observed:
+//! `id = 'x'` returned no row while `'x' = id` returned one).
+
+use nodedb_sql::types::{CollectionInfo, EngineType};
+use nodedb_sql::{SqlCatalog, SqlCatalogError, SqlPlan, plan_sql};
+use nodedb_types::DatabaseId;
+
+struct Catalog;
+
+impl SqlCatalog for Catalog {
+    fn get_collection(
+        &self,
+        _: DatabaseId,
+        name: &str,
+    ) -> std::result::Result<Option<CollectionInfo>, SqlCatalogError> {
+        let info = match name {
+            "articles" => Some(CollectionInfo {
+                name: "articles".into(),
+                engine: EngineType::DocumentStrict,
+                columns: Vec::new(),
+                primary_key: Some("id".into()),
+                has_auto_tier: false,
+                indexes: Vec::new(),
+                bitemporal: false,
+                primary: nodedb_types::PrimaryEngine::Document,
+                vector_primary: None,
+                partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
+            }),
+            _ => None,
+        };
+        Ok(info)
+    }
+
+    fn lookup_array(&self, _name: &str) -> Option<nodedb_sql::types::ArrayCatalogView> {
+        None
+    }
+
+    fn array_exists(&self, _name: &str) -> bool {
+        false
+    }
+}
+
+fn plan_one(sql: &str) -> SqlPlan {
+    let mut plans = plan_sql(sql, &Catalog).expect("planning must succeed");
+    assert_eq!(plans.len(), 1, "expected exactly one plan for: {sql}");
+    plans.pop().unwrap()
+}
+
+/// Control: column-first PK equality is rewritten to a point-lookup.
+#[test]
+fn column_first_pk_equality_is_point_get() {
+    let plan = plan_one("SELECT id FROM articles WHERE id = 'x'");
+    assert!(
+        matches!(plan, SqlPlan::PointGet { .. }),
+        "`id = 'x'` on the primary key must plan as a point-lookup, got {plan:?}"
+    );
+}
+
+/// The defect: literal-first PK equality must ALSO be rewritten to a
+/// point-lookup. `'x' = id` is logically identical to `id = 'x'`; leaving it a
+/// `Scan` routes the same predicate to a different physical operator.
+#[test]
+fn literal_first_pk_equality_is_point_get() {
+    let plan = plan_one("SELECT id FROM articles WHERE 'x' = id");
+    assert!(
+        matches!(plan, SqlPlan::PointGet { .. }),
+        "`'x' = id` on the primary key must plan as the same point-lookup as \
+         `id = 'x'`, but operand order left it as {plan:?}"
+    );
+}

nodedb/tests/crash_recovery.rs

@@ -40,6 +40,67 @@ async fn committed_write_survives_kill_9() {
     );
 }
 
+/// After a hard crash and WAL replay, the primary-key point-lookup index and
+/// the sequential-scan surface must agree for a document_strict collection.
+/// The reported ghost tuples were rows the scan returned but no point-lookup
+/// (`WHERE id = ...`) nor `COUNT(*)` could reach — a PK-index-vs-tuple-store
+/// divergence that surfaced after restart cycles. A single kill -9 + reopen
+/// must never produce that split: every scan-visible row is reachable by its
+/// own primary key, and `COUNT(*)` equals the scanned row count.
+#[tokio::test(flavor = "multi_thread")]
+async fn pk_index_and_scan_agree_after_kill_9() {
+    let mut h = CrashHarness::new();
+    h.spawn();
+    h.wait_ready(Duration::from_secs(20));
+
+    h.exec(
+        "CREATE COLLECTION ghost_check (id TEXT PRIMARY KEY, title TEXT) \
+         WITH (engine='document_strict')",
+    )
+    .await;
+    for i in 0..7u32 {
+        h.exec(&format!(
+            "INSERT INTO ghost_check (id, title) VALUES ('smoke_{i}', 't{i}')"
+        ))
+        .await;
+    }
+
+    h.kill_9();
+    h.reopen();
+
+    let mut scan = h.query_col("SELECT id FROM ghost_check", "id").await;
+    scan.sort();
+    let expected: Vec<String> = (0..7u32).map(|i| format!("smoke_{i}")).collect();
+    assert_eq!(
+        scan, expected,
+        "all 7 committed rows must survive kill -9 + WAL replay (got {scan:?})"
+    );
+
+    // Every scan-visible row must be reachable by its own PK point-lookup;
+    // a scan-visible row unreachable by point-lookup is a ghost tuple.
+    for id in &scan {
+        let point = h
+            .query_col(
+                &format!("SELECT id FROM ghost_check WHERE id = '{id}'"),
+                "id",
+            )
+            .await;
+        assert_eq!(
+            point,
+            vec![id.clone()],
+            "scanned row '{id}' unreachable by point-lookup after crash recovery — a ghost tuple (got {point:?})"
+        );
+    }
+
+    // COUNT(*) must agree with the scan, not exclude the recovered rows.
+    let count = h.query_col_idx("SELECT COUNT(*) FROM ghost_check", 0).await;
+    assert_eq!(
+        count,
+        vec!["7".to_string()],
+        "COUNT(*) must equal the scanned row count after crash recovery (got {count:?})"
+    );
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn kv_write_survives_kill_9() {
     let mut h = CrashHarness::new();

nodedb/tests/native_txn_commit_visibility.rs

@@ -153,14 +153,27 @@ async fn native_committed_txn_row_visible_to_point_and_filtered_reads() {
         &format!("INSERT INTO {coll} (id, name) VALUES ('a1', 'alpha')"),
     )
     .await;
-    assert_eq!(insert.status, ResponseStatus::Ok, "in-tx INSERT: {insert:?}");
+    assert_eq!(
+        insert.status,
+        ResponseStatus::Ok,
+        "in-tx INSERT: {insert:?}"
+    );
 
     let commit = request_drain(&mut stream, 3, OpCode::Commit, TextFields::default()).await;
     assert_eq!(commit.status, ResponseStatus::Ok, "COMMIT op: {commit:?}");
 
     // PK point lookup on the writing connection.
-    let point = sql_drain(&mut stream, 4, &format!("SELECT id FROM {coll} WHERE id = 'a1'")).await;
-    assert_ne!(point.status, ResponseStatus::Error, "point lookup: {point:?}");
+    let point = sql_drain(
+        &mut stream,
+        4,
+        &format!("SELECT id FROM {coll} WHERE id = 'a1'"),
+    )
+    .await;
+    assert_ne!(
+        point.status,
+        ResponseStatus::Error,
+        "point lookup: {point:?}"
+    );
     assert_eq!(
         first_cell(&point),
         Some(Value::String("a1".into())),
@@ -174,7 +187,11 @@ async fn native_committed_txn_row_visible_to_point_and_filtered_reads() {
         &format!("SELECT count(*) FROM {coll} WHERE name = 'alpha'"),
     )
     .await;
-    assert_ne!(count.status, ResponseStatus::Error, "filtered count: {count:?}");
+    assert_ne!(
+        count.status,
+        ResponseStatus::Error,
+        "filtered count: {count:?}"
+    );
     assert_eq!(
         first_cell(&count),
         Some(Value::Integer(1)),
@@ -193,7 +210,12 @@ async fn native_committed_txn_row_visible_to_point_and_filtered_reads() {
 
     // Fresh connection — the miss must not persist across sessions.
     let mut fresh = native_session(&server).await;
-    let point2 = sql_drain(&mut fresh, 1, &format!("SELECT id FROM {coll} WHERE id = 'a1'")).await;
+    let point2 = sql_drain(
+        &mut fresh,
+        1,
+        &format!("SELECT id FROM {coll} WHERE id = 'a1'"),
+    )
+    .await;
     assert_ne!(
         point2.status,
         ResponseStatus::Error,

nodedb/tests/sql_drop_collection.rs

@@ -133,3 +133,66 @@ async fn drop_then_recreate_same_name_starts_empty() {
         "re-created collection must start empty; old rows resurrected: {after:?}"
     );
 }
+
+/// The strict-engine twin of `drop_then_recreate_same_name_starts_empty`.
+/// A `document_strict` collection stores its Binary Tuples under the same
+/// name-derived `{db}:{tenant}:{name}:` prefix that a re-CREATE reuses —
+/// collection identity is not per-creation. DROP must purge every stored
+/// tuple (and its indexes) so a re-created same-name strict collection starts
+/// empty; otherwise the old tuples are inherited by the new incarnation and
+/// scan as all-NULL ghosts.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn drop_then_recreate_document_strict_starts_empty() {
+    let srv = TestServer::start().await;
+
+    srv.exec(
+        "CREATE COLLECTION recycled_strict (id TEXT PRIMARY KEY, title TEXT, body TEXT) \
+         WITH (engine='document_strict')",
+    )
+    .await
+    .unwrap();
+    for i in 0..7u32 {
+        srv.exec(&format!(
+            "INSERT INTO recycled_strict (id, title, body) VALUES ('smoke_{i}', 't{i}', 'b{i}')"
+        ))
+        .await
+        .unwrap();
+    }
+    let before = srv
+        .query_rows("SELECT id FROM recycled_strict")
+        .await
+        .unwrap();
+    assert_eq!(
+        before.len(),
+        7,
+        "7 rows must exist before DROP, got {before:?}"
+    );
+
+    srv.exec("DROP COLLECTION recycled_strict").await.unwrap();
+    srv.exec(
+        "CREATE COLLECTION recycled_strict (id TEXT PRIMARY KEY, title TEXT, body TEXT) \
+         WITH (engine='document_strict')",
+    )
+    .await
+    .unwrap();
+
+    let after = srv
+        .query_rows("SELECT id FROM recycled_strict")
+        .await
+        .unwrap();
+    assert_eq!(
+        after.len(),
+        0,
+        "re-created strict collection must start empty; old tuples resurrected: {after:?}"
+    );
+
+    let count = srv
+        .query_text("SELECT COUNT(*) FROM recycled_strict")
+        .await
+        .unwrap();
+    assert_eq!(
+        count,
+        vec!["0".to_string()],
+        "COUNT(*) on a freshly re-created strict collection must be 0, got {count:?}"
+    );
+}