Share search, graph and document semantics across clients

.github/codeql/suppressions.json

@@ -39,6 +39,24 @@
       "path": "nodedb/src/control/server/shared/ddl/neutral/timeseries/rewrite.rs",
       "sink": "^\\s*let partition_dir = ts_base\\.join\\(dir_name\\);",
       "reason": "dir_name is validated with is_plain_path_component immediately above this join (rejects / \\ : NUL . .. leading/trailing dot-space and control chars); a failing name is skipped with a warning before any path is built, so the join stays one level under ts_base."
+    },
+    {
+      "rule": "rust/cleartext-logging",
+      "path": "nodedb-client/src/pg_cell/text.rs",
+      "sink": "^\\s*Ok\\(v\\) => panic!\\(\"\\{text:\\?\\} as \\{ty\\} must be refused, got \\{v:\\?\\}\"\\),",
+      "reason": "Test-only: a panic inside #[cfg(test)] mod tests prints a decoded test fixture when a decode is wrongly accepted. The value is a hard-coded literal, never user data, and the panic goes to test output, not a log."
+    },
+    {
+      "rule": "rust/cleartext-logging",
+      "path": "nodedb/src/control/server/response_shape/compose/kernel.rs",
+      "sink": "^\\s*panic!\\(\"the whole row is an object: \\{:\\?\\}\", shaped\\.rows\\[0\\]\\);",
+      "reason": "Test-only: a panic inside #[cfg(test)] mod tests prints a row built from hard-coded fixture values when the shape assertion fails. No user data reaches it, and the panic goes to test output, not a log."
+    },
+    {
+      "rule": "rust/cleartext-logging",
+      "path": "nodedb/src/data/executor/strict_format/coerce.rs",
+      "sink": "^\\s*panic!\\(\"(\\{input\\}: expected a decimal|expected an instant), got \\{got:\\?\\}\"\\);",
+      "reason": "Test-only: panics inside #[cfg(test)] mod tests print a coerced fixture value when a coercion returns the wrong variant. The inputs are hard-coded literals, never user data, and the panic goes to test output, not a log."
     }
   ]
 }

docs/graph.md

@@ -74,13 +74,28 @@ GRAPH TRAVERSE FROM 'users:alice' DEPTH 3;
 GRAPH TRAVERSE FROM 'users:alice' DEPTH 2 LABEL 'follows' DIRECTION out;
 ```
 
-Breadth-first search from a start node. Returns discovered nodes at each depth level.
+Breadth-first search from a start node. Returns the discovered nodes with their depth, and the crossed edges with their properties: `{"nodes": [{"id", "depth"}], "edges": [{"from", "to", "label", "properties"}]}`. Nodes at the last depth are not expanded. Their edges to nodes already in the result are included.
 
-| Parameter   | Default | Description            |
-| ----------- | ------- | ---------------------- |
-| `DEPTH`     | 2       | Maximum hop count      |
-| `LABEL`     | (any)   | Filter by edge label   |
-| `DIRECTION` | out     | `in`, `out`, or `both` |
+| Parameter    | Default | Description                                   |
+| ------------ | ------- | --------------------------------------------- |
+| `DEPTH`      | 2       | Maximum hop count                             |
+| `LABEL`      | (any)   | Follow edges with any listed label            |
+| `DIRECTION`  | out     | `in`, `out`, or `both`                        |
+| `EDGE WHERE` | (none)  | Cross only edges whose properties match; last |
+
+### Edge Property Predicate
+
+```sql
+GRAPH TRAVERSE FROM 'users:alice' DEPTH 2 LABEL 'follows' EDGE WHERE since >= 2020 AND active = TRUE;
+GRAPH PATH FROM 'a' TO 'z' MAX_DEPTH 6 EDGE WHERE "kind" IN ('road', 'rail') AND NOT (closed = TRUE);
+```
+
+- `EDGE WHERE` is the last clause. Only `GRAPH TRAVERSE` and `GRAPH PATH` accept it.
+- Terms: `=`, `<>`, `!=`, `>`, `>=`, `<`, `<=`, `IN`, `NOT IN`, `IS NULL`, `IS NOT NULL`, joined by `AND`, `OR`, `NOT`.
+- Each comparison names one property and one literal: `NULL`, `TRUE`, `FALSE`, a quoted string, an integer or a finite float.
+- A missing property matches `IS NULL` and `NOT IN`. It never matches `>`, `>=`, `<`, `<=` or `IN`.
+- An edge without properties evaluates as an empty object.
+- The predicate runs on the core that stores the edge, before the edge counts against the visit cap.
 
 ### Neighbors (1-Hop)
 
@@ -98,7 +113,7 @@ GRAPH PATH FROM 'users:alice' TO 'users:charlie';
 GRAPH PATH FROM 'users:alice' TO 'users:charlie' MAX_DEPTH 5 LABEL 'knows';
 ```
 
