src/Cargo.lock

@@ -1746,6 +1746,7 @@ name = "obikselect"
 version = "0.1.0"
 dependencies = [
  "obicompactvec",
+ "obikalgorithm",
  "obikindex",
  "obisys",
  "tracing",

src/obikselect/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2024"
 
 [dependencies]
 obikindex     = { path = "../obikindex" }
+obikalgorithm = { path = "../obikalgorithm" }
 obicompactvec = { path = "../obicompactvec" }
 obisys        = { path = "../obisys" }
 tracing       = "0.1.44"

src/obikselect/src/lib.rs

@@ -1,11 +1,14 @@
 //! Genome-column selection/aggregation on an `obikindex::KmerIndex`:
 //! projecting/aggregating genome columns (`AggOp`) into new output columns
-//! (`OutputCol`), either into a new index (`select`) or in place
-//! (`select_in_place`). Orthogonal to per-kmer filtering (`obikfilter`),
-//! which operates on rows, not columns. Kept as a separate crate — not a
-//! module of `obikindex` — so the dependency runs one way only.
+//! (`OutputCol`) of a brand new index. Orthogonal to per-kmer filtering
+//! (`obikfilter`), which operates on rows, not columns. Kept as a separate
+//! crate — not a module of `obikindex` — so the dependency runs one way
+//! only. Structured on the model of `obikindexer::algorithms`/
+//! `obikmerge`: [`Select`] implements `obikalgorithm::Algorithm` (two-phase
+//! `new` + setters, then `run`).
 
 mod select;
 mod select_layer;
 
+pub use select::Select;
 pub use select_layer::{AggOp, OutputCol};

src/obikselect/src/select.rs

@@ -1,154 +1,148 @@
-use std::path::Path;
-use std::sync::Arc;
+//! Projecting/aggregating the genome columns of a source
+//! `obikindex::KmerIndex` into a brand new one, according to a list of
+//! [`crate::OutputCol`] specs. Structured exactly like `obikmerge::Merge`:
+//! two-phase construction (`new` + setters), then
+//! `obikalgorithm::Algorithm::run`, via `obikindex::IndexBuilder`
+//! (`create_skeleton` + `finalize_indexed`) — no in-place variant, same as
+//! `Merge`.
 
-use obikindex::IndexBuilder;
-use crate::OutputCol;
-use obisys::{Reporter, Stage, progress_bar, PartitionRunner};
+use std::io;
+use std::path::PathBuf;
+
+use obikalgorithm::Algorithm;
+use obikindex::{GenomeInfo, IndexBuilder, IndexState, KmerIndex};
+use obisys::{PartitionRunner, Progress, Reporter, Stage};
 use tracing::info;
 
-use obikindex::{OKIError, OKIResult};
-use obikindex::KmerIndex;
-use obikindex::{GenomeInfo, IndexMeta};
-use obikindex::IndexState;
+use crate::select_layer::select_partition;
+use crate::OutputCol;
 
-impl KmerIndex {
-    /// Create a new index at `output` by projecting/aggregating the genome columns
-    /// of `src` according to `specs`.
-    ///
-    /// `output_presence` — if true, output uses bit matrices (0/1), regardless of
-    /// whether the source stores counts. The caller is responsible for ensuring all
-    /// specs use logical operators when `output_presence=true` on a count source.
-    pub fn select<P: AsRef<Path>>(
-        output: P,
-        src: &KmerIndex,
-        specs: &[OutputCol],
-        threshold: u32,
+pub struct Select<'a> {
+    src: &'a KmerIndex,
+    output: PathBuf,
+    specs: &'a [OutputCol],
+    threshold: u32,
+    output_presence: bool,
+    force: bool,
+    sparse: bool,
+    reporter: Reporter,
+    on_progress: Option<Box<dyn FnMut(Progress) + Send + 'a>>,
+}
+
+impl<'a> Select<'a> {
+    pub fn new(
+        src: &'a KmerIndex,
+        output: impl Into<PathBuf>,
+        specs: &'a [OutputCol],
         output_presence: bool,
-        force: bool,
-        rep: &mut Reporter,
-    ) -> OKIResult<Self> {
-        let output = output.as_ref();
-
-        if src.state()? != IndexState::Indexed {
-            return Err(OKIError::NotIndexed(src.root_path.clone()));
+    ) -> Self {
+        Self {
+            src,
+            output: output.into(),
+            specs,
+            threshold: 0,
+            output_presence,
+            force: false,
+            sparse: true,
+            reporter: Reporter::new(),
+            on_progress: None,
         }
-
-        KmerIndex::clear_output_for_create(output, force)?;
-
-        let mut config = src.meta.config.clone();
-        config.with_counts = !output_presence;
-        let genomes: Vec<GenomeInfo> = specs
-            .iter()
-            .map(|s| GenomeInfo::new(s.label.clone()))
-            .collect();
-
-        let n_src_genomes = src.meta.genomes().map_err(OKIError::Io)?.len();
-        let n_partitions = src.n_partitions();
-
-        let dst_partition = KmerIndex::create_skeleton(output, config, genomes)?;
-
-        info!(
-            "select: {} partition(s), {} source genome(s) → {} output column(s)",
-            n_partitions,
-            n_src_genomes,
-            specs.len(),
-        );
-
-        let t = Stage::start("select");
-        let pb = progress_bar("select", n_partitions as u64, "partitions");
-
-        let order: Vec<usize> = (0..n_partitions).collect();
-        let runner = PartitionRunner::new();
-        runner
-            .run(
-                &order,
-                |i| {
-                    dst_partition.select_partition(
-                        src,
-                        i,
-                        specs,
-                        n_src_genomes,
-                        threshold,
-                        output_presence,
-                        false,
-                    )
-                },
-                |_, _, _| {
-                    pb.inc(1);
-                },
-            )
-            .map_err(OKIError::Partition)?;
-
-        pb.finish_and_clear();
-        rep.push(t.stop());
-
-        KmerIndex::finalize_indexed(output, rep, false)
     }
 
-    /// Rewrite the genome columns of this index in-place according to `specs`.
-    ///
-    /// The MPHF and unitig files are unchanged; only data matrices are rewritten.
-    pub fn select_in_place(
-        &mut self,
-        specs: &[OutputCol],
-        threshold: u32,
-        output_presence: bool,
-        rep: &mut Reporter,
-    ) -> OKIResult<()> {
-        if self.state()? != IndexState::Indexed {
-            return Err(OKIError::NotIndexed(self.root_path.clone()));
-        }
+    /// Minimum count to consider a genome as "carrying" a k-mer, for
+    /// logical (`Any`/`All`/`None`) aggregation ops (default: `0`).
+    pub fn threshold(mut self, v: u32) -> Self {
+        self.threshold = v;
+        self
+    }
 
-        let n_src_genomes = self.meta.genomes().map_err(OKIError::Io)?.len();
-        let n_partitions = self.n_partitions();
+    /// Remove a pre-existing index at `output` before creating the
+    /// selected one (default: `false`, fails if one already exists).
+    pub fn force(mut self, v: bool) -> Self {
+        self.force = v;
+        self
+    }
 
-        info!(
-            "select (in-place): {} partition(s), {} source genome(s) → {} output column(s)",
-            n_partitions,
-            n_src_genomes,
-            specs.len(),
-        );
+    /// Pack the output's presence matrices in the compact sparse format
+    /// rather than dense (default: `true`).
+    pub fn sparse(mut self, v: bool) -> Self {
+        self.sparse = v;
+        self
+    }
 
-        let t = Stage::start("select");
-        let pb = progress_bar("select", n_partitions as u64, "partitions");
+    /// Progress callback, called once per completed partition.
+    pub fn on_progress(mut self, cb: impl FnMut(Progress) + Send + 'a) -> Self {
+        self.on_progress = Some(Box::new(cb));
+        self
+    }
 
+    /// Per-stage timing report accumulated during `run` — call after `run` returns.
+    pub fn reporter(&self) -> &Reporter {
+        &self.reporter
+    }
+}
+
+impl Algorithm for Select<'_> {
+    type Output = KmerIndex;
+
+    fn run(&mut self) -> obikalgorithm::Result<KmerIndex> {
+        let src = self.src;
+        let output = self.output.clone();
+
+        if src.state()? != IndexState::Indexed {
+            return Err(format!("{}: source index is not fully built", src.dir().display()).into());
+        }
+
+        let n_src_genomes = src.meta().genomes()?.len();
+        info!(
+            "select: {} partition(s), {} source genome(s) → {} output column(s), output={}",
+            src.n_partitions(),
+            n_src_genomes,
+            self.specs.len(),
+            if self.output_presence { "presence" } else { "count" },
+        );
+
+        KmerIndex::clear_output_for_create(&output, self.force)?;
+
+        let mut config = src.meta().config.clone();
+        config.with_counts = !self.output_presence;
+        let genomes: Vec<GenomeInfo> = self
+            .specs
+            .iter()
+            .map(|s| GenomeInfo::new(s.label.clone()))
+            .collect();
+
+        let t = Stage::start("select");
+        let dst = KmerIndex::create_skeleton(&output, config, genomes)?;
+
+        let n_partitions = src.n_partitions();
         let order: Vec<usize> = (0..n_partitions).collect();
         let runner = PartitionRunner::new();
+        let on_progress = &mut self.on_progress;
+        let mut done: u64 = 0;
+
         runner
             .run(
                 &order,
-                |i| {
-                    self.select_partition(
-                        self,
-                        i,
-                        specs,
-                        n_src_genomes,
-                        threshold,
-                        output_presence,
-                        true,
-                    )
-                },
+                |i| select_partition(&dst, src, i, self.specs, self.threshold, self.output_presence),
                 |_, _, _| {
-                    pb.inc(1);
+                    done += 1;
+                    if let Some(cb) = on_progress.as_mut() {
+                        cb(Progress {
+                            position: done,
+                            total: Some(n_partitions as u64),
+                        });
+                    }
                 },
             )
-            .map_err(OKIError::Partition)?;
+            .map_err(|e| io::Error::other(e.to_string()))?;
 
-        pb.finish_and_clear();
-        rep.push(t.stop());
+        self.reporter.push(t.stop());
 
-        let mut config = self.meta.config.clone();
-        config.with_counts = !output_presence;
-        let genomes: Vec<GenomeInfo> = specs
-            .iter()
-            .map(|s| GenomeInfo::new(s.label.clone()))
-            .collect();
-        self.meta.rewrite_config(config, genomes).map_err(OKIError::Io)?;
-        self.meta = Arc::new(IndexMeta::open(self).map_err(OKIError::Io)?);
+        let t = Stage::start("finalize");
+        let dst = KmerIndex::finalize_indexed(&output, &mut self.reporter, self.sparse)?;
+        self.reporter.push(t.stop());
 
-        let t_pack = Stage::start("pack");
-        self.pack_matrices(false)?;
-        rep.push(t_pack.stop());
-        Ok(())
+        Ok(dst)
     }
 }

src/obikselect/src/select_layer.rs

@@ -1,14 +1,21 @@
+//! Per-layer column projection/aggregation — the mechanics
+//! [`crate::select::Select`] runs once per partition-layer. Generic over
+//! source content (`Count`/`Presence`) via `obicompactvec::MatrixGroupOps`
+//! (object-safe: one `&dyn MatrixGroupOps` covers both matrix kinds, so
+//! this never branches on source content beyond the single `match` that
+//! opens it) and over destination content via [`DstBuilder`] — no
+//! duplicated per-op-per-content code path.
+
 use std::fs;
 use std::io;
-use std::path::{Path, PathBuf};
+use std::path::Path;
 
 use obicompactvec::{
     ColGroup, MatrixGroupOps, PersistentBitMatrix, PersistentBitMatrixBuilder,
-    PersistentCompactIntMatrix, PersistentCompactIntMatrixBuilder,
+    PersistentCompactIntMatrix, PersistentCompactIntMatrixBuilder, TempBitVec, TempCompactIntVec,
 };
-use obikindex::{OKIError, OKIResult};
-
-use obikindex::KmerIndex;
+use obikindex::layer::{KmerLayer, LayerContent};
+use obikindex::{KmerIndex, OKIError, OKIResult};
 
 // ── AggOp ─────────────────────────────────────────────────────────────────────
 
@@ -36,9 +43,67 @@ pub struct OutputCol {
     pub op: AggOp,
 }
 
+// ── Group aggregation, generic over source content ───────────────────────────
+
+/// One output column's aggregated values, still content-typed — the shape
+/// `AggOp` naturally produces (`Any`/`All`/`None` → bits, `Sum`/`Min`/`Max`
+/// → ints), independent of whether the *source* was itself count or
+/// presence data.
+enum AggResult {
+    Bit(TempBitVec),
+    Int(TempCompactIntVec),
+}
+
+fn compute_group(mat: &dyn MatrixGroupOps, spec: &OutputCol, threshold: u32) -> io::Result<AggResult> {
+    let g = ColGroup::new(spec.label.clone(), spec.indices.clone());
+    Ok(match spec.op {
+        AggOp::Any => AggResult::Bit(mat.partial_group_any(&g, threshold)?),
+        AggOp::All => AggResult::Bit(mat.partial_group_all(&g, threshold)?),
+        AggOp::None => AggResult::Bit(mat.partial_group_none(&g, threshold)?),
+        AggOp::Sum => AggResult::Int(mat.partial_group_sum(&g)?),
+        AggOp::Min => AggResult::Int(mat.partial_group_min(&g)?),
+        AggOp::Max => AggResult::Int(mat.partial_group_max(&g)?),
+    })
+}
+
+// ── DstBuilder — destination matrix, generic over output content ─────────────
+
+enum DstBuilder {
+    Bit(PersistentBitMatrixBuilder),
+    Int(PersistentCompactIntMatrixBuilder),
+}
+
+impl DstBuilder {
+    fn new(output_presence: bool, n: usize, dir: &Path) -> io::Result<Self> {
+        Ok(if output_presence {
+            DstBuilder::Bit(PersistentBitMatrixBuilder::new(n, dir)?)
+        } else {
+            DstBuilder::Int(PersistentCompactIntMatrixBuilder::new(n, dir)?)
+        })
+    }
+
+    fn add(&mut self, r: AggResult) -> io::Result<()> {
+        match (self, r) {
+            (DstBuilder::Bit(b), AggResult::Bit(v)) => b.add_col_from(&v),
+            (DstBuilder::Bit(b), AggResult::Int(v)) => b.add_col_from_int(&v),
+            (DstBuilder::Int(b), AggResult::Bit(v)) => b.add_col_from_bit(&v),
+            (DstBuilder::Int(b), AggResult::Int(v)) => b.add_col_from(&v),
+        }
+    }
+
+    fn close(self) -> io::Result<()> {
+        match self {
+            DstBuilder::Bit(b) => b.close(),
+            DstBuilder::Int(b) => b.close(),
+        }
+    }
+}
+
 // ── Helpers ───────────────────────────────────────────────────────────────────
 
-/// Copy all plain files (not subdirectories) from `src_dir` to `dst_dir`.
+/// Copy all plain files (not subdirectories) from `src_dir` to `dst_dir` —
+/// carries a layer's kmer identity (mphf/unitigs/evidence/fingerprint)
+/// across unchanged; only the data matrix (a subdirectory) is rebuilt.
 fn copy_layer_files(src_dir: &Path, dst_dir: &Path) -> io::Result<()> {
     for entry in fs::read_dir(src_dir)? {
         let entry = entry?;
@@ -50,248 +115,54 @@ fn copy_layer_files(src_dir: &Path, dst_dir: &Path) -> io::Result<()> {
     Ok(())
 }
 
-// ── fill_builders ─────────────────────────────────────────────────────────────
+// ── select_partition ──────────────────────────────────────────────────────────
 
-fn fill_builders(
+/// Rewrite partition `i`'s data matrices from `src` into `dst` according to
+/// `specs`. Every layer's kmer identity is copied as-is; only the genome
+/// columns are recomputed.
+pub(crate) fn select_partition(
+    dst: &KmerIndex,
+    src: &KmerIndex,
+    i: usize,
     specs: &[OutputCol],
-    src_layer_dir: &Path,
-    src_is_count: bool,
     threshold: u32,
     output_presence: bool,
-    mut dst_bit: Option<&mut PersistentBitMatrixBuilder>,
-    mut dst_int: Option<&mut PersistentCompactIntMatrixBuilder>,
 ) -> OKIResult<()> {
-    if src_is_count {
-        let mat = PersistentCompactIntMatrix::open(src_layer_dir).map_err(OKIError::Io)?;
-        for spec in specs {
-            let g = ColGroup::new(&spec.label, spec.indices.clone());
-            if output_presence {
-                let b = dst_bit.as_deref_mut().unwrap();
-                match spec.op {
-                    AggOp::Any => {
-                        b.add_col_from(&mat.partial_group_any(&g, threshold).map_err(OKIError::Io)?)
-                    }
-                    AggOp::All => {
-                        b.add_col_from(&mat.partial_group_all(&g, threshold).map_err(OKIError::Io)?)
-                    }
-                    AggOp::None => {
-                        b.add_col_from(&mat.partial_group_none(&g, threshold).map_err(OKIError::Io)?)
-                    }
-                    AggOp::Sum => {
-                        b.add_col_from_int(&mat.partial_group_sum(&g).map_err(OKIError::Io)?)
-                    }
-                    AggOp::Min => {
-                        b.add_col_from_int(&mat.partial_group_min(&g).map_err(OKIError::Io)?)
-                    }
-                    AggOp::Max => {
-                        b.add_col_from_int(&mat.partial_group_max(&g).map_err(OKIError::Io)?)
-                    }
-                }
-                .map_err(OKIError::Io)?;
-            } else {
-                let b = dst_int.as_deref_mut().unwrap();
-                match spec.op {
-                    AggOp::Sum => b.add_col_from(&mat.partial_group_sum(&g).map_err(OKIError::Io)?),
-                    AggOp::Min => b.add_col_from(&mat.partial_group_min(&g).map_err(OKIError::Io)?),
-                    AggOp::Max => b.add_col_from(&mat.partial_group_max(&g).map_err(OKIError::Io)?),
-                    AggOp::Any => b.add_col_from_bit(
-                        &mat.partial_group_any(&g, threshold).map_err(OKIError::Io)?,
-                    ),
-                    AggOp::All => b.add_col_from_bit(
-                        &mat.partial_group_all(&g, threshold).map_err(OKIError::Io)?,
-                    ),
-                    AggOp::None => b.add_col_from_bit(
-                        &mat.partial_group_none(&g, threshold).map_err(OKIError::Io)?,
-                    ),
-                }
-                .map_err(OKIError::Io)?;
+    let src_partition = src.partition(i)?;
+    let dst_partition = dst.partition(i)?;
+    let n_layers = src_partition.n_layers();
+
+    for l in 0..n_layers {
+        let src_layer = src_partition.layer(l)?.open()?;
+        let n = src_layer.n();
+
+        let dst_layer_dir = KmerLayer::new(&dst_partition, l)
+            .create()
+            .map_err(OKIError::Io)?
+            .dir()
+            .to_path_buf();
+        copy_layer_files(src_layer.dir(), &dst_layer_dir).map_err(OKIError::Io)?;
+
+        let group_mat: Box<dyn MatrixGroupOps> = match src_layer.content() {
+            LayerContent::Count => {
+                Box::new(PersistentCompactIntMatrix::open(src_layer.dir()).map_err(OKIError::Io)?)
             }
-        }
-    } else {
-        let mat = PersistentBitMatrix::open(src_layer_dir).map_err(OKIError::Io)?;
-        for spec in specs {
-            let g = ColGroup::new(&spec.label, spec.indices.clone());
-            if output_presence {
-                let b = dst_bit.as_deref_mut().unwrap();
-                match spec.op {
-                    AggOp::Any => {
-                        b.add_col_from(&mat.partial_group_any(&g, 1).map_err(OKIError::Io)?)
-                    }
-                    AggOp::All => {
-                        b.add_col_from(&mat.partial_group_all(&g, 1).map_err(OKIError::Io)?)
-                    }
-                    AggOp::None => {
-                        b.add_col_from(&mat.partial_group_none(&g, 1).map_err(OKIError::Io)?)
-                    }
-                    AggOp::Sum => {
-                        b.add_col_from_int(&mat.partial_group_sum(&g).map_err(OKIError::Io)?)
-                    }
-                    AggOp::Min => {
-                        b.add_col_from_int(&mat.partial_group_min(&g).map_err(OKIError::Io)?)
-                    }
-                    AggOp::Max => {
-                        b.add_col_from_int(&mat.partial_group_max(&g).map_err(OKIError::Io)?)
-                    }
-                }
-                .map_err(OKIError::Io)?;
-            } else {
-                let b = dst_int.as_deref_mut().unwrap();
-                match spec.op {
-                    AggOp::Sum => b.add_col_from(&mat.partial_group_sum(&g).map_err(OKIError::Io)?),
-                    AggOp::Min => b.add_col_from(&mat.partial_group_min(&g).map_err(OKIError::Io)?),
-                    AggOp::Max => b.add_col_from(&mat.partial_group_max(&g).map_err(OKIError::Io)?),
-                    AggOp::Any => {
-                        b.add_col_from_bit(&mat.partial_group_any(&g, 1).map_err(OKIError::Io)?)
-                    }
-                    AggOp::All => {
-                        b.add_col_from_bit(&mat.partial_group_all(&g, 1).map_err(OKIError::Io)?)
-                    }
-                    AggOp::None => {
-                        b.add_col_from_bit(&mat.partial_group_none(&g, 1).map_err(OKIError::Io)?)
-                    }
-                }
-                .map_err(OKIError::Io)?;
+            LayerContent::Presence => {
+                Box::new(PersistentBitMatrix::open(src_layer.dir()).map_err(OKIError::Io)?)
             }
-        }
-    }
-    Ok(())
-}
-
-// ── KmerPartition::select_partition ──────────────────────────────────────────
-
-impl KmerIndex {
-    /// Rewrite the data matrices of partition `i` in `src` into `self`.
-    ///
-    /// `specs` defines the output columns (projection/aggregation).
-    /// `output_presence` — if true, all output builders use bit (0/1) format.
-    /// `in_place` — `self` and `src` share the same root; write to temp dirs then swap.
-    pub fn select_partition(
-        &self,
-        src: &KmerIndex,
-        i: usize,
-        specs: &[OutputCol],
-        _n_src_genomes: usize,
-        threshold: u32,
-        output_presence: bool,
-        in_place: bool,
-    ) -> OKIResult<()> {
-        let src_index_dir = src.index_dir(i);
-        if !src_index_dir.exists() {
-            return Ok(());
-        }
-
-        let n_src_layers = src.n_layers(i)?;
-        if n_src_layers == 0 {
-            return Ok(());
-        }
-
-        let dst_index_dir = self.index_dir(i);
-        if !in_place {
-            fs::create_dir_all(&dst_index_dir)?;
-        }
-
-        let data_subdir = if output_presence {
-            "presence"
-        } else {
-            "counts"
         };
 
-        for l in 0..n_src_layers {
-            let src_layer_dir = src.layer_dir(i, l);
-            if !src_layer_dir.exists() {
-                continue;
-            }
+        let data_subdir = if output_presence { "presence" } else { "counts" };
+        let data_dir = dst_layer_dir.join(data_subdir);
+        fs::create_dir_all(&data_dir).map_err(OKIError::Io)?;
 
-            let dst_layer_dir = self.layer_dir(i, l);
-
-            let counts_dir = src_layer_dir.join("counts");
-            let presence_dir = src_layer_dir.join("presence");
-            let src_is_count = counts_dir.exists() && !presence_dir.exists();
-
-            // Determine number of slots and detect implicit layers.
-            let n = if counts_dir.exists() {
-                PersistentCompactIntMatrix::open(&src_layer_dir)
-                    .map_err(OKIError::Io)?
-                    .n()
-            } else if presence_dir.exists() {
-                PersistentBitMatrix::open(&src_layer_dir)
-                    .map_err(OKIError::Io)?
-                    .n()
-            } else {
-                // Implicit single-genome layer: no data matrix needed in output either.
-                if !in_place {
-                    fs::create_dir_all(&dst_layer_dir)?;
-                    copy_layer_files(&src_layer_dir, &dst_layer_dir)?;
-                }
-                continue;
-            };
-
-            // Choose the output data directory (temp name for in-place).
-            let (dst_data_dir, final_data_dir): (PathBuf, PathBuf) = if in_place {
-                let tmp = dst_layer_dir.join(format!("{data_subdir}_new"));
-                let perm = dst_layer_dir.join(data_subdir);
-                (tmp, perm)
-            } else {
-                let perm = dst_layer_dir.join(data_subdir);
-                (perm.clone(), perm)
-            };
-
-            if !in_place {
-                fs::create_dir_all(&dst_layer_dir)?;
-                copy_layer_files(&src_layer_dir, &dst_layer_dir)?;
-            }
-            fs::create_dir_all(&dst_data_dir)?;
-
-            let (mut dst_bit, mut dst_int) = if output_presence {
-                (
-                    Some(PersistentBitMatrixBuilder::new(n, &dst_data_dir).map_err(OKIError::Io)?),
-                    None,
-                )
-            } else {
-                (
-                    None,
-                    Some(
-                        PersistentCompactIntMatrixBuilder::new(n, &dst_data_dir)
-                            .map_err(OKIError::Io)?,
-                    ),
-                )
-            };
-
-            fill_builders(
-                specs,
-                &src_layer_dir,
-                src_is_count,
-                threshold,
-                output_presence,
-                dst_bit.as_mut(),
-                dst_int.as_mut(),
-            )?;
-
-            if output_presence {
-                dst_bit.unwrap().close().map_err(OKIError::Io)?;
-            } else {
-                dst_int.unwrap().close().map_err(OKIError::Io)?;
-            }
-
-            // In-place: swap old data dir for new.
-            if in_place {
-                let old_data_dir = if src_is_count {
-                    dst_layer_dir.join("counts")
-                } else {
-                    dst_layer_dir.join("presence")
-                };
-                if old_data_dir.exists() {
-                    fs::remove_dir_all(&old_data_dir)?;
-                }
-                fs::rename(&dst_data_dir, &final_data_dir)?;
-            }
+        let mut builder = DstBuilder::new(output_presence, n, &data_dir).map_err(OKIError::Io)?;
+        for spec in specs {
+            let r = compute_group(group_mat.as_ref(), spec, threshold).map_err(OKIError::Io)?;
+            builder.add(r).map_err(OKIError::Io)?;
         }
-
-        if !in_place {
-            src.partition_meta(i)?
-                .save(&dst_index_dir)?;
-        }
-
-        Ok(())
+        builder.close().map_err(OKIError::Io)?;
     }
+
+    Ok(())
 }