External sort: split filtered rows into sorted runs, spill each run to a temp file, then k-way merge to produce the final sorted output.
(
&self,
rows: Vec<(String, Vec<u8>)>,
sort_keys: &[(String, bool)],
output_limit: usize,
)
| 16 | /// External sort: split filtered rows into sorted runs, spill each run |
| 17 | /// to a temp file, then k-way merge to produce the final sorted output. |
| 18 | pub(super) fn external_sort( |
| 19 | &self, |
| 20 | rows: Vec<(String, Vec<u8>)>, |
| 21 | sort_keys: &[(String, bool)], |
| 22 | output_limit: usize, |
| 23 | ) -> crate::Result<Vec<(String, Vec<u8>)>> { |
| 24 | // Spill directory for temporary sort run files. Temp files are |
| 25 | // auto-deleted on Drop; the directory persists but is cleaned up |
| 26 | // on the next external_sort call or server restart. |
| 27 | let spill_dir = self |
| 28 | .data_dir |
| 29 | .join(format!("sort-spill/core-{}", self.core_id)); |
| 30 | std::fs::create_dir_all(&spill_dir).map_err(|e| crate::Error::Storage { |
| 31 | engine: "sort".into(), |
| 32 | detail: format!("failed to create sort spill dir: {e}"), |
| 33 | })?; |
| 34 | |
| 35 | let total_rows = rows.len(); |
| 36 | |
| 37 | let mut run_files = Vec::new(); |
| 38 | for chunk in rows.chunks(self.query_tuning.sort_run_size) { |
| 39 | let mut run: Vec<(String, Vec<u8>)> = chunk.to_vec(); |
| 40 | sort_rows(&mut run, sort_keys); |
| 41 | |
| 42 | let file = tempfile::tempfile_in(&spill_dir).map_err(|e| crate::Error::Storage { |
| 43 | engine: "sort".into(), |
| 44 | detail: format!("failed to create sort temp file: {e}"), |
| 45 | })?; |
| 46 | let mut writer = BufWriter::new(file); |
| 47 | |
| 48 | let count = run.len() as u32; |
| 49 | writer |
| 50 | .write_all(&count.to_le_bytes()) |
| 51 | .map_err(|e| crate::Error::Storage { |
| 52 | engine: "sort".into(), |
| 53 | detail: format!("sort spill write: {e}"), |
| 54 | })?; |
| 55 | for (id, val) in &run { |
| 56 | let id_bytes = id.as_bytes(); |
| 57 | writer |
| 58 | .write_all(&(id_bytes.len() as u32).to_le_bytes()) |
| 59 | .map_err(|e| crate::Error::Storage { |
| 60 | engine: "sort".into(), |
| 61 | detail: format!("sort spill write: {e}"), |
| 62 | })?; |
| 63 | writer |
| 64 | .write_all(id_bytes) |
| 65 | .map_err(|e| crate::Error::Storage { |
| 66 | engine: "sort".into(), |
| 67 | detail: format!("sort spill write: {e}"), |
| 68 | })?; |
| 69 | writer |
| 70 | .write_all(&(val.len() as u32).to_le_bytes()) |
| 71 | .map_err(|e| crate::Error::Storage { |
| 72 | engine: "sort".into(), |
| 73 | detail: format!("sort spill write: {e}"), |
| 74 | })?; |
| 75 | writer.write_all(val).map_err(|e| crate::Error::Storage { |