-Cross-core BFS path finding. Returns an ordered list of node IDs, or empty array if no path exists within `MAX_DEPTH` (default 10).
+Cross-core BFS path finding. Returns an ordered list of node IDs, or empty array if no path exists within `MAX_DEPTH` (default 10). `EDGE WHERE` restricts the path to edges whose properties match, tested in each edge's stored direction.
 
 ---
 

docs/query-language.md

@@ -857,11 +857,15 @@ GRAPH INSERT EDGE IN 'edges' FROM 'alice' TO 'bob' TYPE 'knows' PROPERTIES { sin
 -- Traversal
 GRAPH TRAVERSE FROM 'users:alice' DEPTH 3 LABEL 'follows' DIRECTION out;
 
+-- Traversal over edges whose properties match (EDGE WHERE is the last clause)
+GRAPH TRAVERSE FROM 'users:alice' DEPTH 2 LABEL 'follows', 'knows' EDGE WHERE since >= 2020;
+
 -- Neighbors
 GRAPH NEIGHBORS OF 'users:alice' LABEL 'follows' DIRECTION out;
 
 -- Shortest path
 GRAPH PATH FROM 'users:alice' TO 'users:carol' MAX_DEPTH 5;
+GRAPH PATH FROM 'users:alice' TO 'users:carol' MAX_DEPTH 5 EDGE WHERE "weight" > 0.5;
 
 -- Pattern matching (Cypher subset)
 MATCH (u:User)-[follows]->(other:User)

fuzz/src/targets/strict_tuple.rs

@@ -38,10 +38,7 @@ fn schema_from_descriptor(descriptor: &[u8]) -> StrictSchema {
                 4 => ColumnType::Bytes,
                 5 => ColumnType::Timestamp,
                 6 => ColumnType::Timestamptz,
-                7 => ColumnType::Decimal {
-                    precision: 38,
-                    scale: 10,
-                },
+                7 => ColumnType::Decimal(None),
                 8 => ColumnType::Uuid,
                 9 => ColumnType::Vector(
                     descriptor
@@ -147,10 +144,7 @@ fn all_supported_schema() -> StrictSchema {
         ColumnType::Timestamp,
         ColumnType::Timestamptz,
         ColumnType::SystemTimestamp,
-        ColumnType::Decimal {
-            precision: 38,
-            scale: 10,
-        },
+        ColumnType::Decimal(None),
         ColumnType::Uuid,
         ColumnType::Vector(1),
         ColumnType::SparseVector,

nodedb-client-tests/Cargo.toml

@@ -21,3 +21,5 @@ nodedb-types = { workspace = true }
 tokio = { workspace = true, features = ["test-util", "macros", "rt-multi-thread"] }
 tokio-postgres = { workspace = true }
 tempfile = { workspace = true }
+serde_json = { workspace = true }
+sonic-rs = { workspace = true }

nodedb-client-tests/tests/document_declared_key_round_trip.rs

@@ -0,0 +1,230 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! End-to-end tests for a collection whose declared primary key is not `id`.
+//!
+//! The key column is the identity column: every write stores the document id
+//! under it, and every key read and key restriction names it. Both clients
+//! resolve it from the server catalog, so a document put, read, or deleted
+//! through either client is the same row SQL sees under that key.
+
+use std::collections::HashSet;
+
+use nodedb_client::native::pool::PoolConfig;
+use nodedb_client::{Document, NativeClient, NodeDb, NodeDbRemote, SearchResult, Value};
+use nodedb_test_support::pgwire_harness::TestServer;
+
+async fn remote(server: &TestServer) -> NodeDbRemote {
+    NodeDbRemote::connect(&format!(
+        "host=127.0.0.1 port={} user=nodedb dbname=default",
+        server.pg_port
+    ))
+    .await
+    .expect("pgwire connect to harness must succeed")
+}
+
+fn native(server: &TestServer) -> NativeClient {
+    NativeClient::new(PoolConfig::new(
+        format!("127.0.0.1:{}", server.native_port),
+        nodedb_types::protocol::AuthMethod::Trust {
+            username: "nodedb".into(),
+        },
+    ))
+}
+
+async fn sql(remote: &NodeDbRemote, statement: &str) {
+    remote
+        .execute_sql(statement, &[])
+        .await
+        .unwrap_or_else(|e| panic!("{statement}: {e}"));
+}
+
+fn item(id: &str, name: &str, qty: i64) -> Document {
+    let mut doc = Document::new(id);
+    doc.set("name", Value::String(name.into()));
+    doc.set("qty", Value::Integer(qty));
+    doc
+}
+
+/// `written` as it reads back: the declared key is a declared field, so it
+/// reads back holding the document id.
+fn stored(written: &Document) -> Document {
+    let mut doc = written.clone();
+    doc.set("sku", Value::String(written.id.clone()));
+    doc
+}
+
+/// Read `id` through both clients and assert each returns `expected`, or
+/// nothing when `expected` is `None`.
+async fn assert_both_read(
+    remote: &NodeDbRemote,
+    native: &NativeClient,
+    id: &str,
+    expected: Option<&Document>,
+) {
+    let over_pgwire = remote
+        .document_get("items", id)
+        .await
+        .unwrap_or_else(|e| panic!("remote get {id}: {e}"));
+    let over_native = native
+        .document_get("items", id)
+        .await
+        .unwrap_or_else(|e| panic!("native get {id}: {e}"));
+    assert_eq!(over_pgwire.as_ref(), expected, "remote read of {id}");
+    assert_eq!(over_native.as_ref(), expected, "native read of {id}");
+}
+
+#[tokio::test]
+async fn documents_round_trip_under_a_declared_key() {
+    let server = TestServer::start().await;
+    let remote = remote(&server).await;
+    let native = native(&server);
+    sql(
+        &remote,
+        "CREATE COLLECTION items (sku STRING PRIMARY KEY, name STRING) \
+         WITH (engine='document_schemaless')",
+    )
+    .await;
+
+    let by_remote = item("p1", "pen", 3);
+    remote
+        .document_put("items", by_remote.clone())
+        .await
+        .expect("remote put under a declared key");
+    let by_native = item("p2", "ink", 7);
+    native
+        .document_put("items", by_native.clone())
+        .await
+        .expect("native put under a declared key");
+
+    assert_both_read(&remote, &native, "p1", Some(&stored(&by_remote))).await;
+    assert_both_read(&remote, &native, "p2", Some(&stored(&by_native))).await;
+
+    // SQL sees each document under its key column.
+    let keys = remote
+        .execute_sql("SELECT sku FROM items WHERE name = 'ink'", &[])
+        .await
+        .expect("SQL reads the native-written row by its key");
+    assert_eq!(keys.rows, vec![vec![Value::String("p2".into())]]);
+
+    // A scan on a non-key predicate returns the stored fields only: the key
+    // column holds the id, and no `id` column appears beside it.
+    let scanned = remote
+        .execute_sql("SELECT * FROM items WHERE name = 'ink'", &[])
+        .await
+        .expect("SELECT * scan of a declared-key collection");
+    assert!(
+        !scanned.columns.iter().any(|c| c == "id"),
+        "a declared-key scan shows no id column: {:?}",
+        scanned.columns
+    );
+    let sku = scanned
+        .columns
+        .iter()
+        .position(|c| c == "sku")
+        .expect("the scan shows the key column");
+    assert_eq!(scanned.rows.len(), 1);
+    assert_eq!(scanned.rows[0][sku], Value::String("p2".into()));
+    let documents = remote
+        .execute_sql(
+            "SELECT to_jsonb(*) AS document FROM items WHERE name = 'ink'",
+            &[],
+        )
+        .await
+        .expect("to_jsonb(*) scan of a declared-key collection");
+    let [row] = documents.rows.as_slice() else {
+        panic!("one row matches: {:?}", documents.rows);
+    };
+    let Some(Value::String(json)) = row.first() else {
+        panic!("the document cell is JSON text: {row:?}");
+    };
+    let fields: serde_json::Value = sonic_rs::from_str(json).expect("the cell is JSON");
+    assert_eq!(
+        fields,
+        serde_json::json!({"sku": "p2", "name": "ink", "qty": 7}),
+        "a scanned row is exactly its stored fields"
+    );
+
+    // A put replaces the document whole, through either client.
+    let replacement = item("p1", "pencil", 4);
+    native
+        .document_put("items", replacement.clone())
+        .await
+        .expect("native replace of a remote-written document");
+    assert_both_read(&remote, &native, "p1", Some(&stored(&replacement))).await;
+
+    // A key field that names another document is refused.
+    let mut conflicting = item("p3", "cap", 1);
+    conflicting.set("sku", Value::String("other".into()));
+    remote
+        .document_put("items", conflicting.clone())
+        .await
+        .expect_err("remote put whose key field names another id");
+    native
+        .document_put("items", conflicting)
+        .await
+        .expect_err("native put whose key field names another id");
+
+    remote
+        .document_delete("items", "p1")
+        .await
+        .expect("remote delete under a declared key");
+    native
+        .document_delete("items", "p2")
+        .await
+        .expect("native delete under a declared key");
+    assert_both_read(&remote, &native, "p1", None).await;
+    assert_both_read(&remote, &native, "p2", None).await;
+
+    server.graceful_shutdown().await;
+}
+
+fn ids(hits: &[SearchResult]) -> Vec<&str> {
+    hits.iter().map(|h| h.id.as_str()).collect()
+}
+
+#[tokio::test]
+async fn vector_search_restricts_allowed_ids_on_a_declared_key() {
+    let server = TestServer::start().await;
+    let remote = remote(&server).await;
+    let native = native(&server);
+    sql(
+        &remote,
+        "CREATE COLLECTION sku_vecs \
+         FIELDS (sku TEXT PRIMARY KEY, embedding VECTOR(2)) \
+         WITH (engine='vector', m=8, ef_construction=50)",
+    )
+    .await;
+    for (sku, x) in [
+        ("near1", 0.0),
+        ("near2", 0.1),
+        ("near3", 0.2),
+        ("far1", 10.0),
+        ("far2", 20.0),
+        ("far3", 30.0),
+    ] {
+        sql(
+            &remote,
+            &format!("INSERT INTO sku_vecs (sku, embedding) VALUES ('{sku}', ARRAY[{x}, 0.0])"),
+        )
+        .await;
+    }
+
+    // The nearest vectors lie outside the allowed set. The restriction must
+    // apply before the top-k cut, so k rows come back from the allowed set.
+    let allowed: HashSet<String> = ["far1", "far2", "far3"]
+        .iter()
+        .map(|s| s.to_string())
+        .collect();
+    let over_pgwire = remote
+        .vector_search("sku_vecs", &[0.0, 0.0], 2, None, Some(&allowed))
+        .await
+        .expect("remote vector_search with allowed_ids on a declared key");
+    assert_eq!(ids(&over_pgwire), vec!["far1", "far2"], "remote");
+    let over_native = native
+        .vector_search("sku_vecs", &[0.0, 0.0], 2, None, Some(&allowed))
+        .await
+        .expect("native vector_search with allowed_ids on a declared key");
+    assert_eq!(ids(&over_native), vec!["far1", "far2"], "native");
+
+    server.graceful_shutdown().await;
+}