Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ edition = "2021"
readme = "README.md"
# Own version (decoupled from workspace.version): data-beans evolves its CLI /
# QC API independently of the shared utility crates.
version = "0.9.0"
version = "0.9.1"
rust-version = "1.91"
# Ship only the sources, README and license: nothing else in the working tree
# (data stores, saved name lists, scratch files) can end up on crates.io.
Expand Down
97 changes: 62 additions & 35 deletions src/handlers/merging.rs
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,8 @@ pub struct MergeBackendArgs {
long,
required = true,
help = "Output file header",
long_help = "Output file header: {output}.{backend} and {output}.batch.gz.\n\
long_help = "Output file header: {output}.{backend} and {output}.batch.gz,\n\
{output} without a backend suffix it was given.\n\
The zarr backend produces {output}.zarr.zip by default;\n\
pass --no-zip to keep a {output}.zarr directory instead.\n\
The backend will contain everything.\n\
Expand Down Expand Up @@ -129,7 +130,8 @@ pub struct MergeMtxArgs {
long,
required = true,
help = "Output file header",
long_help = "Output file header: {output}.{backend} and {output}.batch.gz.\n\
long_help = "Output file header: {output}.{backend} and {output}.batch.gz,\n\
{output} without a backend suffix it was given.\n\
The zarr backend produces {output}.zarr.zip by default;\n\
pass --no-zip to keep a {output}.zarr directory instead.\n\
The backend will contain all merged data.\n\
Expand Down Expand Up @@ -262,6 +264,20 @@ pub fn run_merge_backend(args: &MergeBackendArgs) -> anyhow::Result<()> {
let num_batches = args.data_files.len();
info!("merging over {} batches ...", num_batches);

let effective_output = apply_zip_flag(&args.output, args.zip, &args.backend);
let (backend, backend_file) =
resolve_backend_file(&effective_output, Some(args.backend.clone()))?;
// The output is cleared before the inputs are read and zipped over its
// name after, so it must not be one of them.
for input in &args.data_files {
for out in [&*effective_output, &*backend_file] {
anyhow::ensure!(
!same_file(input, out),
"the output {out} is the input {input}: give -o another name"
);
}
}

let batch_names = generate_unique_batch_names(&args.data_files)?;
info!("Batch names: {:?}", batch_names);

Expand Down Expand Up @@ -299,11 +315,17 @@ pub fn run_merge_backend(args: &MergeBackendArgs) -> anyhow::Result<()> {
.num_non_zeros()
.ok_or_else(|| anyhow::anyhow!("missing nnz in {}", data_file))?;

if row_names.is_empty() {
if batch_idx == 0 {
row_names = data.row_names()?;
} else {
info!("checking if the row names are consistent");
assert_eq!(row_names, data.row_names()?);
anyhow::ensure!(
row_names == data.row_names()?,
"{} has other row names than {}: merge-backend needs the same rows, in order; \
subset them to common rows first",
data_file,
args.data_files[0]
);
}

let _names = data.column_names()?;
Expand Down Expand Up @@ -335,10 +357,6 @@ pub fn run_merge_backend(args: &MergeBackendArgs) -> anyhow::Result<()> {
total_ncol, num_batches, total_nnz
);

let effective_output = apply_zip_flag(&args.output, args.zip, &args.backend);
let (backend, backend_file) =
resolve_backend_file(&effective_output, Some(args.backend.clone()))?;

if std::path::Path::new(backend_file.as_ref()).exists() {
info!(
"This existing backend file '{}' will be deleted",
Expand Down Expand Up @@ -430,21 +448,41 @@ pub fn run_merge_backend(args: &MergeBackendArgs) -> anyhow::Result<()> {
run_squeeze(&squeeze_args)?;
}

// do the batch mapping at the end
let batch_memb_file = format!("{}.batch.gz", args.output);
let data = open_sparse_matrix(&backend_file, &backend)?;
let default_batch = basename(&args.output)?;
let column_batch_names = data
write_batches_and_finalize(&effective_output, &backend_file, &backend, &batch_map)?;
info!("done");
Ok(())
}

/// Whether paths `a` and `b` name the same file: compared resolved when
/// they exist, as written (made absolute) when not.
fn same_file(a: &str, b: &str) -> bool {
let resolve = |p: &str| std::fs::canonicalize(p).or_else(|_| std::path::absolute(p));
matches!((resolve(a), resolve(b)), (Ok(x), Ok(y)) if x == y)
}

/// The end of a merge: label each column of the merged backend at
/// `backend_file` with its batch (from `batch_map`; a column it misses takes
/// the output's own name) in `{output}.batch.gz`, the output's name without
/// its backend suffix; then finalize the backend as `effective_output`, the
/// output resolved (a `.zarr.zip`, unless `--no-zip`).
pub(crate) fn write_batches_and_finalize(
effective_output: &str,
backend_file: &str,
backend: &SparseIoBackend,
batch_map: &HashMap<Box<str>, Box<str>>,
) -> anyhow::Result<()> {
let stem = strip_backend_suffix(effective_output);
// The name itself, dots and all (`sample.v2`), not cut at its last dot.
let default_batch: Box<str> = std::path::Path::new(stem)
.file_name()
.map_or_else(|| stem.into(), |n| n.to_string_lossy().into());
let labels: Vec<Box<str>> = open_sparse_matrix(backend_file, backend)?
.column_names()?
.iter()
.map(|k| batch_map.get(k).unwrap_or(&default_batch).clone())
.collect::<Vec<_>>();

write_lines(&column_batch_names, &batch_memb_file)?;

finalize_zarr_output(&backend_file, &args.output)?;
info!("done");
Ok(())
.collect();
write_lines(&labels, &format!("{stem}.batch.gz"))?;
finalize_zarr_output(backend_file, effective_output)
}

pub fn run_merge_mtx(args: &MergeMtxArgs) -> anyhow::Result<()> {
Expand Down Expand Up @@ -647,10 +685,10 @@ pub fn run_merge_mtx(args: &MergeMtxArgs) -> anyhow::Result<()> {

let backend = args.backend.clone();
let effective_output = apply_zip_flag(&args.output, args.zip, &backend);
let output = args.output.clone();
let batch_memb_file = (output.to_string() + ".batch.gz").into_boxed_str();

let (_, backend_file) = resolve_backend_file(&effective_output, Some(backend.clone()))?;
// The backend the output's name resolves to (an `.h5` name, or a zarr
// fallback for hdf5 without the feature), not the flag's.
let (backend, backend_file) = resolve_backend_file(&effective_output, Some(backend))?;
let backend_file: String = backend_file.to_string();

if std::path::Path::new(&backend_file).exists() {
Expand Down Expand Up @@ -734,18 +772,7 @@ pub fn run_merge_mtx(args: &MergeMtxArgs) -> anyhow::Result<()> {
run_squeeze(&squeeze_args)?;
}

// do the batch mapping at the end
let data = open_sparse_matrix(&backend_file, &backend)?;
let default_batch = basename(&args.output)?;
let column_batch_names = data
.column_names()?
.iter()
.map(|k| batch_map.get(k).unwrap_or(&default_batch).clone())
.collect::<Vec<_>>();

write_lines(&column_batch_names, &batch_memb_file)?;

finalize_zarr_output(&backend_file, &effective_output)?;
write_batches_and_finalize(&effective_output, &backend_file, &backend, &batch_map)?;
info!("done");
Ok(())
}
Expand Down
88 changes: 77 additions & 11 deletions src/handlers/merging_tests.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
//! merge-backend's first tests: a clean merge round-trips, and a corrupted
//! input FAILS rather than shipping a silently short matrix.
//! merge-backend's first tests: a clean merge round-trips, a zipped merge
//! ends as the archive asked for, backends with other rows are refused with
//! an error, and a corrupted input FAILS rather than shipping a silently
//! short matrix.
//!
//! The failure test has two layers of protection to pin. The block-read loop
//! used to swallow a failed read (`.ok()` in a filter_map), dropping the block;
Expand Down Expand Up @@ -28,6 +30,17 @@ fn write_fixture(path: &str, cols: usize, seed_val: f32) -> anyhow::Result<()> {
Ok(())
}

/// Two inputs in a fresh directory: `a.zarr` (4 columns) and `b.zarr` (3),
/// both over rows g0..g2.
fn two_fixtures() -> anyhow::Result<(tempfile::TempDir, String, String)> {
let dir = tempfile::tempdir()?;
let path = |name: &str| dir.path().join(name).to_string_lossy().into_owned();
let (a, b) = (path("a.zarr"), path("b.zarr"));
write_fixture(&a, 4, 100.0)?;
write_fixture(&b, 3, 900.0)?;
Ok((dir, a, b))
}

fn merge_args(files: &[&str], output: &str) -> MergeBackendArgs {
MergeBackendArgs {
data_files: files.iter().map(|f| Box::from(*f)).collect(),
Expand All @@ -43,16 +56,9 @@ fn merge_args(files: &[&str], output: &str) -> MergeBackendArgs {

#[test]
fn a_clean_merge_round_trips() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let a = dir.path().join("a.zarr");
let b = dir.path().join("b.zarr");
write_fixture(a.to_str().expect("utf8"), 4, 100.0)?;
write_fixture(b.to_str().expect("utf8"), 3, 900.0)?;
let (dir, a, b) = two_fixtures()?;
let out = dir.path().join("m");
run_merge_backend(&merge_args(
&[a.to_str().expect("utf8"), b.to_str().expect("utf8")],
out.to_str().expect("utf8"),
))?;
run_merge_backend(&merge_args(&[&a, &b], out.to_str().expect("utf8")))?;
let merged = open_sparse_matrix(
&format!("{}.zarr", out.to_str().expect("utf8")),
&SparseIoBackend::Zarr,
Expand All @@ -63,6 +69,66 @@ fn a_clean_merge_round_trips() -> anyhow::Result<()> {
Ok(())
}

#[test]
fn a_zipped_merge_ends_as_the_archive_with_its_batches_beside() -> anyhow::Result<()> {
let (dir, a, b) = two_fixtures()?;
let out = dir.path().join("m");
let args = MergeBackendArgs {
zip: true,
..merge_args(&[&a, &b], out.to_str().expect("utf8"))
};
run_merge_backend(&args)?;
// `-o m` asks for m.zarr.zip; the working m.zarr used to be left instead.
assert!(dir.path().join("m.zarr.zip").is_file());
assert!(!dir.path().join("m.zarr").exists());
assert!(dir.path().join("m.batch.gz").is_file());

// A suffixed output keeps its batches beside it under the bare name.
let suffixed = dir.path().join("n.zarr.zip");
run_merge_backend(&MergeBackendArgs {
output: suffixed.to_str().expect("utf8").into(),
..args
})?;
assert!(dir.path().join("n.zarr.zip").is_file());
assert!(!dir.path().join("n.zarr").exists());
assert!(dir.path().join("n.batch.gz").is_file());

// Each column labelled with the input it came from, in order.
let labels = read_lines(dir.path().join("m.batch.gz").to_str().expect("utf8"))?;
let want: Vec<Box<str>> = ["a"; 4]
.into_iter()
.chain(["b"; 3])
.map(Box::from)
.collect();
assert_eq!(labels, want);
Ok(())
}

#[test]
fn an_output_that_is_an_input_is_refused_before_anything_is_cleared() -> anyhow::Result<()> {
let (dir, a, b) = two_fixtures()?;
// `-o a --no-zip` is a.zarr itself, which used to be deleted first.
let out = dir.path().join("a");
let err = run_merge_backend(&merge_args(&[&a, &b], out.to_str().expect("utf8"))).unwrap_err();
assert!(format!("{err:#}").contains("is the input"), "{err:#}");
assert!(open_sparse_matrix(&a, &SparseIoBackend::Zarr)?.num_columns() == Some(4));
Ok(())
}

#[test]
fn backends_with_other_rows_are_refused_not_panicked_on() -> anyhow::Result<()> {
let (dir, a, b) = two_fixtures()?;
let mut other = open_sparse_matrix(&b, &SparseIoBackend::Zarr)?;
let rows: Vec<Box<str>> = ["x0", "x1", "x2"].map(Box::from).into();
other.register_row_names_vec(&rows);
drop(other);
let out = dir.path().join("m");
let err = run_merge_backend(&merge_args(&[&a, &b], out.to_str().expect("utf8"))).unwrap_err();
assert!(format!("{err:#}").contains("other row names"), "{err:#}");
assert!(!dir.path().join("m.zarr").exists());
Ok(())
}

#[test]
fn a_merge_keeps_the_metadata_its_inputs_share() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
Expand Down
23 changes: 7 additions & 16 deletions src/handlers/transformation/squeeze.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
use super::{RowAlignMode, RunSqueezeArgs};
use crate::handlers::explore::{off_screen, open_session};
use crate::handlers::merging::{
find_aligned_rows, generate_unique_batch_names, run_merge_backend, MergeBackendArgs,
find_aligned_rows, generate_unique_batch_names, run_merge_backend, write_batches_and_finalize,
MergeBackendArgs,
};
use crate::hdf5_io::*;
use crate::interactive::cutoff_tui::{choose_cutoffs, choose_cutoffs_in, AxisView, CutoffPicker};
Expand Down Expand Up @@ -296,9 +297,9 @@ fn run_merge_then_squeeze(
all_triplets.len()
);

// `output_prefix` keeps naming the sidecars; the backend itself takes the
// `.zarr.zip` the rest of the tool produces, written as a directory here and
// zipped once everything below has finished with it.
// The backend takes the `.zarr.zip` the rest of the tool produces, written
// as a directory here and zipped once everything below has finished with
// it; its batch labels take its name without the backend suffix.
let (backend, _) = resolve_backend_file(output_prefix, None)?;
let effective_output = apply_zip_flag(output_prefix, cmd_args.zip, &backend);
let (backend, backend_file) = resolve_backend_file(&effective_output, Some(backend))?;
Expand Down Expand Up @@ -349,20 +350,10 @@ fn run_merge_then_squeeze(
final_data.num_columns().unwrap()
);

// Write batch membership file
let batch_memb_file = format!("{}.batch.gz", output_prefix);
drop(final_data);
let batch_map: HashMap<Box<str>, Box<str>> =
column_names.into_iter().zip(column_batch_names).collect();
let default_batch = basename(output_prefix)?;
let final_col_names = final_data.column_names()?;
let final_batch_names: Vec<Box<str>> = final_col_names
.iter()
.map(|k| batch_map.get(k).unwrap_or(&default_batch).clone())
.collect();
write_lines(&final_batch_names, &batch_memb_file)?;

drop(final_data);
finalize_zarr_output(&backend_file, &effective_output)?;
write_batches_and_finalize(&effective_output, &backend_file, &backend, &batch_map)?;

info!("Squeeze and merge (union) complete!");
Ok(())
Expand Down
Loading