refactor(shape): centralize cell-key derivation and cover HTTP

Review follow-up. The rule "read row cells via the per-column keys, never
by display name" was enforced only by each consumer re-deriving the keys
itself, and the HTTP paths did not: `shape_http_payload`,
`ddl_results_to_json`, and the NDJSON stream serialize the row map
straight to JSON, so the `_<n>` suffix for duplicate output names was
reaching clients as an undocumented, untested wire contract.

`ShapedRows::cell_keys()` is now the single source of truth. pgwire and
native read through it instead of each calling `project::cell_keys`, the
`rows` field documents that cells are keyed by it, and all three HTTP
sites document the resulting JSON shape: a duplicate-name projection
emits `{"id": …, "id_1": …}` because a JSON object cannot repeat a key
and dropping the duplicate would silently lose a projected column.
pgwire and native stay positional and still report both columns as `id`.

An accessor rather than a stored field: `ShapedRows` has 146 struct
literal construction sites, so a required field would pull that many
unrelated edits into this change. Derivation is per response, not per
row, so caching buys nothing measurable.

Tests: http_result_projection asserts both cells survive over `/v1/query`
AND `/v1/query/stream`.

Author: mkhairi

nodedb/src/control/server/http/routes/query_stream.rs

@@ -122,6 +122,9 @@ pub(super) fn ndjson_body_stream(
                     return;
                 }
             };
