fix(pgwire): keep both cells when two output columns share a name

`SELECT w.id, b.id FROM w JOIN b ON ...` returned the right table's
value in BOTH columns. The join executor and planner were correct — the
merged row carries `w.id` and `b.id` and `project_row` read each by its
qualified lookup key — but the shaped row is a JSON map keyed by the
column's DISPLAY name, and an unaliased qualified projection displays as
its bare last segment. Both columns keyed on `id`, so the second insert
overwrote the first and every encoder then re-read that one cell by name.

Aliasing (`w.id AS w_id`) produced distinct display names and was
unaffected, which is why the bug only showed on the unaliased spelling.

Shaped rows now store each column's cell under a unique key from
`response_shape::project::cell_keys` — the display name for the first
occurrence, `<name>_<n>` for later duplicates, skipping any candidate
that would shadow another column's name. The three consumers (pgwire
non-streaming and streaming encoders, native conversion) derive the same
keys, so they stay aligned. Wire column names are unchanged: duplicates
still describe as `id`, `id`, matching PostgreSQL. For a duplicate-free
column list the mapping is the identity, so nothing else changes.

Tests: unit coverage for `cell_keys` and duplicate-name `project_row`;
sql_join_correctness gains the unaliased and aliased join projections.

Author: mkhairi

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

