| 311 | } |
| 312 | |
| 313 | async fn execute_slice(&self, args: SliceArgs<'_>) -> crate::Result<Vec<u8>> { |
| 314 | let SliceArgs { |
| 315 | array_id, |
| 316 | slice_msgpack, |
| 317 | attr_projection, |
| 318 | limit, |
| 319 | slice_hilbert_ranges, |
| 320 | prefix_bits, |
| 321 | system_as_of, |
| 322 | valid_at_ms, |
| 323 | } = args; |
| 324 | let coordinator = ArrayCoordinator::for_slice( |
| 325 | self.source_node, |
| 326 | ARRAY_RPC_TIMEOUT_MS, |
| 327 | slice_hilbert_ranges, |
| 328 | prefix_bits, |
| 329 | self.total_shards, |
| 330 | Arc::clone(&self.dispatch), |
| 331 | Arc::clone(&self.circuit_breaker), |
| 332 | ) |
| 333 | .map_err(cluster_err)?; |
| 334 | |
| 335 | let array_id_msgpack = zerompk::to_msgpack_vec(array_id).map_err(encode_err)?; |
| 336 | let req = ArrayShardSliceReq { |
| 337 | array_id_msgpack, |
| 338 | slice_msgpack: slice_msgpack.to_vec(), |
| 339 | attr_projection: attr_projection.to_vec(), |
| 340 | limit, |
| 341 | cell_filter_msgpack: vec![], |
| 342 | // prefix_bits, slice_hilbert_ranges, and shard_hilbert_range are |
| 343 | // stamped per-shard by coord_slice from ArrayCoordParams, which was |
| 344 | // populated by for_slice above. Initialise to defaults here; |
| 345 | // coord_slice overwrites them before serialising the request envelope. |
| 346 | prefix_bits: 0, |
| 347 | slice_hilbert_ranges: vec![], |
| 348 | shard_hilbert_range: None, |
| 349 | system_as_of, |
| 350 | valid_at_ms, |
| 351 | }; |
| 352 | let result = coordinator |
| 353 | .coord_slice(req, limit) |
| 354 | .await |
| 355 | .map_err(cluster_err)?; |
| 356 | |
| 357 | // Encode rows into the structured `ArraySliceResponse` — identical to |
| 358 | // the shape that `dispatch_array_slice` emits for single-node requests. |
| 359 | // This makes local and cluster payloads byte-identical so the pgwire |
| 360 | // handler can use a single decoder regardless of topology. The |
| 361 | // `truncated_before_horizon` flag is OR-reduced across shards by the |
| 362 | // coordinator so the upstream NOTICE is emitted whenever any shard |
| 363 | // dropped data below its system-time horizon. |
| 364 | let mut rows_msgpack = |
| 365 | Vec::with_capacity(5 + result.rows.iter().map(|r| r.len()).sum::<usize>()); |
| 366 | let n = result.rows.len(); |
| 367 | if n <= 15 { |
| 368 | rows_msgpack.push(0x90 | (n as u8)); |
| 369 | } else if n <= u16::MAX as usize { |
| 370 | rows_msgpack.push(0xdc); |