+            // Row maps are keyed by `ShapedRows::cell_keys`, so each NDJSON
+            // line serializes as-is; two output columns sharing a name emit
+            // `{"id": …, "id_1": …}` rather than collapsing to one cell.
             let shaped = shape_decoded_rows(&value, projection.as_ref());
             for row in shaped.rows {
                 if emitted >= limit {

nodedb/src/control/server/http/routes/result_shape.rs

@@ -47,6 +47,11 @@ pub(super) fn shape_http_payload(
         database_id,
         tenant_id,
     )? {
+        // Each row map is already keyed by `ShapedRows::cell_keys`, so it
+        // serializes to JSON as-is. When two output columns share a name the
+        // later one carries a `_<n>` suffix (`SELECT w.id, b.id` →
+        // `{"id": …, "id_1": …}`) — a JSON object cannot repeat a key, and
+        // dropping the duplicate would silently lose a projected column.
         ShapeOutcome::Rows(shaped) => Ok(HttpShaped::Rows(
             shaped
                 .rows
@@ -111,6 +116,8 @@ pub(super) fn ddl_results_to_json(
                     "tag": command,
                 }));
             }
+            // Keyed by `ShapedRows::cell_keys`, same JSON contract as
+            // `shape_http_payload` above (duplicate names take a `_<n>` key).
             DdlResult::Rows(shaped) => {
                 for row in shaped.rows {
                     rows.push(serde_json::Value::Object(row));

nodedb/src/control/server/native/dispatch/conversion.rs

@@ -146,10 +146,10 @@ pub(crate) fn calvin_native_response(
 /// `json_to_value_display`; a column absent from a given row's map becomes
 /// `Value::Null`.
 pub(crate) fn to_native_columns_rows(shaped: &ShapedRows) -> (Vec<String>, Vec<Vec<Value>>) {
-    // Cells live in the row maps under unique per-column keys (display names
-    // may repeat across columns, e.g. `SELECT w.id, b.id`); derive the same
-    // keys the shaper used so every column reads its own cell.
-    let cell_keys = crate::control::server::response_shape::project::cell_keys(&shaped.columns);
+    // Cells live in the row maps under per-column keys (display names may
+    // repeat across columns, e.g. `SELECT w.id, b.id`), so read through the
+    // shared accessor rather than by display name.
+    let cell_keys = shaped.cell_keys();
     let rows = shaped
         .rows
         .iter()

nodedb/src/control/server/pgwire/handler/shape_encode.rs

@@ -146,6 +146,10 @@ pub(in crate::control::server::pgwire) fn shaped_query_response(
     shaped: ShapedRows,
     formats: &[FieldFormat],
 ) -> (Response, Option<String>) {
+    // Cells live in the row maps under per-column keys that differ from the
+    // display names only when two columns share a name; derived here before
+    // the struct is destructured.
+    let keys = shaped.cell_keys();
     let ShapedRows {
         columns,
         column_types,
@@ -164,9 +168,6 @@ pub(in crate::control::server::pgwire) fn shaped_query_response(
         .collect();
     let schema = Arc::new(fields);
 
-    // Cells live in the row maps under unique per-column keys (display names
-    // may repeat across columns); derive the same keys the shaper used.
-    let keys = crate::control::server::response_shape::project::cell_keys(&columns);
     let encoded_rows: Vec<PgWireResult<DataRow>> = rows
         .iter()
         .map(|row| encode_shaped_row(&schema, &keys, &column_types, formats, row))

nodedb/src/control/server/response_shape/types.rs

@@ -228,6 +228,10 @@ pub struct ShapedRows {
     /// type OIDs; the native and http entrypoints ignore it. `Text` is used
     /// wherever the source type is unknown.
     pub column_types: Vec<DdlColType>,
+    /// One map per row. Cells are keyed by [`ShapedRows::cell_keys`], NOT by
+    /// `columns` directly — SQL output names may repeat (`SELECT w.id, b.id`
+    /// displays both as `id`) and a map cannot hold two cells under one key.
+    /// Read cells through `cell_keys` so each column reads its own value.
     pub rows: Vec<serde_json::Map<String, serde_json::Value>>,
     pub notice: Option<String>,
 }
@@ -238,4 +242,24 @@ impl ShapedRows {
     pub fn text_types(n: usize) -> Vec<DdlColType> {
         vec![DdlColType::Text; n]
     }
+
+    /// Per-column keys for reading cells out of [`ShapedRows::rows`], parallel
+    /// to `columns`.
+    ///
+    /// This is the single source of truth every consumer shares — the pgwire
+    /// encoders, the native converter, and the HTTP JSON serializers all key
+    /// rows through it, so the row-map layout can never drift from what a
+    /// consumer expects.
+    ///
+    /// Identical to `columns` unless two output columns share a name, in which
+    /// case later duplicates take a `_<n>` suffix (see
+    /// [`super::project::cell_keys`]). Because the HTTP transports serialize a
+    /// row map directly to JSON, that suffix is user-visible there: a
+    /// duplicate-name `SELECT w.id, b.id` emits `{"id": …, "id_1": …}`, since
+    /// a JSON object likewise cannot carry the same key twice. pgwire and
+    /// native are positional on the wire and still report both columns as
+    /// `id`, matching PostgreSQL.
+    pub fn cell_keys(&self) -> Vec<String> {
+        super::project::cell_keys(&self.columns)
+    }
 }

nodedb/tests/http_result_projection.rs

@@ -220,3 +220,76 @@ async fn http_vector_search_serializes_distance() {
         "distance must be numeric, got {distance:?}"
     );
 }
+
+// ── duplicate output column names ────────────────────────────────────────────
+
+/// POST `sql` to `/v1/query/stream` and return one parsed JSON value per
+/// NDJSON line.
+async fn query_stream_lines(http_port: u16, sql: &str) -> Vec<serde_json::Value> {
+    let url = format!("http://127.0.0.1:{http_port}/v1/query/stream");
+    let resp = reqwest::Client::new()
+        .post(&url)
+        .json(&serde_json::json!({ "sql": sql }))
+        .send()
+        .await
+        .expect("POST /v1/query/stream");
+    assert!(
+        resp.status().is_success(),
+        "stream query must succeed over HTTP: {} — sql={sql}",
+        resp.status()
+    );
+    let body = resp.text().await.expect("read NDJSON body");
+    body.lines()
+        .filter(|l| !l.trim().is_empty())
+        .map(|l| serde_json::from_str(l).unwrap_or_else(|e| panic!("parse NDJSON line {l:?}: {e}")))
+        .collect()
+}
+
+/// Two joined collections both projecting `id` must keep BOTH values in the
+/// HTTP JSON row. A JSON object cannot repeat a key, so the second column
+/// lands under the `_<n>`-suffixed cell key (`id_1`) — dropping it would
+/// silently lose a projected column, which is what the row map did before
+/// cells were keyed independently of the display name.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn http_keeps_both_cells_for_duplicate_column_names() {
+    let srv = TestServer::start().await;
+    srv.exec("CREATE COLLECTION nb_dup_w (id TEXT PRIMARY KEY, wname TEXT) WITH (engine='document_strict')")
+        .await
+        .expect("create nb_dup_w");
+    srv.exec("CREATE COLLECTION nb_dup_b (id TEXT PRIMARY KEY, ref TEXT) WITH (engine='document_strict')")
+        .await
+        .expect("create nb_dup_b");
+    srv.exec("INSERT INTO nb_dup_w (id, wname) VALUES ('w1','alpha')")
+        .await
+        .expect("insert nb_dup_w");
+    srv.exec("INSERT INTO nb_dup_b (id, ref) VALUES ('b1','w1')")
+        .await
+        .expect("insert nb_dup_b");
+
+    let sql = "SELECT w.id, b.id FROM nb_dup_w w JOIN nb_dup_b b ON b.ref = w.id";
+    let rows = query_rows(srv.http_port, sql).await;
+    let stream_rows = query_stream_lines(srv.http_port, sql).await;
+    srv.graceful_shutdown().await;
+
+    assert_eq!(rows.len(), 1, "one joined row expected: {rows:?}");
+    let obj = rows[0].as_object().expect("row is a JSON object");
+    assert_eq!(
+        obj.get("id").and_then(|v| v.as_str()),
+        Some("w1"),
+        "first `id` must carry the left table's value: {obj:?}"
+    );
+    assert_eq!(
+        obj.get("id_1").and_then(|v| v.as_str()),
+        Some("b1"),
+        "the duplicate `id` must survive under the suffixed key: {obj:?}"
+    );
+
+    // The NDJSON streaming path serializes the same row maps, so it carries
+    // the identical contract.
+    assert_eq!(stream_rows.len(), 1, "one streamed row: {stream_rows:?}");
+    let stream_obj = stream_rows[0]
+        .as_object()
+        .expect("streamed row is an object");
+    assert_eq!(stream_obj.get("id").and_then(|v| v.as_str()), Some("w1"));
+    assert_eq!(stream_obj.get("id_1").and_then(|v| v.as_str()), Some("b1"));
+}