Scan columnar rows → standard msgpack.
(&self, tid: u64, collection: &str, limit: usize)
| 75 | |
| 76 | /// Scan columnar rows → standard msgpack. |
| 77 | fn scan_columnar(&self, tid: u64, collection: &str, limit: usize) -> Vec<(String, Vec<u8>)> { |
| 78 | let engine_key = (crate::types::TenantId::new(tid), collection.to_string()); |
| 79 | if let Some(mt) = self.columnar_memtables.get(&engine_key) { |
| 80 | let schema = mt.schema(); |
| 81 | let row_count = (mt.row_count() as usize).min(limit); |
| 82 | let col_meta: Vec<_> = schema |
| 83 | .columns |
| 84 | .iter() |
| 85 | .enumerate() |
| 86 | .map(|(i, (name, ty))| (i, name.clone(), *ty)) |
| 87 | .collect(); |
| 88 | |
| 89 | let mut results = Vec::with_capacity(row_count); |
| 90 | for idx in 0..row_count { |
| 91 | // Build msgpack map directly — no serde_json intermediary. |
| 92 | let mut mp = Vec::with_capacity(col_meta.len() * 32); |
| 93 | msgpack_scan::write_map_header(&mut mp, col_meta.len()); |
| 94 | let mut id = String::new(); |
| 95 | for (col_idx, col_name, col_type) in &col_meta { |
| 96 | msgpack_scan::write_str(&mut mp, col_name); |
| 97 | let col_data = mt.column(*col_idx); |
| 98 | // Check for "id" column to extract the id string. |
| 99 | if col_name == "id" |
| 100 | && let crate::engine::timeseries::columnar_memtable::ColumnData::Symbol(ids) = |
| 101 | col_data |
| 102 | { |
| 103 | let sym_id = ids[idx]; |
| 104 | if let Some(s) = mt.symbol_dict(*col_idx).and_then(|dict| dict.get(sym_id)) |
| 105 | { |
| 106 | id = s.to_string(); |
| 107 | } |
| 108 | } |
| 109 | super::handlers::columnar_read::emit_column_value( |
| 110 | &mut mp, mt, *col_idx, col_type, col_data, idx, |
| 111 | ); |
| 112 | } |
| 113 | results.push((id, mp)); |
| 114 | } |
| 115 | return results; |
| 116 | } |
| 117 | |
| 118 | let Some(engine) = self.columnar_engines.get(&engine_key) else { |
| 119 | return Vec::new(); |
| 120 | }; |
| 121 | |
| 122 | let schema = engine.schema(); |
| 123 | let mut results = Vec::new(); |
| 124 | |
| 125 | // 1. Read from flushed segments (older rows drained from prior memtable flushes). |
| 126 | if let Some(segments) = self.columnar_flushed_segments.get(&engine_key) { |
| 127 | for (seg_idx, seg_bytes) in segments.iter().enumerate() { |
| 128 | if results.len() >= limit { |
| 129 | break; |
| 130 | } |
| 131 | let seg_id = format!("{}", seg_idx as u64 + 1); |
| 132 | let reader = if let Some(ref reg) = self.quarantine_registry { |
| 133 | match crate::storage::quarantine::engines::open_segment_with_quarantine( |
| 134 | reg, seg_bytes, collection, &seg_id, |
no test coverage detected