Fan out a coord-range slice to all target shards and merge the rows. Each shard receives the full slice request with the caller-supplied `limit` pushed down so shards can stop scanning early. The coordinator stamps a per-shard `shard_hilbert_range` so each shard only returns cells whose Hilbert prefix falls within its owned range, preventing duplicate rows in single-node harnesses where all vShar
(
&self,
req: ArrayShardSliceReq,
coordinator_limit: u32,
)
| 137 | /// flag across all shards. If any shard fails the entire operation |
| 138 | /// returns `Err` — partial results are not silently dropped. |
| 139 | pub async fn coord_slice( |
| 140 | &self, |
| 141 | req: ArrayShardSliceReq, |
| 142 | coordinator_limit: u32, |
| 143 | ) -> Result<CoordSliceResult> { |
| 144 | let prefix_bits = self.params.prefix_bits; |
| 145 | let per_shard: Vec<(u32, Vec<u8>)> = self |
| 146 | .params |
| 147 | .shard_ids |
| 148 | .iter() |
| 149 | .map(|&shard_id| { |
| 150 | let shard_hilbert_range = if prefix_bits > 0 { |
| 151 | Some(shard_hilbert_range_for_vshard(shard_id, prefix_bits)) |
| 152 | } else { |
| 153 | None |
| 154 | }; |
| 155 | let per_shard_req = ArrayShardSliceReq { |
| 156 | prefix_bits, |
| 157 | slice_hilbert_ranges: self.params.slice_hilbert_ranges.clone(), |
| 158 | shard_hilbert_range, |
| 159 | ..req.clone() |
| 160 | }; |
| 161 | let bytes = |
| 162 | zerompk::to_msgpack_vec(&per_shard_req).map_err(|e| ClusterError::Codec { |
| 163 | detail: format!("ArrayShardSliceReq serialise: {e}"), |
| 164 | })?; |
| 165 | Ok((shard_id, bytes)) |
| 166 | }) |
| 167 | .collect::<Result<Vec<_>>>()?; |
| 168 | |
| 169 | let fo_params = FanOutPartitionedParams { |
| 170 | source_node: self.params.source_node, |
| 171 | timeout_ms: self.params.timeout_ms, |
| 172 | }; |
| 173 | let raw = fan_out_partitioned( |
| 174 | &fo_params, |
| 175 | super::super::opcodes::ARRAY_SHARD_SLICE_REQ, |
| 176 | &per_shard, |
| 177 | &self.dispatch, |
| 178 | &self.circuit_breaker, |
| 179 | ) |
| 180 | .await?; |
| 181 | let resps = decode_resps::<ArrayShardSliceResp>(&raw)?; |
| 182 | let truncated_before_horizon = |
| 183 | super::super::merge::any_truncated_before_horizon_slice(&resps); |
| 184 | let rows = merge_slice_rows(&resps, coordinator_limit); |
| 185 | Ok(CoordSliceResult { |
| 186 | rows, |
| 187 | truncated_before_horizon, |
| 188 | }) |
| 189 | } |
| 190 | |
| 191 | /// Fan out an aggregate request and reduce partial aggregates from all shards. |
| 192 | /// |
no test coverage detected