From 5dc16b0127e73fbaec431cb8e3e4f0003ee15be1 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 22 Aug 2026 17:16:30 +0200 Subject: [PATCH] refactor(obikselect): implement Algorithm trait for selection Shifts the selection logic from a direct method into a dedicated `Select` struct implementing the `Algorithm` trait. Introduces a two-phase initialization pattern with explicit builder methods and closure-based progress callbacks. Updates the crate's dependency graph to include `obikalgorithm`, refactors layer processing with unified matrix dispatch, and implements dual directory routing for partition execution. --- src/Cargo.lock | 1 + src/obikselect/Cargo.toml | 1 + src/obikselect/src/lib.rs | 11 +- src/obikselect/src/select.rs | 252 ++++++++++----------- src/obikselect/src/select_layer.rs | 347 +++++++++-------------------- 5 files changed, 241 insertions(+), 371 deletions(-) diff --git a/src/Cargo.lock b/src/Cargo.lock index 7f35c589..40c259ba 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -1746,6 +1746,7 @@ name = "obikselect" version = "0.1.0" dependencies = [ "obicompactvec", + "obikalgorithm", "obikindex", "obisys", "tracing", diff --git a/src/obikselect/Cargo.toml b/src/obikselect/Cargo.toml index 4efb38ec..d5f50cdc 100644 --- a/src/obikselect/Cargo.toml +++ b/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" diff --git a/src/obikselect/src/lib.rs b/src/obikselect/src/lib.rs index 30b7192b..78be8dc5 100644 --- a/src/obikselect/src/lib.rs +++ b/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}; diff --git a/src/obikselect/src/select.rs b/src/obikselect/src/select.rs index 7b05aeeb..e26a32ab 100644 --- a/src/obikselect/src/select.rs +++ b/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>( - 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>, +} + +impl<'a> Select<'a> { + pub fn new( + src: &'a KmerIndex, + output: impl Into, + specs: &'a [OutputCol], output_presence: bool, - force: bool, - rep: &mut Reporter, - ) -> OKIResult { - 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 = 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 = (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 { + 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 = 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 = (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 = 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) } } diff --git a/src/obikselect/src/select_layer.rs b/src/obikselect/src/select_layer.rs index 51c768da..cbd265ba 100644 --- a/src/obikselect/src/select_layer.rs +++ b/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 { + 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 { + 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 = 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(()) }