(
local_vshard_id: u32,
payload: &[u8],
executor: &Arc<dyn ArrayLocalExecutor>,
)
| 59 | } |
| 60 | |
| 61 | async fn handle_slice( |
| 62 | local_vshard_id: u32, |
| 63 | payload: &[u8], |
| 64 | executor: &Arc<dyn ArrayLocalExecutor>, |
| 65 | ) -> Result<Vec<u8>> { |
| 66 | let req: ArrayShardSliceReq = |
| 67 | zerompk::from_msgpack(payload).map_err(|e| ClusterError::Codec { |
| 68 | detail: format!("ArrayShardSliceReq decode: {e}"), |
| 69 | })?; |
| 70 | |
| 71 | validate_slice_routing(&req, local_vshard_id)?; |
| 72 | |
| 73 | let rows = executor.exec_slice(&req).await?; |
| 74 | |
| 75 | let truncated = req.limit > 0 && rows.len() >= req.limit as usize; |
| 76 | let resp = ArrayShardSliceResp { |
| 77 | shard_id: local_vshard_id, |
| 78 | rows_msgpack: rows, |
| 79 | truncated, |
| 80 | truncated_before_horizon: false, |
| 81 | }; |
| 82 | serialise(resp) |
| 83 | } |
| 84 | |
| 85 | async fn handle_agg( |
| 86 | local_vshard_id: u32, |
no test coverage detected