use bincode::{Decode, Encode, config, decode_from_slice, serde::encode_to_vec};
use half::f16;
use half::slice::HalfFloatSliceExt;
use indexmap::IndexSet;
use lz4_flex::block::{decompress_into, uncompressed_size};
use lz4_flex::{compress_prepend_size, decompress_size_prepended};
use memmap2::MmapOptions;
use num_traits::FromPrimitive;
use rayon::prelude::*;
use rustc_hash::{FxHashMap, FxHashSet};
use serde::{Deserialize, Serialize};
use std::{
fs::File,
io::{BufWriter, Seek, SeekFrom, Write},
marker::Sync,
path::Path,
sync::Arc,
};
use crate::prelude::*;
use crate::v4_io::{
ChunkCodec, ChunkFormat, V4Ctx, V4Decoded, V4Header, V4Run, decode_v4, decode_v4_into,
gene_norm_table, v4_cell_chunk, v4_gene_chunk,
};
#[derive(Encode, Decode, Serialize, Deserialize, Debug, Clone)]
pub enum RawCounts {
U16(Vec<u16>),
U32(Vec<u32>),
}
pub(crate) const RAW_ELEM_U16: u8 = 2;
pub(crate) const RAW_ELEM_U32: u8 = 4;
const CSR_CHUNK_HEADER_LEN: usize = 32;
const CSC_CHUNK_HEADER_LEN: usize = 36;
const GENE_FLUSH_FREQUENCY: usize = 1_000;
const CELL_FLUSH_FREQUENCY: usize = 100_000;
impl RawCounts {
pub fn get(&self, i: usize) -> u32 {
match self {
RawCounts::U16(v) => v[i] as u32,
RawCounts::U32(v) => v[i],
}
}
pub fn len(&self) -> usize {
match self {
RawCounts::U16(v) => v.len(),
RawCounts::U32(v) => v.len(),
}
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn iter(&self) -> RawCountsIter<'_> {
RawCountsIter {
inner: self,
pos: 0,
}
}
pub fn elem_size(&self) -> u8 {
match self {
RawCounts::U16(_) => RAW_ELEM_U16,
RawCounts::U32(_) => RAW_ELEM_U32,
}
}
pub fn write_bytes(&self, writer: &mut impl Write) -> std::io::Result<()> {
match self {
RawCounts::U16(v) => {
let mut bytes = Vec::with_capacity(v.len() * 2);
for x in v {
bytes.extend_from_slice(&x.to_le_bytes());
}
writer.write_all(&bytes)
}
RawCounts::U32(v) => {
let mut bytes = Vec::with_capacity(v.len() * 4);
for x in v {
bytes.extend_from_slice(&x.to_le_bytes());
}
writer.write_all(&bytes)
}
}
}
pub fn read_from_buffer(
buffer: &[u8],
count: usize,
elem_size: u8,
) -> Result<Self, ScDataErrors> {
let width = match elem_size {
RAW_ELEM_U32 => 4_usize,
RAW_ELEM_U16 | 0 => 2_usize,
other => return Err(ScDataErrors::RawElemSizeInvalid(other)),
};
let needed = count * width;
if buffer.len() < needed {
return Err(ScDataErrors::ChunkPayloadTruncated {
expected: needed,
found: buffer.len(),
});
}
Ok(if width == 4 {
RawCounts::U32(
buffer[..needed]
.chunks_exact(4)
.map(|c| {
u32::from_le_bytes(c.try_into().expect("4-byte chunk by construction"))
})
.collect(),
)
} else {
RawCounts::U16(
buffer[..needed]
.chunks_exact(2)
.map(|c| {
u16::from_le_bytes(c.try_into().expect("2-byte chunk by construction"))
})
.collect(),
)
})
}
pub fn from_u32_auto(data: &[u32]) -> Self {
if data.iter().all(|&v| v <= u16::MAX as u32) {
RawCounts::U16(data.iter().map(|&v| v as u16).collect())
} else {
RawCounts::U32(data.to_vec())
}
}
}
pub struct RawCountsIter<'a> {
inner: &'a RawCounts,
pos: usize,
}
impl<'a> Iterator for RawCountsIter<'a> {
type Item = u32;
fn next(&mut self) -> Option<u32> {
if self.pos < self.inner.len() {
let val = self.inner.get(self.pos);
self.pos += 1;
Some(val)
} else {
None
}
}
}
#[derive(Clone, Debug)]
pub struct CellQuality {
pub cell_indices: Vec<usize>,
pub gene_indices: Vec<usize>,
pub lib_size: Vec<usize>,
pub nnz: Vec<usize>,
}
impl CellQuality {
pub fn set_cell_indices(&mut self, cell_indices: &[usize]) {
self.cell_indices = cell_indices.to_vec();
}
pub fn set_gene_indices(&mut self, gene_indices: &[usize]) {
self.gene_indices = gene_indices.to_vec();
}
}
#[derive(Clone, Debug)]
pub struct MinCellQuality {
pub min_unique_genes: usize,
pub min_lib_size: usize,
pub min_cells: usize,
pub target_size: f32,
}
#[derive(Debug, Clone)]
pub struct CellOnFileQuality {
pub cells_to_keep: Vec<usize>,
pub genes_to_keep: Vec<usize>,
pub cells_to_keep_set: FxHashSet<usize>,
pub genes_to_keep_set: FxHashSet<usize>,
pub cell_old_to_new: FxHashMap<usize, usize>,
pub gene_old_to_new: FxHashMap<usize, usize>,
}
impl CellOnFileQuality {
pub fn new(cells_to_keep: Vec<usize>, genes_to_keep: Vec<usize>) -> Self {
Self {
cells_to_keep,
genes_to_keep,
cells_to_keep_set: FxHashSet::default(),
genes_to_keep_set: FxHashSet::default(),
cell_old_to_new: FxHashMap::default(),
gene_old_to_new: FxHashMap::default(),
}
}
pub fn generate_maps_sets(&mut self) {
let cells_to_keep_set: FxHashSet<usize> = self.cells_to_keep.iter().cloned().collect();
let genes_to_keep_set: FxHashSet<usize> = self.genes_to_keep.iter().cloned().collect();
let cell_old_to_new: FxHashMap<usize, usize> = self
.cells_to_keep
.iter()
.enumerate()
.map(|(new_idx, &old_idx)| (old_idx, new_idx))
.collect();
let gene_old_to_new: FxHashMap<usize, usize> = self
.genes_to_keep
.iter()
.enumerate()
.map(|(new_idx, &old_idx)| (old_idx, new_idx))
.collect();
self.cells_to_keep_set = cells_to_keep_set;
self.genes_to_keep_set = genes_to_keep_set;
self.cell_old_to_new = cell_old_to_new;
self.gene_old_to_new = gene_old_to_new
}
}
pub fn norm_value(x: f32, lib_size: f32, target_size: f32) -> F16 {
F16::from_f32((x / lib_size * target_size).ln_1p())
}
struct ChunkPayload<'a> {
data_raw: &'a [u8],
data_raw_len: usize,
raw_elem_size: u8,
raw_width: usize,
data_norm: &'a [u8],
indices: &'a [u8],
}
impl<'a> ChunkPayload<'a> {
fn slice(
buffer: &'a [u8],
header_len: usize,
lens: (usize, usize, usize),
raw_elem_size: u8,
) -> Result<Self, ScDataErrors> {
let (data_raw_len, data_norm_len, indices_len) = lens;
let raw_width = match raw_elem_size {
RAW_ELEM_U32 => 4,
RAW_ELEM_U16 | 0 => 2,
other => return Err(ScDataErrors::RawElemSizeInvalid(other)),
};
let data_end = header_len + data_raw_len * raw_width;
let norm_end = data_end + data_norm_len * 2;
let total_end = norm_end + indices_len * 4;
if buffer.len() < total_end {
return Err(ScDataErrors::ChunkPayloadTruncated {
expected: total_end,
found: buffer.len(),
});
}
Ok(Self {
data_raw: &buffer[header_len..data_end],
data_raw_len,
raw_elem_size,
raw_width,
data_norm: &buffer[data_end..norm_end],
indices: &buffer[norm_end..total_end],
})
}
fn norm_iter(&self) -> impl Iterator<Item = F16> + '_ {
self.data_norm
.chunks_exact(2)
.map(|c| F16::from_le_bytes(c.try_into().expect("2-byte chunk by construction")))
}
fn index_iter(&self) -> impl Iterator<Item = u32> + '_ {
self.indices
.chunks_exact(4)
.map(|c| u32::from_le_bytes(c.try_into().expect("4-byte chunk by construction")))
}
fn raw_counts(&self) -> Result<RawCounts, ScDataErrors> {
RawCounts::read_from_buffer(self.data_raw, self.data_raw_len, self.raw_elem_size)
}
}
const F16_BLOCK: usize = 64;
pub(crate) trait RunSource {
fn nnz(&self) -> usize;
fn index_at(&self, i: usize) -> u32;
fn raw_at(&self, i: usize) -> u32;
fn norm_at(&self, i: usize) -> f32;
fn extend_indices(&self, out: &mut Vec<u32>);
fn extend_raw<T: FromPrimitive>(&self, out: &mut Vec<T>) -> Result<(), ScDataErrors>;
fn extend_norm(&self, out: &mut Vec<f32>);
}
#[inline]
pub(crate) fn extend_raw_checked<T, I>(out: &mut Vec<T>, vals: I) -> Result<(), ScDataErrors>
where
T: FromPrimitive,
I: Iterator<Item = u32> + Clone,
{
raw_to::<T>(vals.clone().max().unwrap_or(0))?;
out.extend(vals.map(|v| T::from_u32(v).expect("the maximum fits, so every value does")));
Ok(())
}
impl RunSource for ChunkPayload<'_> {
#[inline]
fn nnz(&self) -> usize {
self.indices.len() / 4
}
#[inline(always)]
fn index_at(&self, i: usize) -> u32 {
let b = &self.indices[i * 4..i * 4 + 4];
u32::from_le_bytes([b[0], b[1], b[2], b[3]])
}
#[inline(always)]
fn raw_at(&self, i: usize) -> u32 {
if self.raw_width == 4 {
let b = &self.data_raw[i * 4..i * 4 + 4];
u32::from_le_bytes([b[0], b[1], b[2], b[3]])
} else {
let b = &self.data_raw[i * 2..i * 2 + 2];
u16::from_le_bytes([b[0], b[1]]) as u32
}
}
#[inline(always)]
fn norm_at(&self, i: usize) -> f32 {
let b = &self.data_norm[i * 2..i * 2 + 2];
F16::from_le_bytes([b[0], b[1]]).to_f32()
}
fn extend_indices(&self, out: &mut Vec<u32>) {
out.extend(self.index_iter());
}
fn extend_raw<T: FromPrimitive>(&self, out: &mut Vec<T>) -> Result<(), ScDataErrors> {
if self.raw_width == 4 {
extend_raw_checked(
out,
self.data_raw.chunks_exact(4).map(|c| {
u32::from_le_bytes(c.try_into().expect("4-byte chunk by construction"))
}),
)
} else {
extend_raw_checked(
out,
self.data_raw.chunks_exact(2).map(|c| {
u16::from_le_bytes(c.try_into().expect("2-byte chunk by construction")) as u32
}),
)
}
}
fn extend_norm(&self, out: &mut Vec<f32>) {
let start = out.len();
out.resize(start + self.data_norm.len() / 2, 0.0);
let mut block = [f16::ZERO; F16_BLOCK];
for (src, dst) in self
.data_norm
.chunks(2 * F16_BLOCK)
.zip(out[start..].chunks_mut(F16_BLOCK))
{
for (b, c) in block.iter_mut().zip(src.chunks_exact(2)) {
*b = f16::from_le_bytes(c.try_into().expect("2-byte chunk by construction"));
}
block[..dst.len()].convert_to_f32_slice(dst);
}
}
}
#[inline]
fn header_u32(header: &[u8], at: usize) -> usize {
u32::from_le_bytes([header[at], header[at + 1], header[at + 2], header[at + 3]]) as usize
}
struct CsrChunkView<'a> {
payload: ChunkPayload<'a>,
library_size: usize,
original_index: usize,
to_keep: bool,
}
impl<'a> CsrChunkView<'a> {
fn parse(buffer: &'a [u8]) -> Result<Self, ScDataErrors> {
if buffer.len() < CSR_CHUNK_HEADER_LEN {
return Err(ScDataErrors::ChunkBufferTooSmall {
expected: CSR_CHUNK_HEADER_LEN,
found: buffer.len(),
});
}
let header = &buffer[0..CSR_CHUNK_HEADER_LEN];
let lens = (
header_u32(header, 0),
header_u32(header, 4),
header_u32(header, 8),
);
Ok(Self {
payload: ChunkPayload::slice(buffer, CSR_CHUNK_HEADER_LEN, lens, header[29])?,
library_size: le_u64_at(header, 12) as usize,
original_index: le_u64_at(header, 20) as usize,
to_keep: header[28] != 0,
})
}
}
fn csr_payload(buffer: &[u8]) -> Result<ChunkPayload<'_>, ScDataErrors> {
CsrChunkView::parse(buffer).map(|view| view.payload)
}
fn csc_payload(buffer: &[u8]) -> Result<ChunkPayload<'_>, ScDataErrors> {
if buffer.len() < CSC_CHUNK_HEADER_LEN {
return Err(ScDataErrors::ChunkBufferTooSmall {
expected: CSC_CHUNK_HEADER_LEN,
found: buffer.len(),
});
}
let lens = (
header_u32(buffer, 0),
header_u32(buffer, 4),
header_u32(buffer, 8),
);
ChunkPayload::slice(buffer, CSC_CHUNK_HEADER_LEN, lens, buffer[33])
}
#[derive(Debug)]
pub struct CsrCellChunk {
pub data_raw: RawCounts,
pub data_norm: Vec<F16>,
pub library_size: usize,
pub indices: Vec<u32>,
pub original_index: usize,
pub to_keep: bool,
}
impl CsrCellChunk {
pub fn from_data<T, U>(
data: &[T],
col_idx: &[U],
original_index: usize,
size_factor: f32,
to_keep: bool,
) -> Self
where
T: FloatAndUInt,
U: FloatAndUInt,
{
let sum = data.iter().map(|&x| x.to_f32()).sum::<f32>();
let data_norm: Vec<F16> = data
.iter()
.map(|&x| norm_value(x.to_f32(), sum, size_factor))
.collect();
let data_raw = if data.iter().all(|&x| x.to_u32() <= u16::MAX as u32) {
RawCounts::U16(data.iter().map(|&x| x.to_u32() as u16).collect())
} else {
RawCounts::U32(data.iter().map(|&x| x.to_u32()).collect())
};
Self {
data_raw,
data_norm,
library_size: sum as usize,
indices: col_idx.iter().map(|&x| x.to_u32()).collect::<Vec<u32>>(),
original_index,
to_keep,
}
}
pub fn write_to_bytes(&self, writer: &mut impl Write) -> std::io::Result<()> {
writer.write_all(&self.to_bytes())
}
fn to_bytes(&self) -> Vec<u8> {
let raw_bytes = self.data_raw.len() * self.data_raw.elem_size() as usize;
let mut buffer = Vec::with_capacity(
CSR_CHUNK_HEADER_LEN + raw_bytes + self.data_norm.len() * 2 + self.indices.len() * 4,
);
buffer.extend_from_slice(&(self.data_raw.len() as u32).to_le_bytes());
buffer.extend_from_slice(&(self.data_norm.len() as u32).to_le_bytes());
buffer.extend_from_slice(&(self.indices.len() as u32).to_le_bytes());
buffer.extend_from_slice(&(self.library_size as u64).to_le_bytes());
buffer.extend_from_slice(&(self.original_index as u64).to_le_bytes());
buffer.extend_from_slice(&[self.to_keep as u8, self.data_raw.elem_size(), 0, 0]);
match &self.data_raw {
RawCounts::U16(v) => {
for x in v {
buffer.extend_from_slice(&x.to_le_bytes());
}
}
RawCounts::U32(v) => {
for x in v {
buffer.extend_from_slice(&x.to_le_bytes());
}
}
}
for v in &self.data_norm {
buffer.extend_from_slice(&v.to_bits().to_le_bytes());
}
for i in &self.indices {
buffer.extend_from_slice(&i.to_le_bytes());
}
buffer
}
pub fn to_compressed_bytes(&self) -> std::io::Result<Vec<u8>> {
Ok(compress_prepend_size(&self.to_bytes()))
}
pub fn read_from_buffer(buffer: &[u8]) -> Result<Self, ScDataErrors> {
let view = CsrChunkView::parse(buffer)?;
Ok(Self {
data_raw: view.payload.raw_counts()?,
data_norm: view.payload.norm_iter().collect(),
library_size: view.library_size,
indices: view.payload.index_iter().collect(),
original_index: view.original_index,
to_keep: view.to_keep,
})
}
pub fn generate_chunks_sparse_data<T, U>(
sparse_data: CompressedSparseData2<T, U>,
cell_qc: MinCellQuality,
) -> (Vec<CsrCellChunk>, CellQuality)
where
T: Clone + Default + Into<u32> + Sync,
U: Clone + Default,
{
let n_cells = sparse_data.indptr.len() - 1;
let n_genes = sparse_data.shape.1;
let mut no_cells_exp_gene = vec![0usize; n_genes];
for i in 0..n_cells {
let start_i = sparse_data.indptr[i] as usize;
let end_i = sparse_data.indptr[i + 1] as usize;
for &gene_idx in &sparse_data.indices[start_i..end_i] {
no_cells_exp_gene[gene_idx as usize] += 1;
}
}
let genes_to_keep: Vec<usize> = (0..n_genes)
.filter(|&i| no_cells_exp_gene[i] >= cell_qc.min_cells)
.collect();
let mut gene_index_map = vec![None; n_genes];
for (new_idx, &old_idx) in genes_to_keep.iter().enumerate() {
gene_index_map[old_idx] = Some(new_idx);
}
let results: Vec<_> = (0..n_cells)
.into_par_iter()
.map(|i| {
let start_i = sparse_data.indptr[i] as usize;
let end_i = sparse_data.indptr[i + 1] as usize;
let mut filtered_data = Vec::new();
let mut filtered_indices = Vec::new();
for idx in start_i..end_i {
let old_gene_idx = sparse_data.indices[idx] as usize;
if let Some(new_gene_idx) = gene_index_map[old_gene_idx] {
filtered_data.push(sparse_data.data[idx].clone().into());
filtered_indices.push(new_gene_idx);
}
}
let sum_data_i = filtered_data.iter().sum::<u32>() as usize;
let nnz_i = filtered_indices.len();
let to_keep_i =
(nnz_i >= cell_qc.min_unique_genes) && (sum_data_i >= cell_qc.min_lib_size);
let chunk_i = CsrCellChunk::from_data(
&filtered_data,
&filtered_indices,
i,
cell_qc.target_size,
to_keep_i,
);
(chunk_i, sum_data_i, nnz_i, to_keep_i, i)
})
.collect();
let mut res = Vec::with_capacity(n_cells);
let mut lib_size = Vec::new();
let mut nnz = Vec::new();
let mut cells_to_keep = Vec::new();
for (chunk, lib, n, keep, idx) in results {
res.push(chunk);
if keep {
lib_size.push(lib);
nnz.push(n);
cells_to_keep.push(idx);
}
}
let qc_data = CellQuality {
cell_indices: cells_to_keep,
gene_indices: genes_to_keep,
lib_size,
nnz,
};
(res, qc_data)
}
pub fn get_qc_info(&self) -> (usize, usize) {
(self.indices.len(), self.library_size)
}
pub fn filter_genes(&mut self, genes_to_keep: &[bool]) {
let mut index_map = vec![None; genes_to_keep.len()];
let mut new_idx = 0;
for (old_idx, &keep) in genes_to_keep.iter().enumerate() {
if keep {
index_map[old_idx] = Some(new_idx);
new_idx += 1;
}
}
let mut new_indices = Vec::new();
let mut new_norm = Vec::new();
let mut keep_positions = Vec::new();
for (i, &gene_idx) in self.indices.iter().enumerate() {
if let Some(new_gene_idx) = index_map[gene_idx as usize] {
new_indices.push(new_gene_idx);
new_norm.push(self.data_norm[i]);
keep_positions.push(i);
}
}
let new_raw = match &self.data_raw {
RawCounts::U16(v) => RawCounts::U16(keep_positions.iter().map(|&p| v[p]).collect()),
RawCounts::U32(v) => RawCounts::U32(keep_positions.iter().map(|&p| v[p]).collect()),
};
self.indices = new_indices;
self.data_raw = new_raw;
self.data_norm = new_norm;
}
}
#[derive(Encode, Decode, Serialize, Deserialize, Debug)]
pub struct CscGeneChunk {
pub data_raw: RawCounts,
pub data_norm: Vec<F16>,
pub avg_exp: F16,
pub nnz: usize,
pub indices: Vec<u32>,
pub original_index: usize,
pub to_keep: bool,
}
impl CscGeneChunk {
pub fn from_conversion(
data_raw: RawCounts,
data_norm: &[F16],
col_idx: &[usize],
original_index: usize,
to_keep: bool,
) -> Self {
let avg_exp = F16::from_f32(data_norm.iter().map(|v| v.to_f32()).sum::<f32>());
let nnz = data_raw.len();
Self {
data_raw,
data_norm: data_norm.to_vec(),
avg_exp,
nnz,
indices: col_idx.iter().map(|x| *x as u32).collect::<Vec<u32>>(),
original_index,
to_keep,
}
}
pub fn write_to_bytes(&self, writer: &mut impl Write) -> std::io::Result<()> {
writer.write_all(&(self.data_raw.len() as u32).to_le_bytes())?;
writer.write_all(&(self.data_norm.len() as u32).to_le_bytes())?;
writer.write_all(&(self.indices.len() as u32).to_le_bytes())?;
writer.write_all(&self.avg_exp.to_le_bytes())?;
writer.write_all(&[0, 0])?;
writer.write_all(&(self.nnz as u64).to_le_bytes())?;
writer.write_all(&(self.original_index as u64).to_le_bytes())?;
writer.write_all(&[self.to_keep as u8, self.data_raw.elem_size(), 0, 0])?;
self.data_raw.write_bytes(writer)?;
let mut data_norm_bytes = Vec::with_capacity(self.data_norm.len() * 2);
for v in &self.data_norm {
data_norm_bytes.extend_from_slice(&v.to_bits().to_le_bytes());
}
writer.write_all(&data_norm_bytes)?;
let mut row_indices_bytes = Vec::with_capacity(self.indices.len() * 4);
for i in &self.indices {
row_indices_bytes.extend_from_slice(&i.to_le_bytes());
}
writer.write_all(&row_indices_bytes)?;
Ok(())
}
pub fn to_compressed_bytes(&self) -> std::io::Result<Vec<u8>> {
let mut buffer = Vec::new();
self.write_to_bytes(&mut buffer)?;
Ok(compress_prepend_size(&buffer))
}
pub fn read_from_buffer(buffer: &[u8]) -> Result<Self, ScDataErrors> {
let payload = csc_payload(buffer)?;
Ok(Self {
data_raw: payload.raw_counts()?,
data_norm: payload.norm_iter().collect(),
avg_exp: F16::from_le_bytes([buffer[12], buffer[13]]),
nnz: le_u64_at(buffer, 16) as usize,
indices: payload.index_iter().collect(),
original_index: le_u64_at(buffer, 24) as usize,
to_keep: buffer[32] != 0,
})
}
pub fn filter_selected_cells(&mut self, cells_to_keep: &IndexSet<u32>) {
self.filter_cells_by(|cell| cells_to_keep.get_index_of(&cell).map(|i| i as u32));
}
pub fn filter_selected_cells_dense(&mut self, cell_map: &[u32]) {
self.filter_cells_by(|cell| match cell_map.get(cell as usize) {
Some(&row) if row != INDEX_DROPPED => Some(row),
_ => None,
});
}
#[inline]
fn filter_cells_by(&mut self, row_of: impl Fn(u32) -> Option<u32>) {
let mut kept: Vec<(u32, u32)> = self
.indices
.iter()
.enumerate()
.filter_map(|(pos, &cell)| row_of(cell).map(|row| (row, pos as u32)))
.collect();
if !kept.is_sorted_by_key(|&(row, _)| row) {
kept.sort_unstable_by_key(|&(row, _)| row);
}
self.data_norm = kept
.iter()
.map(|&(_, p)| self.data_norm[p as usize])
.collect();
self.data_raw = match &self.data_raw {
RawCounts::U16(v) => RawCounts::U16(kept.iter().map(|&(_, p)| v[p as usize]).collect()),
RawCounts::U32(v) => RawCounts::U32(kept.iter().map(|&(_, p)| v[p as usize]).collect()),
};
self.indices = kept.iter().map(|&(row, _)| row).collect();
self.nnz = kept.len();
}
pub fn calculate_avg_exp(&self, cells_to_keep: &IndexSet<u32>) -> (usize, f32) {
let sum: f32 = self
.indices
.iter()
.zip(self.data_norm.iter())
.filter(|(cell_idx, _)| cells_to_keep.contains(*cell_idx))
.map(|(_, val)| val.to_f32())
.sum();
let avg = sum / cells_to_keep.len() as f32;
(self.original_index, avg)
}
pub fn to_sparse_axis(&self, n_cells: usize) -> SparseAxis<u32, f32> {
let raw_u32: Vec<u32> = self.data_raw.iter().collect();
SparseAxis::new_csc(
self.indices.iter().map(|x| *x as usize).collect(),
raw_u32,
Some(self.data_norm.iter().map(|x| x.to_f32()).collect()),
n_cells,
)
}
}
#[derive(Encode, Decode, Serialize, Deserialize, Clone)]
pub struct SparseDataHeader {
pub total_cells: usize,
pub total_genes: usize,
pub cell_based: bool,
pub no_chunks: usize,
pub chunk_offsets: Vec<u64>,
pub index_map: FxHashMap<usize, usize>,
}
const FILE_MAGIC: &[u8; 8] = b"SCRNASEQ";
fn le_u64_at(bytes: &[u8], at: usize) -> u64 {
u64::from_le_bytes(
bytes[at..at + 8]
.try_into()
.expect("an 8-byte slice converts to [u8; 8]"),
)
}
#[repr(C)]
#[derive(Encode, Decode, Serialize, Deserialize)]
struct FileHeader {
magic: [u8; 8],
version: u32,
main_header_offset: u64,
cell_based: bool,
target_size: f32,
library_index_offset: u64,
_reserved_1: [u8; 20],
_reserved_2: [u8; 3],
}
impl FileHeader {
fn new(cell_based: bool, target_size: f32) -> Self {
Self {
magic: *FILE_MAGIC,
version: SC_FILE_VERSION,
main_header_offset: 0,
cell_based,
target_size,
library_index_offset: 0,
_reserved_1: [0; 20],
_reserved_2: [0; 3],
}
}
}
pub struct CellPayload {
pub original_index: usize,
pub library_size: usize,
pub bytes: Vec<u8>,
}
pub struct CellGeneSparseWriter {
header: SparseDataHeader,
writer: BufWriter<File>,
chunks_start_pos: u64,
cell_based: bool,
chunks_since_flush: usize,
flush_frequency: usize,
target_size: f32,
library_sizes: Vec<u64>,
version: u32,
codec: ChunkCodec,
cell_library_table: Option<Vec<u64>>,
}
impl CellGeneSparseWriter {
pub fn new<P: AsRef<Path>>(
path_f: P,
cell_based: bool,
total_cells: usize,
total_genes: usize,
target_size: f32,
) -> Result<Self, ScDataErrors> {
let file = File::create(path_f)?;
let mut writer = BufWriter::with_capacity(128 * 1024 * 1024, file);
let file_header = FileHeader::new(cell_based, target_size);
let file_header_enc = encode_to_vec(&file_header, config::standard())
.map_err(|_| ScDataErrors::HeaderEncodeFailed)?;
if file_header_enc.len() < 64 {
writer.write_all(&file_header_enc)?;
writer.write_all(&vec![0u8; 64 - file_header_enc.len()])?;
} else {
writer.write_all(&file_header_enc[..64])?;
}
writer.flush()?;
let chunks_start_pos = 64;
let flush_frequency = if cell_based {
CELL_FLUSH_FREQUENCY
} else {
GENE_FLUSH_FREQUENCY
};
let header = SparseDataHeader {
total_cells,
total_genes,
cell_based,
no_chunks: 0,
chunk_offsets: Vec::new(),
index_map: FxHashMap::default(),
};
Ok(Self {
header,
writer,
chunks_start_pos,
cell_based,
chunks_since_flush: 0_usize,
flush_frequency,
target_size,
library_sizes: Vec::new(),
version: SC_FILE_VERSION,
codec: ChunkCodec::Lz4,
cell_library_table: None,
})
}
pub fn new_v3<P: AsRef<Path>>(
path_f: P,
cell_based: bool,
total_cells: usize,
total_genes: usize,
target_size: f32,
) -> Result<Self, ScDataErrors> {
let mut writer = Self::new(path_f, cell_based, total_cells, total_genes, target_size)?;
writer.version = SC_FILE_VERSION_V3;
Ok(writer)
}
pub fn with_codec(mut self, codec: ChunkCodec) -> Self {
self.codec = codec;
self
}
pub fn format(&self) -> ChunkFormat {
ChunkFormat {
version: self.version,
codec: self.codec,
target_size: self.target_size,
}
}
pub fn set_cell_library_table(&mut self, table: Vec<u64>) -> Result<(), ScDataErrors> {
if self.version != SC_FILE_VERSION {
return Err(ScDataErrors::FileVersionMismatch {
expected: SC_FILE_VERSION,
found: self.version,
});
}
self.check_mode(false)?;
self.cell_library_table = Some(table);
Ok(())
}
fn check_mode(&self, want_cell_based: bool) -> Result<(), ScDataErrors> {
if self.cell_based == want_cell_based {
return Ok(());
}
let label = |b: bool| if b { "cell-based" } else { "gene-based" };
Err(ScDataErrors::ReaderModeMismatch {
actual: label(self.cell_based),
requested: label(want_cell_based),
})
}
fn append_compressed(
&mut self,
compressed: &[u8],
original_index: usize,
) -> Result<(), ScDataErrors> {
let current_pos = self.writer.stream_position()?;
let chunk_offset = current_pos - self.chunks_start_pos;
self.header.chunk_offsets.push(chunk_offset);
self.header
.index_map
.insert(original_index, self.header.no_chunks);
self.writer
.write_all(&(compressed.len() as u64).to_le_bytes())?;
self.writer.write_all(compressed)?;
self.header.no_chunks += 1;
self.chunks_since_flush += 1;
if self.chunks_since_flush >= self.flush_frequency {
self.writer.flush()?;
self.chunks_since_flush = 0;
}
Ok(())
}
fn append_cell(
&mut self,
compressed: &[u8],
original_index: usize,
library_size: usize,
) -> Result<(), ScDataErrors> {
self.append_compressed(compressed, original_index)?;
self.library_sizes.push(library_size as u64);
Ok(())
}
pub fn write_cell_chunk(&mut self, cell_chunk: CsrCellChunk) -> Result<(), ScDataErrors> {
self.check_mode(true)?;
let bytes = self.format().encode_cell(&cell_chunk)?;
self.append_cell(&bytes, cell_chunk.original_index, cell_chunk.library_size)
}
pub fn write_gene_chunk(&mut self, gene_chunk: CscGeneChunk) -> Result<(), ScDataErrors> {
self.check_mode(false)?;
let bytes = self.format().encode_gene(&gene_chunk, &[])?;
self.append_compressed(&bytes, gene_chunk.original_index)
}
pub fn write_compressed_gene_chunks(
&mut self,
chunks: &[(usize, Vec<u8>)],
) -> Result<(), ScDataErrors> {
self.check_mode(false)?;
self.append_compressed_batch(chunks)
}
pub fn write_compressed_cell_chunks(
&mut self,
chunks: &[CellPayload],
) -> Result<(), ScDataErrors> {
self.check_mode(true)?;
for chunk in chunks {
self.append_cell(&chunk.bytes, chunk.original_index, chunk.library_size)?;
}
Ok(())
}
fn append_compressed_batch(&mut self, chunks: &[(usize, Vec<u8>)]) -> Result<(), ScDataErrors> {
for (original_index, payload) in chunks {
self.append_compressed(payload, *original_index)?;
}
Ok(())
}
pub fn finalise(mut self) -> Result<(), ScDataErrors> {
let main_header_offset = self.writer.stream_position()?;
let header_data = encode_to_vec(&self.header, config::standard())
.map_err(|_| ScDataErrors::HeaderEncodeFailed)?;
let header_size = header_data.len() as u64;
self.writer.write_all(&header_size.to_le_bytes())?;
self.writer.write_all(&header_data)?;
let mut file_header = FileHeader::new(self.cell_based, self.target_size);
file_header.version = self.version;
file_header.main_header_offset = main_header_offset;
if self.cell_based {
file_header.library_index_offset = self.write_library_index()?;
} else if let Some(table) = self.cell_library_table.take() {
file_header.library_index_offset = self.write_library_table(&table)?;
}
self.writer.seek(SeekFrom::Start(0))?;
let file_header_enc = encode_to_vec(&file_header, config::standard())
.map_err(|_| ScDataErrors::HeaderEncodeFailed)?;
if file_header_enc.len() < 64 {
self.writer.write_all(&file_header_enc)?;
self.writer
.write_all(&vec![0u8; 64 - file_header_enc.len()])?;
} else {
self.writer.write_all(&file_header_enc[..64])?;
}
self.writer.flush()?;
Ok(())
}
fn write_library_index(&mut self) -> Result<u64, ScDataErrors> {
assert!(self.cell_based, "library-size index on a gene-based file");
assert_eq!(
self.library_sizes.len(),
self.header.no_chunks,
"library-size index has {} entries for {} cell chunks",
self.library_sizes.len(),
self.header.no_chunks
);
let offset = self.writer.stream_position()?;
self.writer
.write_all(&(self.library_sizes.len() as u64).to_le_bytes())?;
for library_size in &self.library_sizes {
self.writer.write_all(&library_size.to_le_bytes())?;
}
Ok(offset)
}
fn write_library_table(&mut self, table: &[u64]) -> Result<u64, ScDataErrors> {
let offset = self.writer.stream_position()?;
self.writer.write_all(&(table.len() as u64).to_le_bytes())?;
for library_size in table {
self.writer.write_all(&library_size.to_le_bytes())?;
}
Ok(offset)
}
pub fn update_header_no_cells(&mut self, no_cells: usize) {
self.header.total_cells = no_cells;
}
pub fn update_header_no_genes(&mut self, no_genes: usize) {
self.header.total_genes = no_genes;
}
}
pub const INDEX_DROPPED: u32 = u32::MAX;
const CELL_WRITE_BATCH: usize = 4_096;
pub fn dense_index_map(kept: &[usize]) -> Vec<u32> {
let mut map = vec![INDEX_DROPPED; kept.iter().max().map_or(0, |&g| g + 1)];
for (new, &old) in kept.iter().enumerate() {
map[old] = new as u32;
}
map
}
pub fn keep_set_index_map(cells_to_keep: &IndexSet<u32>) -> Vec<u32> {
let kept: Vec<usize> = cells_to_keep.iter().map(|&c| c as usize).collect();
dense_index_map(&kept)
}
pub fn dense_gene_map_from_options(mapping: &[Option<usize>]) -> Vec<u32> {
mapping
.iter()
.map(|g| g.map_or(INDEX_DROPPED, |g| g as u32))
.collect()
}
pub fn compress_cell_row(
row: &mut [(u32, u32)],
original_index: usize,
target_size: f32,
format: ChunkFormat,
) -> Result<(usize, CellPayload), ScDataErrors> {
if row.windows(2).any(|w| w[0].0 > w[1].0) {
row.sort_by_key(|&(g, _)| g);
}
let genes: Vec<u32> = row.iter().map(|&(g, _)| g).collect();
let counts: Vec<u32> = row.iter().map(|&(_, c)| c).collect();
let chunk = CsrCellChunk::from_data(&counts, &genes, original_index, target_size, true);
let (nnz, library_size) = chunk.get_qc_info();
let payload = CellPayload {
original_index,
library_size,
bytes: format.encode_cell(&chunk)?,
};
Ok((nnz, payload))
}
pub fn write_cell_rows(
rows: &mut [Vec<(u32, u32)>],
cell_offset: usize,
target_size: f32,
writer: &mut CellGeneSparseWriter,
) -> Result<(Vec<usize>, Vec<usize>), ScDataErrors> {
let mut nnz = Vec::with_capacity(rows.len());
let mut lib_size = Vec::with_capacity(rows.len());
let format = writer.format();
for (batch_idx, batch) in rows.chunks_mut(CELL_WRITE_BATCH).enumerate() {
let first = cell_offset + batch_idx * CELL_WRITE_BATCH;
let built = batch
.par_iter_mut()
.enumerate()
.map(|(k, row)| {
let mut row = std::mem::take(row);
compress_cell_row(&mut row, first + k, target_size, format)
})
.collect::<Result<Vec<_>, ScDataErrors>>()?;
let mut payloads = Vec::with_capacity(built.len());
for (nnz_i, payload) in built {
nnz.push(nnz_i);
lib_size.push(payload.library_size);
payloads.push(payload);
}
writer.write_compressed_cell_chunks(&payloads)?;
}
Ok((nnz, lib_size))
}
pub fn write_cell_chunks_built<F>(
n_chunks: usize,
build: F,
writer: &mut CellGeneSparseWriter,
) -> Result<(Vec<usize>, Vec<usize>), ScDataErrors>
where
F: Fn(usize) -> CsrCellChunk + Sync,
{
let mut nnz = Vec::with_capacity(n_chunks);
let mut lib_size = Vec::with_capacity(n_chunks);
let format = writer.format();
for start in (0..n_chunks).step_by(CELL_WRITE_BATCH) {
let end = (start + CELL_WRITE_BATCH).min(n_chunks);
let built = (start..end)
.into_par_iter()
.map(|k| {
let chunk = build(k);
let (nnz_i, lib_i) = chunk.get_qc_info();
Ok::<_, ScDataErrors>((
nnz_i,
CellPayload {
original_index: chunk.original_index,
library_size: lib_i,
bytes: format.encode_cell(&chunk)?,
},
))
})
.collect::<Result<Vec<_>, _>>()?;
let mut payloads = Vec::with_capacity(built.len());
for (nnz_i, payload) in built {
nnz.push(nnz_i);
lib_size.push(payload.library_size);
payloads.push(payload);
}
writer.write_compressed_cell_chunks(&payloads)?;
}
Ok((nnz, lib_size))
}
pub fn write_cell_chunks_parallel(
chunks: &[CsrCellChunk],
writer: &mut CellGeneSparseWriter,
) -> Result<(), ScDataErrors> {
let format = writer.format();
for batch in chunks.chunks(CELL_WRITE_BATCH) {
let payloads = batch
.par_iter()
.map(|chunk| {
Ok::<_, ScDataErrors>(CellPayload {
original_index: chunk.original_index,
library_size: chunk.library_size,
bytes: format.encode_cell(chunk)?,
})
})
.collect::<Result<Vec<_>, _>>()?;
writer.write_compressed_cell_chunks(&payloads)?;
}
Ok(())
}
pub trait SingleCellReading: Send + Sync {
fn read_cells_parallel(&self, indices: &[usize]) -> Result<Vec<CsrCellChunk>, ScDataErrors>;
fn read_gene_parallel(&self, indices: &[usize]) -> Result<Vec<CscGeneChunk>, ScDataErrors>;
fn read_gene_parallel_raw(&self, indices: &[usize]) -> Result<Vec<CscGeneChunk>, ScDataErrors> {
self.read_gene_parallel(indices)
}
fn read_cells_sparse<T>(
&self,
indices: &[usize],
data_layer: &DataLayerReturn,
) -> Result<CompressedSparseData2<T, f32>, ScDataErrors>
where
T: BixverseNumeric + FromPrimitive;
fn read_genes_sparse<T>(
&self,
indices: &[usize],
cells_to_keep: Option<&IndexSet<u32>>,
data_layer: &DataLayerReturn,
) -> Result<CompressedSparseData2<T, f32>, ScDataErrors>
where
T: BixverseNumeric + FromPrimitive;
fn get_header(&self) -> SparseDataHeader;
fn is_cell_based(&self) -> bool;
fn is_gene_based(&self) -> bool {
!self.is_cell_based()
}
fn target_size(&self) -> Option<f32> {
None
}
fn read_cell(&self, index: usize) -> Result<CsrCellChunk, ScDataErrors> {
self.read_cells_parallel(&[index])?
.into_iter()
.next()
.ok_or(ScDataErrors::ChunkIndexNotFound(index))
}
fn read_gene(&self, index: usize) -> Result<CscGeneChunk, ScDataErrors> {
self.read_gene_parallel(&[index])?
.into_iter()
.next()
.ok_or(ScDataErrors::ChunkIndexNotFound(index))
}
fn read_cells_range(
&self,
start: usize,
end: usize,
) -> Result<Vec<CsrCellChunk>, ScDataErrors> {
let indices: Vec<usize> = (start..end).collect();
self.read_cells_parallel(&indices)
}
fn get_all_cells(&self) -> Result<Vec<CsrCellChunk>, ScDataErrors> {
let indices: Vec<usize> = (0..self.get_header().total_cells).collect();
self.read_cells_parallel(&indices)
}
fn get_all_genes(&self) -> Result<Vec<CscGeneChunk>, ScDataErrors> {
let indices: Vec<usize> = (0..self.get_header().total_genes).collect();
self.read_gene_parallel(&indices)
}
fn read_gene_parallel_filtered(
&self,
indices: &[usize],
cells_to_keep: &IndexSet<u32>,
) -> Result<Vec<CscGeneChunk>, ScDataErrors> {
let mut chunks = self.read_gene_parallel(indices)?;
let cell_map = keep_set_index_map(cells_to_keep);
chunks
.par_iter_mut()
.for_each(|chunk| chunk.filter_selected_cells_dense(&cell_map));
Ok(chunks)
}
fn read_cell_library_sizes(&self, indices: &[usize]) -> Result<Vec<usize>, ScDataErrors> {
Ok(self
.read_cells_parallel(indices)?
.iter()
.map(|chunk| chunk.library_size)
.collect())
}
fn read_gene_nnz(&self, indices: &[usize]) -> Result<Vec<usize>, ScDataErrors> {
Ok(self
.read_gene_parallel(indices)?
.iter()
.map(|chunk| chunk.nnz)
.collect())
}
fn get_all_gene_nnz(&self) -> Result<Vec<usize>, ScDataErrors> {
let indices: Vec<usize> = (0..self.get_header().total_genes).collect();
self.read_gene_nnz(&indices)
}
fn get_clr_offsets(
&self,
indices: &[usize],
batch_size: Option<usize>,
) -> Result<Vec<f64>, ScDataErrors> {
if !self.is_cell_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "gene-based",
requested: "cell-based",
});
}
let batch_size = batch_size.unwrap_or(CELL_BATCH_SIZE);
let n_genes = self.get_header().total_genes as f64;
let mut offsets = Vec::with_capacity(indices.len());
for batch in indices.chunks(batch_size) {
let cells = self.read_cells_parallel(batch)?;
let batch_offsets: Vec<f64> = cells
.par_iter()
.map(|chunk| {
let inv_s = 1.0_f64 / chunk.library_size as f64;
let sum: f64 = chunk
.data_raw
.iter()
.map(|c| (c as f64 * inv_s).ln_1p())
.sum();
sum / n_genes
})
.collect();
offsets.extend_from_slice(&batch_offsets);
}
Ok(offsets)
}
}
pub fn peek_target_size<P: AsRef<Path>>(f_path: P) -> Result<Option<f32>, ScDataErrors> {
use std::io::Read;
let mut buffer = [0u8; 64];
File::open(f_path)?.read_exact(&mut buffer)?;
let (file_header, _) = decode_from_slice::<FileHeader, _>(&buffer, config::standard())
.map_err(|_| ScDataErrors::HeaderDecodeFailed)?;
Ok((file_header.target_size != 0.0).then_some(file_header.target_size))
}
pub fn peek_file_version<P: AsRef<Path>>(f_path: P) -> Result<u32, ScDataErrors> {
use std::io::Read;
let mut buffer = [0u8; 64];
File::open(f_path)?.read_exact(&mut buffer)?;
let (file_header, _) = decode_from_slice::<FileHeader, _>(&buffer, config::standard())
.map_err(|_| ScDataErrors::HeaderDecodeFailed)?;
if &file_header.magic != FILE_MAGIC {
return Err(ScDataErrors::HeaderDecodeFailed);
}
Ok(file_header.version)
}
pub struct ParallelSparseReader {
header: SparseDataHeader,
mmap: Arc<memmap2::Mmap>,
file: File,
chunks_start: u64,
target_size: f32,
library_index: Option<usize>,
identity_index_map: bool,
version: u32,
cell_lib: Option<Arc<Vec<f32>>>,
cell_norm: Option<Arc<Vec<u16>>>,
}
impl ParallelSparseReader {
pub fn new(f_path: &str) -> Result<Self, ScDataErrors> {
let file = File::open(f_path)?;
let file_size = file.metadata()?.len();
let mmap = unsafe {
let mut opts = MmapOptions::new();
if file_size <= 8 * 1024 * 1024 * 1024 {
opts.populate();
}
opts.map(&file)?
};
#[cfg(unix)]
mmap.advise(memmap2::Advice::Random)?;
if mmap.len() < 64 {
return Err(ScDataErrors::HeaderDecodeFailed);
}
let file_header_bytes = &mmap[0..64];
let (file_header, _) =
decode_from_slice::<FileHeader, _>(file_header_bytes, config::standard())
.map_err(|_| ScDataErrors::HeaderDecodeFailed)?;
if &file_header.magic != FILE_MAGIC {
return Err(ScDataErrors::HeaderDecodeFailed);
}
if file_header.version != SC_FILE_VERSION_V3 && file_header.version != SC_FILE_VERSION {
return Err(ScDataErrors::FileVersionMismatch {
expected: SC_FILE_VERSION,
found: file_header.version,
});
}
let main_header_offset = file_header.main_header_offset as usize;
if main_header_offset.saturating_add(8) > mmap.len() {
return Err(ScDataErrors::HeaderDecodeFailed);
}
let header_size = u64::from_le_bytes(
mmap[main_header_offset..main_header_offset + 8]
.try_into()
.expect("8-byte slice, bounds checked above"),
) as usize;
let header_end = main_header_offset
.saturating_add(8)
.saturating_add(header_size);
if header_end > mmap.len() {
return Err(ScDataErrors::HeaderDecodeFailed);
}
let header_bytes = &mmap[main_header_offset + 8..header_end];
let (header, _) =
decode_from_slice::<SparseDataHeader, _>(header_bytes, config::standard())
.map_err(|_| ScDataErrors::HeaderDecodeFailed)?;
let is_v4_gene = file_header.version == SC_FILE_VERSION && !header.cell_based;
let library_index = if is_v4_gene {
None
} else {
Self::check_library_index(&mmap, file_header.library_index_offset, &header)?
};
let cell_lib = if is_v4_gene {
Self::load_cell_library_table(&mmap, file_header.library_index_offset)?
} else {
None
};
let identity_index_map = header.index_map.len() == header.chunk_offsets.len()
&& header
.index_map
.iter()
.all(|(&original, &chunk)| original == chunk);
Ok(Self {
header,
mmap: Arc::new(mmap),
file,
chunks_start: 64,
target_size: file_header.target_size,
library_index,
identity_index_map,
version: file_header.version,
cell_norm: cell_lib
.as_ref()
.filter(|_| file_header.target_size != 0.0)
.map(|lib| Arc::new(gene_norm_table(lib, file_header.target_size))),
cell_lib,
})
}
fn load_cell_library_table(
mmap: &memmap2::Mmap,
offset: u64,
) -> Result<Option<Arc<Vec<f32>>>, ScDataErrors> {
if offset == 0 {
return Ok(None);
}
let offset = offset as usize;
if offset.saturating_add(8) > mmap.len() {
return Err(ScDataErrors::HeaderDecodeFailed);
}
let count = le_u64_at(mmap, offset) as usize;
let end = count
.checked_mul(8)
.and_then(|bytes| bytes.checked_add(offset + 8));
if end.is_none_or(|end| end > mmap.len()) {
return Err(ScDataErrors::HeaderDecodeFailed);
}
Ok(Some(Arc::new(
(0..count)
.map(|i| le_u64_at(mmap, offset + 8 + 8 * i) as f32)
.collect(),
)))
}
pub fn file_version(&self) -> u32 {
self.version
}
fn is_v4(&self) -> bool {
self.version == SC_FILE_VERSION
}
fn v4_ctx(&self) -> V4Ctx<'_> {
V4Ctx {
target_size: self.target_size,
cell_lib: self.cell_lib.as_deref().map(Vec::as_slice),
cell_norm: self.cell_norm.as_deref().map(Vec::as_slice),
is_cell: self.header.cell_based,
}
}
fn check_library_index(
mmap: &memmap2::Mmap,
offset: u64,
header: &SparseDataHeader,
) -> Result<Option<usize>, ScDataErrors> {
if offset == 0 {
return Ok(None);
}
let offset = offset as usize;
if !header.cell_based || offset.saturating_add(8) > mmap.len() {
return Err(ScDataErrors::HeaderDecodeFailed);
}
let count = le_u64_at(mmap, offset) as usize;
let end = count
.checked_mul(8)
.and_then(|bytes| bytes.checked_add(offset + 8));
if count != header.no_chunks || end.is_none_or(|end| end > mmap.len()) {
return Err(ScDataErrors::HeaderDecodeFailed);
}
Ok(Some(offset + 8))
}
fn library_sizes_from_index(
&self,
start: usize,
indices: &[usize],
) -> Result<Vec<usize>, ScDataErrors> {
indices
.iter()
.map(|&original_index| {
let chunk = self
.header
.index_map
.get(&original_index)
.copied()
.filter(|&chunk| chunk < self.header.no_chunks)
.ok_or(ScDataErrors::ChunkIndexNotFound(original_index))?;
Ok(le_u64_at(&self.mmap, start + 8 * chunk) as usize)
})
.collect()
}
#[cfg(test)]
pub(crate) fn has_library_index(&self) -> bool {
self.library_index.is_some()
}
fn chunk_index_of(&self, original_index: usize) -> Result<usize, ScDataErrors> {
if self.identity_index_map {
return if original_index < self.header.chunk_offsets.len() {
Ok(original_index)
} else {
Err(ScDataErrors::ChunkIndexNotFound(original_index))
};
}
self.header
.index_map
.get(&original_index)
.copied()
.ok_or(ScDataErrors::ChunkIndexNotFound(original_index))
}
pub fn count_norm_stored_chunks(&self) -> Result<usize, ScDataErrors> {
if !self.is_v4() {
return Err(ScDataErrors::FileVersionMismatch {
expected: SC_FILE_VERSION,
found: self.version,
});
}
let offsets = &self.header.chunk_offsets;
(0..offsets.len())
.into_par_iter()
.map(|c| {
let at = (self.chunks_start + offsets[c]) as usize + 8;
Ok(V4Header::parse(&self.mmap[at..])?.norm_stored() as usize)
})
.sum()
}
pub(crate) fn visit_cells_v4(
&self,
start: usize,
end: usize,
want_raw: bool,
mut f: impl FnMut(usize, &[u32], &[u32]),
) -> Result<(), ScDataErrors> {
if !self.is_cell_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "gene-based",
requested: "cell-based",
});
}
if !self.is_v4() {
return Err(ScDataErrors::FileVersionMismatch {
expected: SC_FILE_VERSION,
found: self.version,
});
}
self.prefetch(&(start..end).collect::<Vec<_>>());
let ctx = self.v4_ctx();
let mut s = V4Decoded::default();
for cell in start..end {
let (chunk, _) = self.locate_chunk(cell)?;
decode_v4(chunk, &ctx, want_raw, false, &mut s)?;
let raw = if want_raw { &s.raw[..s.idx.len()] } else { &[] };
f(cell, &s.idx, raw);
}
Ok(())
}
fn locate_chunk(&self, original_index: usize) -> Result<(&[u8], usize), ScDataErrors> {
let chunk_index = self.chunk_index_of(original_index)?;
let chunk_offset = (self.chunks_start + self.header.chunk_offsets[chunk_index]) as usize;
let compressed_size = u64::from_le_bytes(
self.mmap[chunk_offset..chunk_offset + 8]
.try_into()
.expect("8-byte slice by construction"),
) as usize;
Ok((
&self.mmap[chunk_offset + 8..chunk_offset + 8 + compressed_size],
chunk_offset,
))
}
fn decompress_chunk(&self, original_index: usize) -> Result<Vec<u8>, ScDataErrors> {
let (compressed, chunk_offset) = self.locate_chunk(original_index)?;
decompress_size_prepended(compressed)
.map_err(|_| ScDataErrors::ChunkDecompressionFailed(chunk_offset as u64))
}
fn decompress_chunk_into(
&self,
original_index: usize,
scratch: &mut Vec<u8>,
) -> Result<(), ScDataErrors> {
let (compressed, chunk_offset) = self.locate_chunk(original_index)?;
let (size, payload) = uncompressed_size(compressed)
.map_err(|_| ScDataErrors::ChunkDecompressionFailed(chunk_offset as u64))?;
scratch.resize(size, 0);
let written = decompress_into(payload, scratch)
.map_err(|_| ScDataErrors::ChunkDecompressionFailed(chunk_offset as u64))?;
scratch.truncate(written);
Ok(())
}
}
#[cfg(unix)]
const PREFETCH_BLOCK: usize = 8 * 1024 * 1024;
#[cfg(unix)]
const PREFETCH_MIN_BYTES: usize = if cfg!(test) { 1 } else { 16 * 1024 * 1024 };
#[cfg(unix)]
const PREFETCH_MERGE_GAP: usize = 256 * 1024;
#[cfg(unix)]
const PREFETCH_SAMPLES: usize = 32;
#[cfg(unix)]
const PREFETCH_SKIP_RESIDENT: f64 = 0.9;
impl ParallelSparseReader {
#[cfg(unix)]
fn prefetch(&self, indices: &[usize]) {
use std::os::unix::fs::FileExt;
let offsets = &self.header.chunk_offsets;
let mut spans: Vec<(usize, usize)> = indices
.iter()
.filter_map(|&i| {
let chunk = self.chunk_index_of(i).ok()?;
let start = (self.chunks_start + offsets[chunk]) as usize;
let end = offsets
.get(chunk + 1)
.map_or(start + 8, |&next| (self.chunks_start + next) as usize);
Some((start, end))
})
.collect();
if spans.iter().map(|&(s, e)| e - s).sum::<usize>() < PREFETCH_MIN_BYTES {
return;
}
spans.sort_unstable();
let mut ranges: Vec<(usize, usize)> = Vec::new();
for (start, end) in spans {
match ranges.last_mut() {
Some(last) if start <= last.1 + PREFETCH_MERGE_GAP => last.1 = last.1.max(end),
_ => ranges.push((start, end)),
}
}
let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }.max(4096) as usize;
let step = (ranges.len() / PREFETCH_SAMPLES).max(1);
let probed: Vec<f64> = ranges
.iter()
.step_by(step)
.filter(|&&(s, e)| s / page * page < self.mmap.len() && e > s)
.map(|&(s, e)| {
let s = s / page * page;
let e = e.min(self.mmap.len());
self.resident_fraction(s, e - s, page)
})
.collect();
if probed.is_empty()
|| probed.iter().sum::<f64>() / probed.len() as f64 >= PREFETCH_SKIP_RESIDENT
{
return;
}
let blocks: Vec<(usize, usize)> = ranges
.iter()
.flat_map(|&(start, end)| {
let start = start / page * page;
(start..end)
.step_by(PREFETCH_BLOCK)
.map(move |s| (s, (s + PREFETCH_BLOCK).min(end)))
})
.filter(|&(s, e)| s < self.mmap.len() && e > s)
.collect();
blocks.par_iter().for_each_init(
|| vec![0u8; PREFETCH_BLOCK],
|buf, &(start, end)| {
let end = end.min(self.mmap.len());
if self.resident_fraction(start, end - start, page) >= PREFETCH_SKIP_RESIDENT {
return;
}
let mut done = 0;
while done < end - start {
match self
.file
.read_at(&mut buf[..end - start - done], (start + done) as u64)
{
Ok(0) | Err(_) => break,
Ok(n) => done += n,
}
}
},
);
}
#[cfg(unix)]
fn resident_fraction(&self, start: usize, len: usize, page: usize) -> f64 {
let mut vec = vec![0u8; len.div_ceil(page)];
let rc = unsafe {
libc::mincore(
self.mmap.as_ptr().add(start) as *mut libc::c_void,
len,
vec.as_mut_ptr().cast(),
)
};
if rc != 0 {
return 0.0;
}
vec.iter().filter(|&&b| b & 1 != 0).count() as f64 / vec.len().max(1) as f64
}
#[cfg(not(unix))]
fn prefetch(&self, _indices: &[usize]) {}
}
const FUSED_PIECES_PER_THREAD: usize = 8;
#[inline(always)]
fn raw_to<T: FromPrimitive>(val: u32) -> Result<T, ScDataErrors> {
T::from_u32(val).ok_or(ScDataErrors::RawCountOverflow {
value: val,
target_type: std::any::type_name::<T>(),
})
}
pub(crate) struct SparseFragment<T> {
data: Option<Vec<T>>,
data_2: Option<Vec<f32>>,
indices: Vec<u32>,
lengths: Vec<usize>,
kept: Vec<(u32, u32)>,
}
impl<T: BixverseNumeric + FromPrimitive> SparseFragment<T> {
fn new(layers: (bool, bool), n_runs: usize) -> Self {
Self {
data: layers.0.then(Vec::new),
data_2: layers.1.then(Vec::new),
indices: Vec::new(),
lengths: Vec::with_capacity(n_runs),
kept: Vec::new(),
}
}
pub(crate) fn push_v4(
&mut self,
chunk: &[u8],
ctx: &V4Ctx,
scratch: &mut V4Decoded,
row_map: Option<&[u32]>,
) -> Result<(), ScDataErrors> {
let (want_raw, want_norm) = self.layers();
if row_map.is_some() {
decode_v4(chunk, ctx, want_raw, want_norm, scratch)?;
return self.push_run(&V4Run::new(scratch), row_map);
}
let h = decode_v4_into(
chunk,
ctx,
want_raw,
&mut scratch.bytes,
&mut scratch.raw,
&mut self.indices,
self.data_2.as_mut(),
)?;
if let Some(data) = self.data.as_mut() {
extend_raw_checked(data, scratch.raw.iter().copied())?;
}
self.lengths.push(h.nnz);
Ok(())
}
pub(crate) fn layers(&self) -> (bool, bool) {
(self.data.is_some(), self.data_2.is_some())
}
#[inline]
pub(crate) fn push_run(
&mut self,
run: &impl RunSource,
row_map: Option<&[u32]>,
) -> Result<(), ScDataErrors> {
let Some(map) = row_map else {
if let Some(data) = self.data.as_mut() {
run.extend_raw(data)?;
}
if let Some(data_2) = self.data_2.as_mut() {
run.extend_norm(data_2);
}
run.extend_indices(&mut self.indices);
self.lengths.push(run.nnz());
return Ok(());
};
self.kept.clear();
self.kept.extend(
(0..run.nnz()).filter_map(|i| match map.get(run.index_at(i) as usize) {
Some(&row) if row != INDEX_DROPPED => Some((row, i as u32)),
_ => None,
}),
);
if !self.kept.is_sorted_by_key(|&(row, _)| row) {
self.kept.sort_unstable_by_key(|&(row, _)| row);
}
if let Some(data) = self.data.as_mut() {
for &(_, i) in &self.kept {
data.push(raw_to(run.raw_at(i as usize))?);
}
}
if let Some(data_2) = self.data_2.as_mut() {
data_2.extend(self.kept.iter().map(|&(_, i)| run.norm_at(i as usize)));
}
self.indices.extend(self.kept.iter().map(|&(row, _)| row));
self.lengths.push(self.kept.len());
Ok(())
}
}
fn wanted_layers(data_layer: &DataLayerReturn) -> (bool, bool) {
(
matches!(
data_layer,
DataLayerReturn::Raw | DataLayerReturn::BothLayers
),
matches!(
data_layer,
DataLayerReturn::Norm | DataLayerReturn::BothLayers
),
)
}
pub(crate) fn read_fused<T, F>(
indices: &[usize],
data_layer: &DataLayerReturn,
cs_type: CompressedSparseFormat,
shape: (usize, usize),
fill: F,
) -> Result<CompressedSparseData2<T, f32>, ScDataErrors>
where
T: BixverseNumeric + FromPrimitive,
F: Fn(usize, &mut V4Decoded, &mut SparseFragment<T>) -> Result<(), ScDataErrors> + Sync,
{
let layers = wanted_layers(data_layer);
let n_pieces = rayon::current_num_threads().max(1) * FUSED_PIECES_PER_THREAD;
let piece = indices.len().div_ceil(n_pieces).max(1);
let fragments: Vec<SparseFragment<T>> = indices
.par_chunks(piece)
.map(|runs| {
let mut scratch = V4Decoded::default();
let mut fragment = SparseFragment::new(layers, runs.len());
for &run in runs {
fill(run, &mut scratch, &mut fragment)?;
}
Ok(fragment)
})
.collect::<Result<Vec<_>, ScDataErrors>>()?;
let nnz_offsets: Vec<usize> = std::iter::once(0)
.chain(fragments.iter().scan(0, |acc, f| {
*acc += f.indices.len();
Some(*acc)
}))
.collect();
let run_offsets: Vec<usize> = std::iter::once(0)
.chain(fragments.iter().scan(0, |acc, f| {
*acc += f.lengths.len();
Some(*acc)
}))
.collect();
let nnz = *nnz_offsets.last().expect("seeded with 0");
let mut data: Vec<T> = if layers.0 {
vec![T::default(); nnz]
} else {
Vec::new()
};
let mut data_2: Option<Vec<f32>> = layers.1.then(|| vec![0.0; nnz]);
let mut out_indices: Vec<u32> = vec![0; nnz];
let mut indptr: Vec<u32> = vec![0; indices.len() + 1];
let data_parts = split_at_offsets(&mut data, &nnz_offsets, layers.0);
let data_2_parts = split_at_offsets(
data_2.as_deref_mut().unwrap_or_default(),
&nnz_offsets,
layers.1,
);
let index_parts = split_at_offsets(&mut out_indices, &nnz_offsets, true);
let indptr_parts = split_at_offsets(&mut indptr[1..], &run_offsets, true);
fragments
.par_iter()
.zip(data_parts)
.zip(data_2_parts)
.zip(index_parts)
.zip(indptr_parts)
.zip(nnz_offsets.par_iter())
.for_each(|(((((fragment, d), d2), idx), ptr), &base)| {
if let Some(src) = fragment.data.as_ref() {
d.copy_from_slice(src);
}
if let Some(src) = fragment.data_2.as_ref() {
d2.copy_from_slice(src);
}
idx.copy_from_slice(&fragment.indices);
let mut offset = base;
for (out, &len) in ptr.iter_mut().zip(&fragment.lengths) {
offset += len;
*out = offset as u32;
}
});
Ok(CompressedSparseData2 {
data,
indices: out_indices,
indptr,
cs_type,
data_2,
shape,
})
}
fn split_at_offsets<'a, X>(buf: &'a mut [X], offsets: &[usize], active: bool) -> Vec<&'a mut [X]> {
let n_parts = offsets.len().saturating_sub(1);
if !active {
return (0..n_parts).map(|_| &mut [][..]).collect();
}
let mut parts = Vec::with_capacity(n_parts);
let mut rest = buf;
for w in offsets.windows(2) {
let (head, tail) = rest.split_at_mut(w[1] - w[0]);
parts.push(head);
rest = tail;
}
parts
}
pub(crate) fn cell_row_map(
cells_to_keep: Option<&IndexSet<u32>>,
total_cells: usize,
) -> (Option<Vec<u32>>, usize) {
match cells_to_keep {
Some(set) => (Some(keep_set_index_map(set)), set.len()),
None => (None, total_cells),
}
}
#[derive(Clone, Debug)]
pub struct CsrBatch {
pub data_raw: Option<Vec<f32>>,
pub data_norm: Option<Vec<f32>>,
pub indices: Vec<u32>,
pub indptr: Vec<u32>,
pub shape: (usize, usize),
}
impl ParallelSparseReader {
pub fn read_cells_csr(
&self,
indices: &[usize],
data_layer: &DataLayerReturn,
) -> Result<CsrBatch, ScDataErrors> {
let csr = self.read_cells_sparse::<f32>(indices, data_layer)?;
Ok(CsrBatch {
data_raw: wanted_layers(data_layer).0.then_some(csr.data),
data_norm: csr.data_2,
indices: csr.indices,
indptr: csr.indptr,
shape: csr.shape,
})
}
}
impl SingleCellReading for ParallelSparseReader {
fn read_cells_parallel(&self, indices: &[usize]) -> Result<Vec<CsrCellChunk>, ScDataErrors> {
if !self.is_cell_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "gene-based",
requested: "cell-based",
});
}
self.prefetch(indices);
if self.is_v4() {
let ctx = self.v4_ctx();
return indices
.par_iter()
.map_init(V4Decoded::default, |scratch, &original_index| {
let (chunk, _) = self.locate_chunk(original_index)?;
v4_cell_chunk(chunk, &ctx, scratch)
})
.collect();
}
indices
.par_iter()
.map(|&original_index| {
let decompressed = self.decompress_chunk(original_index)?;
CsrCellChunk::read_from_buffer(&decompressed)
})
.collect()
}
fn read_cells_sparse<T>(
&self,
indices: &[usize],
data_layer: &DataLayerReturn,
) -> Result<CompressedSparseData2<T, f32>, ScDataErrors>
where
T: BixverseNumeric + FromPrimitive,
{
if !self.is_cell_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "gene-based",
requested: "cell-based",
});
}
self.prefetch(indices);
read_fused(
indices,
data_layer,
CompressedSparseFormat::Csr,
(indices.len(), self.header.total_genes),
|cell, scratch, fragment| {
if self.is_v4() {
let (chunk, _) = self.locate_chunk(cell)?;
return fragment.push_v4(chunk, &self.v4_ctx(), scratch, None);
}
self.decompress_chunk_into(cell, &mut scratch.bytes)?;
let p = csr_payload(&scratch.bytes)?;
fragment.push_run(&p, None)
},
)
}
fn read_genes_sparse<T>(
&self,
indices: &[usize],
cells_to_keep: Option<&IndexSet<u32>>,
data_layer: &DataLayerReturn,
) -> Result<CompressedSparseData2<T, f32>, ScDataErrors>
where
T: BixverseNumeric + FromPrimitive,
{
if !self.is_gene_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "cell-based",
requested: "gene-based",
});
}
let (row_map, n_rows) = cell_row_map(cells_to_keep, self.header.total_cells);
self.prefetch(indices);
read_fused(
indices,
data_layer,
CompressedSparseFormat::Csc,
(n_rows, indices.len()),
|gene, scratch, fragment| {
if self.is_v4() {
let (chunk, _) = self.locate_chunk(gene)?;
return fragment.push_v4(chunk, &self.v4_ctx(), scratch, row_map.as_deref());
}
self.decompress_chunk_into(gene, &mut scratch.bytes)?;
let p = csc_payload(&scratch.bytes)?;
fragment.push_run(&p, row_map.as_deref())
},
)
}
fn read_gene_parallel(&self, indices: &[usize]) -> Result<Vec<CscGeneChunk>, ScDataErrors> {
if !self.is_gene_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "cell-based",
requested: "gene-based",
});
}
self.prefetch(indices);
if self.is_v4() {
let ctx = self.v4_ctx();
return indices
.par_iter()
.map_init(V4Decoded::default, |scratch, &original_index| {
let (chunk, _) = self.locate_chunk(original_index)?;
v4_gene_chunk(chunk, &ctx, scratch)
})
.collect();
}
indices
.par_iter()
.map(|&original_index| {
let decompressed = self.decompress_chunk(original_index)?;
CscGeneChunk::read_from_buffer(&decompressed)
})
.collect()
}
fn read_gene_parallel_raw(&self, indices: &[usize]) -> Result<Vec<CscGeneChunk>, ScDataErrors> {
if !self.is_v4() {
return self.read_gene_parallel(indices);
}
if !self.is_gene_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "cell-based",
requested: "gene-based",
});
}
self.prefetch(indices);
let ctx = self.v4_ctx();
indices
.par_iter()
.map_init(V4Decoded::default, |s, &original_index| {
let (chunk, _) = self.locate_chunk(original_index)?;
let h = decode_v4(chunk, &ctx, true, false, s)?;
Ok(CscGeneChunk {
data_raw: RawCounts::from_u32_auto(&s.raw[..h.nnz]),
data_norm: Vec::new(),
avg_exp: F16::from_bits(h.aux as u16),
nnz: h.nnz,
indices: std::mem::take(&mut s.idx),
original_index: h.original_index,
to_keep: h.to_keep,
})
})
.collect()
}
fn read_gene_parallel_filtered(
&self,
indices: &[usize],
cells_to_keep: &IndexSet<u32>,
) -> Result<Vec<CscGeneChunk>, ScDataErrors> {
if !self.is_gene_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "cell-based",
requested: "gene-based",
});
}
let cell_map = keep_set_index_map(cells_to_keep);
self.prefetch(indices);
if self.is_v4() {
let ctx = self.v4_ctx();
return indices
.par_iter()
.map_init(V4Decoded::default, |scratch, &original_index| {
let (chunk, _) = self.locate_chunk(original_index)?;
let mut chunk = v4_gene_chunk(chunk, &ctx, scratch)?;
chunk.filter_selected_cells_dense(&cell_map);
Ok(chunk)
})
.collect();
}
indices
.par_iter()
.map(|&original_index| {
let decompressed = self.decompress_chunk(original_index)?;
let mut chunk = CscGeneChunk::read_from_buffer(&decompressed)?;
chunk.filter_selected_cells_dense(&cell_map);
Ok(chunk)
})
.collect()
}
fn get_header(&self) -> SparseDataHeader {
self.header.clone()
}
fn read_cell_library_sizes(&self, indices: &[usize]) -> Result<Vec<usize>, ScDataErrors> {
if !self.is_cell_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "gene-based",
requested: "cell-based",
});
}
if let Some(start) = self.library_index {
return self.library_sizes_from_index(start, indices);
}
if self.is_v4() {
return indices
.iter()
.map(|&original_index| {
let (chunk, _) = self.locate_chunk(original_index)?;
Ok(V4Header::parse(chunk)?.aux as usize)
})
.collect();
}
indices
.par_iter()
.map(|&original_index| {
let decompressed = self.decompress_chunk(original_index)?;
Ok(u64::from_le_bytes([
decompressed[12],
decompressed[13],
decompressed[14],
decompressed[15],
decompressed[16],
decompressed[17],
decompressed[18],
decompressed[19],
]) as usize)
})
.collect()
}
fn read_gene_nnz(&self, indices: &[usize]) -> Result<Vec<usize>, ScDataErrors> {
if !self.is_gene_based() {
return Err(ScDataErrors::ReaderModeMismatch {
actual: "cell-based",
requested: "gene-based",
});
}
if self.is_v4() {
return indices
.iter()
.map(|&original_index| {
let (chunk, _) = self.locate_chunk(original_index)?;
Ok(V4Header::parse(chunk)?.nnz)
})
.collect();
}
indices
.par_iter()
.map(|&original_index| {
let decompressed = self.decompress_chunk(original_index)?;
Ok(u64::from_le_bytes([
decompressed[16],
decompressed[17],
decompressed[18],
decompressed[19],
decompressed[20],
decompressed[21],
decompressed[22],
decompressed[23],
]) as usize)
})
.collect()
}
fn is_cell_based(&self) -> bool {
self.header.cell_based
}
fn target_size(&self) -> Option<f32> {
(self.target_size != 0.0).then_some(self.target_size)
}
}
#[derive(Clone, Debug, Copy)]
pub enum DataLayerReturn {
BothLayers,
Raw,
Norm,
}
fn extend_raw_counts<T>(data: &mut Vec<T>, raw: &RawCounts) -> Result<(), ScDataErrors>
where
T: BixverseNumeric + FromPrimitive,
{
let convert = |val: u32| {
T::from_u32(val).ok_or(ScDataErrors::RawCountOverflow {
value: val,
target_type: std::any::type_name::<T>(),
})
};
match raw {
RawCounts::U16(v) => {
for &x in v {
data.push(convert(x as u32)?);
}
}
RawCounts::U32(v) => {
for &x in v {
data.push(convert(x)?);
}
}
}
Ok(())
}
pub fn from_gene_chunks<T>(
chunks: Vec<CscGeneChunk>,
data_layer: &DataLayerReturn,
n_cells: usize,
) -> Result<CompressedSparseData2<T, f32>, ScDataErrors>
where
T: BixverseNumeric + FromPrimitive,
{
let n_genes = chunks.len();
let keep_raw = matches!(
data_layer,
DataLayerReturn::BothLayers | DataLayerReturn::Raw
);
let keep_norm = matches!(
data_layer,
DataLayerReturn::BothLayers | DataLayerReturn::Norm
);
let total_nnz: usize = chunks.iter().map(|c| c.indices.len()).sum();
let mut data: Vec<T> = Vec::with_capacity(if keep_raw { total_nnz } else { 0 });
let mut data_2: Vec<f32> = Vec::with_capacity(if keep_norm { total_nnz } else { 0 });
let mut indices: Vec<u32> = Vec::with_capacity(total_nnz);
let mut indptr: Vec<usize> = Vec::with_capacity(n_genes + 1);
indptr.push(0);
for chunk in chunks {
if keep_raw {
extend_raw_counts(&mut data, &chunk.data_raw)?;
}
if keep_norm {
data_2.extend(chunk.data_norm.iter().map(|v| v.to_f32()));
}
indices.extend_from_slice(&chunk.indices);
indptr.push(indices.len());
}
Ok(CompressedSparseData2 {
data,
indices,
indptr: indptr.index_cast(),
cs_type: CompressedSparseFormat::Csc,
data_2: if keep_norm { Some(data_2) } else { None },
shape: (n_cells, n_genes),
})
}
pub fn from_cell_chunks<T>(
chunks: &[CsrCellChunk],
data_layer: &DataLayerReturn,
n_genes: usize,
) -> Result<CompressedSparseData2<T, f32>, ScDataErrors>
where
T: BixverseNumeric + FromPrimitive,
{
let n_cells = chunks.len();
let keep_raw = matches!(
data_layer,
DataLayerReturn::BothLayers | DataLayerReturn::Raw
);
let keep_norm = matches!(
data_layer,
DataLayerReturn::BothLayers | DataLayerReturn::Norm
);
let total_nnz: usize = chunks.iter().map(|c| c.indices.len()).sum();
let mut data: Vec<T> = Vec::with_capacity(if keep_raw { total_nnz } else { 0 });
let mut data_2: Vec<f32> = Vec::with_capacity(if keep_norm { total_nnz } else { 0 });
let mut indices: Vec<u32> = Vec::with_capacity(total_nnz);
let mut indptr: Vec<usize> = Vec::with_capacity(n_cells + 1);
indptr.push(0);
for chunk in chunks {
if keep_raw {
extend_raw_counts(&mut data, &chunk.data_raw)?;
}
if keep_norm {
data_2.extend(chunk.data_norm.iter().map(|v| v.to_f32()));
}
indices.extend_from_slice(&chunk.indices);
indptr.push(indices.len());
}
Ok(CompressedSparseData2 {
data,
indices,
indptr: indptr.index_cast(),
cs_type: CompressedSparseFormat::Csr,
data_2: if keep_norm { Some(data_2) } else { None },
shape: (n_cells, n_genes),
})
}
impl CscGeneChunk {
pub fn transform_to_clr(&mut self, size_factor: f32) {
let sf_inv = 1.0_f32 / size_factor;
let mut new_sum = 0.0_f32;
for v in self.data_norm.iter_mut() {
let new_val = (v.to_f32().exp_m1() * sf_inv).ln_1p();
new_sum += new_val;
*v = F16::from(f16::from_f32(new_val));
}
self.avg_exp = F16::from(f16::from_f32(new_sum));
}
}
impl CsrCellChunk {
#[allow(clippy::too_many_arguments)]
pub fn add_cells_scrublet(
cell1: &CsrCellChunk,
cell2: &CsrCellChunk,
hvg_indices: &FxHashSet<usize>,
hvg_lib_size_combined: usize,
target_size: f32,
log_transform: bool,
doublet_index: usize,
) -> Self {
let mut gene_counts: FxHashMap<u32, u32> = FxHashMap::default();
for i in 0..cell1.indices.len() {
if hvg_indices.contains(&(cell1.indices[i] as usize)) {
*gene_counts.entry(cell1.indices[i]).or_insert(0) += cell1.data_raw.get(i);
}
}
for i in 0..cell2.indices.len() {
if hvg_indices.contains(&(cell2.indices[i] as usize)) {
*gene_counts.entry(cell2.indices[i]).or_insert(0) += cell2.data_raw.get(i);
}
}
let mut gene_vec: Vec<(u32, u32)> = gene_counts.into_iter().collect();
gene_vec.sort_unstable_by_key(|&(gene, _)| gene);
let mut data_raw: Vec<u32> = Vec::with_capacity(gene_vec.len());
let mut data_norm = Vec::with_capacity(gene_vec.len());
let mut indices = Vec::with_capacity(gene_vec.len());
let norm_factor = target_size / hvg_lib_size_combined as f32;
for (gene, count) in gene_vec {
data_raw.push(count);
let normalised = if log_transform {
(count as f32 * norm_factor).ln_1p()
} else {
count as f32 * norm_factor
};
data_norm.push(F16::from(f16::from_f32(
normalised.clamp(-65504.0, 65504.0),
)));
indices.push(gene);
}
Self {
data_raw: RawCounts::from_u32_auto(&data_raw),
data_norm,
library_size: hvg_lib_size_combined,
indices,
original_index: doublet_index,
to_keep: true,
}
}
}
impl CsrCellChunk {
pub fn from_doublet_simulation(
indices: Vec<u32>,
raw_counts: Vec<u32>,
norm_values: Vec<f32>,
effective_lib_size: usize,
doublet_index: usize,
) -> Self {
debug_assert_eq!(indices.len(), raw_counts.len());
debug_assert_eq!(indices.len(), norm_values.len());
let data_raw = RawCounts::from_u32_auto(&raw_counts);
let data_norm: Vec<F16> = norm_values
.iter()
.map(|&v| F16::from(f16::from_f32(v)))
.collect();
Self {
data_raw,
data_norm,
library_size: effective_lib_size,
indices,
original_index: doublet_index,
to_keep: true,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use half::f16;
struct TempBin(std::path::PathBuf);
impl Drop for TempBin {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.0);
}
}
impl TempBin {
fn new(name: &str) -> Self {
Self(std::env::temp_dir().join(format!("bixverse_data_io_{name}.bin")))
}
fn path(&self) -> &str {
self.0.to_str().expect("temp path is valid UTF-8")
}
}
fn cell_chunk(raw: &[u32], indices: &[u32], original_index: usize) -> CsrCellChunk {
CsrCellChunk {
data_raw: RawCounts::from_u32_auto(raw),
data_norm: raw
.iter()
.map(|&v| F16::from(f16::from_f32(v as f32)))
.collect(),
library_size: raw.iter().map(|&v| v as usize).sum(),
indices: indices.to_vec(),
original_index,
to_keep: true,
}
}
fn gene_chunk(raw: &[u32], indices: &[u32], original_index: usize) -> CscGeneChunk {
CscGeneChunk {
data_raw: RawCounts::from_u32_auto(raw),
data_norm: raw
.iter()
.map(|&v| F16::from(f16::from_f32(v as f32)))
.collect(),
avg_exp: F16::from(f16::from_f32(1.0)),
nnz: raw.len(),
indices: indices.to_vec(),
original_index,
to_keep: true,
}
}
#[test]
fn test_filter_selected_cells_follows_keep_order() {
let keep: IndexSet<u32> = [7_u32, 1, 4, 900].into_iter().collect();
let mut by_set = gene_chunk(&[10, 30, 50, 70], &[1, 3, 5, 7], 0);
by_set.filter_selected_cells(&keep);
assert_eq!(by_set.indices, vec![0, 1]);
assert_eq!(by_set.data_raw.iter().collect::<Vec<u32>>(), vec![70, 10]);
assert_eq!(by_set.data_norm[0].to_f32(), 70.0);
assert_eq!(by_set.nnz, 2);
let mut by_map = gene_chunk(&[10, 30, 50, 70, 90], &[1, 3, 5, 7, 2_000], 0);
by_map.filter_selected_cells_dense(&keep_set_index_map(&keep));
assert_eq!(by_map.indices, by_set.indices);
assert_eq!(
by_map.data_raw.iter().collect::<Vec<u32>>(),
by_set.data_raw.iter().collect::<Vec<u32>>()
);
assert_eq!(by_map.data_norm, by_set.data_norm);
}
#[test]
fn test_cell_chunk_round_trip_preserves_fields() {
let raw = [1_u32, 7, 42, 65_535, 3];
let indices = [0_u32, 5, 9, 100_000, 300_000];
let chunk = cell_chunk(&raw, &indices, 17);
let mut buffer = Vec::new();
chunk.write_to_bytes(&mut buffer).expect("serialise");
let back = CsrCellChunk::read_from_buffer(&buffer).expect("deserialise");
assert_eq!(back.data_raw.iter().collect::<Vec<_>>(), raw.to_vec());
assert_eq!(back.indices, indices.to_vec());
assert_eq!(back.library_size, chunk.library_size);
assert_eq!(back.original_index, 17);
assert!(back.to_keep);
let bits: Vec<u16> = back.data_norm.iter().map(|v| v.to_bits()).collect();
let want: Vec<u16> = chunk.data_norm.iter().map(|v| v.to_bits()).collect();
assert_eq!(bits, want);
}
#[test]
fn test_gene_chunk_round_trip_preserves_fields() {
let raw = [2_u32, 0, 99, 70_000];
let indices = [1_u32, 4, 8, 12];
let chunk = gene_chunk(&raw, &indices, 3);
let mut buffer = Vec::new();
chunk.write_to_bytes(&mut buffer).expect("serialise");
let back = CscGeneChunk::read_from_buffer(&buffer).expect("deserialise");
assert_eq!(back.data_raw.iter().collect::<Vec<_>>(), raw.to_vec());
assert_eq!(back.indices, indices.to_vec());
assert_eq!(back.nnz, 4);
assert_eq!(back.original_index, 3);
assert_eq!(back.avg_exp.to_bits(), chunk.avg_exp.to_bits());
}
#[test]
fn test_gene_chunk_short_buffer_errors_instead_of_panicking() {
for len in CSR_CHUNK_HEADER_LEN..CSC_CHUNK_HEADER_LEN {
let buffer = vec![0_u8; len];
match CscGeneChunk::read_from_buffer(&buffer) {
Err(ScDataErrors::ChunkBufferTooSmall { expected, found }) => {
assert_eq!(expected, CSC_CHUNK_HEADER_LEN);
assert_eq!(found, len);
}
other => panic!("expected ChunkBufferTooSmall for len {len}, got {other:?}"),
}
}
}
#[test]
fn test_chunk_payload_truncation_is_caught() {
let chunk = cell_chunk(&[1, 2, 3], &[0, 1, 2], 0);
let mut buffer = Vec::new();
chunk.write_to_bytes(&mut buffer).expect("serialise");
buffer[0..4].copy_from_slice(&12_u32.to_le_bytes());
assert!(matches!(
CsrCellChunk::read_from_buffer(&buffer),
Err(ScDataErrors::ChunkPayloadTruncated { .. })
));
let gene = gene_chunk(&[1, 2, 3], &[0, 1, 2], 0);
let mut buffer = Vec::new();
gene.write_to_bytes(&mut buffer).expect("serialise");
buffer[8..12].copy_from_slice(&64_u32.to_le_bytes());
assert!(matches!(
CscGeneChunk::read_from_buffer(&buffer),
Err(ScDataErrors::ChunkPayloadTruncated { .. })
));
}
#[test]
fn test_chunk_round_trip_with_unaligned_index_offset() {
let raw = [1_u32, 2, 3];
let indices = [7_u32, 8, 9];
let chunk = cell_chunk(&raw, &indices, 0);
assert_eq!(chunk.data_raw.elem_size(), RAW_ELEM_U16);
let mut buffer = Vec::new();
chunk.write_to_bytes(&mut buffer).expect("serialise");
assert_eq!(buffer.len() % 4, 0);
let back = CsrCellChunk::read_from_buffer(&buffer).expect("deserialise");
assert_eq!(back.indices, indices.to_vec());
assert_eq!(back.data_raw.iter().collect::<Vec<_>>(), raw.to_vec());
}
#[test]
fn test_raw_counts_element_size_discriminant() {
let payload = [1_u8, 0, 2, 0];
let legacy = RawCounts::read_from_buffer(&payload, 2, 0).expect("legacy parses as u16");
assert_eq!(legacy.iter().collect::<Vec<_>>(), vec![1, 2]);
let explicit =
RawCounts::read_from_buffer(&payload, 2, RAW_ELEM_U16).expect("explicit u16 parses");
assert_eq!(explicit.iter().collect::<Vec<_>>(), vec![1, 2]);
assert!(matches!(
RawCounts::read_from_buffer(&payload, 2, 3),
Err(ScDataErrors::RawElemSizeInvalid(3))
));
}
#[test]
fn test_large_raw_count_survives_round_trip() {
let raw = [70_000_u32, 1];
let chunk = cell_chunk(&raw, &[0, 1], 0);
assert_eq!(chunk.data_raw.elem_size(), RAW_ELEM_U32);
let mut buffer = Vec::new();
chunk.write_to_bytes(&mut buffer).expect("serialise");
let back = CsrCellChunk::read_from_buffer(&buffer).expect("deserialise");
let csr = from_cell_chunks::<u32>(&[back], &DataLayerReturn::Raw, 2).expect("assemble");
assert_eq!(csr.data, vec![70_000_u32, 1]);
}
#[test]
fn test_narrow_target_type_reports_overflow() {
let chunk = gene_chunk(&[70_000, 1], &[0, 1], 0);
match from_gene_chunks::<u16>(vec![chunk], &DataLayerReturn::Raw, 2) {
Err(ScDataErrors::RawCountOverflow { value, .. }) => assert_eq!(value, 70_000),
other => panic!("expected RawCountOverflow, got {:?}", other.map(|_| ())),
}
}
#[test]
fn test_cell_based_file_round_trip() {
let temp = TempBin::new("cell_round_trip");
let chunks = vec![
cell_chunk(&[5, 70_000], &[0, 2], 0),
cell_chunk(&[1, 2, 3], &[0, 1, 2], 1),
];
let mut writer =
CellGeneSparseWriter::new(temp.path(), true, 2, 3, 1e4).expect("writer opens");
for chunk in chunks {
writer.write_cell_chunk(chunk).expect("write");
}
writer.finalise().expect("finalise");
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
assert_eq!(reader.target_size(), Some(1e4));
assert!(reader.is_cell_based());
let back = reader.read_cells_parallel(&[0, 1]).expect("read");
assert_eq!(back.len(), 2);
assert_eq!(back[0].data_raw.iter().collect::<Vec<_>>(), vec![5, 70_000]);
assert_eq!(back[1].indices, vec![0, 1, 2]);
}
fn csr_test_store(name: &str) -> (TempBin, ParallelSparseReader) {
csr_test_store_with(name, false)
}
fn csr_test_store_with(name: &str, legacy: bool) -> (TempBin, ParallelSparseReader) {
let temp = TempBin::new(name);
let chunks = vec![
cell_chunk(&[5, 70_000], &[0, 2], 0),
cell_chunk(&[1, 2, 3], &[0, 1, 2], 1),
cell_chunk(&[9], &[3], 2),
];
let ctor = if legacy {
CellGeneSparseWriter::new_v3
} else {
CellGeneSparseWriter::new
};
let mut writer = ctor(temp.path(), true, 3, 4, 1e4).expect("writer opens");
for chunk in chunks {
writer.write_cell_chunk(chunk).expect("write");
}
writer.finalise().expect("finalise");
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
(temp, reader)
}
#[test]
fn test_read_cells_csr_matches_chunk_path() {
let (_temp, reader) = csr_test_store("csr_matches_chunks");
let order = [2usize, 0, 1];
let batch = reader
.read_cells_csr(&order, &DataLayerReturn::BothLayers)
.expect("csr batch");
let chunks = reader.read_cells_parallel(&order).expect("chunk batch");
assert_eq!(batch.shape, (3, 4));
assert_eq!(batch.indptr.len(), order.len() + 1);
assert_eq!(batch.indptr[0], 0);
assert_eq!(
*batch.indptr.last().expect("non-empty"),
batch.indices.len() as u32
);
let raw = batch.data_raw.as_ref().expect("raw requested");
let norm = batch.data_norm.as_ref().expect("norm requested");
for (row, chunk) in chunks.iter().enumerate() {
let lo = batch.indptr[row] as usize;
let hi = batch.indptr[row + 1] as usize;
assert_eq!(&batch.indices[lo..hi], &chunk.indices[..]);
assert_eq!(
raw[lo..hi].to_vec(),
chunk.data_raw.iter().map(|v| v as f32).collect::<Vec<_>>()
);
assert_eq!(
norm[lo..hi].to_vec(),
chunk
.data_norm
.iter()
.map(|v| v.to_f32())
.collect::<Vec<_>>()
);
}
}
#[test]
fn test_read_cells_csr_layer_selection() {
let (_temp, reader) = csr_test_store("csr_layers");
let raw_only = reader
.read_cells_csr(&[0, 1, 2], &DataLayerReturn::Raw)
.expect("raw batch");
assert!(raw_only.data_raw.is_some());
assert!(raw_only.data_norm.is_none());
let norm_only = reader
.read_cells_csr(&[0, 1, 2], &DataLayerReturn::Norm)
.expect("norm batch");
assert!(norm_only.data_raw.is_none());
assert!(norm_only.data_norm.is_some());
}
#[test]
fn test_read_cells_csr_repeats_and_subsets() {
let (_temp, reader) = csr_test_store("csr_repeats");
let batch = reader
.read_cells_csr(&[1, 1], &DataLayerReturn::Raw)
.expect("csr batch");
assert_eq!(batch.shape.0, 2);
assert_eq!(batch.indptr, vec![0, 3, 6]);
assert_eq!(batch.indices, vec![0, 1, 2, 0, 1, 2]);
}
#[test]
fn test_read_cells_csr_rejects_unknown_cell() {
let (_temp, reader) = csr_test_store("csr_unknown_cell");
let err = reader
.read_cells_csr(&[99], &DataLayerReturn::Raw)
.expect_err("index is out of range");
assert!(matches!(err, ScDataErrors::ChunkIndexNotFound(99)));
}
#[test]
fn test_read_cells_csr_rejects_gene_major_store() {
let temp = TempBin::new("csr_gene_major");
let mut writer =
CellGeneSparseWriter::new(temp.path(), false, 2, 1, 1e4).expect("writer opens");
writer
.write_gene_chunk(gene_chunk(&[1, 2], &[0, 1], 0))
.expect("write");
writer.finalise().expect("finalise");
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
let err = reader
.read_cells_csr(&[0], &DataLayerReturn::Raw)
.expect_err("store is gene-major");
assert!(matches!(err, ScDataErrors::ReaderModeMismatch { .. }));
}
fn csc_test_store(name: &str) -> (TempBin, ParallelSparseReader) {
let temp = TempBin::new(name);
let chunks = vec![
gene_chunk(&[5, 70_000, 2], &[0, 3, 5], 0),
gene_chunk(&[1, 2, 3, 4, 5, 6], &[0, 1, 2, 3, 4, 5], 1),
gene_chunk(&[9, 11], &[1, 4], 2),
gene_chunk(&[], &[], 3),
];
let mut writer =
CellGeneSparseWriter::new(temp.path(), false, 6, 4, 1e4).expect("writer opens");
for chunk in chunks {
writer.write_gene_chunk(chunk).expect("write");
}
writer.finalise().expect("finalise");
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
(temp, reader)
}
fn assert_same_sparse<T: BixverseNumeric + std::fmt::Debug + PartialEq>(
got: &CompressedSparseData2<T, f32>,
want: &CompressedSparseData2<T, f32>,
) {
assert_eq!(got.shape, want.shape);
assert_eq!(got.cs_type.is_csc(), want.cs_type.is_csc());
assert_eq!(got.indptr, want.indptr);
assert_eq!(got.indices, want.indices);
assert_eq!(got.data, want.data);
assert_eq!(got.data_2, want.data_2);
}
#[test]
fn test_read_genes_sparse_matches_chunk_path() {
let (_temp, reader) = csc_test_store("csc_matches_chunks");
let genes = [2usize, 0, 3, 1, 0];
let subsets: Vec<Option<IndexSet<u32>>> = vec![
None,
Some([0_u32, 3, 4, 5].into_iter().collect()),
Some([5_u32, 1, 3].into_iter().collect()),
Some(IndexSet::new()),
];
for layer in [
DataLayerReturn::Raw,
DataLayerReturn::Norm,
DataLayerReturn::BothLayers,
] {
for subset in &subsets {
let (chunks, n_cells) = match subset {
Some(set) => (
reader
.read_gene_parallel_filtered(&genes, set)
.expect("chunks"),
set.len(),
),
None => (reader.read_gene_parallel(&genes).expect("chunks"), 6),
};
let want = from_gene_chunks::<u32>(chunks, &layer, n_cells).expect("reference");
let got = reader
.read_genes_sparse::<u32>(&genes, subset.as_ref(), &layer)
.expect("fused");
assert_same_sparse(&got, &want);
}
}
}
#[test]
fn test_read_cells_sparse_matches_chunk_path() {
let (_temp, reader) = csr_test_store("csr_sparse_matches_chunks");
let order = [2usize, 0, 1, 0];
for layer in [
DataLayerReturn::Raw,
DataLayerReturn::Norm,
DataLayerReturn::BothLayers,
] {
let chunks = reader.read_cells_parallel(&order).expect("chunks");
let want = from_cell_chunks::<u32>(&chunks, &layer, 4).expect("reference");
let got = reader
.read_cells_sparse::<u32>(&order, &layer)
.expect("fused");
assert_same_sparse(&got, &want);
}
}
#[test]
fn test_read_genes_sparse_reports_overflow() {
let (_temp, reader) = csc_test_store("csc_overflow");
let keep: IndexSet<u32> = [3_u32].into_iter().collect();
for cells in [None, Some(&keep)] {
match reader.read_genes_sparse::<u16>(&[0], cells, &DataLayerReturn::Raw) {
Err(ScDataErrors::RawCountOverflow { value, .. }) => assert_eq!(value, 70_000),
other => panic!("expected RawCountOverflow, got {:?}", other.map(|_| ())),
}
}
let keep: IndexSet<u32> = [0_u32, 5].into_iter().collect();
let csc = reader
.read_genes_sparse::<u16>(&[0], Some(&keep), &DataLayerReturn::Raw)
.expect("fits u16");
assert_eq!(csc.data, vec![5, 2]);
}
#[test]
fn test_read_genes_sparse_rejects_cell_major_store() {
let (_temp, reader) = csr_test_store("csc_cell_major");
let err = reader
.read_genes_sparse::<f32>(&[0], None, &DataLayerReturn::Raw)
.expect_err("store is cell-major");
assert!(matches!(err, ScDataErrors::ReaderModeMismatch { .. }));
}
#[test]
fn test_decompress_chunk_into_matches_allocating_path() {
let (_temp, reader) = csr_test_store_with("csr_scratch", true);
let mut scratch = vec![0xAAu8; 4096];
for cell in 0..3 {
reader
.decompress_chunk_into(cell, &mut scratch)
.expect("scratch decompress");
let owned = reader.decompress_chunk(cell).expect("owned decompress");
assert_eq!(scratch, owned);
}
}
#[test]
fn test_identity_index_map_is_detected() {
let (_temp, reader) = csr_test_store("csr_identity_map");
assert!(reader.identity_index_map);
assert_eq!(reader.chunk_index_of(2).expect("in range"), 2);
}
fn patch_file_header(path: &str, edit: impl FnOnce(&mut FileHeader)) {
let mut bytes = std::fs::read(path).expect("read back");
let (mut header, _) = decode_from_slice::<FileHeader, _>(&bytes[0..64], config::standard())
.expect("header decodes");
edit(&mut header);
let encoded = encode_to_vec(&header, config::standard()).expect("header encodes");
let mut slot = [0_u8; 64];
slot[..encoded.len()].copy_from_slice(&encoded);
bytes[0..64].copy_from_slice(&slot);
std::fs::write(path, bytes).expect("write back");
}
fn write_valid_file(path: &str) {
let mut writer = CellGeneSparseWriter::new(path, true, 1, 3, 1e4).expect("writer opens");
writer
.write_cell_chunk(cell_chunk(&[1, 2], &[0, 1], 0))
.expect("write");
writer.finalise().expect("finalise");
}
#[test]
fn test_reader_rejects_a_version_mismatch() {
let temp = TempBin::new("version_mismatch");
write_valid_file(temp.path());
patch_file_header(temp.path(), |h| h.version = SC_FILE_VERSION + 1);
match ParallelSparseReader::new(temp.path()) {
Err(ScDataErrors::FileVersionMismatch { expected, found }) => {
assert_eq!(expected, SC_FILE_VERSION);
assert_eq!(found, SC_FILE_VERSION + 1);
}
Err(other) => panic!("expected a version mismatch, got {other:?}"),
Ok(_) => panic!("a mismatched version must not open"),
}
}
#[test]
fn test_reader_rejects_a_bad_magic() {
let temp = TempBin::new("bad_magic");
write_valid_file(temp.path());
patch_file_header(temp.path(), |h| h.magic = *b"NOTOURS!");
assert!(
ParallelSparseReader::new(temp.path()).is_err(),
"a wrong magic string must not open"
);
}
#[test]
fn test_reader_rejects_a_main_header_offset_past_eof() {
let temp = TempBin::new("offset_past_eof");
write_valid_file(temp.path());
patch_file_header(temp.path(), |h| h.main_header_offset = u64::MAX / 2);
assert!(
ParallelSparseReader::new(temp.path()).is_err(),
"an out-of-range main header offset must not open"
);
}
#[test]
fn test_reader_rejects_a_truncated_file() {
let temp = TempBin::new("truncated");
write_valid_file(temp.path());
let bytes = std::fs::read(temp.path()).expect("read back");
std::fs::write(temp.path(), &bytes[..bytes.len() / 2]).expect("truncate");
assert!(
ParallelSparseReader::new(temp.path()).is_err(),
"a truncated file must not open"
);
}
#[test]
fn test_gene_based_file_round_trip() {
let temp = TempBin::new("gene_round_trip");
let mut writer =
CellGeneSparseWriter::new(temp.path(), false, 4, 2, 1e5).expect("writer opens");
writer
.write_gene_chunk(gene_chunk(&[9, 8], &[0, 3], 0))
.expect("write");
writer
.write_gene_chunk(gene_chunk(&[100_000], &[2], 1))
.expect("write");
writer.finalise().expect("finalise");
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
assert_eq!(reader.target_size(), Some(1e5));
assert!(reader.is_gene_based());
let back = reader.read_gene_parallel(&[0, 1]).expect("read");
assert_eq!(back[0].data_raw.iter().collect::<Vec<_>>(), vec![9, 8]);
assert_eq!(back[1].data_raw.iter().collect::<Vec<_>>(), vec![100_000]);
}
#[test]
fn test_writer_rejects_wrong_chunk_orientation() {
let temp = TempBin::new("mode_mismatch");
let mut writer =
CellGeneSparseWriter::new(temp.path(), true, 1, 1, 0.0).expect("writer opens");
match writer.write_gene_chunk(gene_chunk(&[1], &[0], 0)) {
Err(ScDataErrors::ReaderModeMismatch { actual, requested }) => {
assert_eq!(actual, "cell-based");
assert_eq!(requested, "gene-based");
}
other => panic!("expected ReaderModeMismatch, got {other:?}"),
}
}
#[test]
fn test_raw_only_file_reports_no_target_size() {
let temp = TempBin::new("no_target_size");
let mut writer =
CellGeneSparseWriter::new(temp.path(), true, 1, 2, 0.0).expect("writer opens");
writer
.write_cell_chunk(cell_chunk(&[1, 2], &[0, 1], 0))
.expect("write");
writer.finalise().expect("finalise");
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
assert_eq!(reader.target_size(), None);
assert_eq!(peek_target_size(temp.path()).expect("peek"), None);
}
#[derive(Encode, Decode, Serialize, Deserialize)]
struct LegacyFileHeader {
magic: [u8; 8],
version: u32,
main_header_offset: u64,
cell_based: bool,
_reserved_1: [u8; 32],
_reserved_2: [u8; 3],
}
fn encode_slot<T: Serialize>(header: &T) -> [u8; 64] {
let encoded = encode_to_vec(header, config::standard()).expect("encode");
assert!(encoded.len() <= 64, "header must fit the fixed slot");
let mut slot = [0_u8; 64];
slot[..encoded.len()].copy_from_slice(&encoded);
slot
}
#[test]
fn test_legacy_header_decodes_with_zero_target_size() {
let legacy = LegacyFileHeader {
magic: *b"SCRNASEQ",
version: SC_FILE_VERSION_V3,
main_header_offset: 123_456,
cell_based: true,
_reserved_1: [0; 32],
_reserved_2: [0; 3],
};
let slot = encode_slot(&legacy);
let (decoded, _) =
decode_from_slice::<FileHeader, _>(&slot, config::standard()).expect("decode as v3");
assert_eq!(decoded.magic, *b"SCRNASEQ");
assert_eq!(decoded.version, SC_FILE_VERSION_V3);
assert_eq!(decoded.main_header_offset, 123_456);
assert!(decoded.cell_based);
assert_eq!(decoded.target_size, 0.0);
}
#[test]
fn test_new_header_still_decodes_on_the_old_layout() {
let mut header = FileHeader::new(false, 1e4);
header.main_header_offset = 987_654_321;
let slot = encode_slot(&header);
let (decoded, _) = decode_from_slice::<LegacyFileHeader, _>(&slot, config::standard())
.expect("decode as legacy");
assert_eq!(decoded.magic, *b"SCRNASEQ");
assert_eq!(decoded.version, SC_FILE_VERSION);
assert_eq!(decoded.main_header_offset, 987_654_321);
assert!(!decoded.cell_based);
}
#[test]
fn test_target_size_survives_header_round_trip() {
let slot = encode_slot(&FileHeader::new(true, 1e5));
let (decoded, _) =
decode_from_slice::<FileHeader, _>(&slot, config::standard()).expect("decode");
assert_eq!(decoded.target_size, 1e5);
}
#[derive(Encode, Decode, Serialize, Deserialize)]
struct TargetSizeFileHeader {
magic: [u8; 8],
version: u32,
main_header_offset: u64,
cell_based: bool,
target_size: f32,
_reserved_1: [u8; 28],
_reserved_2: [u8; 3],
}
#[test]
fn test_pre_index_header_decodes_with_no_library_index() {
let old = TargetSizeFileHeader {
magic: *b"SCRNASEQ",
version: SC_FILE_VERSION_V3,
main_header_offset: 123_456,
cell_based: true,
target_size: 1e4,
_reserved_1: [0; 28],
_reserved_2: [0; 3],
};
let slot = encode_slot(&old);
let (decoded, _) =
decode_from_slice::<FileHeader, _>(&slot, config::standard()).expect("decode");
assert_eq!(decoded.main_header_offset, 123_456);
assert_eq!(decoded.target_size, 1e4);
assert_eq!(decoded.library_index_offset, 0);
}
#[test]
fn test_header_with_library_index_decodes_on_the_pre_index_layout() {
let mut header = FileHeader::new(true, 1e4);
header.main_header_offset = 987_654_321;
header.library_index_offset = u64::MAX;
let slot = encode_slot(&header);
let (decoded, _) = decode_from_slice::<TargetSizeFileHeader, _>(&slot, config::standard())
.expect("decode as pre-index");
assert_eq!(decoded.magic, *b"SCRNASEQ");
assert_eq!(decoded.version, SC_FILE_VERSION);
assert_eq!(decoded.main_header_offset, 987_654_321);
assert!(decoded.cell_based);
assert_eq!(decoded.target_size, 1e4);
}
fn read_file_header(path: &str) -> FileHeader {
let bytes = std::fs::read(path).expect("read back");
decode_from_slice::<FileHeader, _>(&bytes[0..64], config::standard())
.expect("header decodes")
.0
}
fn write_index_test_file(path: &str, compressed: bool) {
let chunks = vec![
cell_chunk(&[5, 70_000], &[0, 2], 2),
cell_chunk(&[1, 2, 3], &[0, 1, 2], 0),
cell_chunk(&[9], &[1], 1),
];
let mut writer = CellGeneSparseWriter::new(path, true, 3, 3, 1e4).expect("writer opens");
if compressed {
write_cell_chunks_parallel(&chunks, &mut writer).expect("write");
} else {
for chunk in chunks {
writer.write_cell_chunk(chunk).expect("write");
}
}
writer.finalise().expect("finalise");
}
fn library_sizes_both_ways(
reader: &ParallelSparseReader,
indices: &[usize],
) -> (Vec<usize>, Vec<usize>) {
let answered = reader.read_cell_library_sizes(indices).expect("read");
let stored = reader
.read_cells_parallel(indices)
.expect("read chunks")
.iter()
.map(|chunk| chunk.library_size)
.collect();
(answered, stored)
}
#[test]
fn test_library_index_matches_chunk_headers() {
for compressed in [false, true] {
let temp = TempBin::new(&format!("library_index_{compressed}"));
write_index_test_file(temp.path(), compressed);
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
assert!(reader.library_index.is_some());
let (answered, stored) = library_sizes_both_ways(&reader, &[1, 2, 0, 2]);
assert_eq!(answered, vec![9, 70_005, 6, 70_005]);
assert_eq!(answered, stored);
}
}
#[test]
fn test_library_sizes_fall_back_without_an_index() {
let temp = TempBin::new("library_index_absent");
write_index_test_file(temp.path(), false);
patch_file_header(temp.path(), |h| h.library_index_offset = 0);
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
assert!(reader.library_index.is_none());
let (answered, _) = library_sizes_both_ways(&reader, &[0, 1, 2]);
assert_eq!(answered, vec![6, 9, 70_005]);
}
#[test]
fn test_reader_rejects_a_library_index_past_eof() {
let temp = TempBin::new("library_index_past_eof");
write_valid_file(temp.path());
let len = std::fs::metadata(temp.path()).expect("stat").len();
patch_file_header(temp.path(), |h| h.library_index_offset = len);
assert!(matches!(
ParallelSparseReader::new(temp.path()),
Err(ScDataErrors::HeaderDecodeFailed)
));
}
#[test]
fn test_reader_rejects_a_library_index_with_the_wrong_count() {
let temp = TempBin::new("library_index_wrong_count");
write_valid_file(temp.path());
let main_header_offset = read_file_header(temp.path()).main_header_offset;
patch_file_header(temp.path(), |h| h.library_index_offset = main_header_offset);
assert!(matches!(
ParallelSparseReader::new(temp.path()),
Err(ScDataErrors::HeaderDecodeFailed)
));
}
#[test]
fn test_library_index_on_the_ingest_write_path() {
let temp = TempBin::new("library_index_ingest");
let mut rows: Vec<Vec<(u32, u32)>> = (0..300_u32)
.map(|c| {
(0..20_u32)
.filter(|g| (c * 7 + g * 3) % 5 == 0)
.map(|g| (g, 1 + (c + g) % 4 + if c == 7 { 70_000 } else { 0 }))
.collect()
})
.collect();
let mut writer =
CellGeneSparseWriter::new(temp.path(), true, 300, 20, 1e4).expect("writer opens");
let (_, reported) = write_cell_rows(&mut rows, 0, 1e4, &mut writer).expect("write");
writer.finalise().expect("finalise");
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
assert!(reader.library_index.is_some());
let (answered, stored) = library_sizes_both_ways(&reader, &(0..300).collect::<Vec<_>>());
assert_eq!(answered, stored);
assert_eq!(answered, reported);
}
#[test]
fn test_gene_based_file_has_no_library_index() {
let temp = TempBin::new("library_index_gene_file");
let mut writer =
CellGeneSparseWriter::new(temp.path(), false, 2, 1, 1e4).expect("writer opens");
writer
.write_gene_chunk(gene_chunk(&[1, 2], &[0, 1], 0))
.expect("write");
writer.finalise().expect("finalise");
assert_eq!(read_file_header(temp.path()).library_index_offset, 0);
let reader = ParallelSparseReader::new(temp.path()).expect("reader opens");
assert!(reader.library_index.is_none());
}
}