@@ -146,15 +146,18 @@ 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);
     let rows = shaped
         .rows
         .iter()
         .map(|row| {
-            shaped
-                .columns
+            cell_keys
                 .iter()
-                .map(|col| {
-                    row.get(col.as_str())
+                .map(|key| {
+                    row.get(key.as_str())
                         .map(json_to_value_display)
                         .unwrap_or(Value::Null)
                 })

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

@@ -28,22 +28,27 @@ use crate::control::server::response_shape::types::{DdlColType, ShapedRows};
 
 use super::super::ddl_encode::col_type_to_field_with_format;
 
-/// Encode one flat row object into a pgwire `DataRow`, using `columns` (in
-/// order) to look up cells in `row` and `column_types` (parallel to `columns`)
-/// to pick each cell's text rendering.
+/// Encode one flat row object into a pgwire `DataRow`, using `cell_keys` (in
+/// order) to look up cells in `row` and `column_types` (parallel to
+/// `cell_keys`) to pick each cell's text rendering.
+///
+/// `cell_keys` are the per-column unique row-map keys derived from the
+/// display names via `response_shape::project::cell_keys` — identical to the
+/// display names except where those repeat (`SELECT w.id, b.id`), in which
+/// case later duplicates carry a `_n` suffix so both cells survive the map.
 ///
 /// Missing keys and explicit JSON `null` both encode as SQL NULL. Every other
 /// cell renders per its column type via [`encode_typed_cell`]; a
 /// missing/short `column_types` entry defaults to `Text`.
 pub(in crate::control::server::pgwire) fn encode_shaped_row(
     schema: &Arc<Vec<FieldInfo>>,
-    columns: &[String],
+    cell_keys: &[String],
     column_types: &[DdlColType],
     formats: &[FieldFormat],
     row: &serde_json::Map<String, serde_json::Value>,
 ) -> PgWireResult<DataRow> {
     let mut encoder = DataRowEncoder::new(schema.clone());
-    for (idx, name) in columns.iter().enumerate() {
+    for (idx, name) in cell_keys.iter().enumerate() {
         let ct = column_types.get(idx).copied().unwrap_or(DdlColType::Text);
         let format = formats.get(idx).copied().unwrap_or(FieldFormat::Text);
         match row.get(name) {
@@ -159,9 +164,12 @@ 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, &columns, &column_types, formats, row))
+        .map(|row| encode_shaped_row(&schema, &keys, &column_types, formats, row))
         .collect();
 
     let response = Response::Query(QueryResponse::new(

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

@@ -107,6 +107,10 @@ pub(crate) fn streaming_shaped_response(
         .iter()
         .map(|c| c.display_name.clone())
         .collect();
+    // Cells live in the shaped 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(&display_columns);
     // Advertise each projected column's real catalog type so the streaming
     // path's RowDescription OIDs match the non-streaming `shaped_query_response`
     // (and the extended-query Describe path); `column_types` also drives the
@@ -160,7 +164,7 @@ pub(crate) fn streaming_shaped_response(
                     break;
                 }
                 let encoded =
-                    encode_shaped_row(&row_schema, &display_columns, &column_types, &row_formats, row)?;
+                    encode_shaped_row(&row_schema, &cell_keys, &column_types, &row_formats, row)?;
                 emitted += 1;
                 yield encoded;
             }

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

@@ -244,9 +244,15 @@ pub fn shape_decoded_rows(decoded: &JsonValue, projection: Option<&OutputSchema>
             let lookup_keys: Vec<String> = s.columns.iter().map(|c| c.lookup_key.clone()).collect();
             let display_names: Vec<String> =
                 s.columns.iter().map(|c| c.display_name.clone()).collect();
+            // Row cells are stored under per-column unique keys, not display
+            // names: duplicate display names (`SELECT w.id, b.id` → `id`,
+            // `id`) would collide in the row map and collapse both wire
+            // columns to the last value. Encoders re-derive the same keys via
+            // `cell_keys` when reading cells.
+            let keys = super::project::cell_keys(&display_names);
             let projected_rows = rows
                 .iter()
-                .map(|row| project_row(row, &lookup_keys, &display_names))
+                .map(|row| project_row(row, &lookup_keys, &display_names, &keys))
                 .collect();
             // Carry each projected column's real catalog type, aligned in
             // order with `display_names`. Only the pgwire encoder consumes
@@ -280,10 +286,15 @@ pub fn shape_decoded_rows(decoded: &JsonValue, projection: Option<&OutputSchema>
 /// Select and rename one flat row's fields per the projection lists, trying
 /// each candidate key in order: the full lookup key, then the bare
 /// (post-dot) column name, then the SELECT alias.
+///
+/// Cells are inserted under `cell_keys` (unique per column, see
+/// [`super::project::cell_keys`]) rather than the display names, which may
+/// repeat across columns and would otherwise collapse in the output map.
 fn project_row(
     row: &Map<String, JsonValue>,
     lookup_keys: &[String],
     display_names: &[String],
+    cell_keys: &[String],
 ) -> Map<String, JsonValue> {
     let mut out = Map::new();
     for (i, lookup_key) in lookup_keys.iter().enumerate() {
@@ -313,7 +324,8 @@ fn project_row(
             })
             .cloned()
             .unwrap_or(JsonValue::Null);
-        out.insert(display_name.to_string(), value);
+        let cell_key = cell_keys.get(i).map(String::as_str).unwrap_or(display_name);
+        out.insert(cell_key.to_string(), value);
     }
     out
 }
@@ -372,3 +384,44 @@ fn single_result_row(text: String) -> ShapedRows {
         notice: None,
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Two projected columns sharing the display name `id` (`SELECT w.id,
+    /// b.id`) must keep both values in the shaped row instead of collapsing
+    /// to the last table's value.
+    #[test]
+    fn project_row_keeps_both_columns_with_duplicate_display_names() {
+        let mut row = Map::new();
+        row.insert("w.id".to_string(), JsonValue::String("w1".to_string()));
+        row.insert("b.id".to_string(), JsonValue::String("b1".to_string()));
+
+        let lookup_keys = vec!["w.id".to_string(), "b.id".to_string()];
+        let display_names = vec!["id".to_string(), "id".to_string()];
+        let keys = super::super::project::cell_keys(&display_names);
+
+        let out = project_row(&row, &lookup_keys, &display_names, &keys);
+        assert_eq!(out.len(), 2, "both cells must survive the projection");
+        assert_eq!(out.get("id"), Some(&JsonValue::String("w1".to_string())));
+        assert_eq!(out.get("id_1"), Some(&JsonValue::String("b1".to_string())));
+    }
+
+    /// Duplicate-free projections still store cells under the display name
+    /// (`cell_keys` is the identity), so existing readers are unaffected.
+    #[test]
+    fn project_row_uses_display_names_when_unique() {
+        let mut row = Map::new();
+        row.insert("w.id".to_string(), JsonValue::String("w1".to_string()));
+        row.insert("b.title".to_string(), JsonValue::String("t".to_string()));
+
+        let lookup_keys = vec!["w.id".to_string(), "b.title".to_string()];
+        let display_names = vec!["id".to_string(), "title".to_string()];
+        let keys = super::super::project::cell_keys(&display_names);
+
+        let out = project_row(&row, &lookup_keys, &display_names, &keys);
+        assert_eq!(out.get("id"), Some(&JsonValue::String("w1".to_string())));
+        assert_eq!(out.get("title"), Some(&JsonValue::String("t".to_string())));
+    }
+}

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

@@ -55,3 +55,69 @@ pub fn is_scan_wrapper(map: &serde_json::Map<String, serde_json::Value>) -> bool
         && matches!(map.get("id"), Some(serde_json::Value::String(_)))
         && matches!(map.get("data"), Some(serde_json::Value::Object(_)))
 }
+
+/// Unique per-column cell keys for a shaped column list.
+///
+/// SQL output column names may legally repeat — `SELECT w.id, b.id` displays
+/// both columns as `id` — but a shaped row is a JSON map and cannot hold two
+/// cells under one key: inserting the second cell would overwrite the first,
+/// making both wire columns render the last value. The shaper therefore
+/// stores the first occurrence of a name under the name itself and each later
+/// duplicate under `<name>_<n>` (n = 1, 2, …), skipping any candidate that
+/// collides with another column's name. Row writers and every protocol
+/// encoder derive keys through this one function so they always agree; for a
+/// duplicate-free column list this is the identity mapping.
+pub fn cell_keys(columns: &[String]) -> Vec<String> {
+    use std::collections::HashSet;
+
+    let names: HashSet<&str> = columns.iter().map(String::as_str).collect();
+    let mut used: HashSet<String> = HashSet::with_capacity(columns.len());
+    columns
+        .iter()
+        .map(|name| {
+            if used.insert(name.clone()) {
+                return name.clone();
+            }
+            let mut n = 1usize;
+            loop {
+                let candidate = format!("{name}_{n}");
+                if !names.contains(candidate.as_str()) && used.insert(candidate.clone()) {
+                    return candidate;
+                }
+                n += 1;
+            }
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::cell_keys;
+
+    fn keys(cols: &[&str]) -> Vec<String> {
+        cell_keys(&cols.iter().map(|s| s.to_string()).collect::<Vec<_>>())
+    }
+
+    /// Duplicate-free column lists map to themselves.
+    #[test]
+    fn cell_keys_is_identity_without_duplicates() {
+        assert_eq!(keys(&["id", "name", "score"]), ["id", "name", "score"]);
+        assert_eq!(keys(&[]), Vec::<String>::new());
+    }
+
+    /// Later duplicates get a `_n` suffix; the first occurrence keeps the
+    /// bare name (`SELECT w.id, b.id` → `id`, `id_1`).
+    #[test]
+    fn cell_keys_suffixes_later_duplicates() {
+        assert_eq!(keys(&["id", "id"]), ["id", "id_1"]);
+        assert_eq!(keys(&["id", "id", "id"]), ["id", "id_1", "id_2"]);
+    }
+
+    /// A suffix candidate never collides with a real column of that name:
+    /// with columns `id, id, id_1` the second `id` skips `id_1` (taken by a
+    /// real column) and becomes `id_2`.
+    #[test]
+    fn cell_keys_skips_candidates_that_shadow_real_columns() {
+        assert_eq!(keys(&["id", "id", "id_1"]), ["id", "id_2", "id_1"]);
+    }
+}

nodedb/tests/sql_join_correctness.rs

@@ -339,6 +339,55 @@ async fn natural_join_is_not_cartesian() {
     }
 }
 
+// ---------------------------------------------------------------------------
+// Qualified same-name projections
+// ---------------------------------------------------------------------------
+
+/// Unaliased table-qualified projections of same-named columns must each
+/// resolve to their own table's value: `SELECT j_t1.id, j_t2.id` returns the
+/// left value in the first column and the right value in the second, not the
+/// last table's value in both. (Aliased projections already behaved; this
+/// pins the unaliased path, where both output columns display as `id`.)
+#[tokio::test]
+async fn qualified_same_name_columns_resolve_per_table_without_aliases() {
+    let server = TestServer::start().await;
+    setup_join_tables(&server).await;
+
+    let rows = server
+        .query_rows(
+            "SELECT j_t1.id, j_t2.id FROM j_t1 \
+             JOIN j_t2 ON j_t2.t1_id = j_t1.id \
+             WHERE j_t2.id = 'p'",
+        )
+        .await
+        .expect("join with unaliased qualified same-name projections");
+
+    assert_eq!(
+        rows,
+        vec![vec!["a".to_string(), "p".to_string()]],
+        "each qualified `id` projection must carry its own table's value"
+    );
+}
+
+/// The aliased spelling of the same projection stays correct (guard against
+/// the unaliased fix regressing the alias path).
+#[tokio::test]
+async fn qualified_same_name_columns_resolve_per_table_with_aliases() {
+    let server = TestServer::start().await;
+    setup_join_tables(&server).await;
+
+    let rows = server
+        .query_rows(
+            "SELECT j_t1.id AS t1_id, j_t2.id AS t2_id FROM j_t1 \
+             JOIN j_t2 ON j_t2.t1_id = j_t1.id \
+             WHERE j_t2.id = 'p'",
+        )
+        .await
+        .expect("join with aliased qualified same-name projections");
+
+    assert_eq!(rows, vec![vec!["a".to_string(), "p".to_string()]]);
+}
+
 // ---------------------------------------------------------------------------
 // LIMIT with post-join filters
 // ---------------------------------------------------------------------------