fix(control): route single-vShard-homed gather to its owning core

A no-GROUP BY scalar aggregate on a single-vShard-homed collection
(document/kv/columnar/timeseries/spatial/vector/text) plans as
Exchange{Gather} over an Aggregate with no input. In single-node
mode this was broadcast to every core, so every empty non-owning
core seeded its own scalar-aggregate identity row and the merge
returned N rows instead of one. gather_all_vshards now delegates to
gather_single_node, which mirrors the cluster branch's routing:
cluster-partitioned leaves still broadcast, but a single-vShard-homed
plan dispatches to its one owning core alone.

Author: farhan-syah

nodedb/src/control/server/exchange/gather.rs

@@ -323,8 +323,8 @@ pub fn gather_all_cores_stream(
 ///
 /// # Single-node mode
 ///
-/// If `state.gateway` is `None` this degenerates to [`gather_all_cores`] with
-/// unchanged behaviour.
+/// If `state.gateway` is `None`, routing is delegated to
+/// [`super::owning_core::gather_single_node`] (same shape-based routing).
 ///
 /// # Cluster mode — single-vShard-homed sources (document, kv, columnar,
 /// timeseries, spatial, vector, text)
@@ -359,8 +359,18 @@ pub async fn gather_all_vshards(
     txn_id: Option<TxnId>,
 ) -> crate::Result<GatherOutcome> {
     let Some(gateway) = state.gateway.get() else {
-        // Single-node: delegate to the local fan-out path unchanged.
-        return gather_all_cores(state, tenant_id, database_id, plan, trace_id, txn_id).await;
+        // Single-node: route by plan shape (cluster-partitioned leaf → broadcast;
+        // single-vShard-homed collection → its one owning core; else broadcast
+        // fallback), mirroring the cluster branch below.
+        return super::owning_core::gather_single_node(
+            state,
+            tenant_id,
+            database_id,
+            plan,
+            trace_id,
+            txn_id,
+        )
+        .await;
     };
 
     if nodedb_physical::physical_plan::plan_contains_cluster_partitioned_leaf(&plan) {

nodedb/src/control/server/exchange/mod.rs

@@ -11,6 +11,7 @@
 pub mod all_cores;
 pub mod full_scan;
 pub mod gather;
+pub mod owning_core;
 pub mod resolve;
 pub mod streamable;
 

nodedb/src/control/server/exchange/owning_core.rs

@@ -0,0 +1,133 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Route a single-vShard-homed plan to its ONE owning Data-Plane core.
+//!
+//! `gather_single_owning_core` is the single-node sibling of
+//! [`super::gather::gather_all_cores`] for plans whose whole collection lives on
+//! exactly one vShard (document / kv / columnar / timeseries / spatial / vector
+//! / text). Instead of broadcasting the plan to every core — which seeds an
+//! identity scalar-aggregate row on each empty non-owning core and returns one
+//! row per core for a no-`GROUP BY` aggregate — it resolves the collection's
+//! owning vShard and dispatches the bare plan to it alone, exactly mirroring
+//! what the cluster branch of [`super::gather::gather_all_vshards`] does via the
+//! gateway. The owning core already holds every row of a single-vShard-homed
+//! collection, so this returns the identical row set a broadcast would have,
+//! minus the empty cores' spurious contributions.
+
+use crate::bridge::envelope::{ErrorCode, PhysicalPlan, Status};
+use crate::control::server::dispatch_utils::dispatch_to_data_plane_with_txn;
+use crate::control::server::payload_merge::{encode_msgpack_array, extract_msgpack_elements};
+use crate::control::state::SharedState;
+use crate::types::{DatabaseId, TenantId, TraceId, TxnId, VShardId};
+
+use super::gather::{GatherOutcome, gather_all_cores};
+
+/// Single-node routing decision for a resolved `Exchange{Gather}` child plan.
+///
+/// Mirrors the cluster branch of [`super::gather::gather_all_vshards`]:
+///
+/// - A cluster-partitioned leaf (graph traversal / array) spreads its rows
+///   across cores by node-id / tile-id, so it fans to every local core via
+///   [`gather_all_cores`].
+/// - A single-vShard-homed collection (document / kv / columnar / timeseries /
+///   spatial / vector / text) lives wholly on ONE core; the bare plan routes to
+///   that owning core via [`gather_single_owning_core`]. Broadcasting would seed
+///   a scalar-aggregate identity row on every empty non-owning core — so a
+///   no-`GROUP BY` aggregate returns one row PER core instead of one merged row
+///   — and would duplicate a plain scan's rows across cores.
+/// - A plan with no resolvable collection (e.g. `ProviderScan` carrying embedded
+///   rows) keeps the broadcast fallback unchanged.
+pub async fn gather_single_node(
+    state: &SharedState,
+    tenant_id: TenantId,
+    database_id: DatabaseId,
+    plan: PhysicalPlan,
+    trace_id: TraceId,
+    txn_id: Option<TxnId>,
+) -> crate::Result<GatherOutcome> {
+    if nodedb_physical::physical_plan::plan_contains_cluster_partitioned_leaf(&plan) {
+        return gather_all_cores(state, tenant_id, database_id, plan, trace_id, txn_id).await;
+    }
+    if let Some(collection) = plan.collection() {
+        let vshard_id = VShardId::from_collection_in_database(database_id, collection);
+        return gather_single_owning_core(
+            state,
+            tenant_id,
+            database_id,
+            plan,
+            vshard_id,
+            trace_id,
+            txn_id,
+        )
+        .await;
+    }
+    gather_all_cores(state, tenant_id, database_id, plan, trace_id, txn_id).await
+}
+
+/// Dispatch `plan` to the single Data-Plane core that owns `vshard_id` and
+/// gather the one bounded response into a [`GatherOutcome`].
+///
+/// `vshard_id` is the collection's owning vShard
+/// (`VShardId::from_collection_in_database(database_id, collection)`); the
+/// dispatcher's `VShardRouter` resolves it to the one core holding the
+/// collection's rows.
+///
+/// The returned outcome carries that core's own `watermark_lsn` /
+/// `read_version_lsn` and exactly one `shard_watermarks` entry keyed to the
+/// collection's vShard — matching the cluster `dispatch_local` path so an
+/// in-transaction read records the same OCC read-set entry the write-set uses
+/// (writes home to the same `from_collection_in_database` vShard). Aggregate
+/// finalization (`finalize_aggregate`) is a passthrough over the merged array,
+/// so one complete aggregate row in yields one row out.
+pub async fn gather_single_owning_core(
+    state: &SharedState,
+    tenant_id: TenantId,
+    database_id: DatabaseId,
+    plan: PhysicalPlan,
+    vshard_id: VShardId,
+    trace_id: TraceId,
+    txn_id: Option<TxnId>,
+) -> crate::Result<GatherOutcome> {
+    // `Box::pin` breaks an async-fn recursion cycle: `dispatch_to_data_plane_*`
+    // re-enters `resolve_exchange_in_plan`. The plan handed here is the bare,
+    // Exchange-free child of the resolved Gather, so the re-entrant resolve is a
+    // no-op — but the future must be heap-indirected so its size stays finite.
+    let resp = Box::pin(dispatch_to_data_plane_with_txn(
+        state,
+        tenant_id,
+        database_id,
+        vshard_id,
+        plan,
+        trace_id,
+        txn_id,
+    ))
+    .await?;
+
+    // Propagate a Data-Plane error status the same way `gather_all_cores` does:
+    // a `NotFound` from the owning core means "no such slice here" and reads
+    // back as an empty (still validatable) observation; any OTHER error status
+    // must surface as a dispatch error rather than being silently swallowed as
+    // an empty success (e.g. an FTS query-validation rejection, a constraint
+    // error, or a deadline). Without this the owning-core path would drop every
+    // Data-Plane error on a single-vShard read.
+    if resp.status == Status::Error
+        && let Some(ec) = resp.error_code.as_deref()
+        && !matches!(ec, ErrorCode::NotFound)
+    {
+        return Err(crate::Error::Dispatch {
+            detail: format!("{ec:?}"),
+        });
+    }
+
+    let payload_bytes: &[u8] = resp.payload.as_ref();
+    let all_elements = extract_msgpack_elements(payload_bytes);
+    let merged_array = encode_msgpack_array(&all_elements);
+
+    Ok(GatherOutcome {
+        raw: payload_bytes.to_vec(),
+        merged_array,
+        watermark_lsn: resp.watermark_lsn,
+        read_version_lsn: resp.read_version_lsn,
+        shard_watermarks: vec![(vshard_id, resp.watermark_lsn)],
+    })
+}

nodedb/tests/scalar_aggregate_multicore_merge.rs

@@ -0,0 +1,60 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Regression: a no-`GROUP BY` scalar aggregate on a single-vShard-homed
+//! collection must merge to ONE row on a multi-core server.
+//!
+//! A scalar aggregate with no `GROUP BY` plans as
+//! `QueryOp::Aggregate { input: None }` wrapped in `Exchange{Gather}`. In
+//! single-node mode the gather formerly BROADCAST the plan to every core; the
+//! collection's rows live on ONE owning core, so every other (empty) core
+//! seeded its own scalar-aggregate identity row and the coordinator merge is a
+//! passthrough — yielding N rows (N-1 identity rows + 1 real row) instead of
+//! one. The fix routes single-vShard-homed plans to their one owning core.
+//! A single-core harness masks the bug, so these tests drive an 8-core server.
+
+mod common;
+
+use common::pgwire_harness::TestServer;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn scalar_count_star_merges_to_one_row_multicore() {
+    let srv = TestServer::start_multicores(8).await;
+    srv.exec(
+        "CREATE COLLECTION t \
+         COLUMNS (id TEXT PRIMARY KEY, v INTEGER) \
+         WITH (engine='document_strict')",
+    )
+    .await
+    .unwrap();
+    srv.exec("INSERT INTO t (id, v) VALUES ('a',1),('b',2),('c',3)")
+        .await
+        .unwrap();
+
+    let rows = srv.query_rows("SELECT count(*) FROM t").await.unwrap();
+    assert_eq!(
+        rows.len(),
+        1,
+        "scalar count(*) must merge to ONE row across cores, got {rows:?}"
+    );
+    assert_eq!(rows[0][0], "3");
+
+    let rows = srv
+        .query_rows("SELECT count(*) AS c, sum(v) AS s FROM t")
+        .await
+        .unwrap();
+    assert_eq!(
+        rows.len(),
+        1,
+        "aliased scalar aggregate must be one row, got {rows:?}"
+    );
+    assert_eq!(rows[0][0], "3");
+    // `sum` over an INTEGER column renders as a float ("6.0"); the merge
+    // correctness this test guards is the single row + right value, so compare
+    // the value numerically rather than pinning its textual formatting.
+    assert_eq!(
+        rows[0][1].parse::<f64>().expect("numeric sum"),
+        6.0,
+        "aliased sum must carry the merged value, got {:?}",
+        rows[0][1]
+    );
+}