Push zunrplorkwkt #70
Executable
+126
@@ -0,0 +1,126 @@
|
||||
#!/usr/bin/env bash
|
||||
# smoke_test_index2.sh — end-to-end smoke test of `obikmer2 index` (this
|
||||
# working tree) + `obikmer query` (the release binary already on PATH).
|
||||
#
|
||||
# obikmer2 is being rebuilt incrementally from obikmer (see DevDocMD) and
|
||||
# for now only implements `index`. This script checks that an index it
|
||||
# produces is readable by an already-released, PATH-installed `obikmer`
|
||||
# binary — the cross-binary counterpart to scripts/smoke_test_index.sh,
|
||||
# which builds and queries with the same (in-tree) `obikmer`.
|
||||
#
|
||||
# What it does:
|
||||
# 1. builds `obikmer2` (debug, via `cargo run`)
|
||||
# 2. generates a small deterministic random FASTA
|
||||
# 3. runs `obikmer2 index` on it
|
||||
# 4. picks a real k-mer from the source sequence, avoiding low-complexity
|
||||
# substrings (see scripts/smoke_test_index.sh's own note on this)
|
||||
# 5. runs `obikmer query` (PATH binary, not built from this tree) and
|
||||
# checks the k-mer round-trips
|
||||
# 6. prints total-kmers-indexed and a clear PASS/FAIL, exit code matches
|
||||
#
|
||||
# Usage:
|
||||
# scripts/smoke_test_index2.sh [-k KMER_SIZE] [-m MINIMIZER_SIZE] [-p PARTITIONS] [--keep]
|
||||
#
|
||||
# --keep leaves the temp directory in place (path printed) instead of
|
||||
# deleting it on exit, for manual inspection of a failure.
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
K=11
|
||||
M=5
|
||||
PARTITIONS=4
|
||||
KEEP=0
|
||||
|
||||
while [ $# -gt 0 ]; do
|
||||
case "$1" in
|
||||
-k) K="$2"; shift 2 ;;
|
||||
-m) M="$2"; shift 2 ;;
|
||||
-p) PARTITIONS="$2"; shift 2 ;;
|
||||
--keep) KEEP=1; shift ;;
|
||||
*) echo "unknown argument: $1" >&2; exit 2 ;;
|
||||
esac
|
||||
done
|
||||
|
||||
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
WORK="$(mktemp -d -t obikmer2_smoke.XXXXXX)"
|
||||
|
||||
cleanup() {
|
||||
if [ "$KEEP" -eq 1 ]; then
|
||||
echo "kept: $WORK"
|
||||
else
|
||||
rm -rf "$WORK"
|
||||
fi
|
||||
}
|
||||
trap cleanup EXIT
|
||||
|
||||
fail() {
|
||||
echo "FAIL: $1" >&2
|
||||
exit 1
|
||||
}
|
||||
|
||||
command -v obikmer >/dev/null 2>&1 || fail "obikmer not found on PATH"
|
||||
|
||||
# ── 1. generate a small deterministic random FASTA ─────────────────────────
|
||||
python3 - "$WORK/test.fasta" "$K" <<'EOF'
|
||||
import random, sys
|
||||
path, k = sys.argv[1], int(sys.argv[2])
|
||||
random.seed(1234)
|
||||
bases = "ACGT"
|
||||
with open(path, "w") as f:
|
||||
for i in range(3):
|
||||
seq = "".join(random.choice(bases) for _ in range(300))
|
||||
f.write(f">seq{i}\n{seq}\n")
|
||||
EOF
|
||||
|
||||
# ── 2. build + run index (obikmer2, in-tree) ────────────────────────────────
|
||||
cd "$REPO_ROOT/src"
|
||||
INDEX_LOG="$WORK/index.log"
|
||||
if ! cargo run -q -p obikmer2 --bin obikmer2 -- \
|
||||
index -k "$K" -m "$M" --theta 0 -p "$PARTITIONS" \
|
||||
-o "$WORK/out.idx" "$WORK/test.fasta" > "$INDEX_LOG" 2>&1
|
||||
then
|
||||
cat "$INDEX_LOG" >&2
|
||||
fail "obikmer2 index exited non-zero"
|
||||
fi
|
||||
|
||||
N_KMERS="$(grep -o '[0-9]* total kmers indexed' "$INDEX_LOG" | grep -o '^[0-9]*' || true)"
|
||||
[ -n "$N_KMERS" ] || { cat "$INDEX_LOG" >&2; fail "could not find 'N total kmers indexed' in index log"; }
|
||||
[ "$N_KMERS" -gt 0 ] || fail "index reports 0 kmers indexed"
|
||||
|
||||
# ── 3+4. try candidate k-mers spread across the source sequence until one
|
||||
# round-trips — see scripts/smoke_test_index.sh's own note: query's
|
||||
# entropy filter can reject a genuinely-indexed low-complexity
|
||||
# window, that's not a bug, just try the next candidate.
|
||||
readarray -t CANDIDATES < <(python3 - "$WORK/test.fasta" "$K" <<'EOF'
|
||||
import sys
|
||||
path, k = sys.argv[1], int(sys.argv[2])
|
||||
with open(path) as f:
|
||||
seq = "".join(l.strip() for l in f if not l.startswith(">"))
|
||||
for start in range(0, len(seq) - k, 17):
|
||||
print(seq[start:start+k])
|
||||
EOF
|
||||
)
|
||||
[ "${#CANDIDATES[@]}" -gt 0 ] || fail "could not extract any candidate k-mer from the source FASTA"
|
||||
|
||||
QUERY_LOG="$WORK/query.log"
|
||||
FOUND=0
|
||||
for QUERY_KMER in "${CANDIDATES[@]}"; do
|
||||
printf ">q1\n%s\n" "$QUERY_KMER" > "$WORK/query.fasta"
|
||||
# PATH binary, deliberately not built from this tree.
|
||||
if ! obikmer query "$WORK/out.idx" "$WORK/query.fasta" > "$QUERY_LOG" 2>&1
|
||||
then
|
||||
cat "$QUERY_LOG" >&2
|
||||
fail "obikmer query exited non-zero"
|
||||
fi
|
||||
if grep -q '"kmer_count":1' "$QUERY_LOG"; then
|
||||
FOUND=1
|
||||
break
|
||||
fi
|
||||
done
|
||||
|
||||
if [ "$FOUND" -ne 1 ]; then
|
||||
cat "$QUERY_LOG" >&2
|
||||
fail "no candidate k-mer round-tripped (tried ${#CANDIDATES[@]}) — likely a real regression, not a low-complexity fixture"
|
||||
fi
|
||||
|
||||
echo "PASS: obikmer2 index + PATH obikmer query round-trip OK — $N_KMERS kmers indexed, query k-mer '$QUERY_KMER' found"
|
||||
Generated
+2
@@ -1551,6 +1551,8 @@ version = "0.1.0"
|
||||
dependencies = [
|
||||
"ndarray",
|
||||
"obicompactvec",
|
||||
"obikindex",
|
||||
"obikseq",
|
||||
"rayon",
|
||||
"tempfile",
|
||||
]
|
||||
|
||||
@@ -4,6 +4,8 @@ version = "0.1.0"
|
||||
edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
obikindex = { path = "../obikindex"}
|
||||
obikseq = { path = "../obikseq"}
|
||||
obicompactvec = { path = "../obicompactvec" }
|
||||
ndarray = "0.17"
|
||||
rayon = "1"
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
use obikindex::{KmerIndex, layer::KmerLayer};
|
||||
use obikseq::CanonicalKmer;
|
||||
|
||||
use crate::meta_cache::MetaCache;
|
||||
|
||||
pub struct IndexCache<'a> {
|
||||
raw_index: &'a KmerIndex,
|
||||
meta: MetaCache,
|
||||
|
||||
n_partition: usize,
|
||||
n_layer: usize,
|
||||
layer_cache: Vec<KmerLayer>,
|
||||
}
|
||||
|
||||
impl<'a> IndexCache<'a> {
|
||||
/// Opens every layer of every partition once, up front, and keeps them
|
||||
/// alive for the cache's lifetime — `layer(p, l)` stored flat at
|
||||
/// `p * n_layer + l` (`n_layer` uniform across partitions, guaranteed by
|
||||
/// construction).
|
||||
///
|
||||
/// Panics if the index isn't fully built: a cache can only be built on a
|
||||
/// functional index, so any failure here (missing/incomplete layer) is a
|
||||
/// caller error, not a case to design around with a `Result`.
|
||||
pub fn new(index: &'a KmerIndex) -> Self {
|
||||
let n_partition = index.n_partitions();
|
||||
let n_layer = index.n_layers_per_partition().unwrap_or_else(|e| {
|
||||
panic!("IndexCache::new: index is not fully built (n_layers_per_partition failed): {e}")
|
||||
});
|
||||
|
||||
let mut layer_cache = Vec::with_capacity(n_partition * n_layer);
|
||||
for p in 0..n_partition {
|
||||
for l in 0..n_layer {
|
||||
let layer = index
|
||||
.layer(p, l)
|
||||
.and_then(KmerLayer::open)
|
||||
.unwrap_or_else(|e| {
|
||||
panic!(
|
||||
"IndexCache::new: failed to open layer (partition {p}, layer {l}): {e} — index is not fully built"
|
||||
)
|
||||
});
|
||||
layer_cache.push(layer);
|
||||
}
|
||||
}
|
||||
|
||||
IndexCache {
|
||||
raw_index: index,
|
||||
meta: MetaCache::from_meta(&index.meta()),
|
||||
n_partition,
|
||||
n_layer,
|
||||
layer_cache,
|
||||
}
|
||||
}
|
||||
|
||||
/// The index's metadata, snapshotted once at construction — see
|
||||
/// [`MetaCache`]. Read-only
|
||||
#[inline]
|
||||
pub fn meta(&self) -> &MetaCache {
|
||||
&self.meta
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn n_layer(&self) -> usize {
|
||||
self.n_layer
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn n_partition(&self) -> usize {
|
||||
self.n_partition
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn get_layer(&self, partition: usize, layer: usize) -> Option<&KmerLayer> {
|
||||
self.layer_cache.get(partition * self.n_layer + layer)
|
||||
}
|
||||
|
||||
pub fn iter(&self) -> impl Iterator<Item = &KmerLayer> {
|
||||
self.layer_cache.iter()
|
||||
}
|
||||
|
||||
pub fn iter_indexed(&self) -> impl Iterator<Item = (usize, usize, &KmerLayer)> {
|
||||
let n_layer = self.n_layer;
|
||||
self.layer_cache
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(move |(i, layer)| (i / n_layer, i % n_layer, layer))
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn hash(&self, partition: usize, layer: usize, kmer: CanonicalKmer) -> Option<usize> {
|
||||
Some(self.get_layer(partition, layer)?.hash(kmer))
|
||||
}
|
||||
|
||||
pub fn n_kmer(&self, partition: usize, layer: usize) -> Option<usize> {
|
||||
Some(self.get_layer(partition, layer)?.n())
|
||||
}
|
||||
|
||||
/// Membership-checked lookup in one `(partition, layer)` — `None` if
|
||||
/// that layer doesn't carry `kmer`. Unlike [`hash`](Self::hash), this
|
||||
/// goes through `KmerLayer::find_slot`'s evidence check, not a blind
|
||||
/// MPHF lookup.
|
||||
#[inline]
|
||||
pub fn find_in_layer(&self, partition: usize, layer: usize, kmer: CanonicalKmer) -> Option<usize> {
|
||||
self.get_layer(partition, layer)?.find_slot(kmer)
|
||||
}
|
||||
|
||||
/// Membership-checked lookup across every layer of one partition —
|
||||
/// first layer that carries `kmer` wins. `None` if `partition` is out
|
||||
/// of range or no layer of it carries `kmer`.
|
||||
pub fn find_in_partition(
|
||||
&self,
|
||||
partition: usize,
|
||||
kmer: CanonicalKmer,
|
||||
) -> Option<(usize, usize, usize)> {
|
||||
for l in 0..self.n_layer {
|
||||
if let Some(slot) = self.find_in_layer(partition, l, kmer) {
|
||||
return Some((partition, l, slot));
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// Membership-checked lookup across every partition and layer — first
|
||||
/// one that carries `kmer` wins.
|
||||
pub fn find(&self, kmer: CanonicalKmer) -> Option<(usize, usize, usize)> {
|
||||
for p in 0..self.n_partition {
|
||||
if let Some(hit) = self.find_in_partition(p, kmer) {
|
||||
return Some(hit);
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,5 @@
|
||||
//! `CountPartials`/`BitPartials` aggregation across them. No dependency on
|
||||
//! `obikindex` — generic over any `S` implementing those traits.
|
||||
|
||||
mod layered_store;
|
||||
|
||||
pub use layered_store::LayeredStore;
|
||||
pub mod index_cache;
|
||||
pub mod meta_cache;
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
use obikindex::{GenomeInfo, IndexConfig, IndexMeta, IndexState};
|
||||
|
||||
/// A read-only, fully in-memory snapshot of an index's `IndexMeta`, taken
|
||||
/// once at construction. `IndexMeta` itself only caches `config` — `genomes`
|
||||
/// and `state` are re-read from disk on every call, by design, since they
|
||||
/// can legitimately change during a run. This type is the opposite trade:
|
||||
/// everything frozen at snapshot time, no disk access ever, at the cost of
|
||||
/// going stale if the index is mutated after the cache is built — matching
|
||||
/// the "cache can only be built on a functional index, read-only" contract
|
||||
/// the rest of `IndexCache` follows.
|
||||
pub struct MetaCache {
|
||||
config: IndexConfig,
|
||||
genomes: Vec<GenomeInfo>,
|
||||
state: IndexState,
|
||||
}
|
||||
|
||||
impl MetaCache {
|
||||
/// Panics if reading `genomes`/`state` off `meta` fails — same
|
||||
/// discipline as `IndexCache::new`: a cache can only be built on a
|
||||
/// functional index, so a failure here means the index isn't one.
|
||||
pub(crate) fn from_meta(meta: &IndexMeta) -> Self {
|
||||
let genomes = meta
|
||||
.genomes()
|
||||
.unwrap_or_else(|e| panic!("MetaCache::from_meta: failed to read genomes: {e}"));
|
||||
let state = meta
|
||||
.state()
|
||||
.unwrap_or_else(|e| panic!("MetaCache::from_meta: failed to read state: {e}"));
|
||||
MetaCache {
|
||||
config: meta.config().clone(),
|
||||
genomes,
|
||||
state,
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn config(&self) -> &IndexConfig {
|
||||
&self.config
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn genomes(&self) -> &[GenomeInfo] {
|
||||
&self.genomes
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn state(&self) -> IndexState {
|
||||
self.state
|
||||
}
|
||||
}
|
||||
@@ -86,10 +86,17 @@ impl IndexBuilder for KmerIndex {
|
||||
let output = output.as_ref();
|
||||
fs::create_dir_all(output).map_err(OKIError::Io)?;
|
||||
let meta = IndexMeta::create_at(output, config, genomes).map_err(OKIError::Io)?;
|
||||
Ok(KmerIndex {
|
||||
let idx = KmerIndex {
|
||||
root_path: output.to_owned(),
|
||||
meta: Arc::new(meta),
|
||||
})
|
||||
lock: std::sync::Mutex::new(None),
|
||||
};
|
||||
if !idx.acquire_lock() {
|
||||
return Err(OKIError::Io(std::io::Error::other(
|
||||
"failed to acquire index lock right after create_skeleton",
|
||||
)));
|
||||
}
|
||||
Ok(idx)
|
||||
}
|
||||
|
||||
fn finalize_indexed<P: AsRef<Path>>(output: P, rep: &mut Reporter) -> OKIResult<Self> {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use std::fs;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use crate::layer::KmerLayer;
|
||||
use crate::partition::KmerPartition;
|
||||
@@ -15,10 +15,14 @@ use crate::index::meta::{GenomeInfo, IndexConfig, IndexMeta};
|
||||
pub struct KmerIndex {
|
||||
pub(crate) root_path: PathBuf,
|
||||
pub(crate) meta: Arc<IndexMeta>,
|
||||
pub(crate) lock: Mutex<Option<obisys::DirLock>>,
|
||||
}
|
||||
|
||||
impl KmerIndex {
|
||||
/// Create a new index at `path`.
|
||||
/// Create a new index at `path`, holding its directory lock on return
|
||||
/// (see [`acquire_lock`](Self::acquire_lock)) — a freshly created index
|
||||
/// is definitionally about to be written to (scatter/dereplicate/
|
||||
/// count/build), so it is exclusive from the start, unlike `open`.
|
||||
///
|
||||
/// If `genome_info` is `Some`, it is stored immediately.
|
||||
/// If `None`, the genome entry will be added when `mark_scattered` is called.
|
||||
@@ -33,12 +37,26 @@ impl KmerIndex {
|
||||
set_m(config.minimizer_size);
|
||||
let genomes = genome_info.into_iter().collect();
|
||||
let meta = IndexMeta::create_at(&root_path, config, genomes).map_err(OKIError::Io)?;
|
||||
Ok(Self {
|
||||
let idx = Self {
|
||||
root_path,
|
||||
meta: Arc::new(meta),
|
||||
})
|
||||
lock: Mutex::new(None),
|
||||
};
|
||||
if !idx.acquire_lock() {
|
||||
return Err(OKIError::Io(std::io::Error::other(
|
||||
"failed to acquire index lock right after create",
|
||||
)));
|
||||
}
|
||||
Ok(idx)
|
||||
}
|
||||
|
||||
/// Open an existing index for reading — doesn't touch the directory
|
||||
/// lock at all: concurrent readers (`query`, `dump`, a `merge` source,
|
||||
/// ...) must never block on each other. A caller that needs exclusive
|
||||
/// access (resuming an interrupted build) uses
|
||||
/// [`open_lock`](Self::open_lock) instead; either way, `self.lock`
|
||||
/// starts empty — [`test_lock`](Self::test_lock)/
|
||||
/// [`acquire_lock`](Self::acquire_lock) are opt-in from here.
|
||||
pub fn open<P: AsRef<Path>>(path: P) -> OKIResult<Self> {
|
||||
let root_path = path.as_ref().to_owned();
|
||||
let meta = IndexMeta::open_at(&root_path).map_err(OKIError::Io)?;
|
||||
@@ -47,9 +65,54 @@ impl KmerIndex {
|
||||
Ok(Self {
|
||||
root_path,
|
||||
meta: Arc::new(meta),
|
||||
lock: Mutex::new(None),
|
||||
})
|
||||
}
|
||||
|
||||
/// `open`, then hold the directory lock exclusively — for a caller
|
||||
/// that resumes a write pipeline (e.g. `obikmer index` re-run on an
|
||||
/// existing, not-yet-`Indexed` directory) rather than reading
|
||||
/// concurrently with others.
|
||||
pub fn open_lock<P: AsRef<Path>>(path: P) -> OKIResult<Self> {
|
||||
let idx = Self::open(path)?;
|
||||
if !idx.acquire_lock() {
|
||||
return Err(OKIError::Io(std::io::Error::other(
|
||||
"failed to acquire index lock",
|
||||
)));
|
||||
}
|
||||
Ok(idx)
|
||||
}
|
||||
|
||||
/// Block until this index's directory lock is acquired, then hold it
|
||||
/// (released on [`release_lock`](Self::release_lock) or when `self` is
|
||||
/// dropped). Returns `false` only on a genuine I/O failure acquiring
|
||||
/// the lock file itself — contention blocks rather than failing.
|
||||
pub fn acquire_lock(&self) -> bool {
|
||||
match obisys::DirLock::acquire(&self.root_path) {
|
||||
Ok(lock) => {
|
||||
*self.lock.lock().unwrap() = Some(lock);
|
||||
true
|
||||
}
|
||||
Err(_) => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Non-blocking probe of this index's directory lock — doesn't retain
|
||||
/// it either way: `true` if the lock was free (briefly acquired, then
|
||||
/// immediately released), `false` if another process currently holds
|
||||
/// it. For a read-only caller that wants to warn about a build in
|
||||
/// progress without itself becoming exclusive; use
|
||||
/// [`acquire_lock`](Self::acquire_lock)/[`open_lock`](Self::open_lock)
|
||||
/// to actually hold it.
|
||||
pub fn test_lock(&self) -> bool {
|
||||
matches!(obisys::DirLock::try_acquire(&self.root_path), Ok(Some(_)))
|
||||
}
|
||||
|
||||
/// Release this index's directory lock, if held. A no-op if it isn't.
|
||||
pub fn release_lock(&self) {
|
||||
*self.lock.lock().unwrap() = None;
|
||||
}
|
||||
|
||||
/// Return `true` if `path` points to a valid obikmer index directory.
|
||||
///
|
||||
/// A directory is considered an index only when it contains an `index.meta`
|
||||
|
||||
@@ -68,7 +68,7 @@ fn presence_layer_generic_over_sparse_matches_dense() {
|
||||
(kmer.raw().wrapping_add(g as u64)) % 2 == 0
|
||||
}).unwrap();
|
||||
|
||||
let dense_layer = TypedLayer::<PersistentBitMatrix>::open(dir.path(), &mode).unwrap();
|
||||
let dense_layer = TypedLayer::<PersistentBitMatrix>::open(dir.path()).unwrap();
|
||||
assert!(dense_layer.n_cols() >= 1);
|
||||
|
||||
// Build the sparse form directly into the same `presence/` dir the
|
||||
@@ -81,7 +81,7 @@ fn presence_layer_generic_over_sparse_matches_dense() {
|
||||
.close()
|
||||
.unwrap();
|
||||
|
||||
let sparse_layer = TypedLayer::<PersistentSparseBitMatrix>::open(dir.path(), &mode).unwrap();
|
||||
let sparse_layer = TypedLayer::<PersistentSparseBitMatrix>::open(dir.path()).unwrap();
|
||||
|
||||
// Same generic methods, same results, different concrete `D`.
|
||||
assert_eq!(dense_layer.n_cols(), sparse_layer.n_cols());
|
||||
@@ -107,7 +107,7 @@ fn count_layer_reports_count_content_and_columnar_storage() {
|
||||
write_unitigs(dir.path(), &[b"AAAACGT"]);
|
||||
TypedLayer::<PersistentCompactIntMatrix>::build(dir.path(), DEFAULT_BLOCK_BITS, &IndexMode::Exact, |_| 1)
|
||||
.unwrap();
|
||||
let layer = TypedLayer::<PersistentCompactIntMatrix>::open(dir.path(), &IndexMode::Exact).unwrap();
|
||||
let layer = TypedLayer::<PersistentCompactIntMatrix>::open(dir.path()).unwrap();
|
||||
|
||||
assert_eq!(layer.content(), LayerContent::Count);
|
||||
assert_eq!(layer.storage_kind(), obicompactvec::StorageKind::Columnar);
|
||||
@@ -122,7 +122,7 @@ fn presence_layer_reports_presence_content_and_columnar_storage() {
|
||||
TypedLayer::<PersistentBitMatrix>::build_presence(dir.path(), DEFAULT_BLOCK_BITS, &IndexMode::Exact, 2, |kmer, g| {
|
||||
(kmer.raw().wrapping_add(g as u64)) % 2 == 0
|
||||
}).unwrap();
|
||||
let layer = TypedLayer::<PersistentBitMatrix>::open(dir.path(), &IndexMode::Exact).unwrap();
|
||||
let layer = TypedLayer::<PersistentBitMatrix>::open(dir.path()).unwrap();
|
||||
|
||||
assert_eq!(layer.content(), LayerContent::Presence);
|
||||
assert_eq!(layer.storage_kind(), obicompactvec::StorageKind::Columnar);
|
||||
|
||||
@@ -111,5 +111,13 @@ impl KmerPartition {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
// `tests.rs` exercises the old eager `KmerPartition::open`/`.find()`/
|
||||
// `.layers()` cross-layer lookup, removed when `KmerPartition` became the
|
||||
// current lazy `new`/`layer(i)`/`n_layers()` model. That capability's
|
||||
// intended future home is `obikphylo::siblings::cache::PartitionCache` (see
|
||||
// this module's doc comment) — a separate, not-yet-started migration.
|
||||
// Parked here, deliberately excluded from compilation, until that lands;
|
||||
// `tests.rs` itself is left untouched so its assertions can be ported
|
||||
// as-is once `PartitionCache` exists.
|
||||
// #[cfg(test)]
|
||||
// mod tests;
|
||||
|
||||
@@ -10,6 +10,7 @@ use obikseq::SuperKmer;
|
||||
use obiskbuilder::build_superkmers;
|
||||
|
||||
use super::PartitionRouter;
|
||||
use crate::extensions::PrivateBuilder;
|
||||
|
||||
const K: usize = 11;
|
||||
const M: usize = 5;
|
||||
@@ -68,8 +69,8 @@ fn pipeline_counts(seqs: &[&[u8]]) -> (u64, u64) {
|
||||
drop(kp); // ends the borrow of `index` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope
|
||||
Dereplicator::new(&index).run().unwrap();
|
||||
|
||||
let part_dir = index.layer_dir(0, 0);
|
||||
let dedup_path = obikindex::layer::dereplicated_superkmers_path(&part_dir);
|
||||
let layer0 = index.layer0(0).unwrap();
|
||||
let dedup_path = layer0.dereplicated_superkmers_path();
|
||||
if !dedup_path.exists() {
|
||||
return (0, 0);
|
||||
}
|
||||
|
||||
@@ -168,16 +168,6 @@ pub fn run(args: IndexArgs) {
|
||||
let output = args.output.clone();
|
||||
let mut rep = Reporter::new();
|
||||
|
||||
// Locked for the whole build (including a possible --force removal +
|
||||
// recreation below): a second `index` run resuming/overwriting the same
|
||||
// output directory concurrently would otherwise corrupt it. Unlinking
|
||||
// the lock file via --force's remove_dir_all is safe — the held file
|
||||
// descriptor keeps the lock regardless of the directory entry.
|
||||
let _lock = obisys::DirLock::acquire(&output).unwrap_or_else(|e| {
|
||||
eprintln!("error locking output directory {}: {e}", output.display());
|
||||
std::process::exit(1);
|
||||
});
|
||||
|
||||
// ── Resolve evidence kind ────────────────────────────────────────────────
|
||||
let (evidence, effective_kmer_size) = if args.approx {
|
||||
let (z, b, fp) = resolve_approx_params(args.findere_z, args.evidence_bits, args.fp);
|
||||
|
||||
@@ -23,13 +23,13 @@ use super::{DEFAULT_BLOCK_BITS, MAGIC, idx_path};
|
||||
/// both modes. Without `.idx` they fall back to an O(i) sequential scan —
|
||||
/// correct but slower.
|
||||
pub struct UnitigFileReader {
|
||||
mmap: Mmap,
|
||||
mmap: Mmap,
|
||||
block_offsets: Vec<u32>,
|
||||
n_unitigs: usize,
|
||||
n_kmers: usize,
|
||||
k: usize,
|
||||
block_bits: u8,
|
||||
mask: usize, // (1 << block_bits) - 1
|
||||
n_unitigs: usize,
|
||||
n_kmers: usize,
|
||||
k: usize,
|
||||
block_bits: u8,
|
||||
mask: usize, // (1 << block_bits) - 1
|
||||
}
|
||||
|
||||
impl UnitigFileReader {
|
||||
@@ -52,13 +52,13 @@ impl UnitigFileReader {
|
||||
let mmap = unsafe { Mmap::map(&file).map_err(SKError::Io)? };
|
||||
let k = obikseq::params::k();
|
||||
|
||||
let mut offset = 0usize;
|
||||
let mut offset = 0usize;
|
||||
let mut n_unitigs = 0usize;
|
||||
let mut n_kmers = 0usize;
|
||||
let mut n_kmers = 0usize;
|
||||
while offset < mmap.len() {
|
||||
let seql_minus_k = mmap[offset] as usize;
|
||||
n_kmers += seql_minus_k + 1;
|
||||
offset += 1 + (seql_minus_k + k + 3) / 4;
|
||||
n_kmers += seql_minus_k + 1;
|
||||
offset += 1 + (seql_minus_k + k + 3) / 4;
|
||||
n_unitigs += 1;
|
||||
}
|
||||
|
||||
@@ -94,11 +94,22 @@ impl UnitigFileReader {
|
||||
})
|
||||
}
|
||||
|
||||
pub fn len(&self) -> usize { self.n_unitigs }
|
||||
pub fn is_empty(&self) -> bool { self.n_unitigs == 0 }
|
||||
pub fn n_kmers(&self) -> usize { self.n_kmers }
|
||||
pub fn block_bits(&self) -> u8 { self.block_bits }
|
||||
pub fn has_direct_access(&self) -> bool { !self.block_offsets.is_empty() }
|
||||
pub fn len(&self) -> usize {
|
||||
self.n_unitigs
|
||||
}
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.n_unitigs == 0
|
||||
}
|
||||
pub fn n_kmers(&self) -> usize {
|
||||
self.n_kmers
|
||||
}
|
||||
pub fn block_bits(&self) -> u8 {
|
||||
self.block_bits
|
||||
}
|
||||
#[inline]
|
||||
pub fn has_direct_access(&self) -> bool {
|
||||
!self.block_offsets.is_empty()
|
||||
}
|
||||
|
||||
/// Byte offset of record `i` in the mmap.
|
||||
///
|
||||
@@ -106,12 +117,12 @@ impl UnitigFileReader {
|
||||
/// sequential scan otherwise.
|
||||
#[inline]
|
||||
fn chunk_start(&self, i: usize) -> usize {
|
||||
if !self.block_offsets.is_empty() {
|
||||
if self.has_direct_access() {
|
||||
if self.block_bits == 0 {
|
||||
return self.block_offsets[i] as usize;
|
||||
}
|
||||
let block = i >> self.block_bits;
|
||||
let rem = i & self.mask;
|
||||
let rem = i & self.mask;
|
||||
let mut offset = self.block_offsets[block] as usize;
|
||||
for _ in 0..rem {
|
||||
let seql_minus_k = self.mmap[offset] as usize;
|
||||
@@ -136,10 +147,12 @@ impl UnitigFileReader {
|
||||
|
||||
/// Reconstruct chunk `i` as a [`Unitig`].
|
||||
pub fn unitig(&self, i: usize) -> Unitig {
|
||||
let offset = self.chunk_start(i);
|
||||
let seql = self.mmap[offset] as usize + self.k;
|
||||
let offset = self.chunk_start(i);
|
||||
let seql = self.mmap[offset] as usize + self.k;
|
||||
let byte_len = (seql + 3) / 4;
|
||||
let bytes = self.mmap[offset + 1..offset + 1 + byte_len].to_vec().into_boxed_slice();
|
||||
let bytes = self.mmap[offset + 1..offset + 1 + byte_len]
|
||||
.to_vec()
|
||||
.into_boxed_slice();
|
||||
Unitig::new((seql % 4) as u8, bytes)
|
||||
}
|
||||
|
||||
@@ -171,20 +184,26 @@ impl UnitigFileReader {
|
||||
// ── Sequential iterators (O(n) running-offset cursor) ─────────────────────
|
||||
|
||||
pub(crate) fn iter_chunks_sequential(&self) -> impl Iterator<Item = (usize, Unitig)> + '_ {
|
||||
let k = self.k;
|
||||
let k = self.k;
|
||||
let mmap = &*self.mmap;
|
||||
let n = self.n_unitigs;
|
||||
let n = self.n_unitigs;
|
||||
let mut offset = 0usize;
|
||||
(0..n).map(move |chunk_id| {
|
||||
let seql = mmap[offset] as usize + k;
|
||||
let seql = mmap[offset] as usize + k;
|
||||
let byte_len = (seql + 3) / 4;
|
||||
let bytes = mmap[offset + 1..offset + 1 + byte_len].to_vec().into_boxed_slice();
|
||||
let bytes = mmap[offset + 1..offset + 1 + byte_len]
|
||||
.to_vec()
|
||||
.into_boxed_slice();
|
||||
offset += 1 + byte_len;
|
||||
(chunk_id, Unitig::new((seql % 4) as u8, bytes))
|
||||
})
|
||||
}
|
||||
|
||||
/// Iterate all unitigs sequentially. Works without `.idx` (sequential open).
|
||||
/// Iterate all unitigs sequentially, paired with their `chunk_id`.
|
||||
/// Goes through [`iter_chunks_sequential`](Self::iter_chunks_sequential) —
|
||||
/// never touches `chunk_start`/`block_offsets`, so it works identically
|
||||
/// whether opened via `open_sequential` or `open_direct_access` (`.idx`
|
||||
/// loaded or not).
|
||||
pub fn iter_unitigs(&self) -> impl Iterator<Item = (usize, Unitig)> + '_ {
|
||||
self.iter_chunks_sequential()
|
||||
}
|
||||
@@ -194,15 +213,27 @@ impl UnitigFileReader {
|
||||
.flat_map(|(_, u)| u.into_kmers())
|
||||
}
|
||||
|
||||
pub fn iter_canonical_kmers(&self) -> impl Iterator<Item = CanonicalKmer> + '_ {
|
||||
self.iter_chunks_sequential()
|
||||
.flat_map(|(_, u)| u.into_canonical_kmers())
|
||||
}
|
||||
|
||||
/// Sequential scan, `.idx` not required — despite the name, "indexed"
|
||||
/// describes what's returned (each k-mer paired with its `(chunk_id,
|
||||
/// rank)` position), not a dependency on the `.idx` sidecar. Works
|
||||
/// identically whether or not `.idx` exists, since it goes through
|
||||
/// [`iter_chunks_sequential`](Self::iter_chunks_sequential) — never
|
||||
/// `chunk_start`/`block_offsets`. This is what `build_exact_evidence`/
|
||||
/// `build()` use to fill `evidence.bin` from a freshly written
|
||||
/// `unitigs.bin`, before `.idx` exists.
|
||||
pub fn iter_indexed_canonical_kmers(
|
||||
&self,
|
||||
) -> impl Iterator<Item = (CanonicalKmer, usize, usize)> + '_ {
|
||||
self.iter_chunks_sequential()
|
||||
.flat_map(|(chunk_id, u)| {
|
||||
u.into_canonical_kmers()
|
||||
.enumerate()
|
||||
.map(move |(rank, kmer)| (kmer, chunk_id, rank))
|
||||
})
|
||||
self.iter_chunks_sequential().flat_map(|(chunk_id, u)| {
|
||||
u.into_canonical_kmers()
|
||||
.enumerate()
|
||||
.map(move |(rank, kmer)| (kmer, chunk_id, rank))
|
||||
})
|
||||
}
|
||||
|
||||
/// Same streamed sequence as [`iter_indexed_canonical_kmers`](Self::iter_indexed_canonical_kmers),
|
||||
@@ -222,7 +253,9 @@ impl UnitigFileReader {
|
||||
let mmap = &*this.mmap;
|
||||
let seql = mmap[offset] as usize + k;
|
||||
let byte_len = (seql + 3) / 4;
|
||||
let bytes = mmap[offset + 1..offset + 1 + byte_len].to_vec().into_boxed_slice();
|
||||
let bytes = mmap[offset + 1..offset + 1 + byte_len]
|
||||
.to_vec()
|
||||
.into_boxed_slice();
|
||||
offset += 1 + byte_len;
|
||||
(chunk_id, Unitig::new((seql % 4) as u8, bytes))
|
||||
})
|
||||
@@ -238,8 +271,9 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec<u32>)> {
|
||||
let data = std::fs::read(path).map_err(SKError::Io)?;
|
||||
let mut pos = 0;
|
||||
|
||||
let magic_bytes = data.get(pos..pos + 4)
|
||||
.ok_or(SKError::Truncated { context: "unitig index: magic" })?;
|
||||
let magic_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated {
|
||||
context: "unitig index: magic",
|
||||
})?;
|
||||
if magic_bytes != &MAGIC {
|
||||
return Err(SKError::BadMagic {
|
||||
expected: "UIX3",
|
||||
@@ -248,8 +282,9 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec<u32>)> {
|
||||
}
|
||||
pos += 4;
|
||||
|
||||
let bb_bytes = data.get(pos..pos + 4)
|
||||
.ok_or(SKError::Truncated { context: "unitig index: block_bits" })?;
|
||||
let bb_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated {
|
||||
context: "unitig index: block_bits",
|
||||
})?;
|
||||
let block_bits_u32 = u32::from_le_bytes(bb_bytes.try_into().unwrap());
|
||||
if block_bits_u32 > 31 {
|
||||
return Err(SKError::InvalidData {
|
||||
@@ -260,13 +295,15 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec<u32>)> {
|
||||
let block_bits = block_bits_u32 as u8;
|
||||
pos += 4;
|
||||
|
||||
let n_bytes = data.get(pos..pos + 4)
|
||||
.ok_or(SKError::Truncated { context: "unitig index: n_unitigs" })?;
|
||||
let n_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated {
|
||||
context: "unitig index: n_unitigs",
|
||||
})?;
|
||||
let n_unitigs = u32::from_le_bytes(n_bytes.try_into().unwrap()) as usize;
|
||||
pos += 4;
|
||||
|
||||
let nk_bytes = data.get(pos..pos + 8)
|
||||
.ok_or(SKError::Truncated { context: "unitig index: n_kmers" })?;
|
||||
let nk_bytes = data.get(pos..pos + 8).ok_or(SKError::Truncated {
|
||||
context: "unitig index: n_kmers",
|
||||
})?;
|
||||
let n_kmers = u64::from_le_bytes(nk_bytes.try_into().unwrap()) as usize;
|
||||
pos += 8;
|
||||
|
||||
@@ -275,8 +312,9 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec<u32>)> {
|
||||
let n_offsets = n_blocks + 1;
|
||||
let mut block_offsets = Vec::with_capacity(n_offsets);
|
||||
for _ in 0..n_offsets {
|
||||
let off_bytes = data.get(pos..pos + 4)
|
||||
.ok_or(SKError::Truncated { context: "unitig index: block_offsets" })?;
|
||||
let off_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated {
|
||||
context: "unitig index: block_offsets",
|
||||
})?;
|
||||
block_offsets.push(u32::from_le_bytes(off_bytes.try_into().unwrap()));
|
||||
pos += 4;
|
||||
}
|
||||
|
||||
+23
-8
@@ -23,18 +23,33 @@ impl DirLock {
|
||||
/// and the lock file within it if needed). Logs once if the wait is
|
||||
/// non-trivial, so a blocked command doesn't look silently hung.
|
||||
pub fn acquire(dir: &std::path::Path) -> std::io::Result<Self> {
|
||||
std::fs::create_dir_all(dir)?;
|
||||
let lock_path = dir.join(".obikmer.lock");
|
||||
let file = std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.write(true)
|
||||
.open(&lock_path)?;
|
||||
|
||||
let file = Self::open_lock_file(dir)?;
|
||||
if file.try_lock().is_err() {
|
||||
info!(dir = %dir.display(), "waiting for another obikmer process to release this index");
|
||||
file.lock()?;
|
||||
}
|
||||
Ok(Self { _file: file })
|
||||
}
|
||||
|
||||
/// Non-blocking probe: `Ok(Some(lock))` if `dir`'s lock was free and is
|
||||
/// now held by the returned `DirLock`, `Ok(None)` if another process
|
||||
/// currently holds it (not an error — contention is the expected,
|
||||
/// common case this exists to distinguish from a genuine I/O failure).
|
||||
pub fn try_acquire(dir: &std::path::Path) -> std::io::Result<Option<Self>> {
|
||||
let file = Self::open_lock_file(dir)?;
|
||||
match file.try_lock() {
|
||||
Ok(()) => Ok(Some(Self { _file: file })),
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
fn open_lock_file(dir: &std::path::Path) -> std::io::Result<std::fs::File> {
|
||||
std::fs::create_dir_all(dir)?;
|
||||
let lock_path = dir.join(".obikmer.lock");
|
||||
std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.write(true)
|
||||
.open(&lock_path)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user