(
&mut self,
task: &ExecutionTask,
p: SliceParams<'_>,
)
| 44 | |
| 45 | impl CoreLoop { |
| 46 | pub(in crate::data::executor) fn dispatch_array_slice( |
| 47 | &mut self, |
| 48 | task: &ExecutionTask, |
| 49 | p: SliceParams<'_>, |
| 50 | ) -> Response { |
| 51 | let SliceParams { |
| 52 | array_id, |
| 53 | slice_msgpack, |
| 54 | attr_projection, |
| 55 | limit, |
| 56 | cell_filter, |
| 57 | hilbert_range, |
| 58 | system_as_of, |
| 59 | valid_at_ms, |
| 60 | } = p; |
| 61 | |
| 62 | if let Err(resp) = self.ensure_array_open(task, array_id) { |
| 63 | return resp; |
| 64 | } |
| 65 | let slice: Slice = match zerompk::from_msgpack(slice_msgpack) { |
| 66 | Ok(s) => s, |
| 67 | Err(e) => { |
| 68 | return self.response_error( |
| 69 | task, |
| 70 | ErrorCode::Internal { |
| 71 | detail: format!("array slice decode: {e}"), |
| 72 | }, |
| 73 | ); |
| 74 | } |
| 75 | }; |
| 76 | |
| 77 | let schema = match self.array_engine.store(array_id) { |
| 78 | Ok(store) => store.schema().clone(), |
| 79 | Err(e) => { |
| 80 | return self.response_error( |
| 81 | task, |
| 82 | ErrorCode::Unsupported { |
| 83 | detail: format!("array '{}' not open: {e}", array_id.name), |
| 84 | }, |
| 85 | ); |
| 86 | } |
| 87 | }; |
| 88 | |
| 89 | let proj = if attr_projection.is_empty() { |
| 90 | None |
| 91 | } else { |
| 92 | Some(Projection::new( |
| 93 | attr_projection.iter().map(|&i| i as usize).collect(), |
| 94 | )) |
| 95 | }; |
| 96 | let cap = limit as usize; |
| 97 | |
| 98 | // Run through the Ceiling resolver. When no temporal filter is specified |
| 99 | // the cutoff is `i64::MAX` (live read); Ceiling still deduplicates |
| 100 | // multiple system-time versions of the same coord. The response shape |
| 101 | // is `ArraySliceResponse` (rows + truncated_before_horizon flag) for |
| 102 | // both local single-node and cluster shard responses. |
| 103 | let cutoff = system_as_of.unwrap_or(i64::MAX); |
no test coverage detected