Skip to main content

data_beans/sparse_io/
traits.rs

1#![allow(dead_code, unused_imports)]
2
3#[cfg(feature = "tensor")]
4pub use legume_numeric::candle_core::Tensor;
5pub use nalgebra::DMatrix;
6pub use nalgebra_sparse::{csc::CscMatrix, csr::CsrMatrix};
7#[cfg(feature = "ndarray")]
8pub use ndarray::prelude::*;
9
10pub const MAX_ROW_NAME_IDX: usize = 3;
11pub const MAX_COLUMN_NAME_IDX: usize = 10;
12pub const COLUMN_SEP: &str = "@";
13pub const ROW_SEP: &str = "_";
14
15use super::helpers::*;
16use super::meta::Metadata;
17
18use crate::sparse_data_visitors::styled_progress_bar;
19use clap::ValueEnum;
20use indicatif::ParallelProgressIterator;
21use legume_numeric::matrix::mtx_io::*;
22use legume_numeric::matrix::traits::*;
23use log::info;
24use rayon::prelude::*;
25use rustc_hash::FxHashMap as HashMap;
26use std::ops::Range;
27use std::sync::{Arc, Mutex};
28
29#[cfg(test)]
30mod tests;
31
32#[derive(ValueEnum, Clone, Debug, PartialEq)]
33#[clap(rename_all = "lowercase")]
34pub enum SparseIoBackend {
35    Zarr,
36    HDF5,
37}
38
39/// Identifies one of the six 1-D datasets inside a sparse backend.
40/// Used by the streaming write API so we don't have to add six separate
41/// abstract methods per dtype × (csc|csr) × (data|indices|indptr).
42#[derive(Clone, Copy, Debug, PartialEq, Eq)]
43pub enum CsKey {
44    CscData,
45    CscIndices,
46    CscIndptr,
47    CsrData,
48    CsrIndices,
49    CsrIndptr,
50}
51
52/// Entries per slab handed to the backend while a sorted triplet vector is
53/// streamed out as CSC or CSR. Bounds the staging buffers; the triplets
54/// themselves are the only full-size structure alive at that point.
55const SLAB_NNZ: usize = 1 << 20;
56
57/// End of the slab that starts at triplet `start` of a vector sorted on the
58/// major axis `major` (column for CSC, row for CSR): at least `slab_nnz`
59/// entries, or all that remain, and never splitting a major index. Returns
60/// `(end, band_end)` -- the exclusive triplet index and the exclusive major
61/// bound -- so consecutive slabs tile `0..n_major` with no gap, empty
62/// columns or rows included; the last slab runs to `n_major`.
63fn slab_end(
64    triplets: &[(u64, u64, f32)],
65    start: usize,
66    slab_nnz: usize,
67    n_major: usize,
68    major: impl Fn(&(u64, u64, f32)) -> u64,
69) -> (usize, u64) {
70    debug_assert!(slab_nnz > 0);
71    let nnz = triplets.len();
72    let mut end = (start + slab_nnz).min(nnz);
73    while end < nnz && major(&triplets[end]) == major(&triplets[end - 1]) {
74        end += 1;
75    }
76    let band_end = if end == nnz {
77        n_major as u64
78    } else {
79        major(&triplets[end])
80    };
81    (end, band_end)
82}
83
84pub trait SparseIo: Sync + Send {
85    type IndexIter: IntoIterator<Item = usize> + FromIterator<usize>;
86
87    ////////////////////////////
88    // default implementation //
89    ////////////////////////////
90
91    #[cfg(feature = "ndarray")]
92    /// Read columns within the range and return dense `ndarray::Array2`
93    /// * `columns` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
94    ///
95    fn read_columns_ndarray(&self, columns: Self::IndexIter) -> anyhow::Result<Array2<f32>> {
96        let (nrow, ncol, triplets) = self.read_triplets_by_columns(columns)?;
97        Array2::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
98    }
99
100    #[cfg(feature = "tensor")]
101    /// Read columns within the range and return dense `candle_core::Tensor`
102    /// * `columns` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
103    ///
104    fn read_columns_tensor(&self, columns: Self::IndexIter) -> anyhow::Result<Tensor> {
105        let (nrow, ncol, triplets) = self.read_triplets_by_columns(columns)?;
106        Tensor::from_nonzero_triplets(nrow, ncol, &triplets)
107    }
108
109    /// Read columns within the range and return dense `nalgebrea::DMatrix`
110    /// * `columns` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
111    ///
112    fn read_columns_dmatrix(&self, columns: Self::IndexIter) -> anyhow::Result<DMatrix<f32>> {
113        let (nrow, ncol, triplets) = self.read_triplets_by_columns(columns)?;
114        DMatrix::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
115    }
116
117    /// Read columns within the range and return sparse `CsrMatrix`
118    /// * `columns` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
119    ///
120    fn read_columns_csr(&self, columns: Self::IndexIter) -> anyhow::Result<CsrMatrix<f32>> {
121        let (nrow, ncol, triplets) = self.read_triplets_by_columns(columns)?;
122        CsrMatrix::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
123    }
124
125    /// Read columns within the range and return sparse `CsrMatrix`
126    /// * `columns` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
127    ///
128    fn read_columns_csc(&self, columns: Self::IndexIter) -> anyhow::Result<CscMatrix<f32>> {
129        let (nrow, ncol, triplets) = self.read_triplets_by_columns(columns)?;
130        CscMatrix::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
131    }
132
133    /// Zero-copy view of preloaded column-major CSC arrays as
134    /// `(indptr, indices, data)`. Returns `None` when the backend has
135    /// not preloaded columns or doesn't support direct array access.
136    /// Callers (e.g. `SparseIoVec::read_columns_csc`) use this to skip
137    /// the triplet roundtrip when columns are already in memory.
138    fn csc_column_arrays(&self) -> Option<(&[u64], &[u64], &[f32])> {
139        None
140    }
141
142    #[cfg(feature = "ndarray")]
143    /// Read rows within the range and return dense `ndarray::Array2`
144    /// * `rows` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
145    ///
146    fn read_rows_ndarray(&self, rows: Self::IndexIter) -> anyhow::Result<Array2<f32>> {
147        let (nrow, ncol, triplets) = self.read_triplets_by_rows(rows)?;
148        Array2::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
149    }
150
151    #[cfg(feature = "tensor")]
152    /// Read rows within the range and return dense `candle_core::Tensor`
153    /// * `rows` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
154    ///
155    fn read_rows_tensor(&self, rows: Self::IndexIter) -> anyhow::Result<Tensor> {
156        let (nrow, ncol, triplets) = self.read_triplets_by_rows(rows)?;
157        Tensor::from_nonzero_triplets(nrow, ncol, &triplets)
158    }
159
160    /// Read rows within the range and return dense `nalgebra::DMatrix`
161    /// * `rows` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
162    ///
163    fn read_rows_dmatrix(&self, rows: Self::IndexIter) -> anyhow::Result<DMatrix<f32>> {
164        let (nrow, ncol, triplets) = self.read_triplets_by_rows(rows)?;
165        DMatrix::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
166    }
167
168    /// Read rows within the range and return sparse `CsrMatrix`
169    /// * `rows` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
170    ///
171    fn read_rows_csr(&self, rows: Self::IndexIter) -> anyhow::Result<CsrMatrix<f32>> {
172        let (nrow, ncol, triplets) = self.read_triplets_by_rows(rows)?;
173        CsrMatrix::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
174    }
175
176    /// Read rows within the range and return sparse `CscMatrix`
177    /// * `rows` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
178    ///
179    fn read_rows_csc(&self, rows: Self::IndexIter) -> anyhow::Result<CscMatrix<f32>> {
180        let (nrow, ncol, triplets) = self.read_triplets_by_rows(rows)?;
181        CscMatrix::<f32>::from_nonzero_triplets(nrow, ncol, &triplets)
182    }
183
184    /////////////////////////////
185    // `mtx` related functions //
186    /////////////////////////////
187
188    /// Read an mtx file once and populate the backend: the column (CSC) index
189    /// always, the row (CSR) index as well when `index_by_row`. Both are
190    /// streamed out of the same triplet vector, so the file is inflated once
191    /// and the triplets are the only full-size structure alive.
192    /// * `mtx_file`: mtx file to be read into the backend
193    fn import_mtx_file(&mut self, mtx_file: &str, index_by_row: bool) -> anyhow::Result<()> {
194        let (mut mtx_triplets, mtx_shape) = read_mtx_triplets(mtx_file)?;
195        info!("read mtx file: {}", mtx_file);
196        if mtx_triplets.is_empty() {
197            return Err(anyhow::anyhow!("No data in mtx file"));
198        }
199        self.record_mtx_shape(Some(mtx_shape))?;
200        info!("recording the column index");
201        self.record_triplets_by_col(&mut mtx_triplets)?;
202        if index_by_row {
203            info!("recording the row index");
204            self.record_triplets_by_row(&mut mtx_triplets)?;
205        }
206        Ok(())
207    }
208
209    /////////////////////////////////
210    // `dmatrix` related functions //
211    /////////////////////////////////
212
213    /// Add dmatrix to zarr backend by row (CSR format)
214    /// * `array` - 2D array to be added to the backend
215    fn import_dmatrix_by_row(&mut self, matrix: &DMatrix<f32>) -> anyhow::Result<()> {
216        let (nrow, ncol) = matrix.shape();
217        let mut mtx_triplets = dmatrix_to_triplets(matrix);
218        let mtx_shape = (nrow, ncol, mtx_triplets.len());
219        self.record_mtx_shape(Some(mtx_shape))?;
220        self.record_triplets_by_row(&mut mtx_triplets)
221    }
222
223    /// Add dmatrix to zarr backend by column (CSC format)
224    /// * `array` - 2D array to be added to the backend
225    fn import_dmatrix_by_col(&mut self, matrix: &DMatrix<f32>) -> anyhow::Result<()> {
226        let (nrow, ncol) = matrix.shape();
227        let mut mtx_triplets = dmatrix_to_triplets(matrix);
228        let mtx_shape = (nrow, ncol, mtx_triplets.len());
229        self.record_mtx_shape(Some(mtx_shape))?;
230        self.record_triplets_by_col(&mut mtx_triplets)
231    }
232
233    /////////////////////////////////
234    // `ndarray` related functions //
235    /////////////////////////////////
236
237    #[cfg(feature = "ndarray")]
238    /// Add ndarray to zarr backend by row (CSR format)
239    /// * `array` - 2D array to be added to the backend
240    fn import_ndarray_by_row(&mut self, array: &Array2<f32>) -> anyhow::Result<()> {
241        let nrow = array.shape()[0];
242        let ncol = array.shape()[1];
243
244        // dbg!("importing ndarray by row...");
245        let mut mtx_triplets = ndarray_to_triplets(array);
246
247        let nnz = mtx_triplets.len();
248        let mtx_shape = (nrow, ncol, nnz);
249        self.record_mtx_shape(Some(mtx_shape))?;
250
251        // dbg!(format!("populated: {} elements", mtx_triplets.len()));
252
253        self.record_triplets_by_row(&mut mtx_triplets)
254    }
255
256    #[cfg(feature = "ndarray")]
257    /// Add ndarray to zarr backend by column (CSC format)
258    /// * `array` - 2D array to be added to the backend
259    fn import_ndarray_by_col(&mut self, array: &Array2<f32>) -> anyhow::Result<()> {
260        let nrow = array.shape()[0];
261        let ncol = array.shape()[1];
262
263        // dbg!("importing ndarray by column...");
264        let mut mtx_triplets = ndarray_to_triplets(array);
265
266        let nnz = mtx_triplets.len();
267        let mtx_shape = (nrow, ncol, nnz);
268        self.record_mtx_shape(Some(mtx_shape))?;
269
270        // dbg!(format!("populated: {} elements", mtx_triplets.len()));
271
272        self.record_triplets_by_col(&mut mtx_triplets)
273    }
274
275    //////////////////////
276    // backend-specific //
277    //////////////////////
278
279    /// Read rows within the range and return a vector of triplets (row, column, value)
280    /// * `rows` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
281    ///
282    #[allow(clippy::type_complexity)]
283    fn read_triplets_by_rows(
284        &self,
285        rows: Self::IndexIter,
286    ) -> anyhow::Result<(usize, usize, Vec<(u64, u64, f32)>)>;
287
288    /// Read columns within the range and return a vector of triplets (row, col, value)
289    /// * `columns` : range e.g., 0..3 -> [0, 1, 2] or vec![0, 1, 2]
290    ///
291    #[allow(clippy::type_complexity)]
292    fn read_triplets_by_columns(
293        &self,
294        columns: Self::IndexIter,
295    ) -> anyhow::Result<(usize, usize, Vec<(u64, u64, f32)>)>;
296
297    /// Read columns within the range and return a vector of triplets (row, col, value)
298    /// * `col` : usize
299    ///
300    #[allow(clippy::type_complexity)]
301    fn read_triplets_by_single_column(
302        &self,
303        col: usize,
304    ) -> anyhow::Result<(usize, usize, Vec<(u64, u64, f32)>)>;
305
306    /// Export the data to a mtx file. This will take time.
307    /// * `mtx_file`: mtx file to be written
308    fn to_mtx_file(&self, mtx_file: &str) -> anyhow::Result<()>;
309
310    /// Number of rows in the underlying data matrix
311    fn num_rows(&self) -> Option<usize>;
312
313    /// Number of columns in the underlying data matrix
314    fn num_columns(&self) -> Option<usize>;
315
316    /// Number of non-zero elements
317    fn num_non_zeros(&self) -> Option<usize>;
318
319    /// Re-open handles on the CURRENT backend path after its contents were
320    /// replaced from outside (a finished temp file renamed into place). The
321    /// zarr store is path-addressed so this is a cache refresh; hdf5 holds an
322    /// open file handle that would otherwise point at the deleted inode.
323    fn reopen_backend(&mut self) -> anyhow::Result<()>;
324
325    /// Maintained by [`append_csc_slab`](Self::append_csc_slab) — never call
326    /// this yourself: padding the cursor masks exactly the under-append the
327    /// finalize audit exists to catch.
328    #[doc(hidden)]
329    /// Advance the streaming-write cursor by `n` entries. Called by
330    /// [`append_csc_slab`](Self::append_csc_slab); backends keep the count so
331    /// [`finalize_streaming_csc`](Self::finalize_streaming_csc) can audit the
332    /// declared nnz against what was actually appended — the one violation the
333    /// written indptr cannot reveal, because an over-declared tail leaves it
334    /// perfectly monotone with the phantom hiding between the last written
335    /// pointer and the sentinel.
336    fn note_streamed_nnz(&mut self, n: u64);
337
338    /// Entries appended so far in this streaming build.
339    #[doc(hidden)]
340    fn streamed_nnz(&self) -> u64;
341
342    /// Zero the cursor. Called by [`begin_streaming_csc`](Self::begin_streaming_csc).
343    #[doc(hidden)]
344    fn reset_streamed_nnz(&mut self);
345
346    /// The resident by-column indptr, loaded at `open()`. Empty when the
347    /// backend carries no `/by_column/indptr` array — `read_column_indptr`
348    /// silently does nothing on that failure, and the accessors below must
349    /// report that absence rather than read zeros out of it.
350    fn column_indptr(&self) -> &[u64];
351
352    /// Exact nnz of one column, from the resident indptr — no I/O.
353    ///
354    /// `None` for an out-of-range column or when the indptr is absent. This is
355    /// what lets a streaming writer declare a column subset's total nnz up
356    /// front without a counting pass over the data.
357    fn column_nnz(&self, col: usize) -> Option<u64> {
358        let indptr = self.column_indptr();
359        let hi = *indptr.get(col + 1)?;
360        let lo = *indptr.get(col)?;
361        hi.checked_sub(lo)
362    }
363
364    /// Set row names for the matrix
365    /// * `row_name_file`: a file each line contains row name words
366    fn register_row_names_file(&mut self, row_name_file: &str);
367
368    /// Set column names for the matrix
369    /// * `column_name_file`: a file each line contains column name words
370    fn register_column_names_file(&mut self, column_name_file: &str);
371
372    /// Set row names for the matrix
373    /// * `rows`: a vector of row names
374    fn register_row_names_vec(&mut self, rows: &[Box<str>]);
375
376    /// Set column names for the matrix
377    /// * `columns`: a vector of column names
378    fn register_column_names_vec(&mut self, columns: &[Box<str>]);
379
380    /// Add arbitrary names (a vector of strings)
381    /// * `group_name`: group name
382    /// * `name_file`: a file each line contains name words
383    /// * `name_columns`: range of columns to be used for name
384    /// * `name_sep`: separator for name columns
385    fn register_names_file(
386        &mut self,
387        key: &str,
388        name_file: &str,
389        name_columns: Range<usize>,
390        name_sep: &str,
391    ) -> anyhow::Result<()>;
392
393    /// Add arbitrary names (a vector of strings)
394    /// * `group_name`: group name
395    /// * `names`: a file each line contains name words
396    fn register_names_vec(&mut self, key: &str, names: &[Box<str>]) -> anyhow::Result<()>;
397
398    fn row_names(&self) -> anyhow::Result<Vec<Box<str>>>;
399
400    fn column_names(&self) -> anyhow::Result<Vec<Box<str>>>;
401
402    /// Get back the registered names
403    /// * `key`: key for the registered names
404    fn retrieve_registered_names(&self, key: &str) -> anyhow::Result<Vec<Box<str>>>;
405
406    //////////////
407    // metadata //
408    //////////////
409
410    /// The strings stored with the data to say what it is (see
411    /// [`meta`](crate::sparse_io::meta) for the keys programs share); empty
412    /// when none were set.
413    fn metadata(&self) -> Metadata;
414
415    /// Store `meta` as the data's metadata, replacing what was there. It
416    /// needs a writable backend: one being created (set it before zipping),
417    /// or an unzipped zarr store. A zipped zarr store, or an HDF5 file opened
418    /// to read, refuses it.
419    fn set_metadata(&mut self, meta: &Metadata) -> anyhow::Result<()>;
420
421    /// One metadata value, e.g. `meta(meta::SAMPLE)`.
422    fn meta(&self, key: &str) -> Option<String> {
423        self.metadata().remove(key)
424    }
425
426    /// Set one metadata value, keeping the others.
427    fn set_meta(&mut self, key: &str, value: &str) -> anyhow::Result<()> {
428        let mut meta = self.metadata();
429        meta.insert(key.to_string(), value.to_string());
430        self.set_metadata(&meta)
431    }
432
433    /////////////////////////////
434    // major structural change //
435    /////////////////////////////
436
437    /// Select the columns of the data and create a new backend file
438    /// * `columns`: columns to be subsetted
439    /// * `rows`: if something, subset the rows
440    fn subset_columns_rows(
441        &mut self,
442        columns: Option<&Vec<usize>>,
443        rows: Option<&Vec<usize>>,
444    ) -> anyhow::Result<()> {
445        let ncol_data = self
446            .num_columns()
447            .ok_or_else(|| anyhow::anyhow!("missing shape information"))?;
448        let nrow_data = self
449            .num_rows()
450            .ok_or_else(|| anyhow::anyhow!("missing shape information"))?;
451
452        // An empty selection is refused, not honoured: this method DESTROYS the
453        // original backend, and writing a zero-column husk over real data is
454        // almost certainly a caller mistake rather than an intention.
455        // Empty and duplicated selections are refused, not honoured: this
456        // method DESTROYS the original, and a duplicate collapses in the
457        // old→new map while still counting toward the new shape — slabs then
458        // land at wrong offsets and the finalize audit rejects the build with a
459        // message about nnz tiling that names nothing the caller did.
460        let distinct = |sel: &[usize], what: &str| -> anyhow::Result<()> {
461            anyhow::ensure!(!sel.is_empty(), "subset: empty {what} selection");
462            let mut seen = sel.to_vec();
463            seen.sort_unstable();
464            seen.dedup();
465            anyhow::ensure!(
466                seen.len() == sel.len(),
467                "subset: the {what} selection repeats an index ({} of {} are distinct)",
468                seen.len(),
469                sel.len()
470            );
471            Ok(())
472        };
473        if let Some(cols) = columns {
474            distinct(cols, "column")?;
475        }
476        if let Some(rs) = rows {
477            distinct(rs, "row")?;
478        }
479
480        //////////////////////////////////////////////////////
481        // 0. Create a mapping from old to new columns/rows //
482        //////////////////////////////////////////////////////
483
484        let (old2new_cols, new_col_names) =
485            take_subset_indices_names_if_needed(columns, Some(ncol_data), self.column_names()?);
486        let (old2new_rows, new_row_names) =
487            take_subset_indices_names_if_needed(rows, Some(nrow_data), self.row_names()?);
488        let (new_ncol, new_nrow) = (new_col_names.len(), new_row_names.len());
489        anyhow::ensure!(new_ncol > 0, "subset: no column survived the selection");
490        anyhow::ensure!(new_nrow > 0, "subset: no row survived the selection");
491
492        // Old columns in NEW order — the selection's order is the output order.
493        let mut cols_new_order: Vec<(u64, u64)> =
494            old2new_cols.iter().map(|(&o, &n)| (n, o)).collect();
495        cols_new_order.sort_unstable();
496
497        // Dense old-row → new-row map, and whether it preserves order. A
498        // monotone map keeps within-column rows ascending after renumbering, so
499        // no per-column sort is needed; a reordering map costs one small sort
500        // per column.
501        let mut row_map: Vec<Option<u64>> = vec![None; nrow_data];
502        for (&old, &new) in &old2new_rows {
503            row_map[old as usize] = Some(new);
504        }
505        let monotone_rows = row_map.iter().flatten().is_sorted_by(|a, b| a < b);
506
507        ///////////////////////////////////////////////////////
508        // 1. Exact per-new-column nnz, without materialising //
509        ///////////////////////////////////////////////////////
510
511        // No row filter: straight off the resident indptr, zero I/O. With one:
512        // a counting pass — reads every selected column once and keeps counts,
513        // never entries.
514        let full_rows = rows.is_none();
515        let per_col_nnz: Vec<u64> = if full_rows {
516            cols_new_order
517                .iter()
518                .map(|&(_, old)| {
519                    self.column_nnz(old as usize)
520                        .ok_or_else(|| anyhow::anyhow!("subset: no indptr for column {old}"))
521                })
522                .collect::<anyhow::Result<_>>()?
523        } else {
524            // Block reads, never one column at a time: a single-column read
525            // pays the cached-subset machinery per call, which measured two
526            // orders of magnitude slower at imaging scale. Counts only, never
527            // entries.
528            let mut counts = vec![0u64; cols_new_order.len()];
529            let coarse = legume_numeric::matrix::utils::generate_minibatch_intervals(
530                cols_new_order.len(),
531                0,
532                Some(8192),
533            );
534            for (lb, ub) in coarse {
535                let old_cols: Vec<usize> = cols_new_order[lb..ub]
536                    .iter()
537                    .map(|&(_, o)| o as usize)
538                    .collect();
539                let (_, _, triplets) =
540                    self.read_triplets_by_columns(old_cols.into_iter().collect())?;
541                for (i, c_local, _) in triplets {
542                    if row_map[i as usize].is_some() {
543                        counts[lb + c_local as usize] += 1;
544                    }
545                }
546            }
547            counts
548        };
549        let new_nnz: u64 = per_col_nnz.iter().sum();
550
551        ///////////////////////////////////////////////////////////
552        // 2. Stream the survivors into a TEMPORARY sibling file //
553        ///////////////////////////////////////////////////////////
554
555        // Written beside the original (same filesystem, so the final rename is
556        // atomic) and swapped in only when complete. The old implementation
557        // deleted the original FIRST and rewrote it from a RAM buffer, so a
558        // crash mid-write lost the data outright — and that buffer held every
559        // surviving triplet, which is the memory wall this replaces.
560        // CONTRACT of the swap: the original is untouched until the temporary
561        // sibling is complete and finalized; a failure mid-stream leaves the
562        // original intact plus a `{path}.subset_tmp` leftover (cleaned on the
563        // next attempt); the unrecoverable window is only remove→rename below.
564        // A zip-archived backend is refused up front — streaming to a sibling
565        // DIRECTORY and renaming it over the `.zip` name would silently change
566        // the on-disk format under the old extension, and the old code's
567        // "store is read-only" failure was at least loud.
568        let final_path = self.get_backend_file_name().to_string();
569        anyhow::ensure!(
570            !final_path.ends_with(".zip"),
571            "subset: {final_path} is a zip archive; convert it to a directory \
572             backend first (data-beans convert)"
573        );
574        let temp_path = format!("{final_path}.subset_tmp");
575        if std::path::Path::new(&temp_path).exists() {
576            crate::sparse_io::remove_backend_path(&temp_path)?;
577        }
578
579        {
580            let backend_kind = self.backend_type();
581            let mut out = crate::sparse_io::create_sparse_streaming_empty(
582                Some(&temp_path),
583                Some(&backend_kind),
584            )?;
585            out.begin_streaming_csc((new_nrow, new_ncol, new_nnz as usize))?;
586
587            // Blocks bounded by bytes of surviving triplets, from the measured
588            // per-column counts — not by a fixed column count.
589            let blocks = legume_numeric::matrix::utils::byte_budget_intervals(
590                &per_col_nnz,
591                crate::sparse_io::SLAB_BUDGET_BYTES,
592                crate::sparse_io::TRIPLET_BYTES,
593            );
594
595            let mut nnz_offset = 0u64;
596            for (lb, ub) in blocks {
597                // ONE block read per slab (see the counting pass above for why).
598                // The read returns LOCAL column ids in the requested order, rows
599                // ascending within each column.
600                let old_cols: Vec<usize> = cols_new_order[lb..ub]
601                    .iter()
602                    .map(|&(_, o)| o as usize)
603                    .collect();
604                let (_, _, triplets) =
605                    self.read_triplets_by_columns(old_cols.into_iter().collect())?;
606
607                let n_block = ub - lb;
608                let mut per_col: Vec<Vec<(u64, f32)>> = vec![Vec::new(); n_block];
609                for (i, c_local, x) in triplets {
610                    if let Some(new_row) = row_map[i as usize] {
611                        per_col[c_local as usize].push((new_row, x));
612                    }
613                }
614                let mut local_colptr = Vec::with_capacity(n_block);
615                let mut row_indices = Vec::new();
616                let mut values = Vec::new();
617                for entries in &mut per_col {
618                    if !monotone_rows {
619                        // Renumbering scrambled this column's order; restore the
620                        // ascending-rows invariant the writer enforces.
621                        entries.sort_unstable_by_key(|&(r, _)| r);
622                    }
623                    local_colptr.push(row_indices.len() as u64);
624                    for &(r, x) in entries.iter() {
625                        row_indices.push(r);
626                        values.push(x);
627                    }
628                }
629                out.append_csc_slab(lb as u64, nnz_offset, &local_colptr, &row_indices, &values)?;
630                nnz_offset += values.len() as u64;
631            }
632
633            out.finalize_streaming_csc()?;
634            out.build_csr_from_csc_streaming()?;
635            out.register_row_names_vec(&new_row_names);
636            out.register_column_names_vec(&new_col_names);
637            out.set_metadata(&self.metadata())?;
638        }
639
640        ////////////////////////////////////
641        // 3. Swap the finished file in  //
642        ////////////////////////////////////
643
644        self.remove_backend_file()?;
645        std::fs::rename(&temp_path, &final_path)?;
646        self.reopen_backend()?;
647        self.clean_preloaded_columns();
648        self.clean_preloaded_rows();
649        info!("registered new data to {}", self.get_backend_file_name());
650        Ok(())
651    }
652
653    /// Reposition rows in a new order specified by `remap`
654    /// * `row_names_order` - a vector of row names in the new order
655    fn reorder_rows(&mut self, row_names_order: &[Box<str>]) -> anyhow::Result<()> {
656        // The backend is rebuilt from scratch below; its metadata comes along.
657        let meta = self.metadata();
658        let new_col_names = self.column_names()?.clone();
659        let name2new = build_name2index_map(row_names_order);
660
661        let block_size = 100;
662
663        let old2new: HashMap<u64, u64> = self
664            .row_names()?
665            .into_par_iter()
666            .enumerate()
667            .filter_map(|(idx_old, name)| {
668                name2new
669                    .get(&name)
670                    .map(|&idx_new| (idx_old as u64, idx_new as u64))
671            })
672            .collect();
673
674        if let Some(ncol) = self.num_columns() {
675            /////////////////////////////////////////////////////
676            // 1. triplets after filtering and reordering rows //
677            /////////////////////////////////////////////////////
678
679            let arc_triplets = Arc::new(Mutex::new(vec![]));
680
681            let nblock = ncol.div_ceil(block_size);
682
683            info!("remapping triplets ...");
684
685            (0..nblock)
686                .into_par_iter()
687                .progress_with(styled_progress_bar(nblock as u64, "blocks"))
688                .map(|b| {
689                    let lb = (b * block_size) as u64;
690                    let ub = ((b + 1) * block_size).min(ncol) as u64;
691                    (lb, ub)
692                })
693                .for_each(|(lb, ub)| {
694                    let (_, _, _triplets_b) = self
695                        .read_triplets_by_columns(((lb as usize)..(ub as usize)).collect())
696                        .unwrap();
697
698                    let _triplets_b = _triplets_b.into_iter().filter_map(|(i, j_loc, x)| {
699                        let j_glob = j_loc + lb;
700                        old2new.get(&i).map(|&i_new| (i_new, j_glob, x))
701                    });
702
703                    {
704                        let mut triplets = arc_triplets.lock().unwrap();
705                        triplets.extend(_triplets_b);
706                    }
707                });
708
709            /////////////////////////////////////
710            // 2. Remove previous backend file //
711            /////////////////////////////////////
712            self.remove_backend_file()?;
713
714            ///////////////////////////////
715            // 3. populate a new backend //
716            ///////////////////////////////
717            self.initialize_backend()?;
718
719            // populate data from mtx triplets
720            {
721                let mut row_col_val_triplets =
722                    arc_triplets.lock().expect("failed to lock triplets");
723
724                let nnz = row_col_val_triplets.len();
725                debug_assert!(row_col_val_triplets.len() <= nnz); // subset
726                let new_nrow = row_names_order.len();
727                let mtx_shape = (new_nrow, ncol, nnz);
728
729                info!("sorting triplets ...");
730
731                self.record_mtx_shape(Some(mtx_shape))?;
732                self.record_triplets_by_col(&mut row_col_val_triplets)?;
733                self.record_triplets_by_row(&mut row_col_val_triplets)?;
734            }
735            self.read_column_indptr()?;
736            self.read_row_indptr()?;
737
738            self.register_row_names_vec(row_names_order);
739            self.register_column_names_vec(&new_col_names);
740            self.set_metadata(&meta)?;
741            info!("registered new data to {}", self.get_backend_file_name());
742        }
743
744        self.clean_preloaded_columns();
745        self.clean_preloaded_rows();
746        Ok(())
747    }
748    // fn reorder_rows(&mut self, row_names_order: &[Box<str>]) -> anyhow::Result<()>;
749
750    /// Remove backend file
751    fn remove_backend_file(&self) -> anyhow::Result<()>;
752
753    /// Initialize backend
754    fn initialize_backend(&mut self) -> anyhow::Result<()>;
755
756    fn record_mtx_shape(&mut self, mtx_shape: Option<(usize, usize, usize)>) -> anyhow::Result<()>;
757
758    /// Stream the triplets out as CSR slabs; the row-major twin of
759    /// [`record_triplets_by_col`](Self::record_triplets_by_col).
760    fn record_triplets_by_row(
761        &mut self,
762        row_col_val_triplets: &mut Vec<(u64, u64, f32)>,
763    ) -> anyhow::Result<()> {
764        let nrow = self.num_rows().expect("should have `nrow`");
765        let ncol = self.num_columns().expect("should have `ncol`");
766        let nnz = row_col_val_triplets.len();
767
768        if nnz == 0 {
769            let csr_rowptr = vec![0u64; nrow + 1];
770            return self.record_csr_dataset_backend(&[], &[], &csr_rowptr);
771        }
772
773        // One in-place pass on the full key. A stable sort would allocate a
774        // scratch copy of the whole vector, and duplicate coordinates carry no
775        // meaning in coordinate format, so stability buys nothing.
776        row_col_val_triplets.par_sort_unstable_by_key(|&(row, col, _)| (row, col));
777
778        self.begin_streaming_csr((nrow, ncol, nnz))?;
779
780        let mut local_rowptr: Vec<u64> = Vec::new();
781        let mut cols: Vec<u64> = Vec::with_capacity(SLAB_NNZ);
782        let mut vals: Vec<f32> = Vec::with_capacity(SLAB_NNZ);
783
784        let mut start = 0_usize;
785        let mut row_offset = 0_u64;
786        while (row_offset as usize) < nrow {
787            let (end, band_end_row) =
788                slab_end(row_col_val_triplets, start, SLAB_NNZ, nrow, |t| t.0);
789
790            local_rowptr.clear();
791            cols.clear();
792            vals.clear();
793            let mut i = start;
794            for row in row_offset..band_end_row {
795                local_rowptr.push((i - start) as u64);
796                while i < end && row_col_val_triplets[i].0 == row {
797                    cols.push(row_col_val_triplets[i].1);
798                    vals.push(row_col_val_triplets[i].2);
799                    i += 1;
800                }
801            }
802            debug_assert_eq!(i, end, "every entry of the band belongs to one of its rows");
803
804            self.append_csr_slab(row_offset, start as u64, &local_rowptr, &cols, &vals)?;
805            start = end;
806            row_offset = band_end_row;
807        }
808
809        self.finalize_streaming_csr()
810    }
811
812    /// Stream the triplets out as CSC slabs.
813    ///
814    /// After the one in-place sort the triplet vector is the only full-size
815    /// structure alive; the slab staging buffers are bounded by
816    /// [`SLAB_NNZ`], and the backend's streaming audits check the column
817    /// tiling and the appended count on the way through.
818    fn record_triplets_by_col(
819        &mut self,
820        row_col_val_triplets: &mut Vec<(u64, u64, f32)>,
821    ) -> anyhow::Result<()> {
822        let nrow = self.num_rows().expect("should have `nrow`");
823        let ncol = self.num_columns().expect("should have `ncol`");
824        let nnz = row_col_val_triplets.len();
825
826        if nnz == 0 {
827            let csc_colptr = vec![0u64; ncol + 1];
828            return self.record_csc_dataset_backend(&[], &[], &csc_colptr);
829        }
830
831        // See `record_triplets_by_row` for why this is one unstable pass.
832        row_col_val_triplets.par_sort_unstable_by_key(|&(row, col, _)| (col, row));
833
834        self.begin_streaming_csc((nrow, ncol, nnz))?;
835
836        let mut local_colptr: Vec<u64> = Vec::new();
837        let mut rows: Vec<u64> = Vec::with_capacity(SLAB_NNZ);
838        let mut vals: Vec<f32> = Vec::with_capacity(SLAB_NNZ);
839
840        let mut start = 0_usize;
841        let mut col_offset = 0_u64;
842        while (col_offset as usize) < ncol {
843            let (end, band_end_col) =
844                slab_end(row_col_val_triplets, start, SLAB_NNZ, ncol, |t| t.1);
845
846            local_colptr.clear();
847            rows.clear();
848            vals.clear();
849            let mut i = start;
850            for col in col_offset..band_end_col {
851                local_colptr.push((i - start) as u64);
852                while i < end && row_col_val_triplets[i].1 == col {
853                    rows.push(row_col_val_triplets[i].0);
854                    vals.push(row_col_val_triplets[i].2);
855                    i += 1;
856                }
857            }
858            debug_assert_eq!(
859                i, end,
860                "every entry of the band belongs to one of its columns"
861            );
862
863            self.append_csc_slab(col_offset, start as u64, &local_colptr, &rows, &vals)?;
864            start = end;
865            col_offset = band_end_col;
866        }
867
868        self.finalize_streaming_csc()
869    }
870
871    /// CSR data structure in Zarr backend
872    ///
873    /// ```text
874    ///     └── by_row
875    ///         ├── data
876    ///         ├── indices (column indices)
877    ///         └── isndptr (row pointers)
878    /// ```
879    fn record_csr_dataset_backend(
880        &mut self,
881        csr_cols: &[u64],
882        csr_vals: &[f32],
883        csr_rowptr: &[u64],
884    ) -> anyhow::Result<()>;
885
886    /// Helper function to add CSC dataset to HDF5 backend
887    ///
888    /// ```text
889    /// Helper function to record the CSC dataset
890    ///     ├── by_column
891    ///     │   ├── data
892    ///     │   ├── indices (row indices)
893    ///     │   └── indptr (column pointers)
894    /// ```
895    fn record_csc_dataset_backend(
896        &mut self,
897        csc_rows: &[u64],
898        csc_vals: &[f32],
899        csc_colptr: &[u64],
900    ) -> anyhow::Result<()>;
901
902    /// Create a fixed-size 1-D backend dataset of `len` elements for the
903    /// given CSC/CSR slot. No data is written yet.
904    fn cs_create(&mut self, key: CsKey, len: usize) -> anyhow::Result<()>;
905
906    /// Write a `u64` slab at `offset` in the specified dataset.
907    /// Used for CSC/CSR `indices` and `indptr`.
908    fn cs_write_u64(&mut self, key: CsKey, offset: u64, data: &[u64]) -> anyhow::Result<()>;
909
910    /// Write an `f32` slab at `offset` in the specified dataset.
911    /// Used for CSC/CSR `data`.
912    fn cs_write_f32(&mut self, key: CsKey, offset: u64, data: &[f32]) -> anyhow::Result<()>;
913
914    /// Begin a streaming CSC build for a sparse matrix of known shape.
915    /// Pre-creates `/by_column/{data, indices, indptr}` at their final sizes
916    /// so subsequent [`append_csc_slab`](Self::append_csc_slab) calls write
917    /// into disjoint hyperslabs without further allocation.
918    fn begin_streaming_csc(&mut self, shape: (usize, usize, usize)) -> anyhow::Result<()> {
919        // A reused handle must not inherit a previous build's cursor: the
920        // finalize audit compares appended-vs-declared, and a stale count turns
921        // a correct build into a false accusation.
922        self.reset_streamed_nnz();
923        let (_, ncol, nnz) = shape;
924        self.record_mtx_shape(Some(shape))?;
925        self.cs_create(CsKey::CscData, nnz)?;
926        self.cs_create(CsKey::CscIndices, nnz)?;
927        self.cs_create(CsKey::CscIndptr, ncol + 1)?;
928        Ok(())
929    }
930
931    /// Append one contiguous CSC column band.
932    ///
933    /// * `col_offset` — global column index where this band starts
934    /// * `nnz_offset` — global nnz offset where this band's values land
935    /// * `local_colptr` — length `batch_ncol`, values in `[0, batch_nnz]`,
936    ///   will be shifted by `nnz_offset` before writing
937    /// * `row_indices` — length `batch_nnz`
938    /// * `values`      — length `batch_nnz`
939    fn append_csc_slab(
940        &mut self,
941        col_offset: u64,
942        nnz_offset: u64,
943        local_colptr: &[u64],
944        row_indices: &[u64],
945        values: &[f32],
946    ) -> anyhow::Result<()> {
947        // These checks exist because a violation does NOT fail loudly on its
948        // own: unwritten regions read back as the zarr fill value, so a bad
949        // slab yields a backend that opens and reads cleanly while carrying
950        // poisoned or duplicated entries. Cheap (one pass over the slab, no
951        // I/O) next to the compressed writes below.
952        anyhow::ensure!(
953            row_indices.len() == values.len(),
954            "append_csc_slab: {} row indices vs {} values",
955            row_indices.len(),
956            values.len()
957        );
958        anyhow::ensure!(
959            local_colptr.first().copied() == Some(0) || local_colptr.is_empty(),
960            "append_csc_slab: local_colptr must start at 0"
961        );
962        anyhow::ensure!(
963            local_colptr.windows(2).all(|w| w[0] <= w[1]),
964            "append_csc_slab: local_colptr must be monotone non-decreasing"
965        );
966        if let Some(&last) = local_colptr.last() {
967            anyhow::ensure!(
968                last <= values.len() as u64,
969                "append_csc_slab: colptr claims {last} entries, slab holds {}",
970                values.len()
971            );
972        }
973        if let Some(nrow) = self.num_rows() {
974            if let Some(&bad) = row_indices.iter().find(|&&r| r >= nrow as u64) {
975                anyhow::bail!("append_csc_slab: row index {bad} outside the {nrow}-row matrix");
976            }
977        }
978        // Ascending rows within each column: readers document it as an
979        // invariant, and the h5ad export hands the arrays to scipy as-is.
980        for (c, &start) in local_colptr.iter().enumerate() {
981            let end = local_colptr
982                .get(c + 1)
983                .copied()
984                .unwrap_or(values.len() as u64) as usize;
985            anyhow::ensure!(
986                row_indices[start as usize..end]
987                    .windows(2)
988                    .all(|w| w[0] < w[1]),
989                "append_csc_slab: rows within column {} of this band must be \
990                 strictly ascending — repeated rows usually mean duplicate \
991                 (row, col) coordinates in the source (an MTX with repeated \
992                 entries, or a union remap folding rows together)",
993                col_offset as usize + c
994            );
995        }
996
997        let shifted: Vec<u64> = local_colptr.iter().map(|&p| p + nnz_offset).collect();
998        self.cs_write_u64(CsKey::CscIndptr, col_offset, &shifted)?;
999        self.cs_write_u64(CsKey::CscIndices, nnz_offset, row_indices)?;
1000        self.cs_write_f32(CsKey::CscData, nnz_offset, values)?;
1001        self.note_streamed_nnz(values.len() as u64);
1002        Ok(())
1003    }
1004
1005    /// Finalize CSC streaming by writing the final indptr sentinel at
1006    /// position `ncol`, equal to the total nnz.
1007    fn finalize_streaming_csc(&mut self) -> anyhow::Result<()> {
1008        let ncol = self
1009            .num_columns()
1010            .ok_or_else(|| anyhow::anyhow!("ncol not set before finalize_streaming_csc"))?;
1011        let nnz = self
1012            .num_non_zeros()
1013            .ok_or_else(|| anyhow::anyhow!("nnz not set before finalize_streaming_csc"))?;
1014        self.cs_write_u64(CsKey::CscIndptr, ncol as u64, &[nnz as u64])?;
1015        self.read_column_indptr()?;
1016
1017        // The tiling check the per-slab guards cannot do. A gap or overlap in
1018        // the nnz offsets, or an over-declared total, leaves the WRITTEN
1019        // indptr non-monotone or short of the declared nnz — and unwritten
1020        // indptr slots read back as the fill value 0, so column j would claim
1021        // the whole array prefix. `indptr[ncol] - indptr[0]` equals the
1022        // declaration by construction, which is why the old debug_assert on it
1023        // could never fire; the shape of the vector between the endpoints is
1024        // what carries the truth.
1025        let indptr = self.column_indptr();
1026        anyhow::ensure!(
1027            indptr.len() == ncol + 1,
1028            "finalize_streaming_csc: indptr has {} entries, expected {}",
1029            indptr.len(),
1030            ncol + 1
1031        );
1032        anyhow::ensure!(
1033            indptr.first().copied() == Some(0),
1034            "finalize_streaming_csc: indptr[0] = {:?}, expected 0 — the first \
1035             slab was never appended",
1036            indptr.first()
1037        );
1038        if let Some(w) = indptr.windows(2).position(|w| w[0] > w[1]) {
1039            anyhow::bail!(
1040                "finalize_streaming_csc: indptr decreases at column {w} — slabs \
1041                 were appended with a gap or overlap in their nnz offsets"
1042            );
1043        }
1044        // The appended count is the ground truth the indptr cannot carry: an
1045        // over-declared nnz leaves the written indptr perfectly monotone with
1046        // the phantom tail hiding between the last written pointer and the
1047        // sentinel — and the sentinel itself was written by this function, so
1048        // comparing against it can only ever agree.
1049        let appended = self.streamed_nnz();
1050        anyhow::ensure!(
1051            appended == nnz as u64,
1052            "finalize_streaming_csc: {appended} entries appended but {nnz} \
1053             declared — the difference reads back as fill values wearing real \
1054             entries' positions"
1055        );
1056        Ok(())
1057    }
1058
1059    /// Begin a streaming CSR build for a sparse matrix of known shape; the
1060    /// row-major twin of [`begin_streaming_csc`](Self::begin_streaming_csc).
1061    fn begin_streaming_csr(&mut self, shape: (usize, usize, usize)) -> anyhow::Result<()> {
1062        self.reset_streamed_nnz();
1063        let (nrow, _, nnz) = shape;
1064        self.record_mtx_shape(Some(shape))?;
1065        self.cs_create(CsKey::CsrData, nnz)?;
1066        self.cs_create(CsKey::CsrIndices, nnz)?;
1067        self.cs_create(CsKey::CsrIndptr, nrow + 1)?;
1068        Ok(())
1069    }
1070
1071    /// Append one contiguous CSR row band; the row-major twin of
1072    /// [`append_csc_slab`](Self::append_csc_slab), with the same audits.
1073    ///
1074    /// * `row_offset` — global row index where this band starts
1075    /// * `nnz_offset` — global nnz offset where this band's values land
1076    /// * `local_rowptr` — length `batch_nrow`, values in `[0, batch_nnz]`,
1077    ///   will be shifted by `nnz_offset` before writing
1078    /// * `col_indices` — length `batch_nnz`
1079    /// * `values`      — length `batch_nnz`
1080    fn append_csr_slab(
1081        &mut self,
1082        row_offset: u64,
1083        nnz_offset: u64,
1084        local_rowptr: &[u64],
1085        col_indices: &[u64],
1086        values: &[f32],
1087    ) -> anyhow::Result<()> {
1088        anyhow::ensure!(
1089            col_indices.len() == values.len(),
1090            "append_csr_slab: {} column indices vs {} values",
1091            col_indices.len(),
1092            values.len()
1093        );
1094        anyhow::ensure!(
1095            local_rowptr.first().copied() == Some(0) || local_rowptr.is_empty(),
1096            "append_csr_slab: local_rowptr must start at 0"
1097        );
1098        anyhow::ensure!(
1099            local_rowptr.windows(2).all(|w| w[0] <= w[1]),
1100            "append_csr_slab: local_rowptr must be monotone non-decreasing"
1101        );
1102        if let Some(&last) = local_rowptr.last() {
1103            anyhow::ensure!(
1104                last <= values.len() as u64,
1105                "append_csr_slab: rowptr claims {last} entries, slab holds {}",
1106                values.len()
1107            );
1108        }
1109        if let Some(ncol) = self.num_columns() {
1110            if let Some(&bad) = col_indices.iter().find(|&&c| c >= ncol as u64) {
1111                anyhow::bail!(
1112                    "append_csr_slab: column index {bad} outside the {ncol}-column matrix"
1113                );
1114            }
1115        }
1116        for (r, &start) in local_rowptr.iter().enumerate() {
1117            let end = local_rowptr
1118                .get(r + 1)
1119                .copied()
1120                .unwrap_or(values.len() as u64) as usize;
1121            anyhow::ensure!(
1122                col_indices[start as usize..end]
1123                    .windows(2)
1124                    .all(|w| w[0] < w[1]),
1125                "append_csr_slab: columns within row {} of this band must be \
1126                 strictly ascending — repeated columns usually mean duplicate \
1127                 (row, col) coordinates in the source",
1128                row_offset as usize + r
1129            );
1130        }
1131
1132        let shifted: Vec<u64> = local_rowptr.iter().map(|&p| p + nnz_offset).collect();
1133        self.cs_write_u64(CsKey::CsrIndptr, row_offset, &shifted)?;
1134        self.cs_write_u64(CsKey::CsrIndices, nnz_offset, col_indices)?;
1135        self.cs_write_f32(CsKey::CsrData, nnz_offset, values)?;
1136        self.note_streamed_nnz(values.len() as u64);
1137        Ok(())
1138    }
1139
1140    /// Finalize CSR streaming: write the indptr sentinel at position `nrow`,
1141    /// load the row index, and check the appended count against the declared
1142    /// nnz -- the one violation the written indptr cannot reveal.
1143    fn finalize_streaming_csr(&mut self) -> anyhow::Result<()> {
1144        let nrow = self
1145            .num_rows()
1146            .ok_or_else(|| anyhow::anyhow!("nrow not set before finalize_streaming_csr"))?;
1147        let nnz = self
1148            .num_non_zeros()
1149            .ok_or_else(|| anyhow::anyhow!("nnz not set before finalize_streaming_csr"))?;
1150        self.cs_write_u64(CsKey::CsrIndptr, nrow as u64, &[nnz as u64])?;
1151        self.read_row_indptr()?;
1152
1153        let appended = self.streamed_nnz();
1154        anyhow::ensure!(
1155            appended == nnz as u64,
1156            "finalize_streaming_csr: {appended} entries appended but {nnz} \
1157             declared — the slabs did not cover the matrix"
1158        );
1159        Ok(())
1160    }
1161
1162    /// Build `/by_row/{data, indices, indptr}` by transposing the already-
1163    /// written CSC data on disk. Uses two passes over CSC with bounded
1164    /// auxiliary memory (~`24 B × nrow` plus one row-band worth of CSR).
1165    fn build_csr_from_csc_streaming(&mut self) -> anyhow::Result<()> {
1166        let nrow = self
1167            .num_rows()
1168            .ok_or_else(|| anyhow::anyhow!("nrow not set before build_csr_from_csc_streaming"))?;
1169        let ncol = self
1170            .num_columns()
1171            .ok_or_else(|| anyhow::anyhow!("ncol not set before build_csr_from_csc_streaming"))?;
1172        let nnz = self
1173            .num_non_zeros()
1174            .ok_or_else(|| anyhow::anyhow!("nnz not set before build_csr_from_csc_streaming"))?;
1175
1176        if nnz == 0 {
1177            self.cs_create(CsKey::CsrData, 0)?;
1178            self.cs_create(CsKey::CsrIndices, 0)?;
1179            self.cs_create(CsKey::CsrIndptr, nrow + 1)?;
1180            let zeros = vec![0u64; nrow + 1];
1181            self.cs_write_u64(CsKey::CsrIndptr, 0, &zeros)?;
1182            self.read_row_indptr()?;
1183            return Ok(());
1184        }
1185
1186        const COL_BLOCK: usize = 1024;
1187        let n_col_blocks = ncol.div_ceil(COL_BLOCK);
1188        let bar1 = styled_progress_bar(n_col_blocks as u64, "transpose count");
1189        let mut row_counts = vec![0u64; nrow];
1190        let mut col_lo = 0usize;
1191        while col_lo < ncol {
1192            let col_hi = (col_lo + COL_BLOCK).min(ncol);
1193            let cols: Self::IndexIter = (col_lo..col_hi).collect();
1194            let (_, _, triplets) = self.read_triplets_by_columns(cols)?;
1195            for (row_i, _, _) in &triplets {
1196                row_counts[*row_i as usize] += 1;
1197            }
1198            col_lo = col_hi;
1199            bar1.inc(1);
1200        }
1201        bar1.finish_and_clear();
1202
1203        let mut rowptr = vec![0u64; nrow + 1];
1204        let mut acc = 0u64;
1205        for i in 0..nrow {
1206            rowptr[i] = acc;
1207            acc += row_counts[i];
1208        }
1209        rowptr[nrow] = acc;
1210        debug_assert_eq!(acc, nnz as u64);
1211
1212        self.cs_create(CsKey::CsrData, nnz)?;
1213        self.cs_create(CsKey::CsrIndices, nnz)?;
1214        self.cs_create(CsKey::CsrIndptr, nrow + 1)?;
1215        self.cs_write_u64(CsKey::CsrIndptr, 0, &rowptr)?;
1216
1217        // Per-band buffers carry 12 B/nnz (u64 col + f32 val); cap aggregate
1218        // at ~256 MB so the row-banded scatter stays within a fixed budget
1219        // regardless of nnz.
1220        const TRANSPOSE_BAND_BYTES: usize = 256 * 1024 * 1024;
1221        let avg_density = nnz.div_ceil(nrow.max(1));
1222        let band_rows = (TRANSPOSE_BAND_BYTES / (12 * avg_density.max(1)))
1223            .max(1)
1224            .min(nrow);
1225        let n_bands = nrow.div_ceil(band_rows);
1226
1227        let bar2 = styled_progress_bar(n_bands as u64, "transpose scatter");
1228        let mut band_lo = 0usize;
1229        while band_lo < nrow {
1230            let band_hi = (band_lo + band_rows).min(nrow);
1231            let band_nnz_start = rowptr[band_lo];
1232            let band_nnz_end = rowptr[band_hi];
1233            let band_nnz = (band_nnz_end - band_nnz_start) as usize;
1234
1235            if band_nnz == 0 {
1236                band_lo = band_hi;
1237                bar2.inc(1);
1238                continue;
1239            }
1240
1241            let mut out_indices = vec![0u64; band_nnz];
1242            let mut out_values = vec![0f32; band_nnz];
1243            let mut cursor = vec![0u64; band_hi - band_lo];
1244
1245            let mut col_lo = 0usize;
1246            while col_lo < ncol {
1247                let col_hi = (col_lo + COL_BLOCK).min(ncol);
1248                let cols: Self::IndexIter = (col_lo..col_hi).collect();
1249                let (_, _, triplets) = self.read_triplets_by_columns(cols)?;
1250                for &(row_i, col_j_local, x) in &triplets {
1251                    let row_i_us = row_i as usize;
1252                    if row_i_us >= band_lo && row_i_us < band_hi {
1253                        let band_idx = row_i_us - band_lo;
1254                        // col_j_local is already the global column index because
1255                        // read_triplets_by_columns returns columns in the passed order
1256                        // (0..batch for standalone call). We passed col_lo..col_hi,
1257                        // which returns local indices 0..(col_hi - col_lo) — so add col_lo.
1258                        let col_j_global = col_j_local + col_lo as u64;
1259                        let offset_in_band =
1260                            (rowptr[band_lo + band_idx] - band_nnz_start) + cursor[band_idx];
1261                        out_indices[offset_in_band as usize] = col_j_global;
1262                        out_values[offset_in_band as usize] = x;
1263                        cursor[band_idx] += 1;
1264                    }
1265                }
1266                col_lo = col_hi;
1267            }
1268
1269            self.cs_write_u64(CsKey::CsrIndices, band_nnz_start, &out_indices)?;
1270            self.cs_write_f32(CsKey::CsrData, band_nnz_start, &out_values)?;
1271
1272            band_lo = band_hi;
1273            bar2.inc(1);
1274        }
1275        bar2.finish_and_clear();
1276
1277        self.read_row_indptr()?;
1278        Ok(())
1279    }
1280
1281    /// preload row index pointers
1282    fn read_row_indptr(&mut self) -> anyhow::Result<()>;
1283
1284    /// preload column index pointers
1285    fn read_column_indptr(&mut self) -> anyhow::Result<()>;
1286
1287    /// preload all the columns for faster processing
1288    fn preload_columns(&mut self) -> anyhow::Result<()>;
1289
1290    /// unload the memory
1291    fn clean_preloaded_columns(&mut self);
1292
1293    /// preload all the rows for faster processing
1294    fn preload_rows(&mut self) -> anyhow::Result<()>;
1295
1296    /// unload the row memory
1297    fn clean_preloaded_rows(&mut self);
1298
1299    /// backend file name
1300    fn get_backend_file_name(&self) -> &str;
1301
1302    /// backend file type
1303    fn backend_type(&self) -> SparseIoBackend;
1304}