#[cfg(test)]
use std::sync::Mutex as StdMutex;
use std::{
cmp,
collections::{BTreeMap, BTreeSet, HashMap, HashSet, hash_map::Entry},
env, fmt, fs,
fs::File,
io::{self, BufReader, BufWriter, Read, Write},
marker::PhantomData,
mem,
num::NonZeroUsize,
path::{Path, PathBuf},
sync::{Arc, Mutex, atomic::Ordering},
thread::available_parallelism,
time,
};
use arrow::{
compute::{concat_batches, take},
ipc::writer::StreamWriter,
};
use arrow_array::{
Array, ArrayRef, Decimal128Array, FixedSizeListArray, Float32Array, RecordBatch, UInt32Array,
};
use blake3::Hasher as Blake3Hasher;
use bytes::Bytes;
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use datafusion::prelude::Expr;
use futures::{
future::try_join_all,
stream::{self, StreamExt},
};
use object_store::{MultipartUpload, PutPayload, UploadPart};
use rayon::{ThreadPool, ThreadPoolBuilder, prelude::*};
use serde::{Deserialize, Serialize};
use tempfile::NamedTempFile;
use tokio::time::sleep;
use tracing::{debug, error, info, warn};
use uuid::Uuid;
use super::{
build::fanout_shards,
error::BuildError,
handle::{GLOBAL_VECTOR_KMEANS_ITERS, GLOBAL_VECTOR_KMEANS_SEED, Supertable, SupertableInner},
manifest::{
CellVectorSummary, FtsSummaryAgg, ManifestSnapshot, ScalarStatsAgg, SubsectionOffsets,
SuperfileEntry, SuperfileUri, VectorSummary, bloom::BloomBuilder,
},
mutations::{
CommitError, CommitResult, MAX_TARGETS_PER_MUTATION, MutationError, MutationStats,
PendingDelete, PendingUpdate,
},
opann,
options::{DECIMAL128_PRECISION, DECIMAL128_SCALE, SupertableOptions},
utils::vector_split::split_vectors,
wal::{
WalStore,
pipeline::{self, TombstonePhaseOutcome},
state_doc::{
IdSpan, OpKind, RowId, SCHEMA_VERSION, SupertableHandleId, TombstoneEntry,
TombstoneOutcome, WalId, WalState, WalStateDoc,
},
},
};
use crate::{
InfinoError,
config::{self, CentroidAlignment, DrainConsolidate, ThreadCount},
memory::{ConnectionMemoryBudget, Reservation},
runtime_bridge::{bridge_on_runtime, run_on_pool},
storage::{StorageError, StorageProvider},
superfile::{
BuildError as SuperfileBuildError, ReadError, SuperfileReader,
builder::{SuperfileBuilder, VectorConfig},
format::{
CRC_BYTES,
footer::read_kv_metadata,
fts::{HEADER_SIZE_V1_LEGACY as FTS_HEADER_SIZE, U64_BYTES, hdr},
kv,
vec::{
CELL_DIR_ENTRY_SIZE, CLUSTER_IDX_ENTRY_BYTES, DIR_ENTRY_SIZE, DOC_ID_BYTES,
OUTER_HEADER_SIZE, STABLE_ID_BYTES, SUB_HEADER_SIZE, U32_BYTES, cell_dir_entry,
dir_entry, outer_hdr, sub_hdr,
},
},
reader::vector_layout_from_kv,
vector::{
builder::{
MultiCellSubsectionSource, build_merged_subsection_from_fp32,
build_merged_subsection_from_materialized,
build_merged_subsection_from_spilled_materialized,
},
cell_posting::{EncodedCellRow, MaterializedIvfRow, transcode_clamped_components},
distance::Metric,
ivf_merge::{
MergedIvfSubsection, merge_fragment_subsections, route_clusters_into_cells,
},
kmeans::kmeans_with_assignments,
layout::VectorLayout,
quant::BitQuantizer,
reader::{VectorColumnConfig, VectorReader},
rerank_codec::RerankCodec,
rotation::RandomRotation,
spill::{MaterializedRowSpillState, MaterializedRowSpillWriter, SpilledCellRows},
},
},
supertable::{
CommitError as SupertableCommitError, ManifestLoadError,
error::ManifestError,
hidden_deleted::{self, encode_deleted_ids},
manifest::{
ClusterCentroids, RabitqAdmitContext,
commit::{get_current_manifest_etag, manifest_uri},
list::{
CellRoutingParams, DrainedVersionRanges, GlobalVectorIndex, PartitionStrategy,
WIDTH_LAW_KS,
},
options_hash,
part::{self as part_mod, PartId},
},
query::{
dispatch::{open_compaction_input, open_reader},
vector::stable_ids_by_local_for_routing,
},
reader_cache::{DiskCacheStore, disk::mmap_readonly_bytes},
slow_vector_state::{self, CentroidSection, fetch_centroid_section},
wal::{
Lease,
lease::{self, DEFAULT_LEASE_DURATION},
},
},
};
const DRAIN_FINE_RUN_TARGET_BYTES: usize = 2 * 1024 * 1024;
const SUPERFILE_MULTIPART_PART_BYTES: usize = 8 * (1 << 20);
const DRAIN_ID_BATCH_ROWS: usize = 64 * 1024;
const MIB: usize = 1 << 20;
pub(in crate::supertable) const DRAIN_CHECKPOINT_SCHEMA: u32 = 1;
const DRAIN_LOCAL_CHECKPOINT_FILE: &str = "checkpoint.json";
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct DrainCheckpointSource {
superfile_id: String,
uri: String,
birth_version: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct DrainRemoteShard {
shard_id: u32,
superfile_id: String,
cell_counts: Vec<(u32, u32)>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct DrainRemoteCheckpoint {
schema: u32,
epoch_id: String,
options_hash: String,
sources: Vec<DrainCheckpointSource>,
batch_layout: Vec<Vec<u64>>,
shard_count: usize,
completed_shards: Vec<DrainRemoteShard>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct DrainLocalSpill {
n_rows: u32,
n_quants: u32,
dim: usize,
rabitq_len: usize,
rerank_codec_id: u8,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct DrainLocalCell {
n_docs: u32,
subsection_len: u64,
rerank_codec_id: u8,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct DrainLocalCheckpoint {
schema: u32,
epoch_id: String,
batches_done: usize,
spills: HashMap<u32, DrainLocalSpill>,
built_cells: HashMap<u32, DrainLocalCell>,
added_per_cell: HashMap<u32, u32>,
}
impl DrainLocalCheckpoint {
fn new(epoch_id: String) -> Self {
Self {
schema: DRAIN_CHECKPOINT_SCHEMA,
epoch_id,
batches_done: 0,
spills: HashMap::new(),
built_cells: HashMap::new(),
added_per_cell: HashMap::new(),
}
}
}
struct DrainRemoteState {
checkpoint: DrainRemoteCheckpoint,
entries: Vec<Arc<SuperfileEntry>>,
}
#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DrainTestFailurePhase {
AfterBatch,
AfterShard,
}
#[cfg(test)]
struct DrainTestFailure {
phase: DrainTestFailurePhase,
completed: usize,
}
#[cfg(test)]
static DRAIN_TEST_FAILURES: StdMutex<Option<HashMap<String, DrainTestFailure>>> =
StdMutex::new(None);
#[cfg(test)]
fn inject_drain_test_failure(epoch_id: String, phase: DrainTestFailurePhase, completed: usize) {
let mut guard = DRAIN_TEST_FAILURES.lock().expect("drain test failure lock");
let map = guard.get_or_insert_with(HashMap::new);
map.insert(epoch_id, DrainTestFailure { phase, completed });
}
#[cfg(test)]
fn maybe_fail_drain_for_test(
epoch_id: &str,
phase: DrainTestFailurePhase,
completed: usize,
) -> Result<(), BuildError> {
let mut guard = DRAIN_TEST_FAILURES.lock().expect("drain test failure lock");
let Some(map) = guard.as_mut() else {
return Ok(());
};
let should_fail = map
.get(epoch_id)
.is_some_and(|failure| failure.phase == phase && completed >= failure.completed);
if should_fail {
map.remove(epoch_id);
return Err(BuildError::Store(format!(
"injected drain failure after {phase:?} {completed}"
)));
}
Ok(())
}
const BUILD_SCRATCH_DENOM: usize = 2;
const BUILD_SCALAR_NUM: usize = 5;
const BUILD_VECTOR_NUM: usize = 13;
const BUILD_FTS_NUM: usize = 3;
pub struct SupertableWriter {
inner: Arc<SupertableInner>,
buffer: Vec<BufferedBatch>,
buffer_scalar_bytes: usize,
buffer_vector_bytes: usize,
buffer_fts_bytes: usize,
pending_updates: Vec<PendingUpdateEntry>,
pending_deletes: Vec<PendingDeleteEntry>,
}
struct PendingUpdateEntry {
wal_id: WalId,
target_ids: Vec<i128>,
preallocated_superfile_id: uuid::Uuid,
minted_id_spans: Vec<IdSpan>,
new_row_count: u32,
new_row_content_hash: String,
ipc_bytes: Bytes,
}
struct PendingDeleteEntry {
wal_id: WalId,
target_ids: Vec<i128>,
}
impl fmt::Debug for SupertableWriter {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("SupertableWriter")
.field("buffered_batches", &self.buffer.len())
.field("buffered_bytes", &self.buffered_bytes())
.field("manifest_id", &self.inner.manifest.load().manifest_id)
.finish()
}
}
#[derive(Clone)]
struct BufferedBatch {
scalar: RecordBatch,
vectors: Vec<Arc<Float32Array>>,
}
struct VectorColumnView<'a> {
dim: usize,
batches: Vec<&'a [f32]>,
offsets: Vec<usize>,
}
impl<'a> VectorColumnView<'a> {
fn over(buffer: &'a [BufferedBatch], col_idx: usize, dim: usize) -> Self {
let mut batches = Vec::with_capacity(buffer.len());
let mut offsets = Vec::with_capacity(buffer.len() + 1);
let mut total = 0usize;
for buffered in buffer {
offsets.push(total);
let values: &[f32] = buffered.vectors[col_idx].values();
total += values.len() / dim.max(1);
batches.push(values);
}
offsets.push(total);
Self {
dim,
batches,
offsets,
}
}
fn n_rows(&self) -> usize {
self.offsets.last().copied().unwrap_or(0)
}
fn row(&self, local: usize) -> Result<&'a [f32], BuildError> {
let batch = self
.offsets
.partition_point(|&first_row| first_row <= local)
.saturating_sub(1);
let in_batch = local
.checked_sub(self.offsets[batch])
.ok_or_else(|| BuildError::Store(format!("vector row {local} before batch start")))?;
let start = in_batch * self.dim;
self.batches
.get(batch)
.and_then(|values| values.get(start..start + self.dim))
.ok_or_else(|| BuildError::Store(format!("vector row {local} out of buffered range")))
}
}
fn superfiles_per_commit(
total_rows: usize,
buffered_bytes: usize,
pool_threads: usize,
target_bytes: usize,
) -> usize {
let by_bytes = if target_bytes == 0 {
usize::MAX
} else {
buffered_bytes.div_ceil(target_bytes).max(1)
};
by_bytes.min(pool_threads.max(1)).min(total_rows.max(1))
}
fn split_buffer_into_superfile_inputs(
buffer: Vec<BufferedBatch>,
n_superfiles: usize,
vector_dims: &[usize],
) -> Vec<Vec<BufferedBatch>> {
debug_assert!(n_superfiles > 0);
let total_rows: usize = buffer.iter().map(|b| b.scalar.num_rows()).sum();
if total_rows == 0 {
return Vec::new();
}
let base = total_rows / n_superfiles;
let remainder = total_rows % n_superfiles;
let target = |i: usize| if i < remainder { base + 1 } else { base };
let mut pieces: Vec<Vec<BufferedBatch>> = (0..n_superfiles).map(|_| Vec::new()).collect();
let mut piece_idx = 0usize;
let mut piece_remaining = target(0);
for batch in buffer {
let n_rows = batch.scalar.num_rows();
if n_rows == 0 {
continue;
}
let mut row_cursor = 0;
while row_cursor < n_rows {
while piece_remaining == 0 && piece_idx + 1 < n_superfiles {
piece_idx += 1;
piece_remaining = target(piece_idx);
}
let take = cmp::min(piece_remaining, n_rows - row_cursor);
let scalar = batch.scalar.slice(row_cursor, take);
let vectors: Vec<Arc<Float32Array>> = batch
.vectors
.iter()
.enumerate()
.map(|(i, v)| {
let dim = vector_dims[i];
Arc::new(v.slice(row_cursor * dim, take * dim))
})
.collect();
pieces[piece_idx].push(BufferedBatch { scalar, vectors });
row_cursor += take;
piece_remaining -= take;
}
}
pieces.retain(|s| !s.is_empty());
pieces
}
fn schedule_background_storage_reclaim(inner: Arc<SupertableInner>) {
if inner.options.storage.is_none() {
return;
}
#[cfg(not(test))]
{
let rt = inner.query_runtime();
rt.spawn(async move {
sleep(super::gc::DEFAULT_SUPERFILE_RECLAIM_GRACE).await;
if let Err(e) = super::gc::gc_storage_sweep_for_inner(
&inner,
super::gc::DEFAULT_SUPERFILE_RECLAIM_GRACE,
)
.await
{
tracing::debug!("supertable: deferred storage reclaim: {e}");
}
});
}
#[cfg(test)]
{
let _ = inner;
}
}
async fn materialized_ivf_rows_in_doc_order(
vec_reader: &VectorReader,
column: &str,
stable_ids_by_local: &[i128],
tombstones: Option<&roaring::RoaringBitmap>,
) -> Result<Vec<MaterializedIvfRow>, BuildError> {
let mut rows = vec_reader
.materialized_index_rows_async(column)
.await
.ok_or_else(|| {
BuildError::Store(format!(
"IVF maintenance: column '{column}' missing Sq8Residual index"
))
})?;
let n_rows = stable_ids_by_local.len();
let mut by_local = vec![None; n_rows];
for row in &mut rows {
if tombstones.is_some_and(|bm| bm.contains(row.local_doc_id)) {
continue;
}
let slot = row.local_doc_id as usize;
if slot < n_rows {
if row.stable_id == 0 {
row.stable_id = stable_ids_by_local[slot];
row.encoded.stable_id = row.stable_id;
}
by_local[slot] = Some(row.clone());
}
}
Ok(by_local
.into_iter()
.enumerate()
.filter_map(|(i, r)| {
r.map(|mut row| {
row.local_doc_id = i as u32;
row
})
})
.collect())
}
fn split_buffer_by_vector_cell(
buffer: Vec<BufferedBatch>,
cells: &ClusterCentroids,
metric: Metric,
vec_col_idx: usize,
) -> Result<Vec<(u32, Vec<BufferedBatch>)>, BuildError> {
let k = cells.n_cent as usize;
let mut cell_batches: Vec<Vec<BufferedBatch>> = (0..k).map(|_| Vec::new()).collect();
for batch in buffer {
let n_rows = batch.scalar.num_rows();
if n_rows == 0 {
continue;
}
let vecs = batch.vectors[vec_col_idx].values();
let mut assignments = vec![0u32; n_rows];
cells.assign_rows(metric, vecs, &mut assignments);
let mut per_cell_rows: Vec<Vec<usize>> = (0..k).map(|_| Vec::new()).collect();
for (row, &cell) in assignments.iter().enumerate() {
per_cell_rows
.get_mut(cell as usize)
.ok_or_else(|| {
BuildError::Store(format!(
"vector-cell split: row {row} assigned to out-of-range cell {cell} (k={k})"
))
})?
.push(row);
}
for (cell_id, rows) in per_cell_rows.into_iter().enumerate() {
if rows.is_empty() {
continue;
}
let indices = UInt32Array::from(rows.iter().map(|&r| r as u32).collect::<Vec<_>>());
let scalar_cols: Vec<ArrayRef> = (0..batch.scalar.num_columns())
.map(|col_idx| {
take(batch.scalar.column(col_idx), &indices, None).map_err(|e| {
BuildError::Store(format!(
"vector-cell split: take column {col_idx} for cell {cell_id}: {e}"
))
})
})
.collect::<Result<_, _>>()?;
let scalar_batch =
RecordBatch::try_new(batch.scalar.schema(), scalar_cols).map_err(|e| {
BuildError::Store(format!(
"vector-cell split: rebuild batch for cell {cell_id}: {e}"
))
})?;
let vectors: Vec<Arc<Float32Array>> = batch
.vectors
.iter()
.map(|v| -> Result<Arc<Float32Array>, BuildError> {
if v.len() % n_rows != 0 {
return Err(BuildError::Store(format!(
"vector-cell split: {} values do not divide across {n_rows} rows",
v.len()
)));
}
let vdim = v.len() / n_rows;
let mut out = Vec::with_capacity(rows.len() * vdim);
for &r in &rows {
out.extend_from_slice(&v.values()[r * vdim..(r + 1) * vdim]);
}
Ok(Arc::new(Float32Array::from(out)))
})
.collect::<Result<_, _>>()?;
cell_batches[cell_id].push(BufferedBatch {
scalar: scalar_batch,
vectors,
});
}
}
Ok(cell_batches
.into_iter()
.enumerate()
.filter(|(_, batches)| !batches.is_empty())
.map(|(cell_id, batches)| (cell_id as u32, batches))
.collect())
}
fn single_outcome(res: CommitResult) -> Result<MutationStats, InfinoError> {
res.outcomes
.into_iter()
.next()
.ok_or_else(|| InfinoError::Backend("commit produced no mutation outcome".to_string()))
}
fn target_resolve_err(e: ManifestLoadError) -> MutationError {
match e {
ManifestLoadError::PointerVanished | ManifestLoadError::PointerNotFound => {
MutationError::TableGone
}
other => MutationError::TargetResolve(other),
}
}
impl Supertable {
#[cfg_attr(
feature = "detailed-tracing",
tracing::instrument(skip_all, fields(rows = batch.num_rows()))
)]
pub fn append(&self, batch: &RecordBatch) -> Result<(), InfinoError> {
let mut w = self
.writer()
.map_err(|e| InfinoError::from(e).with_context("append", None))?;
w.append(batch)
.map_err(|e| InfinoError::from(e).with_context("append", None))?;
w.commit()
.map_err(|e| InfinoError::from(e).with_context("append", None))?;
Ok(())
}
#[cfg_attr(
feature = "detailed-tracing",
tracing::instrument(skip_all, fields(new_rows = new_rows.num_rows()))
)]
pub fn update(
&self,
predicate: Expr,
new_rows: &RecordBatch,
) -> Result<MutationStats, InfinoError> {
let mut w = self
.writer()
.map_err(|e| InfinoError::from(e).with_context("update", None))?;
let pending = w
.update(predicate, new_rows.clone())
.map_err(|e| InfinoError::from(e).with_context("update", None))?;
if pending.matched == 0 {
return Ok(MutationStats::empty());
}
single_outcome(
w.commit()
.map_err(|e| InfinoError::from(e).with_context("update", None))?,
)
.map_err(|e| e.with_context("update", None))
}
#[cfg_attr(feature = "detailed-tracing", tracing::instrument(skip_all))]
pub fn delete(&self, predicate: Expr) -> Result<MutationStats, InfinoError> {
let mut w = self
.writer()
.map_err(|e| InfinoError::from(e).with_context("delete", None))?;
w.delete(predicate)
.map_err(|e| InfinoError::from(e).with_context("delete", None))?;
single_outcome(
w.commit()
.map_err(|e| InfinoError::from(e).with_context("delete", None))?,
)
.map_err(|e| e.with_context("delete", None))
}
test_visible! {
fn writer(&self) -> Result<SupertableWriter, BuildError> {
if self.inner().options.summary_centroids_from_superfiles {
return Err(BuildError::Store(
"this handle opened in consumer memory mode \
(summary_centroids_from_superfiles): summaries hydrate without fp32, so it \
cannot write — open a writer handle with the mode off"
.into(),
));
}
match self.inner().writer_outstanding.compare_exchange(
false,
true,
Ordering::Acquire,
Ordering::Relaxed,
) {
Ok(_) => Ok(SupertableWriter {
inner: Arc::clone(self.inner()),
buffer: Vec::new(),
buffer_scalar_bytes: 0,
buffer_vector_bytes: 0,
buffer_fts_bytes: 0,
pending_updates: Vec::new(),
pending_deletes: Vec::new(),
}),
Err(_) => Err(BuildError::SupertableInUse),
}
}
}
}
fn bootstrap_centroids_from_batch(
batches: &[BufferedBatch],
vec_dim: usize,
n_cells: usize,
) -> Option<ClusterCentroids> {
let mut vectors = Vec::new();
for batch in batches {
let Some(first) = batch.vectors.first() else {
continue;
};
let vecs = first.values();
let n_rows = batch.scalar.num_rows();
let expected = n_rows.checked_mul(vec_dim)?;
if vecs.len() < expected {
return None;
}
vectors.extend_from_slice(&vecs[..expected]);
}
let n_docs = vectors.len() / vec_dim;
if n_docs == 0 {
return None;
}
let k = n_cells.min(n_docs).max(1);
let (centroids, assignments) = kmeans_with_assignments(
&vectors,
vec_dim,
k,
GLOBAL_VECTOR_KMEANS_ITERS,
GLOBAL_VECTOR_KMEANS_SEED,
);
let mut counts = vec![0u32; k];
for &a in &assignments {
counts[a as usize] += 1;
}
Some(ClusterCentroids::from_fp32(
k as u32,
vec_dim as u32,
¢roids,
counts,
))
}
impl SupertableWriter {
pub fn buffered_batches(&self) -> usize {
self.buffer.len()
}
pub fn buffered_bytes(&self) -> usize {
self.buffer_scalar_bytes + self.buffer_vector_bytes
}
#[cfg_attr(
feature = "detailed-tracing",
tracing::instrument(skip_all, fields(rows = batch.num_rows(), buffered = self.buffer.len()))
)]
pub fn append(&mut self, batch: &RecordBatch) -> Result<(), BuildError> {
let options = &self.inner.options;
let (scalar_no_id, _vector_slices) = split_vectors(batch, options)?;
let mut vectors = Vec::with_capacity(options.vector_columns.len());
for vc in &options.vector_columns {
let col_idx = batch
.schema()
.index_of(&vc.column)
.map_err(|_| BuildError::BatchSchemaMismatch)?;
let fsl = batch
.column(col_idx)
.as_any()
.downcast_ref::<FixedSizeListArray>()
.ok_or(BuildError::BatchSchemaMismatch)?;
let values = fsl.values();
let f32_arr = values
.as_any()
.downcast_ref::<Float32Array>()
.ok_or(BuildError::BatchSchemaMismatch)?
.clone();
vectors.push(Arc::new(f32_arr));
}
let n_rows = scalar_no_id.num_rows();
let mut ids: Vec<i128> = Vec::with_capacity(n_rows);
{
let generator = self
.inner
.id_generator
.lock()
.expect("id_generator mutex poisoned");
for _ in 0..n_rows {
ids.push(generator.next_id());
}
}
let id_array = Decimal128Array::from(ids)
.with_precision_and_scale(DECIMAL128_PRECISION, DECIMAL128_SCALE)
.expect(
"invariant: precision 38 + scale 0 always valid \
for any i128 payload",
);
let mut columns: Vec<ArrayRef> = Vec::with_capacity(scalar_no_id.num_columns() + 1);
columns.push(Arc::new(id_array));
columns.extend(scalar_no_id.columns().iter().cloned());
let scalar = RecordBatch::try_new(options.scalar_schema(), columns)
.map_err(|_| BuildError::BatchSchemaMismatch)?;
let scalar_bytes = scalar.get_array_memory_size();
let vector_bytes = vectors
.iter()
.map(|v| v.len() * mem::size_of::<f32>())
.sum::<usize>();
let fts_bytes = options
.fts_columns
.iter()
.filter_map(|fc| scalar.schema().index_of(&fc.column).ok())
.map(|idx| scalar.column(idx).get_array_memory_size())
.sum::<usize>();
self.buffer.push(BufferedBatch { scalar, vectors });
self.buffer_scalar_bytes += scalar_bytes;
self.buffer_vector_bytes += vector_bytes;
self.buffer_fts_bytes += fts_bytes;
let threshold = (options.commit_threshold_size_mb as usize)
.saturating_mul(1024)
.saturating_mul(1024);
if threshold > 0 && self.buffered_bytes() >= threshold {
self.commit_appends_internal()?;
}
Ok(())
}
pub fn delete(&mut self, predicate: Expr) -> Result<PendingDelete, MutationError> {
let _ = self
.inner
.options
.storage
.as_ref()
.ok_or(MutationError::NoStorageAttached)?;
let supertable = Supertable::from_inner(Arc::clone(&self.inner));
let target_ids = supertable
.reader_strong()
.map_err(target_resolve_err)?
.scan_ids_matching(predicate)
.map_err(MutationError::PredicateEval)?;
let matched = target_ids.len();
if matched > MAX_TARGETS_PER_MUTATION {
return Err(MutationError::MatchCountExceedsCap {
matched,
cap: MAX_TARGETS_PER_MUTATION,
});
}
let wal_id_value = self
.inner
.id_generator
.lock()
.expect("id_generator mutex poisoned")
.next_id();
self.pending_deletes.push(PendingDeleteEntry {
wal_id: WalId(wal_id_value),
target_ids,
});
Ok(PendingDelete { matched })
}
pub fn update(
&mut self,
predicate: Expr,
new_rows: RecordBatch,
) -> Result<PendingUpdate, MutationError> {
let _ = self
.inner
.options
.storage
.as_ref()
.ok_or(MutationError::NoStorageAttached)?;
if new_rows.schema().as_ref() != self.inner.options.schema.as_ref() {
return Err(MutationError::SchemaMismatch(format!(
"expected {:?}, got {:?}",
self.inner.options.schema.fields(),
new_rows.schema().fields()
)));
}
let supertable = Supertable::from_inner(Arc::clone(&self.inner));
let target_ids = supertable
.reader_strong()
.map_err(target_resolve_err)?
.scan_ids_matching(predicate)
.map_err(MutationError::PredicateEval)?;
let matched = target_ids.len();
if matched > MAX_TARGETS_PER_MUTATION {
return Err(MutationError::MatchCountExceedsCap {
matched,
cap: MAX_TARGETS_PER_MUTATION,
});
}
let new_row_count = new_rows.num_rows();
if matched != new_row_count {
return Err(MutationError::CardinalityMismatch {
matched,
new_rows: new_row_count,
});
}
if matched == 0 {
return Ok(PendingUpdate { matched: 0 });
}
let (wal_id_value, minted_id_spans, preallocated_superfile_id) = {
let idgen = self.inner.id_generator.lock().expect("idgen mutex");
let spans = idgen
.reserve_range(matched as u32)
.into_iter()
.map(|(first, last)| IdSpan {
first: RowId(first),
last: RowId(last),
})
.collect::<Vec<_>>();
let wal_id_value = idgen.next_id();
let preallocated = uuid::Uuid::new_v4();
(wal_id_value, spans, preallocated)
};
let ipc_bytes = encode_record_batch_ipc(&new_rows).map_err(|e| {
MutationError::Storage(StorageError::Permanent {
uri: "ipc encode".into(),
source: Box::new(io::Error::other(e)),
})
})?;
let content_hash = blake3::hash(&ipc_bytes).to_hex().to_string();
self.pending_updates.push(PendingUpdateEntry {
wal_id: WalId(wal_id_value),
target_ids,
preallocated_superfile_id,
minted_id_spans,
new_row_count: matched as u32,
new_row_content_hash: content_hash,
ipc_bytes,
});
Ok(PendingUpdate { matched })
}
#[cfg_attr(
feature = "detailed-tracing",
tracing::instrument(skip_all, fields(
buffered = self.buffer.len(),
updates = self.pending_updates.len(),
deletes = self.pending_deletes.len(),
))
)]
pub fn commit(&mut self) -> Result<CommitResult, CommitError> {
if !self.buffer.is_empty() {
self.commit_appends_internal()
.map_err(CommitError::AppendFlush)?;
}
let total_mutations = self.pending_updates.len() + self.pending_deletes.len();
let mut committed_wal_ids: Vec<WalId> = Vec::with_capacity(total_mutations);
let mut outcomes: Vec<MutationStats> = Vec::with_capacity(total_mutations);
let mut updates_to_run = mem::take(&mut self.pending_updates);
let mut update_cursor = 0usize;
while update_cursor < updates_to_run.len() {
let entry = &updates_to_run[update_cursor];
match self.drive_one_update(entry) {
Ok(outcome) => {
committed_wal_ids.push(outcome.wal_id);
outcomes.push(outcome);
update_cursor += 1;
}
Err(cause) => {
let remaining: Vec<PendingUpdateEntry> =
updates_to_run.split_off(update_cursor + 1);
self.pending_updates = remaining;
error!(
committed = outcomes.len(),
total = total_mutations,
error = %cause,
"partial commit: update failed mid-flush"
);
return Err(CommitError::PartialCommit {
committed_wal_ids,
committed: outcomes.len(),
total: total_mutations,
cause: Box::new(cause),
});
}
}
}
let mut deletes_to_run = mem::take(&mut self.pending_deletes);
let mut delete_cursor = 0usize;
while delete_cursor < deletes_to_run.len() {
let entry = &deletes_to_run[delete_cursor];
match self.drive_one_delete(entry) {
Ok(outcome) => {
committed_wal_ids.push(outcome.wal_id);
outcomes.push(outcome);
delete_cursor += 1;
}
Err(cause) => {
let remaining: Vec<PendingDeleteEntry> =
deletes_to_run.split_off(delete_cursor + 1);
self.pending_deletes = remaining;
error!(
committed = outcomes.len(),
total = total_mutations,
error = %cause,
"partial commit: delete failed mid-flush"
);
return Err(CommitError::PartialCommit {
committed_wal_ids,
committed: outcomes.len(),
total: total_mutations,
cause: Box::new(cause),
});
}
}
}
Ok(CommitResult {
wal_ids: committed_wal_ids,
outcomes,
})
}
fn update_wal_doc(&self, entry: &PendingUpdateEntry, now: DateTime<Utc>) -> WalStateDoc {
let lease_span = ChronoDuration::from_std(DEFAULT_LEASE_DURATION)
.expect("default lease duration should be a valid chronoduration");
WalStateDoc {
wal_id: entry.wal_id,
schema_version: SCHEMA_VERSION,
op_kind: OpKind::Update,
state: WalState::Intent,
created_at: now,
lease: Some(Lease {
owner: self.inner.handle_id,
acquired_at: now,
expires_at: now + lease_span,
}),
predicate_repr: "writer.update()".into(),
target_ids: entry.target_ids.iter().map(|&v| RowId(v)).collect(),
new_row_count: Some(entry.new_row_count),
new_row_content_hash: Some(entry.new_row_content_hash.clone()),
preallocated_superfile_id: Some(entry.preallocated_superfile_id),
minted_id_spans: entry.minted_id_spans.clone(),
tombstone_progress: entry
.target_ids
.iter()
.map(|&v| TombstoneEntry {
target_id: RowId(v),
outcome: TombstoneOutcome::Pending,
tombstoned_in_superfile: None,
})
.collect(),
}
}
fn drive_one_update(&self, entry: &PendingUpdateEntry) -> Result<MutationStats, MutationError> {
let storage = self
.inner
.options
.storage
.as_ref()
.ok_or(MutationError::NoStorageAttached)?
.clone();
let wal_doc = self.update_wal_doc(entry, Utc::now());
let wal_store = WalStore::new(Arc::clone(&storage));
let supertable = Supertable::from_inner(Arc::clone(&self.inner));
let wal_id = entry.wal_id;
let ipc_bytes = entry.ipc_bytes.clone();
let owner = self.inner.handle_id;
let drive = async move {
wal_store
.put_arrow(wal_id, ipc_bytes)
.await
.map_err(MutationError::WalStore)?;
let etag = wal_store
.create(&wal_doc)
.await
.map_err(MutationError::WalStore)?;
let append = pipeline::run_append_phase(&supertable, &wal_store, &wal_doc, &etag).await;
let (_outcome, doc_after_append, etag_after_append) = match append {
Ok(appended) => appended,
Err(cause) => {
release_mutation_lease(&wal_store, wal_id, owner).await;
return Err(cause.into());
}
};
let tombstone = pipeline::run_tombstone_phase(
&supertable,
&wal_store,
&doc_after_append,
&etag_after_append,
)
.await;
let (outcome, _post, _post_etag) = match tombstone {
Ok(applied) => applied,
Err(cause) => {
release_mutation_lease(&wal_store, wal_id, owner).await;
return Err(cause.into());
}
};
let (n_t, n_nf) = match outcome {
TombstonePhaseOutcome::Applied {
n_tombstoned,
n_not_found,
}
| TombstonePhaseOutcome::AlreadyComplete {
n_tombstoned,
n_not_found,
} => (n_tombstoned, n_not_found),
};
let _ = wal_store.delete_arrow(wal_id).await;
let _ = wal_store.delete_state(wal_id).await;
Ok::<_, MutationError>((n_t, n_nf))
};
let (n_tombstoned, n_not_found) = bridge_on_runtime(drive, &self.inner.query_runtime())?;
Ok(MutationStats {
wal_id: entry.wal_id,
matched: entry.target_ids.len(),
n_tombstoned,
n_not_found,
})
}
fn delete_wal_doc(&self, entry: &PendingDeleteEntry, now: DateTime<Utc>) -> WalStateDoc {
let lease_span = ChronoDuration::from_std(DEFAULT_LEASE_DURATION)
.expect("default lease duration should be a valid chronoduration");
WalStateDoc {
wal_id: entry.wal_id,
schema_version: SCHEMA_VERSION,
op_kind: OpKind::Delete,
state: WalState::Intent,
created_at: now,
lease: Some(Lease {
owner: self.inner.handle_id,
acquired_at: now,
expires_at: now + lease_span,
}),
predicate_repr: "writer.delete()".into(),
target_ids: entry.target_ids.iter().map(|&v| RowId(v)).collect(),
new_row_count: None,
new_row_content_hash: None,
preallocated_superfile_id: None,
minted_id_spans: Vec::new(),
tombstone_progress: entry
.target_ids
.iter()
.map(|&v| TombstoneEntry {
target_id: RowId(v),
outcome: TombstoneOutcome::Pending,
tombstoned_in_superfile: None,
})
.collect(),
}
}
fn drive_one_delete(&self, entry: &PendingDeleteEntry) -> Result<MutationStats, MutationError> {
let storage = self
.inner
.options
.storage
.as_ref()
.ok_or(MutationError::NoStorageAttached)?
.clone();
let wal_doc = self.delete_wal_doc(entry, Utc::now());
let wal_store = WalStore::new(Arc::clone(&storage));
let supertable = Supertable::from_inner(Arc::clone(&self.inner));
let wal_id = entry.wal_id;
let hidden_inner = self
.inner
.vector_index_table
.as_ref()
.map(|vit| Arc::clone(vit.inner()));
let deleted_ids: Vec<i128> = entry.target_ids.clone();
let owner = self.inner.handle_id;
let drive = async move {
let etag = wal_store
.create(&wal_doc)
.await
.map_err(MutationError::WalStore)?;
let phase =
pipeline::run_tombstone_phase(&supertable, &wal_store, &wal_doc, &etag).await;
let (outcome, _post, _post_etag) = match phase {
Ok(applied) => applied,
Err(cause) => {
release_mutation_lease(&wal_store, wal_id, owner).await;
return Err(cause.into());
}
};
let (n_t, n_nf) = match outcome {
TombstonePhaseOutcome::Applied {
n_tombstoned,
n_not_found,
}
| TombstonePhaseOutcome::AlreadyComplete {
n_tombstoned,
n_not_found,
} => (n_tombstoned, n_not_found),
};
let _ = wal_store.delete_state(wal_id).await;
if let Some(hi) = hidden_inner
&& let Err(e) = record_hidden_deleted_ids(&hi, &deleted_ids).await
{
tracing::warn!(
"supertable: hidden vector-index deleted-set record failed: {e} \
(user-table delete is durable; vector search may transiently \
return deleted rows until the next successful record)"
);
}
Ok::<_, MutationError>((n_t, n_nf))
};
let (n_tombstoned, n_not_found) = bridge_on_runtime(drive, &self.inner.query_runtime())?;
Ok(MutationStats {
wal_id: entry.wal_id,
matched: entry.target_ids.len(),
n_tombstoned,
n_not_found,
})
}
#[cfg_attr(
feature = "detailed-tracing",
tracing::instrument(skip_all, fields(buffered = self.buffer.len()))
)]
fn commit_appends_internal(&mut self) -> Result<(), BuildError> {
if self.buffer.is_empty() {
return Ok::<(), BuildError>(());
}
let _build_guard = reserve_build_scratch(
&self.inner.options.connection_memory_budget,
self.buffer_scalar_bytes,
self.buffer_vector_bytes,
self.buffer_fts_bytes,
)?;
let saved_scalar = self.buffer_scalar_bytes;
let saved_vector = self.buffer_vector_bytes;
let saved_fts = self.buffer_fts_bytes;
let buffer = mem::take(&mut self.buffer);
self.buffer_scalar_bytes = 0;
self.buffer_vector_bytes = 0;
self.buffer_fts_bytes = 0;
match self.commit_appends_with_taken_buffer(&buffer) {
Ok(()) => Ok(()),
Err(e) => {
self.buffer = buffer;
self.buffer_scalar_bytes = saved_scalar;
self.buffer_vector_bytes = saved_vector;
self.buffer_fts_bytes = saved_fts;
Err(e)
}
}
}
fn commit_appends_with_taken_buffer(&self, buffer: &[BufferedBatch]) -> Result<(), BuildError> {
let pending_gvi: Option<GlobalVectorIndex> = if self
.inner
.manifest
.load()
.get_global_vector_index()
.is_none()
&& !buffer.is_empty()
&& let Some(vc) = self.inner.options.vector_columns.first()
&& let Some(grid) = bootstrap_centroids_from_batch(
buffer,
vc.dim,
super::handle::hidden_vector_cell_count(&self.inner.options),
) {
let hidden_cells = super::handle::hidden_vector_cell_count(&self.inner.options);
let user_cells = super::handle::user_vector_cell_count(&self.inner.options);
let user_grid = (user_cells != hidden_cells)
.then(|| bootstrap_centroids_from_batch(buffer, vc.dim, user_cells))
.flatten();
Some(GlobalVectorIndex {
column: vc.column.clone(),
grid,
user_grid,
})
} else {
None
};
let total_rows: usize = buffer.iter().map(|b| b.scalar.num_rows()).sum();
if total_rows == 0 {
return Ok(());
}
let list_metadata = CommitListMetadata {
partition_strategy: None,
global_vector_index: pending_gvi.clone(),
drained_ranges: None,
superseded_cells_additions: None,
};
if !self.inner.options.vector_columns.is_empty() {
let commit_t0 = time::Instant::now();
let pack_grid = pending_gvi
.as_ref()
.cloned()
.or_else(|| self.inner.manifest.load().get_global_vector_index())
.ok_or_else(|| {
BuildError::Store(
"vector columns present but global cell grid missing after Phase A".into(),
)
})?
.into_user_grid();
let metric = self
.inner
.options
.vector_columns
.first()
.map(|vc| vc.metric)
.unwrap_or(Metric::L2Sq);
let (outputs, cell_hints) =
commit_shards_via_drain(buffer, &self.inner, &pack_grid, metric)?;
let build_elapsed = commit_t0.elapsed();
let output_bytes: usize = outputs.iter().map(|output| output.bytes.len()).sum();
let user_batch = prepare_user_superfile_batch(&self.inner, outputs, cell_hints)?;
let prepare_elapsed = commit_t0.elapsed().saturating_sub(build_elapsed);
let data_put_bytes: usize = user_batch
.pending_storage_writes
.iter()
.map(|(_, bytes)| bytes.len())
.sum();
let publish_t0 = time::Instant::now();
bridge_on_runtime(
persist_superfile_publish_batch_async(&self.inner, user_batch, list_metadata),
&self.inner.query_runtime(),
)?;
if crate::storage::io_counters::timeline_enabled() {
eprintln!(
"[supertable commit] build {:.1}ms ({:.1} MiB output) + prepare {:.1}ms + \
publish {:.1}ms ({:.1} MiB data PUT)",
build_elapsed.as_secs_f64() * 1e3,
output_bytes as f64 / (1u64 << 20) as f64,
prepare_elapsed.as_secs_f64() * 1e3,
publish_t0.elapsed().as_secs_f64() * 1e3,
data_put_bytes as f64 / (1u64 << 20) as f64,
);
}
if self.inner.options.storage.is_some() {
schedule_background_storage_reclaim(Arc::clone(&self.inner));
}
return Ok(());
}
let writer_pool = Arc::clone(&self.inner.options.writer_pool);
let n_threads = writer_pool.current_num_threads().max(1);
let buffered_bytes: usize = buffer
.iter()
.map(|b| b.scalar.get_array_memory_size())
.sum();
let target_bytes =
(self.inner.options.superfile_buffer_split_mb as usize).saturating_mul(MIB);
let n_superfiles =
superfiles_per_commit(total_rows, buffered_bytes, n_threads, target_bytes);
let vector_dims: Vec<usize> = self
.inner
.options
.vector_columns
.iter()
.map(|vc| vc.dim)
.collect();
let owned = buffer.to_vec();
let shard_manifest = self.inner.manifest.load_full();
let (shards, cell_hints): (Vec<Vec<BufferedBatch>>, Vec<Option<u32>>) =
if let Some(PartitionStrategy::VectorCell { clusters, .. }) =
shard_manifest.partition_strategy()
{
let metric = self
.inner
.options
.vector_columns
.first()
.map(|vc| vc.metric)
.unwrap_or(Metric::L2Sq);
if clusters.n_cent > 0 && clusters.dim > 0 {
let cell_shards = writer_pool
.install(|| split_buffer_by_vector_cell(owned, clusters, metric, 0))?;
let hints: Vec<Option<u32>> = cell_shards
.iter()
.map(|(cell_id, _)| Some(*cell_id))
.collect();
let shards: Vec<Vec<BufferedBatch>> = cell_shards
.into_iter()
.map(|(_, batches)| batches)
.collect();
(shards, hints)
} else {
let shards =
split_buffer_into_superfile_inputs(owned, n_superfiles, &vector_dims);
let hints = vec![None; shards.len()];
(shards, hints)
}
} else {
let shards = split_buffer_into_superfile_inputs(owned, n_superfiles, &vector_dims);
let hints = vec![None; shards.len()];
(shards, hints)
};
let user_inner = Arc::clone(&self.inner);
let user_options = Arc::clone(&self.inner.options);
let user_global_centroids: Option<std::sync::Arc<[f32]>> =
if config::global().vector.user_centroids == CentroidAlignment::Global {
pending_gvi
.as_ref()
.cloned()
.or_else(|| self.inner.manifest.load().get_global_vector_index())
.filter(|g| g.grid.n_cent > 0 && g.grid.dim > 0)
.map(|g| g.grid.to_fp32().into())
} else {
None
};
let outputs = fanout_shards(&writer_pool, &shards, |slice| {
build_one_shard_with_layout(
slice.as_slice(),
&user_options,
user_options.vector_layout,
user_global_centroids.clone(),
)
})?;
let superfiles = outputs.len();
let user_batch = prepare_user_superfile_batch(&self.inner, outputs, cell_hints)?;
bridge_on_runtime(
persist_superfile_publish_batch_async(&user_inner, user_batch, list_metadata),
&self.inner.query_runtime(),
)?;
if self.inner.options.storage.is_some() {
schedule_background_storage_reclaim(Arc::clone(&self.inner));
}
debug!(superfiles, "published appended superfiles");
Ok(())
}
}
impl Drop for SupertableWriter {
fn drop(&mut self) {
self.inner
.writer_outstanding
.store(false, Ordering::Release);
}
}
pub struct ShardOutput {
bytes: Bytes,
n_docs: u64,
id_min: i128,
id_max: i128,
scalar_stats: HashMap<String, ScalarStatsAgg>,
}
impl ShardOutput {
pub fn new_with_params(
bytes: Bytes,
n_docs: u64,
id_min: i128,
id_max: i128,
scalar_stats: HashMap<String, ScalarStatsAgg>,
) -> Self {
Self {
bytes,
n_docs,
id_min,
id_max,
scalar_stats,
}
}
}
fn reserve_build_scratch(
budget: &Arc<ConnectionMemoryBudget>,
scalar_bytes: usize,
vector_bytes: usize,
fts_bytes: usize,
) -> Result<Reservation, BuildError> {
let estimate = scalar_bytes
.saturating_mul(BUILD_SCALAR_NUM)
.saturating_add(vector_bytes.saturating_mul(BUILD_VECTOR_NUM))
.saturating_add(fts_bytes.saturating_mul(BUILD_FTS_NUM))
/ BUILD_SCRATCH_DENOM;
budget
.try_reserve(estimate)
.map_err(|e| BuildError::OverBudget(format!("during ingest, {e}")))
}
fn build_one_shard_with_layout(
slice: &[BufferedBatch],
options: &SupertableOptions,
vector_layout: crate::superfile::vector::layout::VectorLayout,
provided_centroids: Option<std::sync::Arc<[f32]>>,
) -> Result<ShardOutput, BuildError> {
let mut builder = SuperfileBuilder::new(
options
.builder_options()
.with_vector_layout(vector_layout)
.with_vector_centroids(provided_centroids),
)?;
let scalar_schema = options.scalar_schema();
let id_idx = 0;
let mut id_min = i128::MAX;
let mut id_max = i128::MIN;
let mut n_docs: u64 = 0;
for buffered in slice {
let id_col = buffered
.scalar
.column(id_idx)
.as_any()
.downcast_ref::<Decimal128Array>()
.ok_or_else(|| {
BuildError::IdColumnWrongType(
options.id_column.clone(),
"<id column not Decimal128 at runtime>".to_string(),
)
})?;
for i in 0..id_col.len() {
let v = id_col.value(i);
id_min = id_min.min(v);
id_max = id_max.max(v);
}
n_docs += id_col.len() as u64;
let vector_slices: Vec<&[f32]> = buffered
.vectors
.iter()
.map(|fa| fa.values().as_ref())
.collect();
builder.add_batch(&buffered.scalar, &vector_slices)?;
}
let scalar_batches: Vec<&RecordBatch> = slice.iter().map(|b| &b.scalar).collect();
let scalar_stats = ScalarStatsAgg::from_batches(&scalar_schema, &scalar_batches);
let mut output = NamedTempFile::new()
.map_err(|error| BuildError::Store(format!("shard temp create: {error}")))?;
{
let mut writer = BufWriter::new(output.as_file_mut());
builder.finish_to(&mut writer)?;
writer
.flush()
.map_err(|error| BuildError::Store(format!("shard temp flush: {error}")))?;
}
let bytes = mmap_readonly_bytes(output.path())
.map_err(|error| BuildError::Store(format!("shard mmap: {error}")))?;
let (id_min, id_max) = if n_docs == 0 {
(0, 0)
} else {
(id_min, id_max)
};
Ok(ShardOutput {
bytes,
n_docs,
id_min,
id_max,
scalar_stats,
})
}
pub(crate) fn build_subsection_offsets(bytes: &Bytes) -> Option<SubsectionOffsets> {
let kvs = read_kv_metadata(bytes).ok()?;
let get = |k: &str| -> Option<u64> { kvs.get(k).and_then(|s| s.parse::<u64>().ok()) };
let vec = match (get(kv::VEC_OFFSET), get(kv::VEC_LENGTH)) {
(Some(o), Some(l)) if l > 0 => Some((o, l)),
_ => None,
};
let fts = match (get(kv::FTS_OFFSET), get(kv::FTS_LENGTH)) {
(Some(o), Some(l)) if l > 0 => Some((o, l)),
_ => None,
};
let total_size = bytes.len() as u64;
let layout = vector_layout_from_kv(&kvs);
if layout == VectorLayout::CellPosting {
return Some(SubsectionOffsets {
total_size,
vec,
fts,
vec_open_ranges: Vec::new(),
fts_open_ranges: Vec::new(),
open_blob: Vec::new(),
});
}
let vec_dim = kvs
.get(kv::VEC_COLUMNS)
.and_then(|json| serde_json::from_str::<Vec<VectorColumnConfig>>(json).ok())
.and_then(|cols| match cols.as_slice() {
[only] => Some(only.dim),
_ => None,
});
let vec_open_ranges = vec
.and_then(|(off, len)| vector_open_ranges(bytes, off, len, vec_dim))
.unwrap_or_default();
let fts_open_ranges = fts
.and_then(|(off, len)| fts_open_ranges(bytes, off, len))
.unwrap_or_default();
let open_blob = build_open_blob(bytes, total_size, &vec_open_ranges, &fts_open_ranges);
Some(SubsectionOffsets {
total_size,
vec,
fts,
vec_open_ranges,
fts_open_ranges,
open_blob,
})
}
fn build_open_blob(
bytes: &Bytes,
total_size: u64,
vec_open_ranges: &[(u64, u64)],
fts_open_ranges: &[(u64, u64)],
) -> Vec<(u64, Vec<u8>)> {
const PARQUET_TAIL_SPEC: u64 = 64 * 1024;
let mut blob: Vec<(u64, Vec<u8>)> =
Vec::with_capacity(1 + vec_open_ranges.len() + fts_open_ranges.len());
let parquet_tail_len = PARQUET_TAIL_SPEC.min(total_size);
let parquet_tail_start = total_size.saturating_sub(parquet_tail_len);
let slice = |off: u64, len: u64| -> Option<Vec<u8>> {
let start = off as usize;
let end = start.checked_add(len as usize)?;
bytes.get(start..end).map(|s| s.to_vec())
};
if parquet_tail_len > 0 {
match slice(parquet_tail_start, parquet_tail_len) {
Some(b) => blob.push((parquet_tail_start, b)),
None => return Vec::new(),
}
}
for &(off, len) in vec_open_ranges.iter().chain(fts_open_ranges.iter()) {
match slice(off, len) {
Some(b) => blob.push((off, b)),
None => return Vec::new(),
}
}
blob
}
fn vector_open_ranges(
bytes: &Bytes,
off: u64,
len: u64,
dim: Option<usize>,
) -> Option<Vec<(u64, u64)>> {
let start = off as usize;
let end = start.checked_add(len as usize)?;
let blob = bytes.get(start..end)?;
if blob.len() < OUTER_HEADER_SIZE + CRC_BYTES {
return None;
}
let version =
read_u32_le(blob.get(outer_hdr::VERSION_OFF..outer_hdr::VERSION_OFF + U32_BYTES)?);
if version == crate::superfile::format::vec::VERSION_MULTI_CELL {
return vector_open_ranges_multi_cell(blob, off, dim?);
}
if version != crate::superfile::format::vec::VERSION {
return None;
}
let n_columns =
read_u32_le(blob.get(outer_hdr::N_COLUMNS_OFF..outer_hdr::N_COLUMNS_OFF + U32_BYTES)?)
as usize;
let dir_offset =
read_u64_le(blob.get(outer_hdr::DIR_OFFSET_OFF..outer_hdr::DIR_OFFSET_OFF + U64_BYTES)?)
as usize;
let dir_size = n_columns.checked_mul(DIR_ENTRY_SIZE)?;
let dir_end = dir_offset.checked_add(dir_size)?.checked_add(CRC_BYTES)?;
let dir = blob.get(dir_offset..dir_offset + dir_size)?;
let mut ranges = vec![(off + dir_offset as u64, (dir_size + CRC_BYTES) as u64)];
ranges.push((off, OUTER_HEADER_SIZE as u64));
for i in 0..n_columns {
let entry = i * DIR_ENTRY_SIZE;
let subsection_off = read_u64_le(dir.get(
entry + dir_entry::SUBSECTION_OFF_OFF
..entry + dir_entry::SUBSECTION_OFF_OFF + U64_BYTES,
)?) as usize;
let subsection_len = read_u64_le(dir.get(
entry + dir_entry::SUBSECTION_LEN_OFF
..entry + dir_entry::SUBSECTION_LEN_OFF + U64_BYTES,
)?) as usize;
let codec_meta_off = read_u32_le(dir.get(
entry + dir_entry::CODEC_META_OFF_OFF
..entry + dir_entry::CODEC_META_OFF_OFF + U32_BYTES,
)?) as usize;
let codec_meta_size = read_u32_le(dir.get(
entry + dir_entry::CODEC_META_SIZE_OFF
..entry + dir_entry::CODEC_META_SIZE_OFF + U32_BYTES,
)?) as usize;
if subsection_off.checked_add(SUB_HEADER_SIZE)? > blob.len()
|| subsection_off.checked_add(subsection_len)? > blob.len()
{
return None;
}
ranges.push((off + subsection_off as u64, SUB_HEADER_SIZE as u64));
let sub = blob.get(subsection_off..subsection_off + subsection_len)?;
let centroids_off = read_u64_le(
sub.get(sub_hdr::CENTROIDS_OFF_OFF..sub_hdr::CENTROIDS_OFF_OFF + U64_BYTES)?,
) as usize;
let cluster_idx_off = read_u64_le(
sub.get(sub_hdr::CLUSTER_IDX_OFF_OFF..sub_hdr::CLUSTER_IDX_OFF_OFF + U64_BYTES)?,
) as usize;
let cluster_idx_end = cluster_idx_off.checked_add(
CLUSTER_IDX_ENTRY_BYTES
* read_u32_le(dir.get(
entry + dir_entry::N_CENT_OFF..entry + dir_entry::N_CENT_OFF + U32_BYTES,
)?) as usize,
)?;
if centroids_off < SUB_HEADER_SIZE || cluster_idx_end > subsection_len {
return None;
}
ranges.push((
off + subsection_off as u64 + cluster_idx_off as u64,
(cluster_idx_end - cluster_idx_off) as u64,
));
if codec_meta_size > 0 {
let meta_end = codec_meta_off.checked_add(codec_meta_size)?;
if meta_end > subsection_len {
return None;
}
}
}
if dir_end > blob.len() {
return None;
}
Some(merge_ranges(ranges))
}
fn vector_open_ranges_multi_cell(blob: &[u8], off: u64, dim: usize) -> Option<Vec<(u64, u64)>> {
use crate::superfile::format::vec::U64_BYTES;
if dim == 0 {
return None;
}
let n_cells =
read_u32_le(blob.get(outer_hdr::N_CELLS_OFF..outer_hdr::N_CELLS_OFF + U32_BYTES)?) as usize;
let dir_offset =
read_u64_le(blob.get(outer_hdr::DIR_OFFSET_OFF..outer_hdr::DIR_OFFSET_OFF + U64_BYTES)?)
as usize;
let dir_size = n_cells.checked_mul(CELL_DIR_ENTRY_SIZE)?;
let dir_end = dir_offset.checked_add(dir_size)?.checked_add(CRC_BYTES)?;
if dir_end > blob.len() {
return None;
}
let dir = blob.get(dir_offset..dir_offset + dir_size)?;
let mut ranges = vec![
(off, OUTER_HEADER_SIZE as u64),
(off + dir_offset as u64, (dir_size + CRC_BYTES) as u64),
];
for i in 0..n_cells {
let entry = i * CELL_DIR_ENTRY_SIZE;
let subsection_off = read_u64_le(dir.get(
entry + cell_dir_entry::SUBSECTION_OFF_OFF
..entry + cell_dir_entry::SUBSECTION_OFF_OFF + U64_BYTES,
)?) as usize;
let subsection_len = read_u64_le(dir.get(
entry + cell_dir_entry::SUBSECTION_LEN_OFF
..entry + cell_dir_entry::SUBSECTION_LEN_OFF + U64_BYTES,
)?) as usize;
if subsection_off.checked_add(SUB_HEADER_SIZE)? > blob.len()
|| subsection_off.checked_add(subsection_len)? > blob.len()
{
return None;
}
let sub = blob.get(subsection_off..subsection_off + subsection_len)?;
let centroids_off = read_u64_le(
sub.get(sub_hdr::CENTROIDS_OFF_OFF..sub_hdr::CENTROIDS_OFF_OFF + U64_BYTES)?,
) as usize;
let cluster_idx_off = read_u64_le(
sub.get(sub_hdr::CLUSTER_IDX_OFF_OFF..sub_hdr::CLUSTER_IDX_OFF_OFF + U64_BYTES)?,
) as usize;
let centroids_span = cluster_idx_off.checked_sub(centroids_off)?;
if centroids_off < SUB_HEADER_SIZE || !centroids_span.is_multiple_of(dim * 4) {
return None;
}
let n_cent = centroids_span / (dim * 4);
let cluster_idx_end =
cluster_idx_off.checked_add(n_cent.checked_mul(CLUSTER_IDX_ENTRY_BYTES)?)?;
if cluster_idx_end > subsection_len {
return None;
}
ranges.push((off + subsection_off as u64, SUB_HEADER_SIZE as u64));
ranges.push((
off + (subsection_off + cluster_idx_off) as u64,
(cluster_idx_end - cluster_idx_off) as u64,
));
}
Some(merge_ranges(ranges))
}
fn fts_open_ranges(bytes: &Bytes, off: u64, len: u64) -> Option<Vec<(u64, u64)>> {
let start = off as usize;
let end = start.checked_add(len as usize)?;
let blob = bytes.get(start..end)?;
if blob.len() < FTS_HEADER_SIZE {
return None;
}
let postings_offset =
read_u64_le(blob.get(hdr::POSTINGS_OFFSET_OFF..hdr::POSTINGS_OFFSET_OFF + U64_BYTES)?)
as usize;
let doc_lengths_offset =
read_u64_le(blob.get(hdr::DOC_LENGTHS_DIR_OFF..hdr::DOC_LENGTHS_DIR_OFF + U64_BYTES)?)
as usize;
if postings_offset > blob.len()
|| doc_lengths_offset > blob.len()
|| postings_offset > doc_lengths_offset
{
return None;
}
Some(merge_ranges(vec![
(off, postings_offset as u64),
(
off + doc_lengths_offset as u64,
(blob.len() - doc_lengths_offset) as u64,
),
]))
}
fn merge_ranges(mut ranges: Vec<(u64, u64)>) -> Vec<(u64, u64)> {
ranges.retain(|&(_, len)| len > 0);
ranges.sort_unstable_by_key(|&(off, _)| off);
let mut merged: Vec<(u64, u64)> = Vec::with_capacity(ranges.len());
for (off, len) in ranges {
let end = off + len;
if let Some((last_off, last_len)) = merged.last_mut() {
let last_end = *last_off + *last_len;
if off <= last_end {
*last_len = (*last_len).max(end - *last_off);
continue;
}
}
merged.push((off, len));
}
merged
}
fn read_u32_le(bytes: &[u8]) -> u32 {
u32::from_le_bytes(bytes.try_into().expect("u32 slice length"))
}
fn read_u64_le(bytes: &[u8]) -> u64 {
u64::from_le_bytes(bytes.try_into().expect("u64 slice length"))
}
pub(crate) struct PreparedSuperfile {
pub(crate) entry: Arc<SuperfileEntry>,
pub(crate) bytes_for_store: Option<(SuperfileUri, Bytes)>,
pub(crate) bytes_for_storage: Option<(SuperfileUri, Bytes)>,
pub(crate) bytes_for_cache: Option<(SuperfileUri, Bytes)>,
}
impl PreparedSuperfile {
pub(crate) fn open_reader(&self) -> Option<Result<SuperfileReader, ReadError>> {
let bytes = self
.bytes_for_store
.as_ref()
.or(self.bytes_for_storage.as_ref())
.or(self.bytes_for_cache.as_ref())
.map(|(_, b)| b.clone())?;
Some(SuperfileReader::open(bytes))
}
}
pub(crate) fn build_column_vector_summary(
vec_reader: &VectorReader,
vc: &VectorConfig,
) -> Option<VectorSummary> {
let centroid = vec_reader.summary(&vc.column)?;
let cells: Vec<CellVectorSummary> = vec_reader
.cluster_centroids_by_cell(&vc.column)
.unwrap_or_default()
.into_iter()
.map(|(cell_id, n_cent, dim, fp32, counts)| CellVectorSummary {
cell_id,
clusters: ClusterCentroids::from_fp32(n_cent, dim, &fp32, counts),
})
.collect();
let rotation = RandomRotation::new(vc.dim, vc.rot_seed);
let quant = BitQuantizer::new(vc.dim);
for cell in &cells {
if cell.clusters.dim as usize == vc.dim {
cell.clusters
.prewarm_admit_codes(&rotation, &quant, vc.rot_seed);
}
}
Some(VectorSummary { centroid, cells })
}
pub(super) fn prepare_superfile(
inner: &SupertableInner,
shard: ShardOutput,
) -> Result<Option<PreparedSuperfile>, BuildError> {
prepare_superfile_with_uri(inner, shard, None)
}
pub(super) fn prepare_superfile_with_uri(
inner: &SupertableInner,
shard: ShardOutput,
reuse_uri: Option<SuperfileUri>,
) -> Result<Option<PreparedSuperfile>, BuildError> {
if shard.n_docs == 0 {
return Ok(None);
}
let uri = reuse_uri.unwrap_or_else(SuperfileUri::new_v4);
let bytes_for_storage = inner.options.storage.is_some().then(|| shard.bytes.clone());
let cache_attached = inner.options.disk_cache.is_some() && inner.options.storage.is_some();
let bytes_for_store = (!cache_attached).then(|| shard.bytes.clone());
let bytes_for_cache =
(cache_attached && inner.options.prepopulate_cache_on_commit).then(|| shard.bytes.clone());
let reader =
SuperfileReader::open_with(shard.bytes.clone(), inner.options.superfile_open_options())
.map_err(|e| BuildError::Store(format!("opening superfile for summary: {e}")))?;
let mut fts_summary: HashMap<String, FtsSummaryAgg> = HashMap::new();
if let Some(fts_reader) = reader.fts() {
for fc in &inner.options.fts_columns {
let terms = fts_reader
.iter_column_terms(&fc.column)
.expect("FST bytes valid: superfile just built");
let n_terms_distinct = terms.len() as u32;
let (min_term, max_term) = match (terms.first(), terms.last()) {
(Some(min), Some(max)) => (min.clone(), max.clone()),
_ => (Vec::new(), Vec::new()),
};
let mut bloom_builder = BloomBuilder::sized_for_terms(terms.len());
for term in &terms {
bloom_builder.insert(term);
}
fts_summary.insert(
fc.column.clone(),
FtsSummaryAgg::new_with_params(
bloom_builder.finish(),
n_terms_distinct,
(min_term, max_term),
),
);
}
}
let mut vector_summary: HashMap<String, VectorSummary> = HashMap::new();
if let Some(vec_reader) = reader.vec() {
for vc in &inner.options.vector_columns {
if let Some(summary) = build_column_vector_summary(vec_reader, vc) {
vector_summary.insert(vc.column.clone(), summary);
}
}
}
let subsection_offsets = build_subsection_offsets(&shard.bytes);
let vector_layout = read_vector_layout_from_bytes(&shard.bytes);
if vector_layout == VectorLayout::CellPosting
&& subsection_offsets.as_ref().and_then(|o| o.vec).is_none()
{
let kvs = crate::superfile::format::footer::read_kv_metadata(shard.bytes.as_ref())
.map(|kvs| kvs.keys().cloned().collect::<Vec<_>>())
.unwrap_or_default();
return Err(BuildError::Store(format!(
"cell-posting superfile missing inf.vec offset/length; kv_keys={kvs:?}"
)));
}
let entry = Arc::new(SuperfileEntry {
birth_version: 0,
superfile_id: uuid::Uuid::new_v4(),
uri,
n_docs: shard.n_docs,
id_min: shard.id_min,
id_max: shard.id_max,
scalar_stats: shard.scalar_stats,
fts_summary,
vector_summary,
partition_key: Vec::new(),
partition_hint: None,
subsection_offsets,
vector_layout,
});
Ok(Some(PreparedSuperfile {
entry,
bytes_for_store: bytes_for_store.map(|b| (uri, b)),
bytes_for_storage: bytes_for_storage.map(|b| (uri, b)),
bytes_for_cache: bytes_for_cache.map(|b| (uri, b)),
}))
}
fn finish_superfile_entry(
entry: Arc<SuperfileEntry>,
hint: Option<u32>,
) -> Result<Arc<SuperfileEntry>, BuildError> {
let old = entry.as_ref();
let staged = SuperfileEntry {
birth_version: old.birth_version,
superfile_id: old.superfile_id,
uri: old.uri,
n_docs: old.n_docs,
id_min: old.id_min,
id_max: old.id_max,
scalar_stats: old.scalar_stats.clone(),
fts_summary: old.fts_summary.clone(),
vector_summary: old.vector_summary.clone(),
partition_key: Vec::new(),
partition_hint: hint.or(old.partition_hint),
subsection_offsets: old.subsection_offsets.clone(),
vector_layout: old.vector_layout,
};
Ok(Arc::new(staged))
}
struct SuperfilePublishBatch {
new_entries: Vec<Arc<SuperfileEntry>>,
to_remove: Vec<Arc<SuperfileEntry>>,
pending_storage_writes: Vec<(SuperfileUri, Bytes)>,
pending_cache_inserts: Vec<(SuperfileUri, Bytes)>,
pending_store_inserts: Vec<(SuperfileUri, Bytes)>,
}
fn collect_prepared_superfiles(
_inner: &SupertableInner,
prepared: Vec<PreparedSuperfile>,
) -> Result<SuperfilePublishBatch, BuildError> {
let mut new_entries: Vec<Arc<SuperfileEntry>> = Vec::with_capacity(prepared.len());
let mut pending_storage_writes: Vec<(SuperfileUri, Bytes)> = Vec::new();
let mut pending_cache_inserts: Vec<(SuperfileUri, Bytes)> = Vec::new();
let mut pending_store_inserts: Vec<(SuperfileUri, Bytes)> = Vec::new();
for p in prepared {
if let Some(t) = p.bytes_for_store {
pending_store_inserts.push(t);
}
if let Some(t) = p.bytes_for_storage {
pending_storage_writes.push(t);
}
if let Some(t) = p.bytes_for_cache {
pending_cache_inserts.push(t);
}
new_entries.push(p.entry);
}
Ok(SuperfilePublishBatch {
new_entries,
to_remove: Vec::new(),
pending_storage_writes,
pending_cache_inserts,
pending_store_inserts,
})
}
fn apply_pending_store_inserts(inner: &SupertableInner, inserts: Vec<(SuperfileUri, Bytes)>) {
for (uri, bytes) in inserts {
let _ = inner.options.store.insert(uri, bytes);
}
}
fn prepare_user_superfile_batch_in_scope(
inner: &SupertableInner,
outputs: Vec<ShardOutput>,
hints: Vec<Option<u32>>,
) -> Result<SuperfilePublishBatch, BuildError> {
if outputs.len() != hints.len() {
return Err(BuildError::Store(format!(
"superfile publish inputs out of sync: {} shard outputs for {} partition hints",
outputs.len(),
hints.len()
)));
}
let prepared: Vec<PreparedSuperfile> = outputs
.into_par_iter()
.zip(hints.into_par_iter())
.filter_map(|(shard, hint)| match prepare_superfile(inner, shard) {
Ok(Some(p)) => {
Some(
finish_superfile_entry(p.entry, hint).map(|entry| PreparedSuperfile {
entry,
bytes_for_store: p.bytes_for_store,
bytes_for_storage: p.bytes_for_storage,
bytes_for_cache: p.bytes_for_cache,
}),
)
}
Ok(None) => None,
Err(e) => Some(Err(e)),
})
.collect::<Result<Vec<_>, _>>()?;
collect_prepared_superfiles(inner, prepared)
}
fn prepare_user_superfile_batch(
inner: &SupertableInner,
outputs: Vec<ShardOutput>,
hints: Vec<Option<u32>>,
) -> Result<SuperfilePublishBatch, BuildError> {
inner
.options
.writer_pool
.install(|| prepare_user_superfile_batch_in_scope(inner, outputs, hints))
}
async fn persist_superfile_publish_batch_async(
inner: &SupertableInner,
batch: SuperfilePublishBatch,
list_metadata: CommitListMetadata,
) -> Result<(), BuildError> {
if batch.new_entries.is_empty() {
return Ok(());
}
if let Some(storage) = inner.options.storage.as_ref().cloned() {
let new_manifest = persist_commit_async(
inner,
storage,
batch.new_entries,
&batch.to_remove,
batch.pending_storage_writes,
Vec::new(),
list_metadata,
)
.await
.map_err(BuildError::from)?;
inner.manifest.store(Arc::new(new_manifest));
apply_pending_store_inserts(inner, batch.pending_store_inserts);
if let Some(cache) = inner.options.disk_cache.as_ref() {
warm_cache_inserts(cache, batch.pending_cache_inserts).await;
}
if let (Some(cache), Some(budget)) = (
inner.options.disk_cache.as_ref(),
inner.options.memory_budget_bytes,
) {
cache.sweep_for_budget(budget);
}
return Ok(());
}
let old = inner.manifest.load();
let new = if list_metadata.is_empty() {
old.with_appended(batch.new_entries)
} else {
list_metadata.apply(&old).with_appended(batch.new_entries)
};
for (uri, bytes) in batch.pending_store_inserts {
inner
.options
.store
.insert(uri, bytes)
.map_err(|e| BuildError::Store(format!("store insert for {uri:?}: {e}")))?;
}
inner.manifest.store(Arc::new(new));
Ok(())
}
static MAINT_POOL: std::sync::OnceLock<rayon::ThreadPool> = std::sync::OnceLock::new();
fn maint_pool() -> Result<&'static ThreadPool, BuildError> {
if let Some(pool) = MAINT_POOL.get() {
return Ok(pool);
}
let threads = config::global()
.vector
.maintenance_threads
.resolve_or_default(available_parallelism().map(NonZeroUsize::get).unwrap_or(1));
let pool = ThreadPoolBuilder::new()
.num_threads(threads)
.thread_name(|_| "hidden-maint-cpu".into())
.build()
.map_err(|e| BuildError::Store(format!("hidden maintenance rayon pool: {e}")))?;
Ok(MAINT_POOL.get_or_init(|| pool))
}
test_visible! {
fn maintenance_pool_width() -> usize {
maint_pool().map(ThreadPool::current_num_threads).unwrap_or(1)
}
}
fn drain_batch_superfiles(opts: &SupertableOptions) -> i64 {
opts.drain_batch_superfiles
}
fn spill_row_to_cell(
spills: &mut HashMap<u32, MaterializedRowSpillWriter>,
added: &mut HashMap<u32, u32>,
scratch: &Path,
cell: u32,
row: &MaterializedIvfRow,
) -> Result<(), BuildError> {
let writer = match spills.entry(cell) {
Entry::Occupied(entry) => entry.into_mut(),
Entry::Vacant(entry) => entry.insert(MaterializedRowSpillWriter::create(
scratch,
cell,
row.encoded
.rerank_codec
.dim_from_codes_len(row.encoded.codes.len()),
row.rabitq_code.len(),
)?),
};
writer.append(row)?;
let count = added.entry(cell).or_insert(0);
*count = count.saturating_add(1);
Ok(())
}
fn spill_unfinished_shard_row(
spills: &mut HashMap<u32, MaterializedRowSpillWriter>,
added: &mut HashMap<u32, u32>,
completed_shards: &HashSet<u32>,
shard_count: usize,
scratch: &Path,
cell: u32,
row: &MaterializedIvfRow,
) -> Result<(), BuildError> {
if completed_shards.contains(&(packed_cell_shard(cell, shard_count) as u32)) {
return Ok(());
}
spill_row_to_cell(spills, added, scratch, cell, row)
}
fn drain_checkpoint_source(entry: &SuperfileEntry) -> DrainCheckpointSource {
DrainCheckpointSource {
superfile_id: entry.superfile_id.to_string(),
uri: entry.uri.0.to_string(),
birth_version: entry.birth_version,
}
}
fn drain_epoch_id(
options_hash: &str,
sources: &[DrainCheckpointSource],
batch_layout: &[Vec<u64>],
shard_count: usize,
consolidate: DrainConsolidate,
) -> String {
let mut hasher = Blake3Hasher::new();
hasher.update(&DRAIN_CHECKPOINT_SCHEMA.to_le_bytes());
hasher.update(&(shard_count as u64).to_le_bytes());
hasher.update(options_hash.as_bytes());
hasher.update(match consolidate {
DrainConsolidate::Kmeans => b"kmeans",
DrainConsolidate::Splice => b"splice",
});
for source in sources {
hasher.update(source.superfile_id.as_bytes());
hasher.update(source.uri.as_bytes());
hasher.update(&source.birth_version.to_le_bytes());
}
for batch in batch_layout {
hasher.update(&(batch.len() as u64).to_le_bytes());
for version in batch {
hasher.update(&version.to_le_bytes());
}
}
hasher.finalize().to_hex().to_string()
}
fn drain_scratch_dir(epoch_id: &str) -> PathBuf {
env::temp_dir().join("infino-drain").join(epoch_id)
}
fn drain_local_checkpoint_path(scratch: &Path) -> PathBuf {
scratch.join(DRAIN_LOCAL_CHECKPOINT_FILE)
}
fn load_drain_local_checkpoint(
scratch: &Path,
epoch_id: &str,
) -> Result<Option<DrainLocalCheckpoint>, BuildError> {
let path = drain_local_checkpoint_path(scratch);
let bytes = match fs::read(&path) {
Ok(bytes) => bytes,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
Err(error) => {
return Err(BuildError::Store(format!(
"drain local checkpoint read {}: {error}",
path.display()
)));
}
};
let checkpoint: DrainLocalCheckpoint = serde_json::from_slice(&bytes)
.map_err(|error| BuildError::Store(format!("drain local checkpoint decode: {error}")))?;
if checkpoint.schema != DRAIN_CHECKPOINT_SCHEMA || checkpoint.epoch_id != epoch_id {
return Err(BuildError::Store(format!(
"drain local checkpoint at {} is incompatible (schema {}, epoch {})",
path.display(),
checkpoint.schema,
checkpoint.epoch_id
)));
}
Ok(Some(checkpoint))
}
fn save_drain_local_checkpoint(
scratch: &Path,
checkpoint: &DrainLocalCheckpoint,
) -> Result<(), BuildError> {
fs::create_dir_all(scratch)
.map_err(|error| BuildError::Store(format!("drain scratch create: {error}")))?;
let bytes = serde_json::to_vec(checkpoint)
.map_err(|error| BuildError::Store(format!("drain local checkpoint encode: {error}")))?;
let final_path = drain_local_checkpoint_path(scratch);
let temp_path = scratch.join(format!("{DRAIN_LOCAL_CHECKPOINT_FILE}.tmp"));
{
let mut file = File::create(&temp_path)
.map_err(|error| BuildError::Store(format!("drain checkpoint create: {error}")))?;
file.write_all(&bytes)
.map_err(|error| BuildError::Store(format!("drain checkpoint write: {error}")))?;
file.sync_all()
.map_err(|error| BuildError::Store(format!("drain checkpoint fsync: {error}")))?;
}
fs::rename(&temp_path, &final_path)
.map_err(|error| BuildError::Store(format!("drain checkpoint rename: {error}")))?;
File::open(scratch)
.and_then(|directory| directory.sync_all())
.map_err(|error| BuildError::Store(format!("drain checkpoint dir fsync: {error}")))?;
Ok(())
}
async fn load_drain_remote_checkpoint(
inner: &SupertableInner,
) -> Result<Option<DrainRemoteState>, BuildError> {
let manifest = inner.manifest.load_full();
let Some((uri, hash)) = manifest.slow_vector_state_blob() else {
return Ok(None);
};
let storage = inner
.options
.storage
.as_ref()
.ok_or_else(|| BuildError::Store("drain checkpoint requires storage".into()))?;
let state = slow_vector_state::load_full_state(storage.as_ref(), uri, &hash)
.await
.map_err(|error| BuildError::Store(format!("drain slow-CAS load: {error}")))?;
let Some(pending) = state.pending_drain else {
return Ok(None);
};
if pending_metadata_schema(&pending.metadata) == Some(REPACK_CHECKPOINT_SCHEMA) {
debug!("drain: ignoring foreign repack upload pin in slow-CAS pending state");
return Ok(None);
}
let checkpoint: DrainRemoteCheckpoint = serde_json::from_slice(&pending.metadata)
.map_err(|error| BuildError::Store(format!("drain remote checkpoint decode: {error}")))?;
if checkpoint.schema != DRAIN_CHECKPOINT_SCHEMA {
return Err(BuildError::Store(format!(
"drain remote checkpoint schema {} != supported {}",
checkpoint.schema, DRAIN_CHECKPOINT_SCHEMA
)));
}
if pending.entries.len() != checkpoint.completed_shards.len() {
return Err(BuildError::Store(format!(
"drain slow-CAS has {} pending entries for {} completed shards",
pending.entries.len(),
checkpoint.completed_shards.len()
)));
}
let entry_ids: HashSet<String> = pending
.entries
.iter()
.map(|entry| entry.superfile_id.to_string())
.collect();
if checkpoint
.completed_shards
.iter()
.any(|shard| !entry_ids.contains(&shard.superfile_id))
{
return Err(BuildError::Store(
"drain slow-CAS checkpoint references a missing pending entry".into(),
));
}
Ok(Some(DrainRemoteState {
checkpoint,
entries: pending.entries,
}))
}
async fn save_drain_remote_checkpoint(
inner: &SupertableInner,
state: &mut DrainRemoteState,
) -> Result<(), BuildError> {
let metadata = serde_json::to_vec(&state.checkpoint)
.map_err(|error| BuildError::Store(format!("drain checkpoint encode: {error}")))?;
stamp_slow_vector_state(
inner,
Some(slow_vector_state::PendingDrainState {
metadata,
entries: state.entries.clone(),
}),
)
.await
}
async fn create_drain_remote_checkpoint(
inner: &SupertableInner,
checkpoint: DrainRemoteCheckpoint,
) -> Result<DrainRemoteState, BuildError> {
let mut state = DrainRemoteState {
checkpoint,
entries: Vec::new(),
};
save_drain_remote_checkpoint(inner, &mut state).await?;
Ok(state)
}
fn make_drain_batches(
sources: Vec<Arc<SuperfileEntry>>,
budget: usize,
) -> Vec<(Vec<u64>, Vec<Arc<SuperfileEntry>>)> {
let mut by_version = std::collections::BTreeMap::<u64, Vec<Arc<SuperfileEntry>>>::new();
for source in sources {
by_version
.entry(source.birth_version)
.or_default()
.push(source);
}
let mut batches = Vec::new();
let mut versions = Vec::new();
let mut superfiles = Vec::new();
for (version, mut version_superfiles) in by_version {
if !superfiles.is_empty()
&& superfiles.len().saturating_add(version_superfiles.len()) > budget
{
batches.push((mem::take(&mut versions), mem::take(&mut superfiles)));
}
versions.push(version);
superfiles.append(&mut version_superfiles);
if superfiles.len() >= budget {
batches.push((mem::take(&mut versions), mem::take(&mut superfiles)));
}
}
if !superfiles.is_empty() {
batches.push((versions, superfiles));
}
batches
}
fn drain_batch_layout(batches: &[(Vec<u64>, Vec<Arc<SuperfileEntry>>)]) -> Vec<Vec<u64>> {
batches
.iter()
.map(|(versions, _)| versions.clone())
.collect()
}
const DEFAULT_DRAIN_REPLICA_TARGET_FACTOR: f32 = 1.0;
fn drain_replica_target_factor() -> f32 {
let factor = config::global().vector.drain_replica_target_factor;
if factor.is_finite() && factor > DEFAULT_DRAIN_REPLICA_TARGET_FACTOR {
factor
} else {
DEFAULT_DRAIN_REPLICA_TARGET_FACTOR
}
}
fn drain_replica_extra_budget(n_rows: usize, target_factor: f32) -> usize {
if n_rows == 0 || target_factor <= DEFAULT_DRAIN_REPLICA_TARGET_FACTOR {
return 0;
}
let target_rows = (n_rows as f64 * target_factor as f64).ceil() as usize;
target_rows
.saturating_sub(n_rows)
.min(n_rows.saturating_mul(opann::REPLICA_CLOSURE_MAX_REPLICAS))
}
async fn materialized_user_rows_for_drain(
reader: &SuperfileReader,
column: &str,
stable_ids: &[i128],
tombstones: Option<&roaring::RoaringBitmap>,
) -> Result<Vec<MaterializedIvfRow>, BuildError> {
let vec_reader = reader
.vec()
.ok_or_else(|| BuildError::Store("user superfile missing vector index".into()))?;
if vec_reader.is_multi_cell() {
let cells = vec_reader
.materialized_cells_rows_async(None)
.await
.ok_or_else(|| {
BuildError::Store(format!(
"drain materialize: multi-cell column '{column}' missing Sq8Residual index"
))
})?;
let mut by_id: HashMap<i128, MaterializedIvfRow> = HashMap::new();
for (_, rows) in cells {
for row in rows {
by_id.entry(row.stable_id).or_insert(row);
}
}
if let Some(bm) = tombstones
&& !bm.is_empty()
{
let locals: Vec<u32> = bm.iter().collect();
let id_column = reader.id_column();
let batch = reader
.take_by_local_doc_ids(&locals, &[id_column])
.map_err(|e| BuildError::Store(e.to_string()))?;
let array = batch
.column(0)
.as_any()
.downcast_ref::<Decimal128Array>()
.ok_or_else(|| BuildError::Store("_id column missing".into()))?;
let deleted: HashSet<i128> = array.values().iter().copied().collect();
by_id.retain(|stable_id, _| !deleted.contains(stable_id));
}
let mut rows: Vec<MaterializedIvfRow> = by_id.into_values().collect();
rows.sort_by_key(|row| row.stable_id);
for (local, row) in rows.iter_mut().enumerate() {
row.local_doc_id = local as u32;
}
return Ok(rows);
}
materialized_ivf_rows_in_doc_order(vec_reader, column, stable_ids, tombstones).await
}
pub(in crate::supertable) async fn drain_user_superfiles_to_hidden_cells(
user_inner: Arc<SupertableInner>,
hidden_inner: Arc<SupertableInner>,
) -> Result<(), BuildError> {
if hidden_inner
.compaction_outstanding
.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
.is_err()
{
return Ok(());
}
struct Slot<'a>(&'a std::sync::atomic::AtomicBool);
impl Drop for Slot<'_> {
fn drop(&mut self) {
self.0.store(false, Ordering::Release);
}
}
let _slot = Slot(&hidden_inner.compaction_outstanding);
let Some(gvi) = user_inner.manifest.load_full().get_global_vector_index() else {
return Ok(());
};
let column = gvi.column;
let hidden_manifest = hidden_inner.manifest.load_full();
let hidden_bootstrapped = !hidden_manifest.get_drained_ranges().is_empty();
let (clusters, mut routing) = match hidden_manifest.get_partition_strategy() {
PartitionStrategy::VectorCell {
clusters, routing, ..
} if hidden_bootstrapped => (clusters, routing),
PartitionStrategy::VectorCell { routing, .. } => (gvi.grid, routing),
_ => (gvi.grid, CellRoutingParams::default()),
};
if clusters.n_cent == 0 || clusters.dim == 0 {
return Ok(());
}
let user_manifest = user_inner.manifest.load_full();
let sources = user_manifest
.get_all_superfiles_loaded()
.await
.map_err(|e| BuildError::Store(e.to_string()))?;
if sources.is_empty() {
return Ok(());
}
let batch_cfg = drain_batch_superfiles(&user_inner.options);
if batch_cfg == 0 {
eprintln!("[supertable drain] skipped (drain_batch_superfiles = 0)");
return Ok(());
}
let storage = hidden_inner
.options
.storage
.clone()
.ok_or_else(|| BuildError::Store("hidden drain requires storage".into()))?;
let shard_count = packed_cell_shard_count(&hidden_inner.options);
let consolidate = user_inner.options.drain_consolidate;
let budget = if batch_cfg < 0 {
usize::MAX
} else {
(batch_cfg as usize).max(1)
};
let drained = hidden_inner.manifest.load_full().get_drained_ranges();
let user_strategy = user_manifest.get_partition_strategy();
let current_options_hash =
options_hash::compute_options_hash(user_inner.options.as_ref(), &user_strategy).to_hex();
let (batches, mut remote_state) = if let Some(remote_state) =
load_drain_remote_checkpoint(&hidden_inner).await?
{
if remote_state.checkpoint.shard_count != shard_count {
return Err(BuildError::Store(format!(
"drain checkpoint shard count {} != configured writer width {shard_count}",
remote_state.checkpoint.shard_count
)));
}
if remote_state.checkpoint.options_hash != current_options_hash {
return Err(BuildError::Store(format!(
"drain checkpoint options hash {} != current {}",
remote_state.checkpoint.options_hash, current_options_hash
)));
}
let n_drained = remote_state
.checkpoint
.sources
.iter()
.filter(|source| drained.contains(source.birth_version))
.count();
if n_drained == remote_state.checkpoint.sources.len() {
let scratch = drain_scratch_dir(&remote_state.checkpoint.epoch_id);
if let Err(error) = fs::remove_dir_all(&scratch)
&& error.kind() != io::ErrorKind::NotFound
{
tracing::warn!("drain local checkpoint cleanup failed: {error}");
}
refresh_slow_vector_state(&hidden_inner).await?;
schedule_background_storage_reclaim(Arc::clone(&hidden_inner));
return Ok(());
}
if n_drained != 0 {
return Err(BuildError::Store(
"drain checkpoint source versions are only partially committed".into(),
));
}
let source_by_id: HashMap<String, Arc<SuperfileEntry>> = sources
.iter()
.map(|entry| (entry.superfile_id.to_string(), Arc::clone(entry)))
.collect();
let mut selected = Vec::with_capacity(remote_state.checkpoint.sources.len());
for source in &remote_state.checkpoint.sources {
let entry = source_by_id.get(&source.superfile_id).ok_or_else(|| {
BuildError::Store(format!(
"drain checkpoint source {} is missing from the user manifest",
source.superfile_id
))
})?;
if entry.uri.0.to_string() != source.uri || entry.birth_version != source.birth_version
{
return Err(BuildError::Store(format!(
"drain checkpoint source {} no longer matches the user manifest",
source.superfile_id
)));
}
selected.push(Arc::clone(entry));
}
let batches = make_drain_batches(selected, budget);
let batch_layout = drain_batch_layout(&batches);
if batch_layout != remote_state.checkpoint.batch_layout {
return Err(BuildError::Store(
"drain checkpoint batch layout differs from current configuration".into(),
));
}
let epoch_id = drain_epoch_id(
¤t_options_hash,
&remote_state.checkpoint.sources,
&batch_layout,
shard_count,
consolidate,
);
if epoch_id != remote_state.checkpoint.epoch_id {
return Err(BuildError::Store(
"drain checkpoint epoch hash is invalid".into(),
));
}
(batches, remote_state)
} else {
let mut selected: Vec<Arc<SuperfileEntry>> = sources
.iter()
.filter(|entry| !drained.contains(entry.birth_version))
.cloned()
.collect();
if selected.is_empty() {
eprintln!(
"[supertable drain] nothing to drain: all {} user superfile(s) already drained",
sources.len()
);
return Ok(());
}
selected.sort_unstable_by(|left, right| {
left.birth_version
.cmp(&right.birth_version)
.then_with(|| left.superfile_id.cmp(&right.superfile_id))
});
let source_refs: Vec<DrainCheckpointSource> = selected
.iter()
.map(|entry| drain_checkpoint_source(entry))
.collect();
let batches = make_drain_batches(selected, budget);
let batch_layout = drain_batch_layout(&batches);
let epoch_id = drain_epoch_id(
¤t_options_hash,
&source_refs,
&batch_layout,
shard_count,
consolidate,
);
let checkpoint = DrainRemoteCheckpoint {
schema: DRAIN_CHECKPOINT_SCHEMA,
epoch_id,
options_hash: current_options_hash,
sources: source_refs,
batch_layout,
shard_count,
completed_shards: Vec::new(),
};
let remote_state = create_drain_remote_checkpoint(&hidden_inner, checkpoint).await?;
(batches, remote_state)
};
let store = user_inner.options.store.clone();
let storage_opt = user_inner.options.storage.clone();
let (metric, drain_rot_seed) = hidden_inner
.options
.vector_columns
.first()
.map(|c| (c.metric, c.rot_seed))
.unwrap_or((Metric::L2Sq, 0));
let assign_skip =
!hidden_bootstrapped && config::global().vector.user_centroids == CentroidAlignment::Global;
let column_name = column.clone();
let drain_t0 = std::time::Instant::now();
let drain_rss0 = proc_rss_mib();
let n_batches = batches.len();
let mut running_clusters = clusters;
let drain_scratch = drain_scratch_dir(&remote_state.checkpoint.epoch_id);
fs::create_dir_all(&drain_scratch)
.map_err(|error| BuildError::Store(format!("drain scratch create: {error}")))?;
let mut local_checkpoint =
load_drain_local_checkpoint(&drain_scratch, &remote_state.checkpoint.epoch_id)?
.unwrap_or_else(|| DrainLocalCheckpoint::new(remote_state.checkpoint.epoch_id.clone()));
if local_checkpoint.batches_done > n_batches {
return Err(BuildError::Store(format!(
"drain local checkpoint completed {} of only {n_batches} batches",
local_checkpoint.batches_done
)));
}
let mut completed_shards = HashSet::new();
let mut new_entries = Vec::new();
let mut added_per_cell = local_checkpoint.added_per_cell.clone();
let pending_entry_by_id: HashMap<String, Arc<SuperfileEntry>> = remote_state
.entries
.iter()
.map(|entry| (entry.superfile_id.to_string(), Arc::clone(entry)))
.collect();
for remote_shard in &remote_state.checkpoint.completed_shards {
if !completed_shards.insert(remote_shard.shard_id) {
return Err(BuildError::Store(format!(
"drain checkpoint repeats shard {}",
remote_shard.shard_id
)));
}
let entry = pending_entry_by_id
.get(&remote_shard.superfile_id)
.cloned()
.ok_or_else(|| {
BuildError::Store(format!(
"drain checkpoint shard {} entry {} is missing",
remote_shard.shard_id, remote_shard.superfile_id
))
})?;
if entry.partition_hint != Some(remote_shard.shard_id) {
return Err(BuildError::Store(format!(
"drain checkpoint shard {} entry has partition hint {:?}",
remote_shard.shard_id, entry.partition_hint
)));
}
storage
.head(&superfile_storage_path(&entry.uri))
.await
.map_err(|error| {
BuildError::Store(format!(
"drain checkpoint shard {} object is unavailable: {error}",
remote_shard.shard_id
))
})?;
for &(cell, count) in &remote_shard.cell_counts {
match added_per_cell.insert(cell, count) {
Some(existing) if existing != count => {
return Err(BuildError::Store(format!(
"drain checkpoint cell {cell} count {count} != local count {existing}"
)));
}
_ => {}
}
}
new_entries.push(entry);
}
let clean_uncheckpointed_drain = local_checkpoint.batches_done == 0
&& completed_shards.is_empty()
&& local_checkpoint.spills.is_empty();
let mut width_law = clean_uncheckpointed_drain.then(|| {
opann::WidthLawCalibration::new(
running_clusters.dim as usize,
metric,
user_inner.options.target_recall,
)
});
let transcode_clamp_baseline = transcode_clamped_components();
let mut cell_spills = HashMap::new();
for (&cell, spill) in &local_checkpoint.spills {
if completed_shards.contains(&(packed_cell_shard(cell, shard_count) as u32)) {
continue;
}
let rerank_codec = RerankCodec::from_codec_id(spill.rerank_codec_id).ok_or_else(|| {
BuildError::Store(format!(
"cell {cell}: checkpoint has unknown codec id {}",
spill.rerank_codec_id
))
})?;
cell_spills.insert(
cell,
MaterializedRowSpillWriter::resume(
&drain_scratch,
cell,
MaterializedRowSpillState {
n_rows: spill.n_rows,
n_quants: spill.n_quants,
dim: spill.dim,
rabitq_len: spill.rabitq_len,
rerank_codec,
},
)?,
);
}
let mut packed_cells = Vec::new();
for (&cell, state) in &local_checkpoint.built_cells {
let cell_shard = packed_cell_shard(cell, shard_count) as u32;
if completed_shards.contains(&cell_shard) {
continue;
}
packed_cells.push(restore_spilled_packed_cell(&drain_scratch, cell, state)?);
}
for (batch_idx, (_, batch_sources)) in batches.iter().enumerate() {
if batch_idx < local_checkpoint.batches_done {
continue;
}
let batch_t0 = std::time::Instant::now();
let gets_before = if crate::storage::io_counters::timeline_enabled() {
let snap = storage_opt.as_ref().map(|s| s.usage_meter().snapshot());
crate::storage::io_counters::timeline_reset();
snap
} else {
None
};
let read_concurrency = drain_read_concurrency();
let readers: Vec<(Arc<SuperfileReader>, Vec<i128>)> =
stream::iter(batch_sources.iter().map(|entry| {
let entry = Arc::clone(entry);
let store = Arc::clone(&store);
let storage_opt = storage_opt.clone();
let manifest = Arc::clone(&user_manifest);
async move {
let reader = match store.reader(&entry.uri) {
Ok(r) if r.is_fully_resident() => r,
_ => {
let storage = storage_opt.as_ref().ok_or_else(|| {
BuildError::Store(
"drain requires storage to load user superfiles".into(),
)
})?;
let (bytes, _) = storage
.get(&entry.uri.storage_path())
.await
.map_err(|e| BuildError::Store(e.to_string()))?;
Arc::new(
SuperfileReader::open(bytes)
.map_err(|e| BuildError::Store(e.to_string()))?,
)
}
};
let stable_ids =
stable_ids_by_local_for_routing(&manifest, &entry, &reader, &None)
.await
.map_err(|e| BuildError::Store(e.to_string()))?;
Ok::<_, BuildError>((reader, stable_ids))
}
}))
.buffered(read_concurrency)
.collect::<Vec<_>>()
.await
.into_iter()
.collect::<Result<Vec<_>, BuildError>>()?;
if crate::storage::io_counters::timeline_enabled() {
let spans = crate::storage::io_counters::timeline_take();
let range_gets = match (storage_opt.as_ref(), gets_before.as_ref()) {
(Some(s), Some(before)) => s.usage_meter().snapshot().since(before).get_count,
_ => 0,
};
let min_start = spans.iter().map(|s| s.start_us).min().unwrap_or(0);
let max_end = spans.iter().map(|s| s.end_us).max().unwrap_or(0);
let wall_us = max_end.saturating_sub(min_start);
let sum_us: u64 = spans
.iter()
.map(|s| s.end_us.saturating_sub(s.start_us))
.sum();
let bytes: u64 = spans.iter().map(|s| s.len).sum();
let concurrency = if wall_us > 0 {
sum_us as f64 / wall_us as f64
} else {
0.0
};
eprintln!(
"[supertable drain] batch {}/{} materialize I/O: {} object reads, {:.1} MiB, wall {:.1}ms, Σdur {:.1}ms, implied concurrency {:.1}x ({} range-gets)",
batch_idx + 1,
n_batches,
spans.len(),
bytes as f64 / (1u64 << 20) as f64,
wall_us as f64 / 1e3,
sum_us as f64 / 1e3,
concurrency,
range_gets,
);
}
let batch_log = match consolidate {
DrainConsolidate::Splice => {
let column_name_ref = column_name.as_str();
let stable_ids_per_input: Vec<Vec<i128>> =
readers.iter().map(|(_, ids)| ids.clone()).collect();
let routed: HashMap<u32, (MergedIvfSubsection, Vec<i128>)> =
hidden_inner.options.writer_pool.install(
|| -> Result<HashMap<u32, (MergedIvfSubsection, Vec<i128>)>, BuildError> {
let inputs: Vec<(&VectorReader, &str)> = readers
.iter()
.map(|(r, _)| {
r.vec()
.ok_or_else(|| {
BuildError::Store(
"user superfile missing vector index".into(),
)
})
.map(|vr| (vr, column_name_ref))
})
.collect::<Result<_, _>>()?;
let clusters_ref = &running_clusters;
route_clusters_into_cells(
&inputs,
&stable_ids_per_input,
|centroid: &[f32]| {
let mut assign = [0u32];
clusters_ref.assign_rows(metric, centroid, &mut assign);
vec![assign[0]]
},
)
.map_err(|e| e.into())
},
)?;
let n_cells = routed.len();
let dim = running_clusters.dim as usize;
let mut routed: Vec<_> = routed.into_iter().collect();
routed.sort_unstable_by_key(|(cell_id, _)| *cell_id);
for (cell_id, (subsection, stable_ids)) in routed {
accumulate_splice_cell(
&mut packed_cells,
&mut local_checkpoint,
&mut added_per_cell,
&completed_shards,
shard_count,
drain_scratch.as_path(),
cell_id,
subsection,
stable_ids,
dim,
metric,
)?;
}
format!(
"splice: route+accumulate {:.1}ms, {n_cells} cell(s)",
batch_t0.elapsed().as_secs_f64() * 1e3,
)
}
DrainConsolidate::Kmeans => {
let column_for_mat = column_name.clone();
let tombstone_cache = user_inner.tombstone_cache.clone();
let now = time::Instant::now();
let row_sets: Vec<Vec<MaterializedIvfRow>> =
stream::iter(readers.iter().zip(batch_sources.iter()).map(
|((reader, stable_ids), entry)| {
let column_for_mat = column_for_mat.clone();
let tombstone_cache = tombstone_cache.clone();
let entry = Arc::clone(entry);
async move {
let bitmap = tombstone_cache
.as_ref()
.map(|t| t.bitmap_for(entry.superfile_id, now))
.transpose()
.map_err(|e| BuildError::Store(e.to_string()))?;
materialized_user_rows_for_drain(
reader,
&column_for_mat,
stable_ids,
bitmap.as_deref(),
)
.await
}
},
))
.buffered(commit_write_concurrency())
.collect::<Vec<_>>()
.await
.into_iter()
.collect::<Result<Vec<_>, BuildError>>()?;
let t_mat = batch_t0.elapsed().as_secs_f64() * 1e3;
let all_rows: Vec<MaterializedIvfRow> = row_sets.into_iter().flatten().collect();
let n_batch_rows = all_rows.len();
for writer in cell_spills.values_mut() {
writer.begin_batch();
}
let replica_target = drain_replica_target_factor();
let mut seen_stable_ids: HashSet<i128> = HashSet::with_capacity(n_batch_rows);
let distinct_rows: Vec<&MaterializedIvfRow> = all_rows
.iter()
.filter(|row| seen_stable_ids.insert(row.stable_id))
.collect();
if assign_skip
&& drain_replica_extra_budget(distinct_rows.len(), replica_target) == 0
{
for row in &distinct_rows {
spill_unfinished_shard_row(
&mut cell_spills,
&mut added_per_cell,
&completed_shards,
shard_count,
drain_scratch.as_path(),
row.cluster,
row,
)?;
if let Some(cal) = width_law.as_mut() {
cal.offer(row);
}
}
} else {
let replica_extra_budget =
drain_replica_extra_budget(distinct_rows.len(), replica_target);
let clusters_ref = &running_clusters;
let admit_ctx =
RabitqAdmitContext::new(clusters_ref.dim as usize, drain_rot_seed);
let window = opann::assignment_shortlist_window(clusters_ref.n_cent as usize);
let assignments: Vec<opann::BoundaryAssignment> =
hidden_inner.options.writer_pool.install(|| {
distinct_rows
.par_iter()
.map(|row| {
opann::boundary_assignment_encoded(
clusters_ref,
metric,
&row.encoded,
&admit_ctx,
window,
)
})
.collect()
});
let mut replica_candidates: Vec<(usize, u32, f32)> = assignments
.iter()
.enumerate()
.flat_map(|(row_idx, assignment)| {
assignment
.replicas
.iter()
.flatten()
.map(move |&(cell, margin)| (row_idx, cell, margin))
})
.collect();
replica_candidates.sort_by(|a, b| a.2.total_cmp(&b.2));
for (row_idx, cell, _) in
replica_candidates.into_iter().take(replica_extra_budget)
{
spill_unfinished_shard_row(
&mut cell_spills,
&mut added_per_cell,
&completed_shards,
shard_count,
drain_scratch.as_path(),
cell,
distinct_rows[row_idx],
)?;
}
for (row, assignment) in distinct_rows.iter().zip(&assignments) {
spill_unfinished_shard_row(
&mut cell_spills,
&mut added_per_cell,
&completed_shards,
shard_count,
drain_scratch.as_path(),
assignment.primary,
row,
)?;
if let Some(cal) = width_law.as_mut() {
cal.offer(row);
}
}
}
let mut checkpointed_spills = HashMap::with_capacity(cell_spills.len());
for (&cell, writer) in &mut cell_spills {
let state = writer.checkpoint().map_err(BuildError::from)?;
checkpointed_spills.insert(
cell,
DrainLocalSpill {
n_rows: state.n_rows,
n_quants: state.n_quants,
dim: state.dim,
rabitq_len: state.rabitq_len,
rerank_codec_id: state.rerank_codec.codec_id(),
},
);
}
local_checkpoint.spills = checkpointed_spills;
let t_spill = batch_t0.elapsed().as_secs_f64() * 1e3;
format!(
"kmeans: materialize {:.1}ms + {} {:.1}ms, {} batch row(s) -> {} cell spill(s)",
t_mat,
if assign_skip {
"group(assign-skip)+spill"
} else {
"assign+spill"
},
t_spill - t_mat,
n_batch_rows,
cell_spills.len(),
)
}
};
local_checkpoint.batches_done = batch_idx + 1;
local_checkpoint.added_per_cell = added_per_cell.clone();
save_drain_local_checkpoint(&drain_scratch, &local_checkpoint)?;
#[cfg(test)]
maybe_fail_drain_for_test(
&remote_state.checkpoint.epoch_id,
DrainTestFailurePhase::AfterBatch,
local_checkpoint.batches_done,
)?;
eprintln!(
"[supertable drain] batch {}/{} ({} sf, {batch_log})",
batch_idx + 1,
n_batches,
batch_sources.len(),
);
}
{
let build_t0 = time::Instant::now();
let scratch = drain_scratch.as_path();
let n_cells_total = added_per_cell.len();
let total_rows: u64 = added_per_cell.values().map(|count| u64::from(*count)).sum();
let n_superfiles = shard_count;
let mut cell_counts_by_shard: HashMap<u32, Vec<(u32, u32)>> = HashMap::new();
for (&cell, &count) in &added_per_cell {
let shard = packed_cell_shard(cell, n_superfiles) as u32;
cell_counts_by_shard
.entry(shard)
.or_default()
.push((cell, count));
}
for counts in cell_counts_by_shard.values_mut() {
counts.sort_unstable_by_key(|(cell, _)| *cell);
}
let expected_shards = cell_counts_by_shard.len();
crate::superfile::vector::builder::build_phase_timers::reset();
let mut sources: Vec<(u32, DrainCellSource)> = packed_cells
.into_iter()
.map(|cell| (cell.cell_id, DrainCellSource::Packed(cell)))
.collect();
match consolidate {
DrainConsolidate::Splice => {
if !cell_spills.is_empty() {
return Err(BuildError::Store(
"splice drain must not leave materialized row spills".into(),
));
}
}
DrainConsolidate::Kmeans => {
sources.extend(
cell_spills
.into_iter()
.map(|(cell, writer)| {
writer
.finish()
.map(|spill| (cell, DrainCellSource::Rows(spill)))
.map_err(BuildError::from)
})
.collect::<Result<Vec<_>, BuildError>>()?,
);
}
}
if sources.is_empty() && !added_per_cell.is_empty() {
return Err(BuildError::Store(
"drain has cell counts but no cell build sources".into(),
));
}
let mut shard_sources = group_cells_by_packed_shard(sources, n_superfiles);
shard_sources.retain(|(shard_id, _)| !completed_shards.contains(shard_id));
let checkpoint = Arc::new(Mutex::new(local_checkpoint));
let vector_config = hidden_inner
.options
.vector_columns
.first()
.cloned()
.ok_or_else(|| BuildError::Store("drain pack requires a vector column".into()))?;
if let Some(mut cal) = width_law.take() {
let rot_seed = vector_config.rot_seed;
let pool_hint =
opann::rerank_pool_hint(&routing.width_for_k, running_clusters.n_cent as usize);
let clusters_for_freeze = running_clusters;
let (frozen, clusters_back) = run_on_pool(None, "width-law freeze", move || {
cal.freeze(&clusters_for_freeze, rot_seed, pool_hint);
(cal, clusters_for_freeze)
})
.await
.map_err(|e| BuildError::Store(format!("width-law freeze: {e}")))?;
width_law = Some(frozen);
running_clusters = clusters_back;
}
let width_law_ref = width_law.as_ref();
let prepared_shards: Vec<PreparedSuperfile> = fanout_shards(
&hidden_inner.options.writer_pool,
&shard_sources,
|(shard_id, cells)| {
let mut packed = Vec::with_capacity(cells.len());
for (cell_id, source) in cells {
let cell = match source {
DrainCellSource::Packed(cell) => cell.clone(),
DrainCellSource::Rows(spill) => {
if let Some(cal) = width_law_ref {
cal.score_cell(*cell_id, spill)?;
}
let cell = build_spilled_packed_cell_from_rows(
scratch,
*cell_id,
spill,
&vector_config,
)?;
{
let mut state = checkpoint.lock().map_err(|_| {
BuildError::Store("drain checkpoint lock poisoned".into())
})?;
state.spills.remove(cell_id);
state.built_cells.insert(
*cell_id,
DrainLocalCell {
n_docs: cell.n_docs,
subsection_len: cell.subsection_len,
rerank_codec_id: cell.rerank_codec.codec_id(),
},
);
save_drain_local_checkpoint(&drain_scratch, &state)?;
}
spill.remove_files();
cell
}
};
packed.push((*cell_id, cell));
}
let prepared: PreparedSuperfile =
build_prepared_from_spilled_cells(&hidden_inner, scratch, *shard_id, &packed)?;
if let Some(cal) = width_law_ref {
match prepared.open_reader() {
Some(reader) => {
let reader = reader.map_err(|e| {
BuildError::Store(format!("depth-law shard reopen: {e}"))
})?;
if let Some(views) = reader
.vec()
.and_then(|v| v.cell_fine_calibration_views(&vector_config.column))
{
cal.observe_shard_views(&views);
}
}
None => warn!(
"drain depth-law observation skipped for shard {shard_id}: \
bytes not retained (cache-attached, no prepopulation)"
),
}
}
Ok::<_, BuildError>(prepared)
},
)?;
local_checkpoint = checkpoint
.lock()
.map_err(|_| BuildError::Store("drain checkpoint lock poisoned".into()))?
.clone();
if prepared_shards.len() + completed_shards.len() > n_superfiles {
return Err(BuildError::Store(format!(
"drain produced {} packed shards for {n_superfiles} workers",
prepared_shards.len() + completed_shards.len()
)));
}
let publish = collect_prepared_superfiles(&hidden_inner, prepared_shards)?;
if !publish.to_remove.is_empty() {
return Err(BuildError::Store(
"drain prepared removals while publishing new worker shards".into(),
));
}
let entry_by_uri: HashMap<SuperfileUri, Arc<SuperfileEntry>> = publish
.new_entries
.iter()
.map(|entry| (entry.uri, Arc::clone(entry)))
.collect();
let pending_cache_inserts = publish.pending_cache_inserts;
let pending_store_inserts = publish.pending_store_inserts;
let multipart_threshold = hidden_inner.options.put_multipart_threshold_bytes;
let put_futures = publish
.pending_storage_writes
.into_iter()
.map(|(uri, bytes)| {
let storage = Arc::clone(&storage);
async move {
put_new_superfile_bytes(&storage, multipart_threshold, uri, bytes)
.await
.map(|()| uri)
.map_err(|error| BuildError::Store(error.to_string()))
}
});
let mut uploads = stream::iter(put_futures).buffer_unordered(commit_write_concurrency());
while let Some(uploaded) = uploads.next().await {
let uri = uploaded?;
let entry = entry_by_uri.get(&uri).cloned().ok_or_else(|| {
BuildError::Store(format!("uploaded drain shard {} has no entry", uri.0))
})?;
let shard_id = entry.partition_hint.ok_or_else(|| {
BuildError::Store(format!(
"uploaded drain shard {} has no partition hint",
uri.0
))
})?;
let cell_counts = cell_counts_by_shard
.get(&shard_id)
.cloned()
.ok_or_else(|| {
BuildError::Store(format!(
"uploaded drain shard {shard_id} has no cell counts"
))
})?;
remote_state.entries.push(Arc::clone(&entry));
remote_state
.checkpoint
.completed_shards
.push(DrainRemoteShard {
shard_id,
superfile_id: entry.superfile_id.to_string(),
cell_counts: cell_counts.clone(),
});
remote_state
.checkpoint
.completed_shards
.sort_unstable_by_key(|shard| shard.shard_id);
save_drain_remote_checkpoint(&hidden_inner, &mut remote_state).await?;
#[cfg(test)]
maybe_fail_drain_for_test(
&remote_state.checkpoint.epoch_id,
DrainTestFailurePhase::AfterShard,
remote_state.checkpoint.completed_shards.len(),
)?;
completed_shards.insert(shard_id);
new_entries.push(entry);
for (cell, _) in cell_counts {
local_checkpoint.spills.remove(&cell);
if let Some(state) = local_checkpoint.built_cells.remove(&cell)
&& let Ok(packed) = restore_spilled_packed_cell(&drain_scratch, cell, &state)
{
remove_spilled_packed_cell(&packed);
}
}
save_drain_local_checkpoint(&drain_scratch, &local_checkpoint)?;
}
if new_entries.len() != expected_shards {
return Err(BuildError::Store(format!(
"drain has {} completed shards but expected {expected_shards}",
new_entries.len()
)));
}
let n_shard_files = new_entries.len();
let mut cell_updates: HashMap<u32, u32> = HashMap::new();
for (cell, added) in &added_per_cell {
let base = running_clusters
.counts
.get(*cell as usize)
.copied()
.unwrap_or(0);
cell_updates.insert(*cell, base.saturating_add(*added));
}
running_clusters = opann::apply_cell_updates(&running_clusters, &cell_updates);
let mut new_drained = hidden_inner.manifest.load_full().get_drained_ranges();
let drained_max = batches
.iter()
.flat_map(|(versions, _)| versions.iter().copied())
.max()
.unwrap_or(0);
let lo = new_drained.prefix_end().map(|end| end + 1).unwrap_or(0);
new_drained.insert_range(lo.min(drained_max), drained_max);
if let Some(cal) = width_law.take()
&& let Some(laws) = cal.finish(&running_clusters)
{
for (slot, measured) in routing.width_for_k.iter_mut().zip(laws.width_for_k) {
*slot = (*slot).max(measured);
}
for (slot, measured) in routing.fine_for_k.iter_mut().zip(laws.fine_for_k) {
*slot = (*slot).max(measured);
}
opann::merge_rerank_with_pools(
&mut routing.rerank_for_k,
&mut routing.rerank_pool_cells,
&laws.rerank_for_k,
laws.pool_cells,
);
opann::clear_rerank_beyond_pool(
&routing.width_for_k,
&mut routing.rerank_for_k,
&routing.rerank_pool_cells,
);
info!(
"supertable drain: probe laws at k={WIDTH_LAW_KS:?}: width measured {:?} stamped {:?}; fine depth measured {:?} stamped {:?}; rerank measured {:?} stamped {:?}",
laws.width_for_k,
routing.width_for_k,
laws.fine_for_k,
routing.fine_for_k,
laws.rerank_for_k,
routing.rerank_for_k
);
}
let list_metadata = CommitListMetadata {
partition_strategy: Some(PartitionStrategy::VectorCell {
column: column.clone(),
clusters: running_clusters.clone(),
routing,
}),
drained_ranges: Some(new_drained),
global_vector_index: None,
superseded_cells_additions: None,
};
let no_removals: Vec<Arc<SuperfileEntry>> = Vec::new();
let new_manifest = persist_commit_async(
&hidden_inner,
Arc::clone(&storage),
new_entries,
&no_removals,
Vec::new(),
Vec::new(),
list_metadata,
)
.await
.map_err(BuildError::from)?;
hidden_inner.manifest.store(Arc::new(new_manifest));
apply_pending_store_inserts(&hidden_inner, pending_store_inserts);
if !pending_cache_inserts.is_empty()
&& let Some(cache) = hidden_inner.options.disk_cache.as_ref()
{
warm_cache_after_commit(&hidden_inner, cache, pending_cache_inserts);
}
if let Err(error) = fs::remove_dir_all(&drain_scratch)
&& error.kind() != io::ErrorKind::NotFound
{
tracing::warn!("drain local checkpoint cleanup failed: {error}");
}
eprintln!(
"[supertable drain] cell build: {} row(s), {} cell(s) -> {} packed shard superfile(s) for {} worker(s), {:.1}ms",
total_rows,
n_cells_total,
n_shard_files,
n_superfiles,
build_t0.elapsed().as_secs_f64() * 1e3,
);
if crate::superfile::vector::builder::build_phase_timers::enabled() {
let (train_ms, assign_ms, calib_ms) =
crate::superfile::vector::builder::build_phase_timers::snapshot_ms();
eprintln!(
"[supertable drain] cell build phases (summed CPU, {n_cells_total} cells): train {train_ms:.1}ms + assign {assign_ms:.1}ms + calibrate {calib_ms:.1}ms",
);
}
}
eprintln!(
"[supertable drain] done ({}, {} batch(es), budget {} sf): total {:.1}ms; RSS {} -> {} MiB",
match consolidate {
DrainConsolidate::Kmeans => "kmeans",
DrainConsolidate::Splice => "splice",
},
n_batches,
batch_cfg,
drain_t0.elapsed().as_secs_f64() * 1e3,
drain_rss0
.map(|v| format!("{v:.0}"))
.unwrap_or_else(|| "?".into()),
proc_rss_mib()
.map(|v| format!("{v:.0}"))
.unwrap_or_else(|| "?".into()),
);
let clamped_components = transcode_clamped_components() - transcode_clamp_baseline;
if clamped_components > 0 {
eprintln!(
"[supertable drain] BUG: {clamped_components} component(s) saturated their \
destination Sq8 quantizer during this drain's re-encodes (#512 failure \
mode). Cosine: an ingest path bypassed normalization; L2/NegDot: a \
destination grid failed to cover its inputs. Affected rows' recall \
silently degrades — find the source and rebuild the table.",
);
}
refresh_slow_vector_state(&hidden_inner).await?;
schedule_background_storage_reclaim(Arc::clone(&hidden_inner));
Ok(())
}
async fn load_materialized_rows_from_ivf_superfile(
inner: &SupertableInner,
entry: &Arc<SuperfileEntry>,
column: &str,
now: time::Instant,
only_cells: Option<&[u32]>,
) -> Result<Vec<MaterializedIvfRow>, BuildError> {
let (reader, bitmap) = open_ivf_reader_with_tombstones(inner, entry, now).await?;
let vec_reader = reader
.vec()
.ok_or_else(|| BuildError::Store("IVF cell superfile missing vector index".into()))?;
if vec_reader.is_multi_cell() {
let groups =
group_multicell_rows(vec_reader, column, only_cells, bitmap.as_deref()).await?;
return Ok(groups.into_iter().flat_map(|(_, rows)| rows).collect());
}
let manifest = inner.manifest.load_full();
let stable_ids = stable_ids_by_local_for_routing(&manifest, entry, &reader, &None)
.await
.map_err(|e| BuildError::Store(e.to_string()))?;
materialized_ivf_rows_in_doc_order(vec_reader, column, &stable_ids, bitmap.as_deref()).await
}
async fn open_ivf_reader_with_tombstones(
inner: &SupertableInner,
entry: &Arc<SuperfileEntry>,
now: time::Instant,
) -> Result<(Arc<SuperfileReader>, Option<Arc<roaring::RoaringBitmap>>), BuildError> {
let storage = inner
.options
.storage
.as_ref()
.ok_or_else(|| BuildError::Store("cell maintenance requires storage".into()))?;
let disk_cache = inner.options.disk_cache.as_ref();
let bitmap = inner
.tombstone_cache
.as_ref()
.map(|t| t.bitmap_for(entry.superfile_id, now))
.transpose()
.map_err(|e| BuildError::Store(e.to_string()))?;
let reader = open_reader(&inner.options.store, disk_cache, Some(storage), entry, true)
.await
.map_err(|e| BuildError::Store(e.to_string()))?;
Ok((reader, bitmap))
}
async fn group_multicell_rows(
vec_reader: &VectorReader,
column: &str,
only_cells: Option<&[u32]>,
tombstones: Option<&roaring::RoaringBitmap>,
) -> Result<Vec<(u32, Vec<MaterializedIvfRow>)>, BuildError> {
let cells = vec_reader
.materialized_cells_rows_async(only_cells)
.await
.ok_or_else(|| {
BuildError::Store(format!(
"IVF maintenance: multi-cell column '{column}' missing Sq8Residual index"
))
})?;
let mut file_doc_base_by_cell: HashMap<u32, u32> = HashMap::new();
let mut running = 0u32;
for (ci, &cell_id) in vec_reader.packed_cell_ids().iter().enumerate() {
file_doc_base_by_cell.insert(cell_id, running);
let n = vec_reader
.vector_columns_config()
.nth(ci)
.map(|c| c.n_docs)
.unwrap_or(0);
running = running.saturating_add(n);
}
let mut out = Vec::with_capacity(cells.len());
for (cell_id, mut rows) in cells {
let base = file_doc_base_by_cell.get(&cell_id).copied().unwrap_or(0);
if let Some(bm) = tombstones {
rows.retain(|r| !bm.contains(base + r.local_doc_id));
}
out.push((cell_id, rows));
}
Ok(out)
}
async fn cell_doc_counts_for_entry(
inner: &SupertableInner,
entry: &Arc<SuperfileEntry>,
superseded: Option<&BTreeSet<u32>>,
) -> Result<Vec<(u32, u32)>, BuildError> {
let storage = inner
.options
.storage
.as_ref()
.ok_or_else(|| BuildError::Store("cell maintenance requires storage".into()))?;
let reader = open_reader(
&inner.options.store,
inner.options.disk_cache.as_ref(),
Some(storage),
entry,
true,
)
.await
.map_err(|e| BuildError::Store(e.to_string()))?;
let v = reader
.vec()
.ok_or_else(|| BuildError::Store("IVF entry missing vector index".into()))?;
let is_superseded = |cell: u32| superseded.is_some_and(|s| s.contains(&cell));
if v.is_multi_cell() {
Ok(v.packed_cell_ids()
.iter()
.filter(|&&cell| !is_superseded(cell))
.filter_map(|&cell| {
let n = v.packed_cell_n_docs(cell)?;
Some((cell, n))
})
.collect())
} else {
let cell = entry.partition_hint.unwrap_or(0);
if is_superseded(cell) {
Ok(vec![])
} else {
Ok(vec![(cell, entry.n_docs as u32)])
}
}
}
fn proc_rss_mib() -> Option<f64> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
for line in status.lines() {
if let Some(rest) = line.strip_prefix("VmRSS:") {
let kb: f64 = rest.trim().trim_end_matches("kB").trim().parse().ok()?;
return Some(kb / 1024.0);
}
}
None
}
#[derive(Clone)]
struct SpilledPackedCell {
cell_id: u32,
n_docs: u32,
rerank_codec: RerankCodec,
subsection_len: u64,
subsection_path: PathBuf,
stable_ids_path: PathBuf,
}
enum DrainCellSource {
Packed(SpilledPackedCell),
Rows(SpilledCellRows),
}
impl MultiCellSubsectionSource for SpilledPackedCell {
fn cell_id(&self) -> u32 {
self.cell_id
}
fn n_docs(&self) -> u32 {
self.n_docs
}
fn len(&self) -> u64 {
self.subsection_len
}
fn rerank_codec(&self) -> RerankCodec {
self.rerank_codec
}
fn write_to(&self, output: &mut dyn Write) -> Result<(), SuperfileBuildError> {
let file = File::open(&self.subsection_path)?;
let copied = io::copy(&mut BufReader::new(file), output)?;
if copied != self.subsection_len {
return Err(SuperfileBuildError::VectorSchemaMismatch(format!(
"cell {} subsection spill length {copied} != expected {}",
self.cell_id, self.subsection_len
)));
}
Ok(())
}
}
impl MultiCellSubsectionSource for &SpilledPackedCell {
fn cell_id(&self) -> u32 {
(*self).cell_id()
}
fn n_docs(&self) -> u32 {
(*self).n_docs()
}
fn len(&self) -> u64 {
(*self).len()
}
fn rerank_codec(&self) -> RerankCodec {
(*self).rerank_codec()
}
fn write_to(&self, output: &mut dyn Write) -> Result<(), SuperfileBuildError> {
(*self).write_to(output)
}
}
fn spill_packed_cell(
scratch: &Path,
cell_id: u32,
subsection: MergedIvfSubsection,
stable_ids: &[i128],
) -> Result<SpilledPackedCell, BuildError> {
if stable_ids.len() != subsection.n_docs as usize {
return Err(BuildError::Store(format!(
"cell {cell_id}: stable_ids len {} != subsection n_docs {}",
stable_ids.len(),
subsection.n_docs
)));
}
let subsection_path = scratch.join(format!("cell-{cell_id}.ivf"));
let subsection_temp = scratch.join(format!("cell-{cell_id}.ivf.tmp"));
{
let mut subsection_file = File::create(&subsection_temp)
.map_err(|error| BuildError::Store(format!("cell subsection create: {error}")))?;
subsection_file
.write_all(&subsection.bytes)
.map_err(|error| BuildError::Store(format!("cell subsection write: {error}")))?;
subsection_file
.sync_all()
.map_err(|error| BuildError::Store(format!("cell subsection fsync: {error}")))?;
}
fs::rename(&subsection_temp, &subsection_path)
.map_err(|error| BuildError::Store(format!("cell subsection rename: {error}")))?;
let subsection_len = subsection.bytes.len() as u64;
let stable_ids_path = scratch.join(format!("cell-{cell_id}.ids"));
let stable_ids_temp = scratch.join(format!("cell-{cell_id}.ids.tmp"));
{
let ids_file = File::create(&stable_ids_temp)
.map_err(|error| BuildError::Store(format!("cell ids create: {error}")))?;
let mut writer = BufWriter::new(ids_file);
for stable_id in stable_ids {
writer
.write_all(&stable_id.to_le_bytes())
.map_err(|error| BuildError::Store(format!("cell ids write: {error}")))?;
}
writer
.flush()
.map_err(|error| BuildError::Store(format!("cell ids flush: {error}")))?;
writer
.get_ref()
.sync_all()
.map_err(|error| BuildError::Store(format!("cell ids fsync: {error}")))?;
}
fs::rename(&stable_ids_temp, &stable_ids_path)
.map_err(|error| BuildError::Store(format!("cell ids rename: {error}")))?;
Ok(SpilledPackedCell {
cell_id,
n_docs: subsection.n_docs,
rerank_codec: subsection.rerank_codec,
subsection_len,
subsection_path,
stable_ids_path,
})
}
fn build_spilled_packed_cell_from_rows(
scratch: &Path,
cell_id: u32,
spill: &SpilledCellRows,
vector_config: &VectorConfig,
) -> Result<SpilledPackedCell, BuildError> {
let subsection_path = scratch.join(format!("cell-{cell_id}.ivf"));
let subsection_temp = scratch.join(format!("cell-{cell_id}.ivf.tmp"));
let stable_ids_path = scratch.join(format!("cell-{cell_id}.ids"));
let stable_ids_temp = scratch.join(format!("cell-{cell_id}.ids.tmp"));
let (cell_config, cell_n_cent) = drain_cell_vector_config(vector_config, spill.n_rows());
let built = build_merged_subsection_from_spilled_materialized(
cell_config,
cell_n_cent,
spill,
&subsection_temp,
&stable_ids_temp,
scratch,
)?;
fs::rename(&subsection_temp, &subsection_path)
.map_err(|error| BuildError::Store(format!("cell subsection rename: {error}")))?;
fs::rename(&stable_ids_temp, &stable_ids_path)
.map_err(|error| BuildError::Store(format!("cell ids rename: {error}")))?;
Ok(SpilledPackedCell {
cell_id,
n_docs: built.n_docs,
rerank_codec: built.rerank_codec,
subsection_len: built.subsection_len,
subsection_path,
stable_ids_path,
})
}
fn restore_spilled_packed_cell(
scratch: &Path,
cell_id: u32,
state: &DrainLocalCell,
) -> Result<SpilledPackedCell, BuildError> {
let rerank_codec = RerankCodec::from_codec_id(state.rerank_codec_id).ok_or_else(|| {
BuildError::Store(format!(
"cell {cell_id}: checkpoint has unknown codec id {}",
state.rerank_codec_id
))
})?;
let subsection_path = scratch.join(format!("cell-{cell_id}.ivf"));
let stable_ids_path = scratch.join(format!("cell-{cell_id}.ids"));
let subsection_size = fs::metadata(&subsection_path)
.map_err(|error| BuildError::Store(format!("cell subsection metadata: {error}")))?
.len();
if subsection_size != state.subsection_len {
return Err(BuildError::Store(format!(
"cell {cell_id}: checkpointed subsection length {} != file length {subsection_size}",
state.subsection_len
)));
}
let ids_size = fs::metadata(&stable_ids_path)
.map_err(|error| BuildError::Store(format!("cell ids metadata: {error}")))?
.len();
let expected_ids_size = u64::from(state.n_docs) * STABLE_ID_BYTES as u64;
if ids_size != expected_ids_size {
return Err(BuildError::Store(format!(
"cell {cell_id}: checkpointed ids length {expected_ids_size} != file length {ids_size}"
)));
}
Ok(SpilledPackedCell {
cell_id,
n_docs: state.n_docs,
rerank_codec,
subsection_len: state.subsection_len,
subsection_path,
stable_ids_path,
})
}
fn remove_spilled_packed_cell(cell: &SpilledPackedCell) {
let _ = fs::remove_file(&cell.subsection_path);
let _ = fs::remove_file(&cell.stable_ids_path);
}
fn load_merged_from_spilled(
cell: &SpilledPackedCell,
dim: usize,
) -> Result<(MergedIvfSubsection, Vec<i128>), BuildError> {
let bytes = fs::read(&cell.subsection_path)
.map_err(|error| BuildError::Store(format!("cell subsection spill read: {error}")))?;
if bytes.len() as u64 != cell.subsection_len {
return Err(BuildError::Store(format!(
"cell {}: spill length {} != expected {}",
cell.cell_id,
bytes.len(),
cell.subsection_len
)));
}
if bytes.len() < SUB_HEADER_SIZE + CRC_BYTES {
return Err(BuildError::Store(format!(
"cell {}: spilled subsection too short",
cell.cell_id
)));
}
let centroids_off = u64::from_le_bytes(
bytes[sub_hdr::CENTROIDS_OFF_OFF..sub_hdr::CENTROIDS_OFF_OFF + 8]
.try_into()
.expect("8-byte centroids off"),
) as usize;
let cluster_idx_off = u64::from_le_bytes(
bytes[sub_hdr::CLUSTER_IDX_OFF_OFF..sub_hdr::CLUSTER_IDX_OFF_OFF + 8]
.try_into()
.expect("8-byte cluster idx off"),
) as usize;
let summary_off = u64::from_le_bytes(
bytes[sub_hdr::SUMMARY_OFF_OFF..sub_hdr::SUMMARY_OFF_OFF + 8]
.try_into()
.expect("8-byte summary off"),
) as usize;
let codec_meta_size = u32::from_le_bytes(
bytes[sub_hdr::CODEC_META_SIZE_OFF..sub_hdr::CODEC_META_SIZE_OFF + U32_BYTES]
.try_into()
.expect("4-byte codec meta size"),
) as usize;
if cluster_idx_off < centroids_off || !(cluster_idx_off - centroids_off).is_multiple_of(dim * 4)
{
return Err(BuildError::Store(format!(
"cell {}: invalid centroid region for dim {dim}",
cell.cell_id
)));
}
let n_cent = (cluster_idx_off - centroids_off) / (dim * 4);
let codec_meta_off = cluster_idx_off + n_cent * CLUSTER_IDX_ENTRY_BYTES;
let ids = read_spilled_stable_ids(cell)?;
Ok((
MergedIvfSubsection {
bytes,
n_cent,
n_docs: cell.n_docs,
rerank_codec: cell.rerank_codec,
summary_offset_in_sub: summary_off,
codec_meta_offset_in_sub: if codec_meta_size == 0 {
0
} else {
codec_meta_off
},
codec_meta_size,
},
ids,
))
}
fn accumulate_splice_cell(
packed_cells: &mut Vec<SpilledPackedCell>,
local_checkpoint: &mut DrainLocalCheckpoint,
added_per_cell: &mut HashMap<u32, u32>,
completed_shards: &HashSet<u32>,
shard_count: usize,
scratch: &Path,
cell_id: u32,
subsection: MergedIvfSubsection,
stable_ids: Vec<i128>,
dim: usize,
metric: Metric,
) -> Result<(), BuildError> {
let shard = packed_cell_shard(cell_id, shard_count) as u32;
if completed_shards.contains(&shard) {
return Ok(());
}
let (subsection, stable_ids) =
match packed_cells.iter().position(|cell| cell.cell_id == cell_id) {
Some(idx) => {
let existing = packed_cells.swap_remove(idx);
let (left, left_ids) = load_merged_from_spilled(&existing, dim)?;
remove_spilled_packed_cell(&existing);
local_checkpoint.built_cells.remove(&cell_id);
merge_fragment_subsections(&left, &left_ids, &subsection, &stable_ids, dim, metric)?
}
None => (subsection, stable_ids),
};
let n_docs = subsection.n_docs;
let packed = spill_packed_cell(scratch, cell_id, subsection, &stable_ids)?;
local_checkpoint.built_cells.insert(
cell_id,
DrainLocalCell {
n_docs: packed.n_docs,
subsection_len: packed.subsection_len,
rerank_codec_id: packed.rerank_codec.codec_id(),
},
);
packed_cells.push(packed);
added_per_cell.insert(cell_id, n_docs);
Ok(())
}
fn read_spilled_stable_ids(cell: &SpilledPackedCell) -> Result<Vec<i128>, BuildError> {
let mut reader = BufReader::new(
File::open(&cell.stable_ids_path)
.map_err(|error| BuildError::Store(format!("cell ids spill open: {error}")))?,
);
let mut ids = Vec::with_capacity(cell.n_docs as usize);
let mut encoded = [0u8; STABLE_ID_BYTES];
for _ in 0..cell.n_docs {
reader
.read_exact(&mut encoded)
.map_err(|error| BuildError::Store(format!("cell ids spill read: {error}")))?;
ids.push(i128::from_le_bytes(encoded));
}
Ok(ids)
}
fn packed_cell_shard_count(options: &SupertableOptions) -> usize {
options.writer_pool.current_num_threads().max(1)
}
fn packed_cell_shard(cell: u32, shard_count: usize) -> usize {
debug_assert!(shard_count > 0);
(cell as usize) % shard_count
}
fn group_cells_by_packed_shard<T>(
cells: Vec<(u32, T)>,
shard_count: usize,
) -> Vec<(u32, Vec<(u32, T)>)> {
debug_assert!(shard_count > 0);
let mut buckets: Vec<Vec<(u32, T)>> = (0..shard_count).map(|_| Vec::new()).collect();
for (cell, payload) in cells {
buckets[packed_cell_shard(cell, shard_count)].push((cell, payload));
}
buckets
.into_iter()
.enumerate()
.filter(|(_, cells)| !cells.is_empty())
.map(|(shard, mut cells)| {
cells.sort_unstable_by_key(|(cell, _)| *cell);
(shard as u32, cells)
})
.collect()
}
#[derive(Clone, Copy)]
enum PackRow<'a> {
Fp32 { stable_id: i128, vector: &'a [f32] },
}
struct AssignedCellGroup<'a> {
cell_id: u32,
members: Vec<(i128, bool, PackRow<'a>)>,
}
struct PackedCellGroup {
cell_id: u32,
subsection: MergedIvfSubsection,
#[cfg(test)]
stable_ids: Vec<i128>,
}
fn pack_row_stable_id(row: PackRow<'_>) -> i128 {
match row {
PackRow::Fp32 { stable_id, .. } => stable_id,
}
}
fn assign_cells<'a>(
rows: &[PackRow<'a>],
clusters: &ClusterCentroids,
metric: Metric,
rot_seed: u64,
replica_target_factor: f32,
) -> Result<Vec<AssignedCellGroup<'a>>, BuildError> {
if rows.is_empty() {
return Ok(Vec::new());
}
let replica_extra_budget = drain_replica_extra_budget(rows.len(), replica_target_factor);
let admit_ctx = RabitqAdmitContext::new(clusters.dim as usize, rot_seed);
let window = opann::assignment_shortlist_window(clusters.n_cent as usize);
let assignments: Vec<opann::BoundaryAssignment> = rows
.par_iter()
.map(|row| match *row {
PackRow::Fp32 { vector, .. } => {
opann::boundary_assignment_fp32(clusters, metric, vector, &admit_ctx, window)
}
})
.collect();
let mut replica_candidates: Vec<(usize, u32, f32)> = assignments
.iter()
.enumerate()
.flat_map(|(row_idx, assignment)| {
assignment
.replicas
.iter()
.flatten()
.map(move |&(cell, margin)| (row_idx, cell, margin))
})
.collect();
replica_candidates.sort_by(|a, b| a.2.total_cmp(&b.2));
let mut buckets: HashMap<u32, Vec<(i128, bool, PackRow<'a>)>> = HashMap::new();
for (row_idx, cell, _) in replica_candidates.into_iter().take(replica_extra_budget) {
let row = rows[row_idx];
buckets
.entry(cell)
.or_default()
.push((pack_row_stable_id(row), false, row));
}
for (row, assignment) in rows.iter().zip(&assignments) {
buckets
.entry(assignment.primary)
.or_default()
.push((pack_row_stable_id(*row), true, *row));
}
let mut out = Vec::with_capacity(buckets.len());
for (cell_id, mut members) in buckets {
members.sort_by_key(|(stable_id, is_primary, _)| (!*is_primary, *stable_id));
out.push(AssignedCellGroup { cell_id, members });
}
out.sort_unstable_by_key(|group| group.cell_id);
Ok(out)
}
fn drain_cell_vector_config(cfg: &VectorConfig, n_rows: usize) -> (VectorConfig, usize) {
debug_assert!(n_rows > 0);
let dim = cfg.dim;
let rerank_codec = if cfg.rerank_codec.is_ivf_mergeable() {
cfg.rerank_codec
} else {
RerankCodec::Sq8Residual
};
let rabitq_bytes = dim.div_ceil(u8::BITS as usize);
let rerank_bytes = rerank_codec.per_vector_bytes(dim);
let row_stride =
rabitq_bytes + DOC_ID_BYTES + rerank_bytes + STABLE_ID_BYTES + mem::size_of::<f32>();
let rows_per_run = (DRAIN_FINE_RUN_TARGET_BYTES / row_stride.max(1)).max(1);
let n_cent = n_rows.div_ceil(rows_per_run).clamp(1, n_rows);
let cell_cfg = VectorConfig {
rerank_codec,
provided_centroids: None,
..cfg.clone()
};
(cell_cfg, n_cent)
}
fn drain_pack_assigned_cell(
group: AssignedCellGroup<'_>,
cfg: &VectorConfig,
) -> Result<PackedCellGroup, BuildError> {
let AssignedCellGroup { cell_id, members } = group;
if members.is_empty() {
return Err(BuildError::Store(format!(
"cell {cell_id}: assign produced an empty bucket"
)));
}
let dim = cfg.dim;
let (cell_cfg, cell_n_cent) = drain_cell_vector_config(cfg, members.len());
let stable_ids: Vec<i128> = members.iter().map(|(stable_id, _, _)| *stable_id).collect();
let mut corpus = Vec::with_capacity(members.len() * dim);
for (_, _, row) in &members {
match *row {
PackRow::Fp32 { vector, .. } => corpus.extend_from_slice(vector),
}
}
let subsection =
build_merged_subsection_from_fp32(cell_cfg, cell_n_cent, Arc::new(corpus), &stable_ids)?;
Ok(PackedCellGroup {
cell_id,
subsection,
#[cfg(test)]
stable_ids,
})
}
fn build_one_shard_from_packed_cells(
cells: Vec<(u32, MergedIvfSubsection, Vec<i128>)>,
options: &SupertableOptions,
) -> Result<ShardOutput, BuildError> {
if cells.is_empty() {
return Err(BuildError::NoDocsToBuild);
}
let mut cells = cells;
cells.sort_by_key(|(cell_id, _, _)| *cell_id);
let mut stable_ids: Vec<i128> = Vec::new();
let mut subsections: Vec<(u32, MergedIvfSubsection)> = Vec::with_capacity(cells.len());
for (cell_id, sub, ids) in cells {
if ids.len() != sub.n_docs as usize {
return Err(BuildError::Store(format!(
"cell {cell_id}: stable_ids len {} != subsection n_docs {}",
ids.len(),
sub.n_docs
)));
}
stable_ids.extend_from_slice(&ids);
subsections.push((cell_id, sub));
}
let id_array = Decimal128Array::from_iter_values(stable_ids.iter().copied())
.with_precision_and_scale(
crate::supertable::options::DECIMAL128_PRECISION,
crate::supertable::options::DECIMAL128_SCALE,
)
.expect("invariant: precision 38 + scale 0 always valid for any i128 payload");
let scalar = RecordBatch::try_new(
options.scalar_schema(),
vec![Arc::new(id_array) as ArrayRef],
)
.map_err(|_| BuildError::BatchSchemaMismatch)?;
let mut builder = SuperfileBuilder::new(
options
.builder_options()
.with_vector_layout(VectorLayout::MultiCellIvf),
)?;
builder.add_batch_ids_only(&scalar)?;
builder.set_prebuilt_multi_cell_ivfs(subsections)?;
let id_min = stable_ids.iter().copied().min().unwrap_or(0);
let id_max = stable_ids.iter().copied().max().unwrap_or(0);
let n_docs = stable_ids.len() as u64;
let scalar_stats = ScalarStatsAgg::from_batches(&options.scalar_schema(), &[&scalar]);
let mut output = NamedTempFile::new()
.map_err(|error| BuildError::Store(format!("compacted shard temp create: {error}")))?;
{
let mut writer = BufWriter::new(output.as_file_mut());
builder.finish_to(&mut writer)?;
writer
.flush()
.map_err(|error| BuildError::Store(format!("compacted shard temp flush: {error}")))?;
}
let bytes = mmap_readonly_bytes(output.path())
.map_err(|error| BuildError::Store(format!("compacted shard mmap: {error}")))?;
Ok(ShardOutput {
bytes,
n_docs,
id_min,
id_max,
scalar_stats,
})
}
fn build_prepared_from_packed_cells(
inner: &SupertableInner,
shard_id: u32,
cells: Vec<(u32, MergedIvfSubsection, Vec<i128>)>,
) -> Result<PreparedSuperfile, BuildError> {
let shard = build_one_shard_from_packed_cells(cells, &inner.options)?;
let prepared = prepare_superfile(inner, shard)?.ok_or(BuildError::NoDocsToBuild)?;
let entry = finish_superfile_entry(prepared.entry, Some(shard_id))?;
Ok(PreparedSuperfile {
entry,
bytes_for_store: prepared.bytes_for_store,
bytes_for_storage: prepared.bytes_for_storage,
bytes_for_cache: prepared.bytes_for_cache,
})
}
fn build_prepared_from_spilled_cells(
inner: &SupertableInner,
scratch: &Path,
shard_id: u32,
cells: &[(u32, SpilledPackedCell)],
) -> Result<PreparedSuperfile, BuildError> {
if cells.is_empty() {
return Err(BuildError::NoDocsToBuild);
}
let mut ordered: Vec<&SpilledPackedCell> = cells.iter().map(|(_, cell)| cell).collect();
ordered.sort_unstable_by_key(|cell| cell.cell_id);
let n_docs = ordered
.iter()
.map(|cell| cell.n_docs as usize)
.sum::<usize>();
let scalar_schema = inner.options.scalar_schema();
let mut scalar_stats = HashMap::new();
let mut builder = SuperfileBuilder::new(
inner
.options
.builder_options()
.with_vector_layout(VectorLayout::MultiCellIvf),
)?;
let mut id_min = i128::MAX;
let mut id_max = i128::MIN;
let mut ids_seen = 0usize;
for cell in &ordered {
let mut reader = BufReader::new(
File::open(&cell.stable_ids_path)
.map_err(|error| BuildError::Store(format!("cell ids spill open: {error}")))?,
);
let mut remaining = cell.n_docs as usize;
while remaining > 0 {
let take = remaining.min(DRAIN_ID_BATCH_ROWS);
let mut ids = Vec::with_capacity(take);
let mut encoded = [0u8; STABLE_ID_BYTES];
for _ in 0..take {
reader
.read_exact(&mut encoded)
.map_err(|error| BuildError::Store(format!("cell ids spill read: {error}")))?;
let id = i128::from_le_bytes(encoded);
id_min = id_min.min(id);
id_max = id_max.max(id);
ids.push(id);
}
let id_array = Decimal128Array::from_iter_values(ids)
.with_precision_and_scale(DECIMAL128_PRECISION, DECIMAL128_SCALE)
.expect("invariant: precision 38 + scale 0 always valid for any i128 payload");
let scalar =
RecordBatch::try_new(scalar_schema.clone(), vec![Arc::new(id_array) as ArrayRef])
.map_err(|_| BuildError::BatchSchemaMismatch)?;
ScalarStatsAgg::merge(
&mut scalar_stats,
&ScalarStatsAgg::from_batch(&scalar_schema, &scalar),
);
builder.add_batch_ids_only(&scalar)?;
ids_seen += take;
remaining -= take;
}
}
if ids_seen != n_docs {
return Err(BuildError::Store(format!(
"shard {shard_id}: stable id count {ids_seen} != expected {n_docs}"
)));
}
let mut output = NamedTempFile::new_in(scratch)
.map_err(|error| BuildError::Store(format!("packed shard temp create: {error}")))?;
builder.finish_multi_cell_sources_to(&ordered, BufWriter::new(output.as_file_mut()))?;
output
.as_file_mut()
.flush()
.map_err(|error| BuildError::Store(format!("packed shard temp flush: {error}")))?;
let bytes = mmap_readonly_bytes(output.path())
.map_err(|error| BuildError::Store(format!("packed shard mmap: {error}")))?;
let (id_min, id_max) = if n_docs == 0 {
(0, 0)
} else {
(id_min, id_max)
};
let shard = ShardOutput {
bytes,
n_docs: n_docs as u64,
id_min,
id_max,
scalar_stats,
};
let prepared = prepare_superfile(inner, shard)?.ok_or(BuildError::NoDocsToBuild)?;
let entry = finish_superfile_entry(prepared.entry, Some(shard_id))?;
Ok(PreparedSuperfile {
entry,
bytes_for_store: prepared.bytes_for_store,
bytes_for_storage: prepared.bytes_for_storage,
bytes_for_cache: prepared.bytes_for_cache,
})
}
fn commit_shards_via_drain(
buffer: &[BufferedBatch],
inner: &SupertableInner,
clusters: &ClusterCentroids,
metric: Metric,
) -> Result<(Vec<ShardOutput>, Vec<Option<u32>>), BuildError> {
let stage_t0 = time::Instant::now();
let vc = inner
.options
.vector_columns
.first()
.cloned()
.ok_or_else(|| BuildError::Store("drain-commit requires a vector column".into()))?;
let dim = vc.dim;
if dim != clusters.dim as usize {
return Err(BuildError::Store(format!(
"commit vector dim {dim} does not match global grid dim {}",
clusters.dim
)));
}
let mut stable_ids: Vec<i128> = Vec::new();
let mut scalar_batches: Vec<&RecordBatch> = Vec::with_capacity(buffer.len());
for buffered in buffer {
let id_col = buffered
.scalar
.column(0)
.as_any()
.downcast_ref::<Decimal128Array>()
.ok_or_else(|| {
BuildError::IdColumnWrongType(
inner.options.id_column.clone(),
"<id column not Decimal128 at runtime>".to_string(),
)
})?;
for i in 0..id_col.len() {
stable_ids.push(id_col.value(i));
}
scalar_batches.push(&buffered.scalar);
}
if stable_ids.is_empty() {
return Ok((Vec::new(), Vec::new()));
}
let vector_views: Vec<VectorColumnView<'_>> = inner
.options
.vector_columns
.iter()
.enumerate()
.map(|(col_idx, col)| VectorColumnView::over(buffer, col_idx, col.dim))
.collect();
let primary_view = vector_views
.first()
.ok_or_else(|| BuildError::Store("drain-commit missing vector values".into()))?;
if primary_view.n_rows() != stable_ids.len() {
return Err(BuildError::Store(format!(
"commit vector rows {} != id rows {}",
primary_view.n_rows(),
stable_ids.len()
)));
}
let scalar_schema = inner.options.scalar_schema();
let source_scalar = concat_batches(&scalar_schema, scalar_batches.iter().copied())
.map_err(|err| BuildError::Store(err.to_string()))?;
let local_by_id: HashMap<i128, u32> = stable_ids
.iter()
.enumerate()
.map(|(local, &id)| (id, local as u32))
.collect();
let flatten_elapsed = stage_t0.elapsed();
let rows: Vec<PackRow<'_>> = stable_ids
.iter()
.enumerate()
.map(|(local, &stable_id)| {
Ok(PackRow::Fp32 {
stable_id,
vector: primary_view.row(local)?,
})
})
.collect::<Result<_, BuildError>>()?;
let replica_target = drain_replica_target_factor();
let assigned = inner
.options
.writer_pool
.install(|| assign_cells(&rows, clusters, metric, vc.rot_seed, replica_target))?;
let assign_elapsed = stage_t0.elapsed().saturating_sub(flatten_elapsed);
let assigned_cells: Vec<(u32, AssignedCellGroup<'_>)> = assigned
.into_iter()
.map(|group| (group.cell_id, group))
.collect();
let packed_shards =
group_cells_by_packed_shard(assigned_cells, packed_cell_shard_count(&inner.options));
let options = &inner.options;
let shard_outputs = fanout_shards(&inner.options.writer_pool, &packed_shards, |task| {
let (shard_id, cells) = task;
build_one_packed_shard_via_drain(
cells,
&source_scalar,
&vector_views,
&local_by_id,
options,
&vc,
)
.map(|output| output.map(|output| (*shard_id, output)))
})?;
let fanout_elapsed = stage_t0
.elapsed()
.saturating_sub(flatten_elapsed)
.saturating_sub(assign_elapsed);
if crate::storage::io_counters::timeline_enabled() {
eprintln!(
"[supertable commit] flatten {:.1}ms + assign {:.1}ms + shard pack/finish {:.1}ms",
flatten_elapsed.as_secs_f64() * 1e3,
assign_elapsed.as_secs_f64() * 1e3,
fanout_elapsed.as_secs_f64() * 1e3,
);
}
let mut outputs = Vec::with_capacity(shard_outputs.len());
let mut cell_hints = Vec::with_capacity(shard_outputs.len());
for entry in shard_outputs.into_iter().flatten() {
cell_hints.push(Some(entry.0));
outputs.push(entry.1);
}
Ok((outputs, cell_hints))
}
fn build_one_packed_shard_via_drain(
cells: &[(u32, AssignedCellGroup<'_>)],
source_scalar: &RecordBatch,
vector_views: &[VectorColumnView<'_>],
local_by_id: &HashMap<i128, u32>,
options: &SupertableOptions,
vc: &VectorConfig,
) -> Result<Option<ShardOutput>, BuildError> {
let mut ordered_locals: Vec<u32> = Vec::new();
for (_, group) in cells {
for (member_id, is_primary, _) in &group.members {
if !*is_primary {
continue;
}
let local = local_by_id.get(member_id).copied().ok_or_else(|| {
BuildError::Store(format!(
"primary stable_id {member_id} missing from commit rows"
))
})?;
ordered_locals.push(local);
}
}
if ordered_locals.is_empty() {
return Ok(None);
}
let (packed_groups, body_and_fts) = rayon::join(
|| {
cells
.iter()
.map(|(cell_id, group)| {
let owned = AssignedCellGroup {
cell_id: *cell_id,
members: group.members.clone(),
};
drain_pack_assigned_cell(owned, vc)
})
.collect::<Result<Vec<_>, BuildError>>()
},
|| build_shard_parquet_and_fts(source_scalar, vector_views, &ordered_locals, options),
);
let packed_groups = packed_groups?;
let (mut builder, id_min, id_max, n_docs, scalar_stats) = body_and_fts?;
let subsections: Vec<(u32, MergedIvfSubsection)> = packed_groups
.into_iter()
.map(|g| (g.cell_id, g.subsection))
.collect();
builder.set_prebuilt_multi_cell_ivfs(subsections)?;
let mut output = NamedTempFile::new()
.map_err(|error| BuildError::Store(format!("compacted shard temp create: {error}")))?;
{
let mut writer = BufWriter::new(output.as_file_mut());
builder.finish_to(&mut writer)?;
writer
.flush()
.map_err(|error| BuildError::Store(format!("compacted shard temp flush: {error}")))?;
}
let bytes = mmap_readonly_bytes(output.path())
.map_err(|error| BuildError::Store(format!("compacted shard mmap: {error}")))?;
Ok(Some(ShardOutput {
bytes,
n_docs,
id_min,
id_max,
scalar_stats,
}))
}
#[allow(clippy::type_complexity)]
fn build_shard_parquet_and_fts(
source_scalar: &RecordBatch,
vector_views: &[VectorColumnView<'_>],
ordered_locals: &[u32],
options: &SupertableOptions,
) -> Result<
(
SuperfileBuilder,
i128,
i128,
u64,
HashMap<String, ScalarStatsAgg>,
),
BuildError,
> {
let take_indices = UInt32Array::from(ordered_locals.to_vec());
let columns: Vec<ArrayRef> = source_scalar
.columns()
.iter()
.map(|column| take(column.as_ref(), &take_indices, None))
.collect::<Result<Vec<_>, _>>()
.map_err(|err| BuildError::Store(err.to_string()))?;
let scalar = RecordBatch::try_new(source_scalar.schema(), columns)
.map_err(|_| BuildError::BatchSchemaMismatch)?;
let mut ordered_vectors: Vec<Vec<f32>> = Vec::with_capacity(vector_views.len());
for view in vector_views {
let mut ordered = Vec::with_capacity(ordered_locals.len() * view.dim);
for &local in ordered_locals {
ordered.extend_from_slice(view.row(local as usize)?);
}
ordered_vectors.push(ordered);
}
let vector_slices: Vec<&[f32]> = ordered_vectors.iter().map(Vec::as_slice).collect();
let mut builder = SuperfileBuilder::new(
options
.builder_options()
.with_vector_layout(VectorLayout::MultiCellIvf),
)?;
builder.add_batch(&scalar, &vector_slices)?;
let scalar_schema = options.scalar_schema();
let scalar_stats = ScalarStatsAgg::from_batches(&scalar_schema, &[&scalar]);
let id_col = scalar
.column(0)
.as_any()
.downcast_ref::<Decimal128Array>()
.ok_or_else(|| {
BuildError::IdColumnWrongType(
options.id_column.clone(),
"<id column not Decimal128 at runtime>".to_string(),
)
})?;
let mut id_min = i128::MAX;
let mut id_max = i128::MIN;
for i in 0..id_col.len() {
let v = id_col.value(i);
id_min = id_min.min(v);
id_max = id_max.max(v);
}
let n_docs = id_col.len() as u64;
let (id_min, id_max) = if n_docs == 0 {
(0, 0)
} else {
(id_min, id_max)
};
Ok((builder, id_min, id_max, n_docs, scalar_stats))
}
const MIN_ROWS_TO_SPLIT_CELL: usize = 2;
const SPLIT_RESIDENT_BYTES_PER_ROW_DIM: u64 = 7;
fn split_batch_memory_budget_bytes() -> u64 {
split_batch_window_bytes(config::global().vector.compaction_max_memory_mb)
}
fn split_batch_window_bytes(configured_mib: u64) -> u64 {
const MIB: u64 = 1024 * 1024;
const SPLIT_BATCH_FALLBACK_BUDGET_MIB: u64 = 4096;
let budget_mib = if configured_mib == 0 {
SPLIT_BATCH_FALLBACK_BUDGET_MIB
} else {
configured_mib
};
budget_mib.saturating_mul(MIB)
}
fn estimate_split_resident_bytes(physical_rows: u64, dim: u32) -> u64 {
physical_rows
.saturating_mul(u64::from(dim))
.saturating_mul(SPLIT_RESIDENT_BYTES_PER_ROW_DIM)
}
fn select_split_batch(
cell_counts: &HashMap<u32, u64>,
unsplittable: &HashSet<u32>,
dim: u32,
budget_bytes: u64,
max_cells: usize,
) -> Vec<u32> {
let candidates = split_candidates(cell_counts, unsplittable);
let mut batch: Vec<u32> = Vec::new();
let mut estimated_bytes = 0u64;
for (cell, n) in candidates {
if batch.len() >= max_cells {
break;
}
let cost = estimate_split_resident_bytes(n, dim);
if !batch.is_empty() && estimated_bytes.saturating_add(cost) > budget_bytes {
continue;
}
batch.push(cell);
estimated_bytes = estimated_bytes.saturating_add(cost);
}
batch
}
pub(in crate::supertable) async fn scan_cell_parents(
inner: &SupertableInner,
manifest: &ManifestSnapshot,
only_cells: Option<&[u32]>,
) -> Result<(HashMap<u32, u64>, HashMap<u32, Vec<Arc<SuperfileEntry>>>), BuildError> {
let superseded_map = manifest.get_superseded_cells();
let mut cell_counts: HashMap<u32, u64> = HashMap::new();
let mut parents_by_cell: HashMap<u32, Vec<Arc<SuperfileEntry>>> = HashMap::new();
for entry in manifest.superfiles.iter() {
let superseded = superseded_map.and_then(|m| m.get(&entry.superfile_id));
for (cell, n) in cell_doc_counts_for_entry(inner, entry, superseded).await? {
if only_cells.is_some_and(|want| !want.contains(&cell)) {
continue;
}
*cell_counts.entry(cell).or_default() += u64::from(n);
parents_by_cell
.entry(cell)
.or_default()
.push(Arc::clone(entry));
}
}
Ok((cell_counts, parents_by_cell))
}
struct ExtractedCellRows {
cell: u32,
parent_ids: Vec<Uuid>,
rows: Vec<MaterializedIvfRow>,
}
struct PlannedCellSplit {
cell: u32,
parent_ids: Vec<Uuid>,
rows: Vec<MaterializedIvfRow>,
sub_centroids: Vec<f32>,
k: usize,
assign: Vec<u32>,
}
struct BuiltCellSplit {
cell: u32,
parent_ids: Vec<Uuid>,
child_ids: Vec<u32>,
child_counts: Vec<u32>,
prepared: Vec<(u32, PreparedSuperfile)>,
}
pub(in crate::supertable) struct SplitBatchOutcome {
pub(in crate::supertable) per_cell: Vec<(u32, Option<Vec<(u32, u64)>>)>,
pub(in crate::supertable) new_entries_by_cell: Vec<(u32, Arc<SuperfileEntry>)>,
}
impl SplitBatchOutcome {
fn no_op_cells(cells: Vec<u32>) -> Self {
Self {
per_cell: cells.into_iter().map(|cell| (cell, None)).collect(),
new_entries_by_cell: Vec::new(),
}
}
}
const SPLIT_BULK_REPACK_MIN_CANDIDATE_FRACTION: f64 = 0.25;
pub(in crate::supertable) const REPACK_CHECKPOINT_SCHEMA: u32 = 0x5250_4B31;
#[derive(Serialize, Deserialize)]
struct RepackCheckpoint {
schema: u32,
}
#[derive(Deserialize)]
struct PendingMetadataSchemaProbe {
schema: u32,
}
fn pending_metadata_schema(metadata: &[u8]) -> Option<u32> {
serde_json::from_slice::<PendingMetadataSchemaProbe>(metadata)
.ok()
.map(|probe| probe.schema)
}
fn repack_scratch_dir() -> PathBuf {
env::temp_dir()
.join("infino-repack")
.join(Uuid::new_v4().to_string())
}
fn split_candidates(
cell_counts: &HashMap<u32, u64>,
unsplittable: &HashSet<u32>,
) -> Vec<(u32, u64)> {
let mut candidates: Vec<(u32, u64)> = cell_counts
.iter()
.filter(|&(cell, &n)| {
opann::split_candidate(n)
&& (n as usize) >= MIN_ROWS_TO_SPLIT_CELL
&& !unsplittable.contains(cell)
})
.map(|(&cell, &n)| (cell, n))
.collect();
candidates.sort_unstable_by_key(|&(cell, n)| (cmp::Reverse(n), cell));
candidates
}
fn live_split_extraction_jobs(
cells: &[u32],
parents_by_cell: &HashMap<u32, Vec<Arc<SuperfileEntry>>>,
superseded_map: Option<&BTreeMap<Uuid, BTreeSet<u32>>>,
) -> Vec<(u32, Vec<Arc<SuperfileEntry>>)> {
cells
.iter()
.map(|&cell| {
let parents: Vec<Arc<SuperfileEntry>> = parents_by_cell
.get(&cell)
.map(|entries| {
entries
.iter()
.filter(|entry| {
!superseded_map
.and_then(|m| m.get(&entry.superfile_id))
.is_some_and(|s| s.contains(&cell))
})
.map(Arc::clone)
.collect()
})
.unwrap_or_default();
(cell, parents)
})
.collect()
}
async fn extract_split_cell_rows(
inner: &SupertableInner,
column: &str,
now: time::Instant,
jobs: Vec<(u32, Vec<Arc<SuperfileEntry>>)>,
) -> Result<Vec<ExtractedCellRows>, BuildError> {
let extraction = jobs.into_iter().map(|(cell, parents)| {
let column = column.to_owned();
async move {
let only_cell = [cell];
let mut rows: Vec<MaterializedIvfRow> = Vec::new();
for entry in &parents {
let mut entry_rows = load_materialized_rows_from_ivf_superfile(
inner,
entry,
&column,
now,
Some(&only_cell),
)
.await?;
rows.append(&mut entry_rows);
}
Ok::<ExtractedCellRows, BuildError>(ExtractedCellRows {
cell,
parent_ids: parents.iter().map(|entry| entry.superfile_id).collect(),
rows,
})
}
});
let results: Vec<Result<ExtractedCellRows, BuildError>> = stream::iter(extraction)
.buffered(drain_read_concurrency())
.collect()
.await;
results.into_iter().collect()
}
fn plan_split_wave(
plan_inputs: Vec<ExtractedCellRows>,
clusters: &ClusterCentroids,
metric: Metric,
modality_d: f64,
) -> Vec<Result<PlannedCellSplit, u32>> {
plan_inputs
.into_par_iter()
.map(|extracted| {
let ExtractedCellRows {
cell,
parent_ids,
rows,
} = extracted;
let split_refs: Vec<&EncodedCellRow> = rows.iter().map(|r| &r.encoded).collect();
let Some((k, self_tune)) =
opann::cell_split_plan(&split_refs, clusters.dim as usize, cell, modality_d)
else {
return Err(cell);
};
let (sub_centroids, assign) =
opann::plan_sq8_split_kway(&split_refs, clusters, cell, metric, k, self_tune);
drop(split_refs);
let dim = clusters.dim as usize;
let well_formed = !sub_centroids.is_empty()
&& sub_centroids.len() % dim == 0
&& assign.len() == rows.len();
debug_assert!(
well_formed,
"planner shape for cell {cell}: {} centroid floats (dim {dim}), \
{} assignments for {} rows",
sub_centroids.len(),
assign.len(),
rows.len()
);
if !well_formed {
warn!(
cell,
"cell split: malformed planner output; skipping the cell"
);
return Err(cell);
}
let k = sub_centroids.len() / dim;
Ok(PlannedCellSplit {
cell,
parent_ids,
rows,
sub_centroids,
k,
assign,
})
})
.collect()
}
async fn pin_uploaded_superfiles(
inner: &SupertableInner,
entries: Vec<Arc<SuperfileEntry>>,
probe_existing: bool,
) -> Result<(), BuildError> {
if probe_existing {
let manifest = inner.manifest.load_full();
if let (Some((uri, hash)), Some(storage)) = (
manifest.slow_vector_state_blob(),
inner.options.storage.as_ref(),
) && let Ok(state) =
slow_vector_state::load_full_state(storage.as_ref(), uri, &hash).await
&& let Some(pending) = state.pending_drain
&& pending_metadata_schema(&pending.metadata) == Some(DRAIN_CHECKPOINT_SCHEMA)
{
warn!(
"split upload pin replacing a stale drain checkpoint (the drain phase \
precedes the split pass, so a surviving drain pin is unconsumable)"
);
}
}
let metadata = serde_json::to_vec(&RepackCheckpoint {
schema: REPACK_CHECKPOINT_SCHEMA,
})
.map_err(|error| BuildError::Store(format!("split upload pin encode: {error}")))?;
stamp_slow_vector_state(
inner,
Some(slow_vector_state::PendingDrainState { metadata, entries }),
)
.await
}
async fn unpin_after_failed_publish(inner: &SupertableInner, error: BuildError) -> BuildError {
if let Err(unpin) = stamp_slow_vector_state(inner, None).await {
debug!("split upload unpin after failed publish: {unpin}");
}
error
}
struct RepackedSplit {
cell: u32,
parent_ids: Vec<Uuid>,
child_ids: Vec<u32>,
child_counts: Vec<u32>,
packed: Vec<(u32, SpilledPackedCell)>,
}
fn build_split_subcell(
inner: &SupertableInner,
shard_count: usize,
cell_id: u32,
mut rows: Vec<MaterializedIvfRow>,
) -> Result<Option<PreparedSuperfile>, BuildError> {
if rows.is_empty() {
return Ok(None);
}
for (i, row) in rows.iter_mut().enumerate() {
row.local_doc_id = i as u32;
}
let stable_ids: Vec<i128> = rows.iter().map(|r| r.stable_id).collect();
let base_cfg = inner
.options
.vector_columns
.first()
.cloned()
.ok_or_else(|| BuildError::Store("missing vector column".into()))?;
let (cfg, cell_n_cent) = drain_cell_vector_config(&base_cfg, rows.len());
let sub = build_merged_subsection_from_materialized(cfg, cell_n_cent, rows)?;
let shard_id = packed_cell_shard(cell_id, shard_count) as u32;
build_prepared_from_packed_cells(inner, shard_id, vec![(cell_id, sub, stable_ids)]).map(Some)
}
pub(in crate::supertable) async fn split_overflow_cell_batch(
inner: &Arc<SupertableInner>,
manifest: &ManifestSnapshot,
batch_cells: &[u32],
modality_d: f64,
parents_by_cell: &HashMap<u32, Vec<Arc<SuperfileEntry>>>,
) -> Result<SplitBatchOutcome, BuildError> {
let (clusters, column, routing, metric) = match manifest.get_partition_strategy() {
PartitionStrategy::VectorCell {
clusters,
column,
routing,
} => {
let Some(vec_col) = inner.options.vector_columns.first() else {
return Ok(SplitBatchOutcome::no_op_cells(batch_cells.to_vec()));
};
(clusters, column, routing, vec_col.metric)
}
_ => return Ok(SplitBatchOutcome::no_op_cells(batch_cells.to_vec())),
};
if clusters.n_cent == 0 || clusters.dim == 0 {
return Ok(SplitBatchOutcome::no_op_cells(batch_cells.to_vec()));
}
let now = time::Instant::now();
let storage = inner
.options
.storage
.clone()
.ok_or_else(|| BuildError::Store("cell split requires storage".into()))?;
let superseded_map = manifest.get_superseded_cells();
let mut noop_cells: Vec<u32> = Vec::new();
let mut eligible_cells: Vec<u32> = Vec::new();
for &cell in batch_cells {
if cell >= clusters.n_cent {
noop_cells.push(cell);
} else {
eligible_cells.push(cell);
}
}
let jobs = live_split_extraction_jobs(&eligible_cells, parents_by_cell, superseded_map);
let extracted = extract_split_cell_rows(inner, &column, now, jobs).await?;
let mut plan_inputs: Vec<ExtractedCellRows> = Vec::new();
for item in extracted {
if item.rows.len() < MIN_ROWS_TO_SPLIT_CELL {
noop_cells.push(item.cell);
} else {
plan_inputs.push(item);
}
}
let plan_clusters = clusters.clone();
let planned_or_noop: Vec<Result<PlannedCellSplit, u32>> = run_on_pool(
Some(maint_pool()?),
"cell split batch planning",
move || plan_split_wave(plan_inputs, &plan_clusters, metric, modality_d),
)
.await
.map_err(|e| BuildError::Store(format!("cell split batch planning: {e}")))?;
let mut planned: Vec<PlannedCellSplit> = Vec::new();
for item in planned_or_noop {
match item {
Ok(split) => planned.push(split),
Err(cell) => noop_cells.push(cell),
}
}
if planned.is_empty() {
return Ok(SplitBatchOutcome::no_op_cells(noop_cells));
}
planned.sort_unstable_by_key(|split| split.cell);
let fold_inputs: Vec<(u32, &[f32], usize)> = planned
.iter()
.map(|split| (split.cell, split.sub_centroids.as_slice(), split.k))
.collect();
let (updated_clusters, ids_per_split) =
opann::insert_split_centroids_batch(&clusters, &fold_inputs);
drop(fold_inputs);
let shard_count = packed_cell_shard_count(&inner.options);
let build_inner = Arc::clone(inner);
let build_jobs: Vec<(PlannedCellSplit, Vec<u32>)> =
planned.into_iter().zip(ids_per_split).collect();
let built: Vec<BuiltCellSplit> = run_on_pool(
Some(maint_pool()?),
"cell split batch child builds",
move || {
build_jobs
.into_par_iter()
.map(|(split, child_ids)| {
let PlannedCellSplit {
cell,
parent_ids,
rows,
assign,
..
} = split;
let mut groups: Vec<Vec<MaterializedIvfRow>> =
(0..child_ids.len()).map(|_| Vec::new()).collect();
for (row, &side) in rows.into_iter().zip(assign.iter()) {
debug_assert!(
(side as usize) < child_ids.len(),
"planner assignment {side} outside {} children",
child_ids.len()
);
groups[(side as usize).min(child_ids.len() - 1)].push(row);
}
let child_counts: Vec<u32> = groups.iter().map(|g| g.len() as u32).collect();
let mut prepared: Vec<(u32, PreparedSuperfile)> = Vec::new();
for (group, &child_id) in groups.into_iter().zip(child_ids.iter()) {
if let Some(p) =
build_split_subcell(&build_inner, shard_count, child_id, group)?
{
prepared.push((child_id, p));
}
}
Ok(BuiltCellSplit {
cell,
parent_ids,
child_ids,
child_counts,
prepared,
})
})
.collect::<Result<Vec<BuiltCellSplit>, BuildError>>()
},
)
.await
.map_err(|e| BuildError::Store(format!("cell split batch child builds: {e}")))??;
if built.iter().all(|b| b.prepared.is_empty()) {
noop_cells.extend(built.iter().map(|b| b.cell));
return Ok(SplitBatchOutcome::no_op_cells(noop_cells));
}
let count_updates: HashMap<u32, u32> = built
.iter()
.flat_map(|b| {
b.child_ids
.iter()
.copied()
.zip(b.child_counts.iter().copied())
})
.collect();
if let Some(&bad) = count_updates
.keys()
.find(|&&cell| cell >= updated_clusters.n_cent)
{
return Err(BuildError::Store(format!(
"cell split batch: child id {bad} outside the folded grid ({} cells)",
updated_clusters.n_cent
)));
}
let updated_clusters = opann::apply_cell_count_updates(&updated_clusters, &count_updates);
let mut superseded_additions: BTreeMap<Uuid, BTreeSet<u32>> = BTreeMap::new();
let mut committed_cells: Vec<(u32, Vec<(u32, u64)>)> = Vec::with_capacity(built.len());
let mut new_entries_by_cell: Vec<(u32, Arc<SuperfileEntry>)> = Vec::new();
let mut all_prepared: Vec<PreparedSuperfile> = Vec::new();
for b in built {
for parent in &b.parent_ids {
superseded_additions
.entry(*parent)
.or_default()
.insert(b.cell);
}
committed_cells.push((
b.cell,
b.child_ids
.iter()
.copied()
.zip(b.child_counts.iter().map(|&c| u64::from(c)))
.collect(),
));
for (cell, p) in b.prepared {
new_entries_by_cell.push((cell, Arc::clone(&p.entry)));
all_prepared.push(p);
}
}
let SuperfilePublishBatch {
new_entries,
to_remove: _,
pending_storage_writes,
pending_cache_inserts: _,
pending_store_inserts,
} = collect_prepared_superfiles(inner, all_prepared)?;
pin_uploaded_superfiles(inner, new_entries.clone(), true).await?;
let multipart_threshold = inner.options.put_multipart_threshold_bytes;
let uploads = pending_storage_writes.into_iter().map(|(uri, bytes)| {
let storage = Arc::clone(&storage);
async move {
put_new_superfile_bytes(&storage, multipart_threshold, uri, bytes)
.await
.map_err(|error| BuildError::Store(error.to_string()))
}
});
let mut in_flight = stream::iter(uploads).buffer_unordered(commit_write_concurrency());
while let Some(upload) = in_flight.next().await {
if let Err(error) = upload {
drop(in_flight);
return Err(unpin_after_failed_publish(inner, error).await);
}
}
drop(in_flight);
let list_metadata = CommitListMetadata {
partition_strategy: Some(PartitionStrategy::VectorCell {
column: column.clone(),
clusters: updated_clusters,
routing,
}),
drained_ranges: None,
global_vector_index: None,
superseded_cells_additions: Some(superseded_additions),
};
let no_removals: Vec<Arc<SuperfileEntry>> = Vec::new();
let new_manifest = match persist_commit_async(
inner,
Arc::clone(&storage),
new_entries,
&no_removals,
Vec::new(),
Vec::new(),
list_metadata,
)
.await
{
Ok(manifest) => manifest,
Err(error) => {
return Err(unpin_after_failed_publish(inner, BuildError::from(error)).await);
}
};
inner.manifest.store(Arc::new(new_manifest));
apply_pending_store_inserts(inner, pending_store_inserts);
schedule_background_storage_reclaim(Arc::clone(inner));
for (cell, children) in &committed_cells {
debug!(
cell = *cell,
rows = children.iter().map(|(_, n)| *n).sum::<u64>(),
k = children.len(),
child_min = children.iter().map(|(_, n)| *n).min().unwrap_or(0),
child_max = children.iter().map(|(_, n)| *n).max().unwrap_or(0),
"cell split committed"
);
}
debug!(
cells = committed_cells.len(),
children = new_entries_by_cell.len(),
noops = noop_cells.len(),
wall_ms = now.elapsed().as_millis() as u64,
"cell split batch committed"
);
let mut per_cell: Vec<(u32, Option<Vec<(u32, u64)>>)> = committed_cells
.into_iter()
.map(|(cell, children)| (cell, Some(children)))
.collect();
per_cell.extend(noop_cells.into_iter().map(|cell| (cell, None)));
Ok(SplitBatchOutcome {
per_cell,
new_entries_by_cell,
})
}
pub(in crate::supertable) async fn split_repack_bulk(
inner: &Arc<SupertableInner>,
manifest: &ManifestSnapshot,
candidates: Vec<(u32, u64)>,
modality_d: f64,
parents_by_cell: &HashMap<u32, Vec<Arc<SuperfileEntry>>>,
) -> Result<SplitBatchOutcome, BuildError> {
let all_cells = || {
candidates
.iter()
.map(|&(cell, _)| cell)
.collect::<Vec<u32>>()
};
let (clusters, column, routing, metric) = match manifest.get_partition_strategy() {
PartitionStrategy::VectorCell {
clusters,
column,
routing,
} => {
let Some(vec_col) = inner.options.vector_columns.first() else {
return Ok(SplitBatchOutcome::no_op_cells(all_cells()));
};
(clusters, column, routing, vec_col.metric)
}
_ => return Ok(SplitBatchOutcome::no_op_cells(all_cells())),
};
if clusters.n_cent == 0 || clusters.dim == 0 {
return Ok(SplitBatchOutcome::no_op_cells(all_cells()));
}
let Some(base_cfg) = inner.options.vector_columns.first().cloned() else {
return Ok(SplitBatchOutcome::no_op_cells(all_cells()));
};
let now = time::Instant::now();
let storage = inner
.options
.storage
.clone()
.ok_or_else(|| BuildError::Store("cell split requires storage".into()))?;
let superseded_map = manifest.get_superseded_cells();
let dim = clusters.dim;
let initial_n_cent = clusters.n_cent;
let budget_bytes = split_batch_memory_budget_bytes();
struct RepackScratchGuard {
path: PathBuf,
}
impl Drop for RepackScratchGuard {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.path);
}
}
let scratch = repack_scratch_dir();
fs::create_dir_all(&scratch)
.map_err(|error| BuildError::Store(format!("repack scratch create: {error}")))?;
let _scratch_guard = RepackScratchGuard {
path: scratch.clone(),
};
let mut running_clusters = clusters;
let mut noop_cells: Vec<u32> = Vec::new();
let mut committed_cells: Vec<(u32, Vec<(u32, u64)>)> = Vec::new();
let mut superseded_additions: BTreeMap<Uuid, BTreeSet<u32>> = BTreeMap::new();
let mut count_updates: HashMap<u32, u32> = HashMap::new();
let mut packed_children: Vec<(u32, SpilledPackedCell)> = Vec::new();
let cell_bytes: HashMap<u32, u64> = candidates.iter().copied().collect();
let mut queue: Vec<(u32, u64)> = candidates;
while !queue.is_empty() {
let mut wave_cells: Vec<u32> = Vec::new();
let mut wave_bytes = 0u64;
let mut deferred: Vec<(u32, u64)> = Vec::new();
for (cell, n) in queue.drain(..) {
if cell >= initial_n_cent {
noop_cells.push(cell);
continue;
}
let cost = estimate_split_resident_bytes(n, dim);
if wave_cells.is_empty() || wave_bytes.saturating_add(cost) <= budget_bytes {
wave_cells.push(cell);
wave_bytes = wave_bytes.saturating_add(cost);
} else {
deferred.push((cell, n));
}
}
queue = deferred;
if wave_cells.is_empty() {
continue;
}
let wave_reservation: Option<Reservation> = match inner
.options
.connection_memory_budget
.try_reserve(usize::try_from(wave_bytes).unwrap_or(usize::MAX))
{
Ok(reservation) => Some(reservation),
Err(_) if wave_cells.len() > 1 => {
let requeue: Vec<(u32, u64)> = wave_cells
.split_off(1)
.into_iter()
.map(|cell| (cell, cell_bytes.get(&cell).copied().unwrap_or(0)))
.collect();
let mut restored = requeue;
restored.append(&mut queue);
queue = restored;
let single = estimate_split_resident_bytes(
cell_bytes.get(&wave_cells[0]).copied().unwrap_or(0),
dim,
);
inner
.options
.connection_memory_budget
.try_reserve(usize::try_from(single).unwrap_or(usize::MAX))
.ok()
}
Err(_) => None,
};
if wave_reservation.is_none() {
debug!(
cell = wave_cells[0],
"repack: budget denied; single-cell wave proceeds unreserved"
);
}
let jobs = live_split_extraction_jobs(&wave_cells, parents_by_cell, superseded_map);
let extracted = extract_split_cell_rows(inner, &column, now, jobs).await?;
let mut plan_inputs: Vec<ExtractedCellRows> = Vec::new();
for item in extracted {
if item.rows.len() < MIN_ROWS_TO_SPLIT_CELL {
noop_cells.push(item.cell);
} else {
plan_inputs.push(item);
}
}
if plan_inputs.is_empty() {
continue;
}
let plan_clusters = running_clusters.clone();
let planned_or_noop: Vec<Result<PlannedCellSplit, u32>> =
run_on_pool(Some(maint_pool()?), "repack wave planning", move || {
plan_split_wave(plan_inputs, &plan_clusters, metric, modality_d)
})
.await
.map_err(|e| BuildError::Store(format!("repack wave planning: {e}")))?;
let mut planned: Vec<PlannedCellSplit> = Vec::new();
for item in planned_or_noop {
match item {
Ok(split) => planned.push(split),
Err(cell) => noop_cells.push(cell),
}
}
if planned.is_empty() {
continue;
}
planned.sort_unstable_by_key(|split| split.cell);
let fold_inputs: Vec<(u32, &[f32], usize)> = planned
.iter()
.map(|split| (split.cell, split.sub_centroids.as_slice(), split.k))
.collect();
let (next_clusters, ids_per_split) =
opann::insert_split_centroids_batch(&running_clusters, &fold_inputs);
drop(fold_inputs);
running_clusters = next_clusters;
let wave_scratch = scratch.clone();
let wave_cfg = base_cfg.clone();
let build_jobs: Vec<(PlannedCellSplit, Vec<u32>)> =
planned.into_iter().zip(ids_per_split).collect();
let repacked: Vec<RepackedSplit> =
run_on_pool(Some(maint_pool()?), "repack wave child packs", move || {
build_jobs
.into_par_iter()
.map(|(split, child_ids)| {
let PlannedCellSplit {
cell,
parent_ids,
rows,
assign,
..
} = split;
let mut groups: Vec<Vec<MaterializedIvfRow>> =
(0..child_ids.len()).map(|_| Vec::new()).collect();
for (row, &side) in rows.into_iter().zip(assign.iter()) {
debug_assert!(
(side as usize) < child_ids.len(),
"planner assignment {side} outside {} children",
child_ids.len()
);
groups[(side as usize).min(child_ids.len() - 1)].push(row);
}
let child_counts: Vec<u32> =
groups.iter().map(|g| g.len() as u32).collect();
let mut packed: Vec<(u32, SpilledPackedCell)> = Vec::new();
for (group, &child_id) in groups.into_iter().zip(child_ids.iter()) {
if group.is_empty() {
continue;
}
let mut spills: HashMap<u32, MaterializedRowSpillWriter> =
HashMap::new();
let mut added: HashMap<u32, u32> = HashMap::new();
for row in &group {
spill_row_to_cell(
&mut spills,
&mut added,
&wave_scratch,
child_id,
row,
)?;
}
drop(group);
let spill = spills
.remove(&child_id)
.ok_or_else(|| {
BuildError::Store(
"repack: spill writer missing for child".into(),
)
})?
.finish()?;
let packed_cell = build_spilled_packed_cell_from_rows(
&wave_scratch,
child_id,
&spill,
&wave_cfg,
)?;
spill.remove_files();
packed.push((child_id, packed_cell));
}
Ok(RepackedSplit {
cell,
parent_ids,
child_ids,
child_counts,
packed,
})
})
.collect::<Result<Vec<RepackedSplit>, BuildError>>()
})
.await
.map_err(|e| BuildError::Store(format!("repack wave child packs: {e}")))??;
for split in repacked {
for parent in &split.parent_ids {
superseded_additions
.entry(*parent)
.or_default()
.insert(split.cell);
}
count_updates.extend(
split
.child_ids
.iter()
.copied()
.zip(split.child_counts.iter().copied()),
);
committed_cells.push((
split.cell,
split
.child_ids
.iter()
.copied()
.zip(split.child_counts.iter().map(|&c| u64::from(c)))
.collect(),
));
packed_children.extend(split.packed);
}
}
if packed_children.is_empty() {
return Ok(SplitBatchOutcome::no_op_cells(noop_cells));
}
if let Some(&bad) = count_updates
.keys()
.find(|&&cell| cell >= running_clusters.n_cent)
{
return Err(BuildError::Store(format!(
"repack: child id {bad} outside the folded grid ({} cells)",
running_clusters.n_cent
)));
}
let final_clusters = opann::apply_cell_count_updates(&running_clusters, &count_updates);
let shard_count = packed_cell_shard_count(&inner.options);
let buckets = group_cells_by_packed_shard(packed_children, shard_count);
let build_inner = Arc::clone(inner);
let build_scratch = scratch.clone();
let bucket_cells: HashMap<u32, Vec<u32>> = buckets
.iter()
.map(|(shard, cells)| (*shard, cells.iter().map(|(cell, _)| *cell).collect()))
.collect();
let prepared: Vec<(u32, PreparedSuperfile)> =
run_on_pool(Some(maint_pool()?), "repack shard assembly", move || {
buckets
.par_iter()
.map(|(shard, cells)| {
build_prepared_from_spilled_cells(&build_inner, &build_scratch, *shard, cells)
.map(|prepared| (*shard, prepared))
})
.collect::<Result<Vec<(u32, PreparedSuperfile)>, BuildError>>()
})
.await
.map_err(|e| BuildError::Store(format!("repack shard assembly: {e}")))??;
let mut new_entries_by_cell: Vec<(u32, Arc<SuperfileEntry>)> = Vec::new();
for (shard, prepared_shard) in &prepared {
if let Some(cells) = bucket_cells.get(shard) {
for &cell in cells {
new_entries_by_cell.push((cell, Arc::clone(&prepared_shard.entry)));
}
}
}
let prepared: Vec<PreparedSuperfile> = prepared.into_iter().map(|(_, p)| p).collect();
let SuperfilePublishBatch {
new_entries,
to_remove: _,
pending_storage_writes,
pending_cache_inserts: _,
pending_store_inserts,
} = collect_prepared_superfiles(inner, prepared)?;
pin_uploaded_superfiles(inner, new_entries.clone(), true).await?;
let multipart_threshold = inner.options.put_multipart_threshold_bytes;
let uploads = pending_storage_writes.into_iter().map(|(uri, bytes)| {
let storage = Arc::clone(&storage);
async move {
put_new_superfile_bytes(&storage, multipart_threshold, uri, bytes)
.await
.map_err(|error| BuildError::Store(error.to_string()))
}
});
let mut in_flight = stream::iter(uploads).buffer_unordered(commit_write_concurrency());
while let Some(landed) = in_flight.next().await {
if let Err(error) = landed {
drop(in_flight);
return Err(unpin_after_failed_publish(inner, error).await);
}
}
drop(in_flight);
let list_metadata = CommitListMetadata {
partition_strategy: Some(PartitionStrategy::VectorCell {
column: column.clone(),
clusters: final_clusters,
routing,
}),
drained_ranges: None,
global_vector_index: None,
superseded_cells_additions: Some(superseded_additions),
};
let no_removals: Vec<Arc<SuperfileEntry>> = Vec::new();
let new_manifest = match persist_commit_async(
inner,
Arc::clone(&storage),
new_entries,
&no_removals,
Vec::new(),
Vec::new(),
list_metadata,
)
.await
{
Ok(manifest) => manifest,
Err(error) => {
return Err(unpin_after_failed_publish(inner, BuildError::from(error)).await);
}
};
inner.manifest.store(Arc::new(new_manifest));
apply_pending_store_inserts(inner, pending_store_inserts);
schedule_background_storage_reclaim(Arc::clone(inner));
debug!(
cells = committed_cells.len(),
children = new_entries_by_cell.len(),
shards = bucket_cells.len(),
noops = noop_cells.len(),
wall_ms = now.elapsed().as_millis() as u64,
"cell split bulk repack committed"
);
let mut per_cell: Vec<(u32, Option<Vec<(u32, u64)>>)> = committed_cells
.into_iter()
.map(|(cell, children)| (cell, Some(children)))
.collect();
per_cell.extend(noop_cells.into_iter().map(|cell| (cell, None)));
Ok(SplitBatchOutcome {
per_cell,
new_entries_by_cell,
})
}
pub(in crate::supertable) async fn split_overflow_cell(
inner: Arc<SupertableInner>,
split_cell: u32,
modality_d: f64,
) -> Result<Option<Vec<(u32, u64)>>, BuildError> {
let manifest = inner.manifest.load_full();
if !matches!(
manifest.get_partition_strategy(),
PartitionStrategy::VectorCell { .. }
) {
return Ok(None);
}
let only_cell = [split_cell];
let (_cell_counts, parents_by_cell) =
scan_cell_parents(&inner, &manifest, Some(&only_cell)).await?;
let outcome =
split_overflow_cell_batch(&inner, &manifest, &only_cell, modality_d, &parents_by_cell)
.await?;
Ok(outcome
.per_cell
.into_iter()
.next()
.and_then(|(_, result)| result))
}
fn apply_split_outcome_to_pass(
outcome: SplitBatchOutcome,
cell_counts: &mut HashMap<u32, u64>,
unsplittable: &mut HashSet<u32>,
parents_by_cell: &mut HashMap<u32, Vec<Arc<SuperfileEntry>>>,
splits_committed: &mut usize,
) {
for (cell, entry) in outcome.new_entries_by_cell {
parents_by_cell.entry(cell).or_default().push(entry);
}
for (cell, result) in outcome.per_cell {
match result {
Some(child_counts) => {
*splits_committed += 1;
for (child, docs) in child_counts {
cell_counts.insert(child, docs);
if !opann::split_overflow_needed(docs) {
unsplittable.insert(child);
}
}
}
None => {
unsplittable.insert(cell);
}
}
}
}
pub(in crate::supertable) async fn split_overflow_cells(
inner: Arc<SupertableInner>,
) -> Result<(), BuildError> {
const MAX_SPLITS_PER_OPTIMIZE: usize = 4096;
let manifest = inner.manifest.load_full();
let (dim, n_cent) = match manifest.get_partition_strategy() {
PartitionStrategy::VectorCell { clusters, .. } => (clusters.dim, clusters.n_cent),
_ => return Ok(()),
};
let (mut cell_counts, mut parents_by_cell) = scan_cell_parents(&inner, &manifest, None).await?;
let mut unsplittable: HashSet<u32> = HashSet::new();
let mut splits_committed = 0usize;
let budget_bytes = split_batch_memory_budget_bytes();
let eligible = split_candidates(&cell_counts, &unsplittable);
if !eligible.is_empty()
&& eligible.len() as f64 >= f64::from(n_cent) * SPLIT_BULK_REPACK_MIN_CANDIDATE_FRACTION
{
let outcome = split_repack_bulk(
&inner,
&manifest,
eligible,
opann::cell_split_modality_d(),
&parents_by_cell,
)
.await?;
apply_split_outcome_to_pass(
outcome,
&mut cell_counts,
&mut unsplittable,
&mut parents_by_cell,
&mut splits_committed,
);
}
loop {
let mut batch = select_split_batch(
&cell_counts,
&unsplittable,
dim,
budget_bytes,
MAX_SPLITS_PER_OPTIMIZE.saturating_sub(splits_committed),
);
if batch.is_empty() {
break;
}
let estimated_bytes: u64 = batch
.iter()
.map(|cell| {
estimate_split_resident_bytes(cell_counts.get(cell).copied().unwrap_or(0), dim)
})
.sum();
let reservation: Option<Reservation> = match inner
.options
.connection_memory_budget
.try_reserve(usize::try_from(estimated_bytes).unwrap_or(usize::MAX))
{
Ok(reservation) => Some(reservation),
Err(_) => {
if batch.len() > 1 {
batch.truncate(1);
let n = cell_counts.get(&batch[0]).copied().unwrap_or(0);
let single_bytes = usize::try_from(estimate_split_resident_bytes(n, dim))
.unwrap_or(usize::MAX);
inner
.options
.connection_memory_budget
.try_reserve(single_bytes)
.ok()
} else {
None
}
}
};
if reservation.is_none() {
debug!(
cell = batch[0],
"cell split: budget denied; single split proceeds unreserved"
);
}
let batch_manifest = inner.manifest.load_full();
let outcome = split_overflow_cell_batch(
&inner,
&batch_manifest,
&batch,
opann::cell_split_modality_d(),
&parents_by_cell,
)
.await?;
drop(reservation);
apply_split_outcome_to_pass(
outcome,
&mut cell_counts,
&mut unsplittable,
&mut parents_by_cell,
&mut splits_committed,
);
if splits_committed >= MAX_SPLITS_PER_OPTIMIZE {
warn!(
"cell split: hit per-optimize split bound ({MAX_SPLITS_PER_OPTIMIZE}); \
over-cap cells remain and will converge on the next optimize"
);
break;
}
}
if splits_committed > 0 {
let over_cap = cell_counts
.values()
.filter(|&&n| opann::split_overflow_needed(n))
.count();
let max_cell = cell_counts.values().copied().max().unwrap_or(0);
debug!(
splits = splits_committed,
cells = cell_counts.len(),
over_cap,
max_cell,
unsplittable = unsplittable.len(),
"cell split pass done"
);
}
Ok(())
}
pub(in crate::supertable) async fn recalibrate_probe_laws(
inner: &Arc<SupertableInner>,
) -> Result<bool, BuildError> {
let manifest = inner.manifest.load_full();
let (clusters, column, routing, metric, rot_seed) = match manifest.get_partition_strategy() {
PartitionStrategy::VectorCell {
clusters,
column,
routing,
} => {
let Some(vec_col) = inner
.options
.vector_columns
.iter()
.find(|cfg| cfg.column == column)
else {
return Err(BuildError::Store(format!(
"vector routing column {column:?} missing from table options"
)));
};
(clusters, column, routing, vec_col.metric, vec_col.rot_seed)
}
_ => return Ok(false),
};
if clusters.n_cent == 0 || clusters.dim == 0 {
return Ok(false);
}
if routing.width_for_k.iter().all(|&w| w == 0) {
return Ok(false);
}
let Some(storage) = inner.options.storage.clone() else {
return Err(BuildError::Store(
"probe-law recalibration requires configured storage".into(),
));
};
let now = time::Instant::now();
let superseded_map = manifest.get_superseded_cells();
let mut work: Vec<(Arc<SuperfileEntry>, Vec<(u32, u32)>)> = Vec::new();
let mut total_docs = 0u64;
for entry in manifest.superfiles.iter() {
let superseded = superseded_map.and_then(|m| m.get(&entry.superfile_id));
let cells: Vec<(u32, u32)> = cell_doc_counts_for_entry(inner, entry, superseded)
.await?
.into_iter()
.filter(|&(_, n)| n > 0)
.collect();
if !cells.is_empty() {
total_docs += cells.iter().map(|&(_, n)| u64::from(n)).sum::<u64>();
work.push((Arc::clone(entry), cells));
}
}
if work.is_empty() || total_docs == 0 {
return Ok(false);
}
let scan_ids: HashSet<Uuid> = manifest.superfiles.iter().map(|e| e.superfile_id).collect();
let mut cal =
opann::WidthLawCalibration::new(clusters.dim as usize, metric, inner.options.target_recall);
let sample_count = total_docs.min(opann::WIDTH_LAW_QUERY_SAMPLE as u64);
let sample_ordinals: Vec<u64> = (0..sample_count)
.map(|i| ((u128::from(i) * u128::from(total_docs)) / u128::from(sample_count)) as u64)
.collect();
let mut picks: BTreeMap<(usize, u32), Vec<u32>> = BTreeMap::new();
let mut base = 0u64;
let mut next_pick = 0usize;
'outer: for (ei, (_, cells)) in work.iter().enumerate() {
for &(cell, n) in cells {
let end = base + u64::from(n);
while next_pick < sample_ordinals.len() && sample_ordinals[next_pick] < end {
picks
.entry((ei, cell))
.or_default()
.push((sample_ordinals[next_pick] - base) as u32);
next_pick += 1;
}
if next_pick == sample_ordinals.len() {
break 'outer;
}
base = end;
}
}
for (&(ei, cell), ordinals) in &picks {
let (entry, cells) = &work[ei];
let rows =
load_materialized_rows_from_ivf_superfile(inner, entry, &column, now, Some(&[cell]))
.await?;
if rows.is_empty() {
continue;
}
let phys = cells
.iter()
.find(|&&(c, _)| c == cell)
.map(|&(_, n)| u64::from(n))
.unwrap_or(0)
.max(1);
let mut last_idx = usize::MAX;
for &ordinal in ordinals {
let idx = ((u128::from(ordinal) * rows.len() as u128) / u128::from(phys)) as usize;
let idx = idx.min(rows.len() - 1);
if idx == last_idx {
continue;
}
last_idx = idx;
cal.offer(&rows[idx]);
}
}
let pool = maint_pool()?;
let pool_hint = opann::rerank_pool_hint(&routing.width_for_k, clusters.n_cent as usize);
let clusters_for_freeze = clusters;
let (cal, clusters) = run_on_pool(Some(pool), "recalibration freeze", move || {
cal.freeze(&clusters_for_freeze, rot_seed, pool_hint);
(cal, clusters_for_freeze)
})
.await
.map_err(|e| BuildError::Store(format!("recalibration freeze: {e}")))?;
let cal = Arc::new(cal);
let chunk_cells = pool.current_num_threads().max(1);
for (entry, cells) in &work {
for chunk in cells.chunks(chunk_cells) {
let mut loaded: Vec<(u32, Vec<MaterializedIvfRow>)> = Vec::with_capacity(chunk.len());
for &(cell, _) in chunk {
let rows = load_materialized_rows_from_ivf_superfile(
inner,
entry,
&column,
now,
Some(&[cell]),
)
.await?;
loaded.push((cell, rows));
}
let chunk_cal = Arc::clone(&cal);
run_on_pool(Some(pool), "recalibration score", move || {
let result = loaded
.par_iter()
.try_for_each(|(cell, rows)| chunk_cal.score_rows(*cell, rows));
drop(chunk_cal);
result
})
.await
.map_err(|e| BuildError::Store(format!("recalibration score: {e}")))??;
}
let reader = open_compaction_input(
&inner.options.store,
inner.options.disk_cache.as_ref(),
inner.options.storage.as_ref(),
entry,
)
.await
.map_err(|e| BuildError::Store(e.to_string()))?;
if let Some(views) = reader
.vec()
.and_then(|v| v.cell_fine_calibration_views(&column))
{
let observe_cal = Arc::clone(&cal);
run_on_pool(Some(pool), "recalibration depth observation", move || {
observe_cal.observe_shard_views(&views);
drop(observe_cal);
})
.await
.map_err(|e| BuildError::Store(format!("recalibration depth observation: {e}")))?;
}
}
let cal = Arc::into_inner(cal)
.ok_or_else(|| BuildError::Store("recalibration state still shared".into()))?;
let Some(laws) = run_on_pool(Some(pool), "recalibration finish", move || {
cal.finish(&clusters)
})
.await
.map_err(|e| BuildError::Store(format!("recalibration finish: {e}")))?
else {
return Ok(false);
};
let max_retries = inner.options.max_commit_retries.max(1);
for attempt in 0..max_retries {
let manifest = inner.manifest.load_full();
let (clusters, fresh_routing) = match manifest.get_partition_strategy() {
PartitionStrategy::VectorCell {
clusters, routing, ..
} => (clusters, routing),
_ => return Ok(false),
};
let mut routing = fresh_routing;
let evidence_current = manifest
.superfiles
.iter()
.map(|e| e.superfile_id)
.collect::<HashSet<Uuid>>()
== scan_ids;
for (slot, measured) in routing.width_for_k.iter_mut().zip(laws.width_for_k) {
if evidence_current {
if measured > 0 {
*slot = measured;
}
} else {
*slot = (*slot).max(measured);
}
}
for (slot, measured) in routing.fine_for_k.iter_mut().zip(laws.fine_for_k) {
*slot = (*slot).max(measured);
}
opann::merge_rerank_with_pools(
&mut routing.rerank_for_k,
&mut routing.rerank_pool_cells,
&laws.rerank_for_k,
laws.pool_cells,
);
opann::clear_rerank_beyond_pool(
&routing.width_for_k,
&mut routing.rerank_for_k,
&routing.rerank_pool_cells,
);
if routing == fresh_routing {
return Ok(false);
}
let list_metadata = CommitListMetadata {
partition_strategy: Some(PartitionStrategy::VectorCell {
column: column.clone(),
clusters: clusters.clone(),
routing,
}),
drained_ranges: None,
global_vector_index: None,
superseded_cells_additions: None,
};
let base = Arc::new(list_metadata.apply(&manifest));
let no_removals: Vec<Arc<SuperfileEntry>> = Vec::new();
match try_commit_attempt(
Arc::clone(&storage),
Arc::clone(&inner.options),
base,
&[],
&no_removals,
NewEntryBirthVersions::StampCommit,
&mut Vec::new(),
&mut Vec::new(),
)
.await
{
Ok(new_manifest) => {
inner.manifest.store(Arc::new(new_manifest));
info!(
"supertable optimize: probe laws recalibrated over {} cells at k={WIDTH_LAW_KS:?}: width {:?}, fine depth {:?}, rerank {:?}",
clusters.n_cent, routing.width_for_k, routing.fine_for_k, routing.rerank_for_k
);
return Ok(true);
}
Err(SupertableCommitError::WriteContentionExhausted) if attempt + 1 < max_retries => {
refresh_inner_state_async(inner, &storage)
.await
.map_err(BuildError::from)?;
sleep(backoff_delay(attempt)).await;
}
Err(e) => {
inner.note_commit_error(&e);
return Err(BuildError::from(e));
}
}
}
Err(BuildError::from(
SupertableCommitError::WriteContentionExhausted,
))
}
pub(super) fn backoff_delay(attempt: u32) -> time::Duration {
const BASE_MS: u64 = 10;
const CAP_MS: u64 = 1000;
const MAX_SHIFT: u32 = 6;
const JITTER_RANGE_PCT: i64 = 30;
const JITTER_MODULUS: u64 = 61;
const PERCENT_DIVISOR: i64 = 100;
let exp = BASE_MS.saturating_mul(1u64 << attempt.min(MAX_SHIFT));
let capped = exp.min(CAP_MS);
let nanos = time::SystemTime::now()
.duration_since(time::UNIX_EPOCH)
.map(|d| d.subsec_nanos() as u64)
.unwrap_or(0);
let jitter_pct = (nanos % JITTER_MODULUS) as i64 - JITTER_RANGE_PCT;
let adjusted = ((capped as i64) + (capped as i64 * jitter_pct / PERCENT_DIVISOR)).max(1) as u64;
time::Duration::from_millis(adjusted)
}
pub(in crate::supertable) async fn refresh_slow_vector_state(
inner: &SupertableInner,
) -> Result<(), BuildError> {
stamp_slow_vector_state(inner, None).await
}
async fn previous_centroid_section(
options: &SupertableOptions,
storage: &dyn StorageProvider,
manifest: &ManifestSnapshot,
) -> Option<Arc<CentroidSection>> {
let reference = manifest.slow_vector_state_centroids_blob()?.clone();
let slot = Arc::clone(&options.centroid_section_cache);
let mut guard = slot.lock().await;
if let Some(section) = guard.as_ref()
&& section.uri() == reference.uri
{
return Some(Arc::clone(section));
}
match fetch_centroid_section(storage, &reference, manifest.get_all_superfiles()).await {
Ok(section) => {
let section = Arc::new(section);
*guard = Some(Arc::clone(§ion));
Some(section)
}
Err(error) => {
tracing::warn!(
"previous centroid section {} unavailable ({error}); republish must compose \
from resident fp32 only",
reference.uri
);
None
}
}
}
pub(in crate::supertable) async fn stamp_slow_vector_state(
inner: &SupertableInner,
pending_drain: Option<slow_vector_state::PendingDrainState>,
) -> Result<(), BuildError> {
let Some(storage) = inner.options.storage.clone() else {
return Ok(());
};
let max_retries = inner.options.max_commit_retries.max(1);
let mut next_id_floor: u64 = 0;
for attempt in 0..max_retries {
let old = inner.manifest.load_full();
let old = if next_id_floor > 0 {
Arc::new(old.with_next_manifest_id_floor(next_id_floor))
} else {
old
};
let entries = old.get_all_superfiles();
if entries.is_empty() && pending_drain.is_none() {
return Ok(());
}
let previous_section =
previous_centroid_section(&inner.options, storage.as_ref(), &old).await;
let published = match pending_drain.as_ref() {
Some(pending) => {
slow_vector_state::write_state_with_pending_drain(
storage.as_ref(),
entries,
pending,
previous_section.as_deref(),
)
.await
}
None => {
slow_vector_state::write_state(
storage.as_ref(),
entries,
previous_section.as_deref(),
)
.await
}
}
.map_err(|e| BuildError::Store(e.to_string()))?;
if let Some((cur_uri, cur_hash)) = old.slow_vector_state_blob()
&& cur_uri == published.uri
&& cur_hash == published.content_hash
&& old.slow_vector_state_centroids_blob() == Some(&published.centroids)
{
return Ok(());
}
let new_manifest =
old.with_slow_vector_state(published.uri, published.content_hash, published.centroids);
let attempted_id = new_manifest.get_manifest_id();
let prev_etag = get_current_manifest_etag(&storage, Arc::clone(&old))
.await
.inspect_err(|e| inner.note_commit_error(e))
.map_err(BuildError::from)?;
match new_manifest
.write(storage.as_ref(), prev_etag.as_deref(), &[])
.await
{
Ok(()) => {
inner.manifest.store(Arc::new(new_manifest));
return Ok(());
}
Err(SupertableCommitError::WriteContentionExhausted) if attempt + 1 < max_retries => {
next_id_floor = next_id_floor.max(
refresh_and_orphaned_id_floor(inner, &storage, attempted_id)
.await
.map_err(|e| BuildError::Store(e.to_string()))?,
);
sleep(backoff_delay(attempt)).await;
}
Err(e) => return Err(BuildError::Store(e.to_string())),
}
}
Err(BuildError::Store(
"slow vector-state refresh: write contention exhausted".into(),
))
}
async fn release_mutation_lease(wal_store: &WalStore, wal_id: WalId, owner: SupertableHandleId) {
if let Err(e) = lease::try_release(wal_store, wal_id, owner).await {
debug!(
error = %e,
"supertable: could not hand back the WAL lease after a failed mutation; \
recovery picks the WAL up once the lease expires"
);
}
}
async fn record_hidden_deleted_ids(
inner: &SupertableInner,
new_deleted: &[i128],
) -> Result<(), BuildError> {
if new_deleted.is_empty() {
return Ok(());
}
let Some(storage) = inner.options.storage.clone() else {
return Ok(());
};
let max_retries = inner.options.max_commit_retries.max(1);
let mut next_id_floor: u64 = 0;
for attempt in 0..max_retries {
let old = inner.manifest.load_full();
let old = if next_id_floor > 0 {
Arc::new(old.with_next_manifest_id_floor(next_id_floor))
} else {
old
};
let mut ids = hidden_deleted::deleted_user_ids(&old)
.map_err(|e| BuildError::Store(e.to_string()))?
.as_ref()
.clone();
let before = ids.len();
ids.extend_from_slice(new_deleted);
ids.sort_unstable();
ids.dedup();
if ids.len() == before {
return Ok(());
}
let bytes = encode_deleted_ids(&ids);
let new_manifest = old.with_deleted_user_ids(bytes);
let attempted_id = new_manifest.get_manifest_id();
let prev_etag = get_current_manifest_etag(&storage, Arc::clone(&old))
.await
.inspect_err(|e| inner.note_commit_error(e))
.map_err(BuildError::from)?;
match new_manifest
.write(storage.as_ref(), prev_etag.as_deref(), &[])
.await
{
Ok(()) => {
inner.manifest.store(Arc::new(new_manifest));
return Ok(());
}
Err(SupertableCommitError::WriteContentionExhausted) if attempt + 1 < max_retries => {
next_id_floor = next_id_floor.max(
refresh_and_orphaned_id_floor(inner, &storage, attempted_id)
.await
.map_err(|e| BuildError::Store(e.to_string()))?,
);
sleep(backoff_delay(attempt)).await;
}
Err(e) => return Err(BuildError::Store(e.to_string())),
}
}
Err(BuildError::Store(
"deleted-set record: write contention exhausted".into(),
))
}
#[derive(Debug, Default, Clone)]
pub(crate) struct CommitListMetadata {
pub(crate) partition_strategy: Option<PartitionStrategy>,
pub(crate) global_vector_index: Option<GlobalVectorIndex>,
pub(crate) drained_ranges: Option<DrainedVersionRanges>,
pub(crate) superseded_cells_additions: Option<BTreeMap<Uuid, BTreeSet<u32>>>,
}
impl CommitListMetadata {
pub(crate) fn empty() -> Self {
Self::default()
}
pub(crate) fn is_empty(&self) -> bool {
self.partition_strategy.is_none()
&& self.global_vector_index.is_none()
&& self.drained_ranges.is_none()
&& self.superseded_cells_additions.is_none()
}
pub(crate) fn apply(&self, base: &ManifestSnapshot) -> ManifestSnapshot {
let mut out = base.with_drained_ranges(base.get_drained_ranges());
if let Some(strategy) = self.partition_strategy.clone() {
out = out.with_partition_strategy(strategy);
}
if let Some(index) = self.global_vector_index.clone() {
out = out.with_global_vector_index(index);
}
if let Some(ranges) = self.drained_ranges.clone() {
out = out.with_drained_ranges(ranges);
}
if let Some(additions) = &self.superseded_cells_additions {
out = out.with_superseded_cells_added(additions);
}
out
}
}
pub(in crate::supertable) async fn persist_commit_async(
inner: &SupertableInner,
storage: Arc<dyn StorageProvider>,
new_entries: Vec<Arc<SuperfileEntry>>,
entries_to_remove: &[Arc<SuperfileEntry>],
mut pending_storage_writes: Vec<(SuperfileUri, Bytes)>,
mut pending_storage_replaces: Vec<(SuperfileUri, Bytes)>,
list_metadata: CommitListMetadata,
) -> Result<ManifestSnapshot, SupertableCommitError> {
let storage_async = Arc::clone(&storage);
let opts = Arc::clone(&inner.options);
let max_retries = opts.max_commit_retries.max(1);
let drive = async move {
let mut last_err: Option<SupertableCommitError> = None;
let mut next_id_floor: u64 = 0;
for attempt in 0..max_retries {
let old = inner.manifest.load_full();
let old = if next_id_floor > 0 {
Arc::new(old.with_next_manifest_id_floor(next_id_floor))
} else {
old
};
let base = if list_metadata.is_empty() {
old
} else {
Arc::new(list_metadata.apply(&old))
};
let attempted_id = base.get_next_manifest_id();
let pending_writes = &mut pending_storage_writes;
let pending_replaces = &mut pending_storage_replaces;
match try_commit_attempt(
Arc::clone(&storage_async),
Arc::clone(&opts),
base,
&new_entries,
entries_to_remove,
NewEntryBirthVersions::StampCommit,
pending_writes,
pending_replaces,
)
.await
{
Ok(new_manifest) => return Ok(new_manifest),
Err(SupertableCommitError::WriteContentionExhausted)
if attempt + 1 < max_retries =>
{
next_id_floor = next_id_floor.max(
refresh_and_orphaned_id_floor(inner, &storage_async, attempted_id).await?,
);
last_err = Some(SupertableCommitError::WriteContentionExhausted);
sleep(backoff_delay(attempt)).await;
}
Err(e) => {
inner.note_commit_error(&e);
return Err(e);
}
}
}
Err(last_err.unwrap_or(SupertableCommitError::WriteContentionExhausted))
};
drive.await
}
pub(in crate::supertable) fn persist_commit(
inner: &SupertableInner,
storage: Arc<dyn StorageProvider>,
new_entries: Vec<Arc<SuperfileEntry>>,
entries_to_remove: &[Arc<SuperfileEntry>],
pending_storage_writes: Vec<(SuperfileUri, Bytes)>,
pending_storage_replaces: Vec<(SuperfileUri, Bytes)>,
list_metadata: CommitListMetadata,
) -> Result<(), SupertableCommitError> {
let drive = persist_commit_async(
inner,
storage,
new_entries,
entries_to_remove,
pending_storage_writes,
pending_storage_replaces,
list_metadata,
);
let new_manifest = bridge_on_runtime(drive, &inner.query_runtime())?;
inner.manifest.store(Arc::new(new_manifest));
inner.reconcile_tombstone_seqs();
Ok(())
}
async fn put_superfile_replace(
storage: &Arc<dyn StorageProvider>,
path: &str,
bytes: Bytes,
) -> Result<(), StorageError> {
match storage.head(path).await {
Ok(meta) => storage
.put_if_match(path, bytes, meta.etag.as_deref())
.await
.map(|_| ()),
Err(StorageError::NotFound { .. }) => storage.put_atomic(path, bytes).await.map(|_| ()),
Err(e) => Err(e),
}
}
fn commit_write_concurrency() -> usize {
std::thread::available_parallelism()
.map(|n| n.get() / 2)
.unwrap_or(1)
.max(1)
}
const DRAIN_READ_CONCURRENCY_CAP: usize = 64;
fn drain_read_concurrency() -> usize {
if let ThreadCount::Fixed(n) = config::global().vector.drain_read_concurrency
&& n > 0
{
return n;
}
std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
.clamp(
crate::config::DEFAULT_PREFETCH_CONCURRENCY,
DRAIN_READ_CONCURRENCY_CAP,
)
}
#[cfg_attr(
feature = "detailed-tracing",
tracing::instrument(skip_all, fields(superfiles = pending_storage_writes.len()))
)]
pub async fn write_superfile_list(
storage: &Arc<dyn StorageProvider>,
opts: &Arc<SupertableOptions>,
pending_storage_writes: &mut Vec<(SuperfileUri, Bytes)>,
pending_storage_replaces: &mut Vec<(SuperfileUri, Bytes)>,
) -> Result<(), SupertableCommitError> {
write_superfile_list_with_threshold(
storage,
opts,
opts.put_multipart_threshold_bytes,
pending_storage_writes,
pending_storage_replaces,
)
.await
}
async fn put_new_superfile_bytes(
storage: &Arc<dyn StorageProvider>,
multipart_threshold: u64,
uri: SuperfileUri,
bytes: Bytes,
) -> Result<(), SupertableCommitError> {
let path = superfile_storage_path(&uri);
let result = if (bytes.len() as u64) >= multipart_threshold {
put_superfile_multipart(storage.as_ref(), &path, bytes).await
} else {
storage.put_atomic(&path, bytes).await.map(|_| ())
};
match result {
Ok(()) | Err(StorageError::PreconditionFailed { .. }) => Ok(()),
Err(error) => Err(SupertableCommitError::from(error)),
}
}
async fn write_superfile_list_with_threshold(
storage: &Arc<dyn StorageProvider>,
_opts: &Arc<SupertableOptions>,
put_multipart_threshold_bytes: u64,
pending_storage_writes: &mut Vec<(SuperfileUri, Bytes)>,
pending_storage_replaces: &mut Vec<(SuperfileUri, Bytes)>,
) -> Result<(), SupertableCommitError> {
let write_concurrency = commit_write_concurrency();
let replace_futs = pending_storage_replaces
.iter()
.enumerate()
.map(|(i, (uri, bytes))| {
let storage = Arc::clone(storage);
let uri = *uri;
let bytes = bytes.clone();
async move {
let path = superfile_storage_path(&uri);
put_superfile_replace(&storage, &path, bytes)
.await
.map(|()| i)
.map_err(SupertableCommitError::from)
}
});
let mut err = None;
let mut successful_replace_idx = Vec::with_capacity(pending_storage_replaces.len());
for r in stream::iter(replace_futs)
.buffer_unordered(write_concurrency)
.collect::<Vec<_>>()
.await
{
match r {
Ok(i) => successful_replace_idx.push(i),
Err(e) => err = Some(e),
}
}
successful_replace_idx.sort_unstable_by(|a, b| b.cmp(a));
for idx in successful_replace_idx {
pending_storage_replaces.remove(idx);
}
if let Some(e) = err {
return Err(e);
}
let multipart_threshold = put_multipart_threshold_bytes;
let put_futs = pending_storage_writes
.iter()
.enumerate()
.map(|(i, (uri, bytes))| {
let storage = Arc::clone(storage);
let uri = *uri;
let bytes = bytes.clone();
async move {
put_new_superfile_bytes(&storage, multipart_threshold, uri, bytes)
.await
.map(|()| i)
}
});
let mut err = None;
let mut successful_writes_idx = Vec::with_capacity(pending_storage_writes.len());
for r in stream::iter(put_futs)
.buffer_unordered(write_concurrency)
.collect::<Vec<_>>()
.await
{
match r {
Ok(i) => successful_writes_idx.push(i),
Err(e) => err = Some(e),
}
}
successful_writes_idx.sort_unstable_by(|a, b| b.cmp(a));
for idx in successful_writes_idx {
pending_storage_writes.remove(idx);
}
if let Some(e) = err {
return Err(e);
}
Ok(())
}
#[derive(Clone, Copy)]
pub(crate) enum NewEntryBirthVersions {
StampCommit,
Preserve,
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn try_commit_attempt(
storage: Arc<dyn StorageProvider>,
opts: Arc<SupertableOptions>,
current_manifest: Arc<ManifestSnapshot>,
new_entries: &[Arc<SuperfileEntry>],
entries_to_remove: &[Arc<SuperfileEntry>],
birth_versions: NewEntryBirthVersions,
pending_storage_writes: &mut Vec<(SuperfileUri, Bytes)>,
pending_storage_replaces: &mut Vec<(SuperfileUri, Bytes)>,
) -> Result<ManifestSnapshot, SupertableCommitError> {
write_superfile_list(
&storage,
&opts,
pending_storage_writes,
pending_storage_replaces,
)
.await?;
let (mut new_manifest, parts_to_write) = match birth_versions {
NewEntryBirthVersions::StampCommit => {
current_manifest
.update(new_entries, entries_to_remove)
.await?
}
NewEntryBirthVersions::Preserve => {
current_manifest
.update_preserving_birth_versions(new_entries, entries_to_remove)
.await?
}
};
if matches!(
new_manifest.partition_strategy(),
Some(PartitionStrategy::VectorCell { .. })
) {
let entries = new_manifest.get_all_superfiles();
if !entries.is_empty() {
let previous_section =
previous_centroid_section(&opts, storage.as_ref(), current_manifest.as_ref()).await;
let published = slow_vector_state::write_state(
storage.as_ref(),
entries,
previous_section.as_deref(),
)
.await
.map_err(|e| {
SupertableCommitError::ManifestError(ManifestError::ManifestLoadError(
ManifestLoadError::SlowStateHydration(e.to_string()),
))
})?;
new_manifest = new_manifest.with_slow_vector_state_ref(
published.uri,
published.content_hash,
published.centroids,
);
}
}
let prev_etag = get_current_manifest_etag(&storage, current_manifest).await?;
let encoded_refs: Vec<&[u8]> = parts_to_write
.iter()
.flat_map(|ep| [Some(ep.encoded.as_slice()), ep.routing_encoded.as_deref()])
.flatten()
.collect();
new_manifest
.write(storage.as_ref(), prev_etag.as_deref(), &encoded_refs)
.await?;
let _ = PhantomData::<(PartId, part_mod::ContentHash)>;
Ok(new_manifest)
}
pub(in crate::supertable) async fn refresh_inner_state_async(
inner: &SupertableInner,
storage: &Arc<dyn StorageProvider>,
) -> Result<(), SupertableCommitError> {
let current = inner.manifest.load_full();
let manifest = match ManifestSnapshot::load(Some(current), storage.clone(), None).await {
Ok(manifest) => manifest,
Err(ManifestLoadError::PointerNotFound) => return Ok(()),
Err(ManifestLoadError::AlreadyLoaded) => return Ok(()),
Err(err) => {
return Err(SupertableCommitError::ManifestError(
ManifestError::ManifestLoadError(err),
));
}
};
inner.manifest.store(manifest);
inner.reconcile_tombstone_seqs();
Ok(())
}
pub(in crate::supertable) async fn refresh_and_orphaned_id_floor(
inner: &SupertableInner,
storage: &Arc<dyn StorageProvider>,
attempted_id: u64,
) -> Result<u64, SupertableCommitError> {
const MAX_ORPHAN_RUN_PROBES: u64 = 32;
refresh_inner_state_async(inner, storage).await?;
let refreshed_id = inner.manifest.load_full().get_manifest_id();
if refreshed_id >= attempted_id {
return Ok(0);
}
let probe_limit = attempted_id.saturating_add(MAX_ORPHAN_RUN_PROBES);
let mut next_free_id = attempted_id;
while next_free_id < probe_limit {
match storage.head(&manifest_uri(next_free_id)).await {
Ok(_) => next_free_id += 1,
Err(StorageError::NotFound { .. }) => break,
Err(e) => {
warn!(
error = %e,
manifest_id = next_free_id,
"orphaned-list probe failed; retrying at the last established id"
);
break;
}
}
}
Ok(if next_free_id > attempted_id {
next_free_id
} else {
0
})
}
pub(in crate::supertable) async fn stamp_tombstone_seqs(
inner: &SupertableInner,
touched: &[Uuid],
) -> Result<(), SupertableCommitError> {
let Some(storage) = inner.options.storage.clone() else {
return Ok(());
};
let max_retries = inner.options.max_commit_retries.max(1);
let mut next_id_floor: u64 = 0;
for attempt in 0..max_retries {
let old = inner.manifest.load_full();
let old = if next_id_floor > 0 {
Arc::new(old.with_next_manifest_id_floor(next_id_floor))
} else {
old
};
let Some(new_manifest) = old.with_tombstone_seqs_bumped(touched) else {
return Ok(());
};
let attempted_id = new_manifest.get_manifest_id();
let prev_etag = match get_current_manifest_etag(&storage, Arc::clone(&old)).await {
Ok(etag) => etag,
Err(SupertableCommitError::WriteContentionExhausted) if attempt + 1 < max_retries => {
refresh_inner_state_async(inner, &storage).await?;
sleep(backoff_delay(attempt)).await;
continue;
}
Err(e) => {
inner.note_commit_error(&e);
return Err(e);
}
};
match new_manifest
.write(storage.as_ref(), prev_etag.as_deref(), &[])
.await
{
Ok(()) => {
inner.manifest.store(Arc::new(new_manifest));
inner.reconcile_tombstone_seqs();
return Ok(());
}
Err(SupertableCommitError::WriteContentionExhausted) if attempt + 1 < max_retries => {
next_id_floor = next_id_floor
.max(refresh_and_orphaned_id_floor(inner, &storage, attempted_id).await?);
sleep(backoff_delay(attempt)).await;
}
Err(e) => return Err(e),
}
}
Err(SupertableCommitError::WriteContentionExhausted)
}
fn encode_record_batch_ipc(batch: &RecordBatch) -> Result<Bytes, String> {
let mut out: Vec<u8> = Vec::new();
{
let mut writer = StreamWriter::try_new(&mut out, &batch.schema())
.map_err(|e| format!("ipc writer init: {e}"))?;
writer.write(batch).map_err(|e| format!("ipc write: {e}"))?;
writer.finish().map_err(|e| format!("ipc finish: {e}"))?;
}
Ok(Bytes::from(out))
}
fn superfile_storage_path(uri: &SuperfileUri) -> String {
uri.storage_path()
}
pub(in crate::supertable) async fn put_bytes_multipart_or_atomic(
storage: &dyn StorageProvider,
path: &str,
bytes: Bytes,
multipart_threshold: u64,
) -> Result<(), StorageError> {
if (bytes.len() as u64) >= multipart_threshold {
put_superfile_multipart(storage, path, bytes).await
} else {
storage.put_atomic(path, bytes).await.map(|_| ())
}
}
async fn put_superfile_multipart(
storage: &dyn StorageProvider,
path: &str,
bytes: Bytes,
) -> Result<(), StorageError> {
match storage.head(path).await {
Ok(_) => return Err(StorageError::PreconditionFailed { uri: path.into() }),
Err(StorageError::NotFound { .. }) => {}
Err(e) => return Err(e),
}
let mut upload = storage.put_multipart(path).await?;
let total = bytes.len();
let part_concurrency = commit_write_concurrency().max(1);
let mut parts: Vec<UploadPart> = Vec::with_capacity(part_concurrency);
let mut offset = 0;
while offset < total {
let end = cmp::min(offset + SUPERFILE_MULTIPART_PART_BYTES, total);
let chunk = bytes.slice(offset..end);
parts.push(upload.put_part(PutPayload::from_bytes(chunk)));
offset = end;
if parts.len() == part_concurrency {
flush_superfile_multipart_parts(&mut upload, path, &mut parts).await?;
}
}
flush_superfile_multipart_parts(&mut upload, path, &mut parts).await?;
if let Err(e) = upload.complete().await {
let _ = upload.abort().await;
return Err(StorageError::Permanent {
uri: path.into(),
source: Box::new(e),
});
}
Ok(())
}
async fn flush_superfile_multipart_parts(
upload: &mut Box<dyn MultipartUpload>,
path: &str,
parts: &mut Vec<UploadPart>,
) -> Result<(), StorageError> {
if parts.is_empty() {
return Ok(());
}
if let Err(error) = try_join_all(mem::take(parts)).await {
let _ = upload.abort().await;
return Err(StorageError::Permanent {
uri: path.into(),
source: Box::new(error),
});
}
Ok(())
}
pub(in crate::supertable) async fn finalize_compaction_commit(
inner: Arc<SupertableInner>,
_storage: &Arc<dyn crate::storage::StorageProvider>,
_new_entries: &[Arc<SuperfileEntry>],
_entries_to_remove: &[Arc<SuperfileEntry>],
pending_cache_inserts: Vec<(SuperfileUri, Bytes)>,
) {
schedule_background_storage_reclaim(Arc::clone(&inner));
if !pending_cache_inserts.is_empty()
&& let Some(cache) = inner.options.disk_cache.as_ref().cloned()
{
warm_cache_after_commit(&inner, &cache, pending_cache_inserts);
}
if let (Some(cache), Some(budget)) = (
inner.options.disk_cache.as_ref(),
inner.options.memory_budget_bytes,
) {
cache.sweep_for_budget(budget);
}
}
async fn warm_cache_inserts(cache: &Arc<DiskCacheStore>, inserts: Vec<(SuperfileUri, Bytes)>) {
for (uri, bytes) in inserts {
if let Err(e) = cache.insert_warm(&uri, bytes).await {
tracing::warn!(
"supertable: warm cache pre-population failed for {}: {} \
(superfile is durable in storage; first query will cold-fetch)",
uri.0,
e
);
}
}
}
fn warm_cache_after_commit(
inner: &SupertableInner,
cache: &Arc<DiskCacheStore>,
pending: Vec<(SuperfileUri, Bytes)>,
) {
let cache = Arc::clone(cache);
bridge_on_runtime(warm_cache_inserts(&cache, pending), &inner.query_runtime());
}
pub(crate) fn read_vector_layout_from_bytes(bytes: &Bytes) -> VectorLayout {
match read_kv_metadata(bytes.as_ref()) {
Ok(kvs) => vector_layout_from_kv(&kvs),
Err(_) => VectorLayout::Ivf,
}
}
#[cfg(test)]
mod tests {
use std::{
sync::Arc,
time::{Duration, Instant},
};
use arrow_array::{
Array, Decimal128Array, FixedSizeListArray, Float32Array, LargeStringArray, RecordBatch,
};
use arrow_schema::{DataType, Field, Schema};
use datafusion::prelude::{col, lit};
use figment::{
Figment,
providers::{Format, Yaml},
};
use rayon::ThreadPoolBuilder;
use tempfile::TempDir;
use super::*;
use crate::{
config::Config,
superfile::{
builder::{FtsConfig, VectorConfig},
fts::reader::{Bm25Stats, BoolMode},
vector::{distance::Metric, rerank_codec::RerankCodec},
},
supertable::{
SupertableOptions,
handle::Supertable,
storage::LocalFsStorageProvider,
wal::{recovery::scan_and_recover, state_doc::SupertableHandleId},
},
test_helpers::{
build_title_batch, default_supertable_options, default_tokenizer as tok,
fault_storage::{FaultKind, FaultOp, FaultStorage},
},
};
const COMMIT_AS_DRAIN_TEST_DIM: usize = 16;
const COMMIT_AS_DRAIN_TEST_ROWS: usize = 8;
const COMMIT_AS_DRAIN_TEST_ROT_SEED: u64 = 7;
const BOUNDARY_STUB_TARGET_FACTOR: f32 = 2.0;
const TEST_SPLIT_BYTES: usize = 64 * MIB;
#[test]
fn superfiles_per_commit_follows_bytes_capped_by_pool() {
const T: usize = TEST_SPLIT_BYTES;
assert_eq!(superfiles_per_commit(1_000_000, 10 << 20, 192, T), 1);
assert_eq!(superfiles_per_commit(1_000_000, 1 << 30, 192, T), 16);
assert_eq!(superfiles_per_commit(1_000_000, 1 << 30, 8, T), 8);
assert_eq!(superfiles_per_commit(3, 1 << 30, 192, T), 3);
assert_eq!(superfiles_per_commit(1, 1, 0, T), 1);
}
#[test]
fn superfiles_per_commit_split_boundaries() {
const T: usize = TEST_SPLIT_BYTES;
assert_eq!(superfiles_per_commit(1_000_000, T, 192, T), 1);
assert_eq!(superfiles_per_commit(1_000_000, T + 1, 192, T), 2);
assert_eq!(superfiles_per_commit(1_000_000, 4 * T, 192, T), 4);
assert_eq!(superfiles_per_commit(1_000_000, 4 * T + 1, 192, T), 5);
assert_eq!(superfiles_per_commit(10, 0, 192, T), 1);
for bytes in [T + 1, 2 * T - 1, 3 * T + T / 2, 10 * T + 1] {
let n = superfiles_per_commit(1_000_000, bytes, 192, T);
assert!(bytes.div_ceil(n) >= T / 2, "bytes={bytes} n={n}");
}
}
#[test]
fn superfiles_per_commit_zero_split_restores_thread_fanout() {
assert_eq!(superfiles_per_commit(1_000_000, 10 << 20, 192, 0), 192);
assert_eq!(superfiles_per_commit(1_000_000, 1 << 30, 8, 0), 8);
assert_eq!(superfiles_per_commit(3, 1 << 30, 192, 0), 3);
}
#[test]
fn superfile_buffer_split_config_knob_reaches_commit_fanout() {
let batch = build_simple_batch(0, 50_000); for (target_mb, expect) in [(1u64, 2usize), (0, 2)] {
let opts = options_id_title()
.with_writer_pool(writer_pool_with(2))
.with_superfile_buffer_split_mb(target_mb);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
w.append(&batch).expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(
r.n_superfiles(),
expect,
"target_mb={target_mb} should shard to the 2-thread pool cap"
);
}
let opts = options_id_title()
.with_writer_pool(writer_pool_with(2))
.with_superfile_buffer_split_mb(4096);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
w.append(&batch).expect("append");
w.commit().expect("commit");
assert_eq!(st.reader().expect("reader").n_superfiles(), 1);
}
#[test]
fn each_small_commit_produces_exactly_one_superfile() {
let opts = options_id_title().with_writer_pool(writer_pool_with(4));
let st = Supertable::create(opts).expect("create");
for round in 0..3u64 {
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(round * 10, 5))
.expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(
r.n_superfiles(),
(round + 1) as usize,
"one new superfile per small commit"
);
}
let r = st.reader().expect("reader");
assert_eq!(r.n_docs_total(), 15);
}
#[test]
fn single_piece_fts_commit_is_searchable() {
let opts = options_id_title().with_writer_pool(writer_pool_with(4));
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 100)).expect("append");
w.commit().expect("commit");
drop(w);
let r = st.reader().expect("reader");
assert_eq!(r.n_superfiles(), 1, "small FTS commit stays one piece");
let hits = st
.bm25_search(
"title",
"alpha",
10,
BoolMode::Or,
Bm25Stats::PerSuperfile,
None,
)
.expect("bm25 over one-piece commit");
let n: usize = hits.iter().map(|b| b.num_rows()).sum();
assert!(n > 0, "single-shard FTS index must return hits");
}
#[test]
fn supertable_writer_debug_renders() {
let directory = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let table = Supertable::create(
options_title_emb_serial(COMMIT_AS_DRAIN_TEST_DIM).with_storage(storage),
)
.expect("create");
let writer = table.writer().expect("writer");
let rendered = format!("{writer:?}");
assert!(
rendered.contains("SupertableWriter"),
"debug must render the writer, got {rendered}"
);
}
#[test]
fn split_buffer_by_vector_cell_routes_rows_to_nearest_cell() {
use std::collections::HashMap;
use arrow_array::{Float32Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
let dim = 4usize;
let centroids = vec![1.0f32, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0];
let cells = ClusterCentroids::from_fp32(2, dim as u32, ¢roids, vec![1u32; 2]);
let scalar = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("t", DataType::Utf8, false)])),
vec![Arc::new(StringArray::from(vec!["a", "b", "c", "d"]))],
)
.expect("scalar batch");
let vectors = Float32Array::from(vec![
0.9, 0.1, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 0.9, 0.1, 0.0, 0.0, 1.0, 0.0, 0.0, ]);
let batch = BufferedBatch {
scalar,
vectors: vec![Arc::new(vectors)],
};
let out = split_buffer_by_vector_cell(vec![batch], &cells, Metric::Cosine, 0)
.expect("split buffer by vector cell");
let mut rows_by_cell: HashMap<u32, usize> = HashMap::new();
for (cell, batches) in &out {
rows_by_cell.insert(*cell, batches.iter().map(|b| b.scalar.num_rows()).sum());
}
assert_eq!(
rows_by_cell.get(&0).copied(),
Some(2),
"two rows must route to the e_0 cell"
);
assert_eq!(
rows_by_cell.get(&1).copied(),
Some(2),
"two rows must route to the e_1 cell"
);
}
#[test]
fn drain_local_checkpoint_round_trips_and_rejects_other_epoch() {
let directory = TempDir::new().expect("tempdir");
let mut checkpoint = DrainLocalCheckpoint::new("epoch-a".into());
checkpoint.batches_done = 2;
checkpoint.spills.insert(
7,
DrainLocalSpill {
n_rows: 11,
n_quants: 3,
dim: 16,
rabitq_len: 2,
rerank_codec_id: RerankCodec::Sq8FixedResidual.codec_id(),
},
);
checkpoint.built_cells.insert(
2,
DrainLocalCell {
n_docs: 9,
subsection_len: 1_024,
rerank_codec_id: RerankCodec::Sq8FixedResidual.codec_id(),
},
);
checkpoint.added_per_cell.insert(2, 9);
checkpoint.added_per_cell.insert(7, 11);
save_drain_local_checkpoint(directory.path(), &checkpoint).expect("save");
let loaded = load_drain_local_checkpoint(directory.path(), "epoch-a")
.expect("load")
.expect("checkpoint");
assert_eq!(loaded, checkpoint);
assert!(
load_drain_local_checkpoint(directory.path(), "epoch-b").is_err(),
"an incompatible local epoch must fail loud"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drain_remote_checkpoint_lives_in_slow_cas_state() {
let directory = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let table =
Supertable::create(options_id_title_serial().with_storage(Arc::clone(&storage)))
.expect("create table");
let mut writer = table.writer().expect("writer");
writer
.append(&build_simple_batch(0, 2))
.expect("append visible entry");
writer.commit().expect("commit visible entry");
drop(writer);
let pending_entry = Arc::clone(&table.reader().expect("reader").manifest().superfiles[0]);
let sources = vec![DrainCheckpointSource {
superfile_id: "source-id".into(),
uri: "source-uri".into(),
birth_version: 4,
}];
let batch_layout = vec![vec![4]];
let options_hash = "options".to_string();
let checkpoint = DrainRemoteCheckpoint {
schema: DRAIN_CHECKPOINT_SCHEMA,
epoch_id: drain_epoch_id(
&options_hash,
&sources,
&batch_layout,
2,
DrainConsolidate::Kmeans,
),
options_hash,
sources,
batch_layout,
shard_count: 2,
completed_shards: Vec::new(),
};
let mut state = create_drain_remote_checkpoint(table.inner(), checkpoint.clone())
.await
.expect("create");
let loaded = load_drain_remote_checkpoint(table.inner())
.await
.expect("load")
.expect("checkpoint");
assert_eq!(loaded.checkpoint, checkpoint);
state.entries.push(Arc::clone(&pending_entry));
state.checkpoint.completed_shards.push(DrainRemoteShard {
shard_id: 1,
superfile_id: pending_entry.superfile_id.to_string(),
cell_counts: vec![(3, 10)],
});
save_drain_remote_checkpoint(table.inner(), &mut state)
.await
.expect("CAS update");
let updated = load_drain_remote_checkpoint(table.inner())
.await
.expect("reload")
.expect("checkpoint");
assert_eq!(updated.checkpoint.completed_shards.len(), 1);
assert_eq!(updated.entries.len(), 1);
assert_eq!(updated.entries[0].superfile_id, pending_entry.superfile_id);
refresh_slow_vector_state(table.inner())
.await
.expect("replace checkpoint with settled slow state");
assert!(
load_drain_remote_checkpoint(table.inner())
.await
.expect("load settled state")
.is_none(),
"settled slow-CAS state must not retain a drain checkpoint"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drain_resumes_from_last_local_batch_checkpoint() {
let directory = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let options = options_title_emb_serial(COMMIT_AS_DRAIN_TEST_DIM)
.with_storage(storage)
.with_drain_batch_superfiles(1);
let table = Supertable::create(options).expect("create");
for _ in 0..2 {
let mut writer = table.writer().expect("writer");
writer
.append(&build_axis_vector_batch(
COMMIT_AS_DRAIN_TEST_ROWS,
COMMIT_AS_DRAIN_TEST_DIM,
))
.expect("append");
writer.commit().expect("commit");
}
let (hidden, epoch_id) = current_drain_epoch(&table).await;
inject_drain_test_failure(epoch_id.clone(), DrainTestFailurePhase::AfterBatch, 1);
let first = drain_user_superfiles_to_hidden_cells(
Arc::clone(table.inner()),
Arc::clone(hidden.inner()),
)
.await;
assert!(first.is_err(), "first drain must stop at the failpoint");
let local = load_drain_local_checkpoint(&drain_scratch_dir(&epoch_id), &epoch_id)
.expect("load local checkpoint")
.expect("local checkpoint");
assert_eq!(local.batches_done, 1);
drain_user_superfiles_to_hidden_cells(
Arc::clone(table.inner()),
Arc::clone(hidden.inner()),
)
.await
.expect("resume drain");
assert!(
!drain_scratch_dir(&epoch_id).exists(),
"successful final CAS removes local checkpoint scratch"
);
assert!(
load_drain_remote_checkpoint(hidden.inner())
.await
.expect("load settled slow state")
.is_none(),
"settled slow-CAS state contains no pending drain"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drain_preserves_uploaded_shard_across_node_replacement() {
let directory = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let options = options_title_emb_serial(COMMIT_AS_DRAIN_TEST_DIM)
.with_storage(storage)
.with_writer_pool(writer_pool_with(2))
.with_drain_batch_superfiles(1);
let table = Supertable::create(options).expect("create");
let mut writer = table.writer().expect("writer");
writer
.append(&build_axis_vector_batch(
4 * COMMIT_AS_DRAIN_TEST_ROWS,
COMMIT_AS_DRAIN_TEST_DIM,
))
.expect("append");
writer.commit().expect("commit");
drop(writer);
let (hidden, epoch_id) = current_drain_epoch(&table).await;
inject_drain_test_failure(epoch_id.clone(), DrainTestFailurePhase::AfterShard, 1);
let first = drain_user_superfiles_to_hidden_cells(
Arc::clone(table.inner()),
Arc::clone(hidden.inner()),
)
.await;
assert!(first.is_err(), "first drain must stop after one shard");
let checkpoint = load_drain_remote_checkpoint(hidden.inner())
.await
.expect("load pending slow state")
.expect("pending drain");
assert_eq!(checkpoint.checkpoint.completed_shards.len(), 1);
let preserved_id = checkpoint.entries[0].superfile_id;
let preserved_path = checkpoint.entries[0].uri.storage_path();
hidden
.gc_async(Duration::ZERO)
.await
.expect("GC with active checkpoint");
hidden
.options()
.storage
.as_ref()
.expect("hidden storage")
.head(&preserved_path)
.await
.expect("checkpointed shard remains live through GC");
fs::remove_dir_all(drain_scratch_dir(&epoch_id)).expect("drop local scratch");
drain_user_superfiles_to_hidden_cells(
Arc::clone(table.inner()),
Arc::clone(hidden.inner()),
)
.await
.expect("replacement-node resume");
assert!(
hidden
.reader()
.expect("reader")
.manifest()
.superfiles
.iter()
.any(|entry| entry.superfile_id == preserved_id),
"final manifest must reuse the shard recorded in slow-CAS"
);
assert!(
load_drain_remote_checkpoint(hidden.inner())
.await
.expect("load settled state")
.is_none()
);
}
fn schema_id_title() -> Arc<Schema> {
Arc::new(Schema::new(vec![Field::new(
"title",
DataType::LargeUtf8,
false,
)]))
}
fn fixed_list_f32(dim: usize) -> DataType {
DataType::FixedSizeList(
Arc::new(Field::new("item", DataType::Float32, true)),
dim as i32,
)
}
fn schema_id_title_emb(dim: usize) -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("title", DataType::LargeUtf8, false),
Field::new("emb", fixed_list_f32(dim), false),
]))
}
fn options_id_title() -> SupertableOptions {
SupertableOptions::new(
schema_id_title(),
vec![FtsConfig {
column: "title".into(),
positions: false,
}],
vec![],
Some(tok()),
)
.expect("valid options")
}
fn options_id_title_serial() -> SupertableOptions {
let pool = Arc::new(
ThreadPoolBuilder::new()
.num_threads(1)
.build()
.expect("build pool"),
);
options_id_title().with_writer_pool(pool)
}
fn writer_pool_with(n: usize) -> Arc<rayon::ThreadPool> {
Arc::new(
ThreadPoolBuilder::new()
.num_threads(n)
.build()
.expect("build pool"),
)
}
fn build_simple_batch(_start: u64, n: usize) -> RecordBatch {
let titles =
LargeStringArray::from((0..n).map(|i| format!("doc {i} alpha")).collect::<Vec<_>>());
RecordBatch::try_new(schema_id_title(), vec![Arc::new(titles)]).expect("build batch")
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn splice_drain_concatenates_same_cell_across_batches() {
use crate::superfile::reader::VectorSearchOptions;
let directory = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let options = options_title_emb_serial(COMMIT_AS_DRAIN_TEST_DIM)
.with_storage(storage)
.with_drain_consolidate(DrainConsolidate::Splice)
.with_drain_batch_superfiles(1);
let table = Supertable::create(options).expect("create");
for _ in 0..2 {
let mut writer = table.writer().expect("writer");
writer
.append(&build_axis_vector_batch(
COMMIT_AS_DRAIN_TEST_ROWS,
COMMIT_AS_DRAIN_TEST_DIM,
))
.expect("append");
writer.commit().expect("commit");
}
let (hidden, _epoch) = current_drain_epoch(&table).await;
drain_user_superfiles_to_hidden_cells(
Arc::clone(table.inner()),
Arc::clone(hidden.inner()),
)
.await
.expect("splice drain across batches");
assert!(
hidden.reader().expect("reader").n_superfiles() > 0,
"splice drain must populate the hidden cell index"
);
let mut q = vec![0.0f32; COMMIT_AS_DRAIN_TEST_DIM];
q[0] = 1.0;
let hits = table
.reader()
.expect("reader")
.vector_hits(
"emb",
&q,
COMMIT_AS_DRAIN_TEST_ROWS * 2,
VectorSearchOptions::new().with_nprobe(32),
None,
)
.expect("search");
assert!(
!hits.is_empty(),
"docs survive the cross-batch splice concatenate"
);
}
const OPEN_RANGES_FIXTURE_ROWS: usize = 5_000;
const OPEN_RANGES_FIXTURE_CEILING_BYTES: u64 = 16 * 1024;
#[test]
fn multi_cell_open_ranges_exclude_row_proportional_regions() {
let dim = COMMIT_AS_DRAIN_TEST_DIM;
let dir = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(dir.path()).expect("provider"));
let st = Supertable::create(options_title_emb_serial(dim).with_storage(storage))
.expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_axis_vector_batch(OPEN_RANGES_FIXTURE_ROWS, dim))
.expect("append");
w.commit().expect("commit");
drop(w);
let mut checked = 0usize;
for entry in walkdir(dir.path()) {
let bytes = Bytes::from(fs::read(&entry).expect("read superfile"));
let Some(offsets) = build_subsection_offsets(&bytes) else {
continue;
};
if offsets.vec_open_ranges.is_empty() {
continue;
}
let staged: u64 = offsets.vec_open_ranges.iter().map(|&(_, len)| len).sum();
assert!(
staged <= OPEN_RANGES_FIXTURE_CEILING_BYTES,
"{entry:?}: staged vector open bytes {staged} scale with rows \
(ceiling {OPEN_RANGES_FIXTURE_CEILING_BYTES})"
);
checked += 1;
}
assert!(checked > 0, "fixture must produce vector superfiles");
}
fn walkdir(root: &std::path::Path) -> Vec<std::path::PathBuf> {
let mut out = Vec::new();
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
let Ok(entries) = fs::read_dir(&dir) else {
continue;
};
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
stack.push(path);
} else if path.to_string_lossy().ends_with(".sf.parquet") {
out.push(path);
}
}
}
out
}
fn options_title_emb_serial(dim: usize) -> SupertableOptions {
SupertableOptions::new(
schema_id_title_emb(dim),
vec![],
vec![VectorConfig {
column: "emb".into(),
dim,
rot_seed: 7,
metric: Metric::L2Sq,
rerank_codec: RerankCodec::Fp32,
provided_centroids: None,
}],
Some(tok()),
)
.expect("valid options")
.with_writer_pool(writer_pool_with(1))
}
fn build_axis_vector_batch(n: usize, dim: usize) -> RecordBatch {
let titles =
LargeStringArray::from((0..n).map(|i| format!("doc {i} beta")).collect::<Vec<_>>());
let mut flat = Vec::with_capacity(n * dim);
for row in 0..n {
for d in 0..dim {
flat.push(if d == row % dim { 1.0 } else { 0.0 });
}
}
let values = Arc::new(Float32Array::from(flat));
let list = FixedSizeListArray::try_new(
Arc::new(Field::new("item", DataType::Float32, true)),
dim as i32,
values,
None,
)
.expect("fixed-size list");
RecordBatch::try_new(
schema_id_title_emb(dim),
vec![Arc::new(titles), Arc::new(list)],
)
.expect("vector batch")
}
async fn current_drain_epoch(table: &Supertable) -> (Arc<Supertable>, String) {
let hidden = table
.inner()
.vector_index_table
.as_ref()
.expect("hidden table")
.clone();
let user_manifest = table.inner().manifest.load_full();
let drained = hidden.inner().manifest.load_full().get_drained_ranges();
let mut sources: Vec<Arc<SuperfileEntry>> = user_manifest
.get_all_superfiles_loaded()
.await
.expect("load user sources")
.into_iter()
.filter(|entry| !drained.contains(entry.birth_version))
.collect();
sources.sort_unstable_by(|left, right| {
left.birth_version
.cmp(&right.birth_version)
.then_with(|| left.superfile_id.cmp(&right.superfile_id))
});
let batch_cfg = drain_batch_superfiles(&table.inner().options);
let budget = if batch_cfg < 0 {
usize::MAX
} else {
(batch_cfg as usize).max(1)
};
let source_refs: Vec<DrainCheckpointSource> = sources
.iter()
.map(|entry| drain_checkpoint_source(entry))
.collect();
let batches = make_drain_batches(sources, budget);
let batch_layout = drain_batch_layout(&batches);
let strategy = user_manifest.get_partition_strategy();
let options_hash =
options_hash::compute_options_hash(table.inner().options.as_ref(), &strategy).to_hex();
let shard_count = packed_cell_shard_count(&hidden.inner().options);
(
hidden,
drain_epoch_id(
&options_hash,
&source_refs,
&batch_layout,
shard_count,
table.inner().options.drain_consolidate,
),
)
}
fn committed_reader(st: &Supertable) -> (Arc<SuperfileEntry>, Arc<SuperfileReader>) {
let entry = Arc::clone(&st.reader().expect("reader").manifest().superfiles[0]);
let reader = st
.options()
.store
.reader(&entry.uri)
.expect("committed superfile reader");
(entry, reader)
}
#[test]
fn reserve_build_scratch_weights_each_input_class() {
let budget = ConnectionMemoryBudget::measured();
let (scalar, vector, fts) = (1000usize, 2000usize, 400usize);
let guard = reserve_build_scratch(&budget, scalar, vector, fts)
.expect("measured budget never denies");
let expected =
(BUILD_SCALAR_NUM * scalar + BUILD_VECTOR_NUM * vector + BUILD_FTS_NUM * fts)
/ BUILD_SCRATCH_DENOM;
assert_eq!(budget.used(), expected);
drop(guard);
assert_eq!(budget.used(), 0, "reservation released on drop");
let reserved = |s, v, f| {
let _guard = reserve_build_scratch(&budget, s, v, f).expect("measured");
budget.used()
};
let scalar_only = reserved(1000, 0, 0);
let vector_only = reserved(0, 1000, 0);
let scalar_plus_fts = reserved(1000, 0, 1000);
assert!(
vector_only > scalar_only,
"a vector byte must reserve more than a scalar byte ({vector_only} vs {scalar_only})"
);
assert!(
scalar_plus_fts > scalar_only,
"the FTS term is additive on top of scalar ({scalar_plus_fts} vs {scalar_only})"
);
}
#[test]
fn append_over_budget_is_refused() {
let mut opts = options_id_title_serial();
opts.connection_memory_budget = ConnectionMemoryBudget::with_limit(1);
let st = Supertable::create(opts).expect("create");
let err = st
.append(&build_simple_batch(0, 8))
.expect_err("build over a 0-byte gate is refused");
let InfinoError::OverBudget(msg) = err else {
panic!("expected InfinoError::OverBudget, got {err:?}");
};
assert!(
msg.contains("ingest"),
"over-budget message should identify ingest: {msg}"
);
assert_eq!(st.reader().expect("reader").n_docs_total(), 0);
assert!(st.options().connection_memory_budget.denials() >= 1);
assert_eq!(st.options().connection_memory_budget.peak(), 0);
}
#[test]
fn append_under_measured_budget_runs_and_tracks_peak() {
let mut opts = options_id_title_serial();
opts.connection_memory_budget = ConnectionMemoryBudget::measured();
let st = Supertable::create(opts).expect("create");
st.append(&build_simple_batch(0, 8))
.expect("measured budget never refuses");
assert_eq!(st.reader().expect("reader").n_docs_total(), 8);
let budget = &st.options().connection_memory_budget;
assert_eq!(budget.denials(), 0);
assert!(
budget.peak() > 0,
"the build must reserve against the budget"
);
}
#[test]
fn append_under_ample_bounded_budget_runs() {
const AMPLE_BUDGET_BYTES: u64 = 1 << 30; let mut opts = options_id_title_serial();
opts.connection_memory_budget = ConnectionMemoryBudget::with_limit(AMPLE_BUDGET_BYTES);
let st = Supertable::create(opts).expect("create");
st.append(&build_simple_batch(0, 8))
.expect("under-budget append runs under a bounded budget");
assert_eq!(st.reader().expect("reader").n_docs_total(), 8);
let budget = &st.options().connection_memory_budget;
assert_eq!(budget.denials(), 0);
assert!(budget.limit().is_some(), "bounded, not measured");
}
#[test]
fn over_budget_commit_preserves_the_buffer() {
let mut opts = options_id_title_serial();
opts.connection_memory_budget = ConnectionMemoryBudget::with_limit(1);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 8)).expect("append buffers");
assert_eq!(w.buffered_batches(), 1);
let err = w
.commit()
.expect_err("commit over a 0-byte gate is refused");
assert!(
matches!(err, CommitError::AppendFlush(BuildError::OverBudget(_))),
"got {err:?}"
);
assert_eq!(w.buffered_batches(), 1);
}
#[test]
fn auto_flush_over_budget_is_refused_from_append() {
const AUTO_FLUSH_TRIP_ROWS: usize = 40_000;
let mut opts = options_id_title_serial().with_commit_threshold_size_mb(1);
opts.connection_memory_budget = ConnectionMemoryBudget::with_limit(1);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
let err = w
.append(&build_simple_batch(0, AUTO_FLUSH_TRIP_ROWS))
.expect_err("auto-flush over a 0-byte gate is refused");
assert!(matches!(err, BuildError::OverBudget(_)), "got {err:?}");
}
#[test]
fn vector_ingest_over_budget_is_refused() {
let dim = 16;
let mut opts = options_with_vector(dim);
opts.connection_memory_budget = ConnectionMemoryBudget::with_limit(1);
let st = Supertable::create(opts).expect("create");
let err = st
.append(&build_vector_batch(0, 8, dim))
.expect_err("vector build over a 0-byte gate is refused");
assert!(matches!(err, InfinoError::OverBudget(_)), "got {err:?}");
assert_eq!(st.reader().expect("reader").n_docs_total(), 0);
}
#[test]
fn vector_ingest_reserves_and_runs_under_measured() {
let dim = 16;
let mut opts = options_with_vector(dim);
opts.connection_memory_budget = ConnectionMemoryBudget::measured();
let st = Supertable::create(opts).expect("create");
st.append(&build_vector_batch(0, 8, dim))
.expect("measured vector ingest runs");
assert_eq!(st.reader().expect("reader").n_docs_total(), 8);
let budget = &st.options().connection_memory_budget;
assert_eq!(budget.denials(), 0);
assert!(
budget.peak() > 0,
"the vector build must reserve against the budget"
);
}
#[test]
fn recalibration_skips_non_vector_cell_tables() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let before = st.inner().manifest.load_full().get_manifest_id();
let stamped = st
.block_on_query(recalibrate_probe_laws(st.inner()))
.expect("recalibrate on a user table is a clean no-op");
assert!(!stamped, "no VectorCell strategy, nothing to restamp");
assert_eq!(
st.inner().manifest.load_full().get_manifest_id(),
before,
"the no-op must not commit"
);
}
#[test]
fn writer_slot_is_exclusive() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let _w = st.writer().expect("first writer");
let err = st.writer().expect_err("second writer should fail");
assert!(matches!(err, BuildError::SupertableInUse));
}
#[test]
fn writer_slot_releases_on_drop() {
let st = Supertable::create(options_id_title_serial()).expect("create");
{
let _w = st.writer().expect("first writer");
}
let _w2 = st.writer().expect("second writer after drop");
}
#[test]
fn consumer_memory_mode_handle_refuses_writer() {
let opts = options_id_title_serial().with_summary_centroids_from_superfiles(true);
let st = Supertable::create(opts).expect("create");
let err = st
.writer()
.expect_err("consumer-mode handle must not write");
assert!(
err.to_string().contains("consumer memory mode"),
"unexpected refusal: {err}"
);
}
#[test]
fn append_then_commit_publishes_one_superfile() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 4)).expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(r.manifest_id(), 1);
assert_eq!(r.n_superfiles(), 1);
assert_eq!(r.n_docs_total(), 4);
}
#[test]
fn commit_with_empty_buffer_is_noop() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
w.commit().expect("commit-empty");
assert_eq!(st.manifest_id(), 0, "no manifest swap on empty commit");
assert_eq!(st.reader().expect("reader").n_superfiles(), 0);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn superfile_is_queryable_via_store() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 4)).expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
let superfile = &r.manifest().superfiles[0];
let store = &st.options().store;
let sf_reader = store.reader(&superfile.uri).expect("reader");
let hits = sf_reader
.bm25_hits_async("title", "alpha", 10, BoolMode::Or)
.await
.expect("bm25");
assert_eq!(hits.len(), 4);
}
#[test]
fn superfile_entry_records_id_range_and_n_docs() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(100, 3)).expect("a");
w.append(&build_simple_batch(50, 2)).expect("b");
w.commit().expect("commit");
let r = st.reader().expect("reader");
let seg = &r.manifest().superfiles[0];
assert_eq!(seg.n_docs, 5);
assert!(seg.id_min > 0);
assert!(seg.id_max > seg.id_min, "id_max should exceed id_min");
}
#[test]
fn superfile_entry_carries_fts_summary() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 4)).expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
let seg = &r.manifest().superfiles[0];
let fts = seg
.fts_summary
.get("title")
.expect("title FTS summary present");
assert!(
fts.n_terms_distinct >= 3,
"expected ≥ 3 distinct terms, got {}",
fts.n_terms_distinct,
);
assert!(fts.may_contain(b"alpha"));
assert!(fts.may_contain(b"doc"));
let (min_term, max_term) = fts.term_range.as_ref().expect("non-empty FST has a range");
assert!(!min_term.is_empty());
assert!(!max_term.is_empty());
assert!(min_term <= max_term, "min_term <= max_term invariant");
}
fn build_vector_batch(_start: u64, n: usize, dim: usize) -> RecordBatch {
let titles = LargeStringArray::from((0..n).map(|i| format!("doc {i}")).collect::<Vec<_>>());
let mut flat = Vec::with_capacity(n * dim);
for i in 0..n {
for j in 0..dim {
flat.push(((i + j) as f32) / 100.0);
}
}
let item_field = Arc::new(Field::new("item", DataType::Float32, true));
let values = Float32Array::from(flat);
let fsl = FixedSizeListArray::try_new(item_field, dim as i32, Arc::new(values), None)
.expect("FSL");
RecordBatch::try_new(
schema_id_title_emb(dim),
vec![Arc::new(titles), Arc::new(fsl)],
)
.expect("batch")
}
fn options_with_vector(dim: usize) -> SupertableOptions {
let pool = Arc::new(
ThreadPoolBuilder::new()
.num_threads(1)
.build()
.expect("build pool"),
);
SupertableOptions::new(
schema_id_title_emb(dim),
vec![],
vec![VectorConfig {
column: "emb".into(),
dim,
rot_seed: 7,
metric: Metric::Cosine,
rerank_codec: RerankCodec::Fp32,
provided_centroids: None,
}],
None,
)
.expect("valid options")
.with_writer_pool(pool)
}
#[test]
fn superfile_entry_carries_vector_summary() {
let dim = 16;
let st = Supertable::create(options_with_vector(dim)).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_vector_batch(0, 8, dim)).expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
let seg = &r.manifest().superfiles[0];
let vs = seg
.vector_summary
.get("emb")
.expect("emb vector summary present");
assert_eq!(vs.centroid.len(), dim);
assert!(
vs.cells.iter().any(|cell| !cell.clusters.is_empty()),
"cluster centroids must be populated"
);
assert!(vs.cells.iter().all(|cell| {
cell.clusters.dim as usize == dim
&& cell.clusters.n_cent >= 1
&& cell.clusters.counts.len() == cell.clusters.n_cent as usize
&& cell.clusters.centroids.len() == cell.clusters.n_cent as usize * dim
}));
let total: u64 = vs
.cells
.iter()
.flat_map(|cell| cell.clusters.counts.iter())
.map(|&count| count as u64)
.sum();
assert!(total >= seg.n_docs, "counts {total} < rows {}", seg.n_docs);
let budget_cap = (seg.n_docs as f64
* f64::from(config::global().vector.drain_replica_target_factor.max(1.0)))
.ceil() as u64;
assert!(
total <= budget_cap,
"counts {total} exceed replica budget cap {budget_cap}"
);
}
#[test]
fn grid_commit_writes_multicell_parquet_in_vector_order() {
let dir = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(dir.path()).expect("provider"));
let st = Supertable::create(
options_title_emb_serial(COMMIT_AS_DRAIN_TEST_DIM).with_storage(storage),
)
.expect("create");
assert!(
!st.reader()
.expect("reader")
.options()
.vector_columns
.is_empty(),
"fixture must declare vector columns so commit takes the assign-pack path"
);
let mut w = st.writer().expect("writer");
w.append(&build_axis_vector_batch(
COMMIT_AS_DRAIN_TEST_ROWS,
COMMIT_AS_DRAIN_TEST_DIM,
))
.expect("append");
w.commit().expect("commit");
let (entry, reader) = committed_reader(&st);
assert_eq!(entry.vector_layout, VectorLayout::MultiCellIvf);
assert_eq!(entry.n_docs, COMMIT_AS_DRAIN_TEST_ROWS as u64);
let vec_reader = reader.vec().expect("vector reader");
assert!(vec_reader.is_multi_cell());
let vector_locals: Vec<u32> = (0..vec_reader.n_docs() as u32).collect();
let vector_ids = vec_reader
.inline_stable_ids_for_locals(&vector_locals)
.expect("inline stable ids");
let parquet_locals: Vec<u32> = (0..entry.n_docs as u32).collect();
let parquet_batch = reader
.take_by_local_doc_ids(&parquet_locals, &["_id"])
.expect("read parquet ids");
let parquet_ids = parquet_batch
.column(0)
.as_any()
.downcast_ref::<Decimal128Array>()
.expect("decimal ids")
.values()
.to_vec();
let mut seen = HashSet::new();
let first_occurrence: Vec<i128> = vector_ids
.iter()
.copied()
.filter(|id| seen.insert(*id))
.collect();
assert_eq!(parquet_ids, first_occurrence);
let parquet_set: HashSet<i128> = parquet_ids.iter().copied().collect();
assert!(
vector_ids.iter().all(|id| parquet_set.contains(id)),
"every stub must duplicate a parquet row"
);
}
#[test]
fn assign_pack_boundary_replicas_are_vector_only_stubs() {
let dim = COMMIT_AS_DRAIN_TEST_DIM;
let mut centroids = vec![0.0f32; dim * 2];
centroids[dim] = 1.0;
let clusters = ClusterCentroids::from_fp32(2, dim as u32, ¢roids, vec![0, 0]);
let vectors = [
vec![0.49; dim],
vec![0.51; dim],
vec![0.48; dim],
vec![0.52; dim],
];
let stable_ids = [10_i128, 11, 12, 13];
let rows: Vec<PackRow<'_>> = vectors
.iter()
.zip(stable_ids)
.map(|(vector, stable_id)| PackRow::Fp32 { stable_id, vector })
.collect();
let assigned = assign_cells(
&rows,
&clusters,
Metric::L2Sq,
COMMIT_AS_DRAIN_TEST_ROT_SEED,
BOUNDARY_STUB_TARGET_FACTOR,
)
.expect("assign");
let postings: usize = assigned.iter().map(|group| group.members.len()).sum();
let primaries: usize = assigned
.iter()
.flat_map(|group| group.members.iter())
.filter(|(_, is_primary, _)| *is_primary)
.count();
assert_eq!(primaries, rows.len());
assert!(
postings > primaries,
"boundary replicas add vector postings, not primary rows"
);
let cfg = VectorConfig {
column: "emb".into(),
dim,
rot_seed: 7,
metric: Metric::L2Sq,
rerank_codec: RerankCodec::Sq8Residual,
provided_centroids: None,
};
for group in assigned {
let n_members = group.members.len();
let packed = drain_pack_assigned_cell(group, &cfg).expect("drain pack");
assert_eq!(packed.stable_ids.len(), n_members);
assert_eq!(packed.subsection.n_docs as usize, n_members);
}
}
#[test]
fn drain_fine_centroids_follow_two_mib_run_target() {
const DIM: usize = 1024;
const COMMIT_CELL_ROWS: usize = 98;
const DRAINED_CELL_ROWS: usize = 1_562;
let cfg = VectorConfig {
column: "emb".into(),
dim: DIM,
rot_seed: 7,
metric: Metric::L2Sq,
rerank_codec: RerankCodec::Sq8Residual,
provided_centroids: None,
};
assert_eq!(
drain_cell_vector_config(&cfg, COMMIT_CELL_ROWS).1,
1,
"a small commit delta fits one ~2 MiB fine run"
);
assert_eq!(
drain_cell_vector_config(&cfg, DRAINED_CELL_ROWS).1,
2,
"a fully drained cell needs two ~2 MiB fine runs"
);
}
#[test]
fn commit_superfile_count_follows_bytes_not_pool_size() {
for n_threads in [1usize, 2, 4] {
let opts = options_id_title().with_writer_pool(writer_pool_with(n_threads));
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
for i in 0..n_threads * 2 {
w.append(&build_simple_batch(i as u64 * 10, 3))
.expect("append");
}
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(
r.n_superfiles(),
1,
"small buffer must stay one superfile on a {n_threads}-thread pool",
);
assert_eq!(r.n_docs_total(), (n_threads * 2 * 3) as u64);
}
}
#[test]
fn commit_splits_wide_buffer_up_to_pool_width() {
const ROWS: usize = 100_000;
let opts = options_id_title()
.with_writer_pool(writer_pool_with(2))
.with_commit_threshold_size_mb(4096);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
let titles = LargeStringArray::from(
(0..ROWS)
.map(|i| format!("doc {i} {}", "x".repeat(1024)))
.collect::<Vec<_>>(),
);
let batch = RecordBatch::try_new(schema_id_title(), vec![Arc::new(titles)]).expect("batch");
w.append(&batch).expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(
r.n_superfiles(),
2,
"~100 MiB buffer splits, pinned to 2 by the pool cap"
);
assert_eq!(r.n_docs_total(), ROWS as u64);
}
#[test]
fn apply_config_with_fixed_writer_threads_sizes_the_pool() {
let yaml = r#"
commit_threshold_size_mb: 1024
supertable:
reader_threads: 1
writer_threads: 4
"#;
let cfg =
Config::from_figment(Figment::new().merge(Yaml::string(yaml))).expect("parse config");
let opts = options_id_title().apply_config(&cfg).expect("apply_config");
assert_eq!(
opts.writer_pool.current_num_threads(),
4,
"writer_threads=4 should size the pool to 4"
);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
for i in 0..8u64 {
w.append(&build_simple_batch(i * 10, 3)).expect("append");
}
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(r.n_superfiles(), 1, "small buffer stays one superfile");
assert_eq!(r.n_docs_total(), 24);
}
#[test]
fn append_auto_flushes_when_buffer_crosses_threshold() {
let opts = options_id_title_serial().with_commit_threshold_size_mb(1);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
let batch = build_simple_batch(0, 50_000);
w.append(&batch).expect("append");
assert_eq!(st.manifest_id(), 1, "auto-flush should fire");
assert_eq!(w.buffered_batches(), 0, "buffer drained on auto-flush");
w.commit().expect("commit-empty");
assert_eq!(st.manifest_id(), 1);
}
#[test]
fn append_does_not_auto_flush_when_threshold_zero() {
let opts = options_id_title_serial().with_commit_threshold_size_mb(0);
let st = Supertable::create(opts).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 50_000)).expect("append");
assert_eq!(st.manifest_id(), 0, "no auto-flush at threshold=0");
assert!(w.buffered_batches() > 0);
}
#[ignore = "known O(n) regression: manifest part rewrite on every commit"]
#[test]
fn commit_latency_is_constant_with_localfs() {
const N: usize = 100;
const DOCS_PER_COMMIT: usize = 64;
const MAX_GROWTH_FACTOR: f64 = 2.0;
let dir = TempDir::new().expect("tempdir");
let storage = Arc::new(LocalFsStorageProvider::new(dir.path()).expect("provider"));
let opts = options_id_title_serial().with_storage(storage);
let st = Supertable::create(opts).expect("create");
let mut latencies_ms: Vec<u128> = Vec::with_capacity(N);
for i in 0..N {
let batch = build_simple_batch(i as u64, DOCS_PER_COMMIT);
let t0 = Instant::now();
st.append(&batch).expect("append");
latencies_ms.push(t0.elapsed().as_millis());
}
let avg = |slice: &[u128]| slice.iter().sum::<u128>() as f64 / slice.len() as f64;
let first5_avg = avg(&latencies_ms[..5]);
let last5_avg = avg(&latencies_ms[N - 5..]);
let ratio = last5_avg / first5_avg.max(1.0);
println!(
"first-5 avg: {first5_avg:.1}ms last-5 avg: {last5_avg:.1}ms ratio: {ratio:.1}x"
);
assert!(
ratio <= MAX_GROWTH_FACTOR,
"commit latency grew {ratio:.1}x from first-5 ({first5_avg:.1}ms) to \
last-5 ({last5_avg:.1}ms) — O(n) growth in manifest rewrite path"
);
}
#[test]
fn each_commit_appends_to_existing_superfiles() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 2)).expect("a1");
w.commit().expect("c1");
w.append(&build_simple_batch(10, 3)).expect("a2");
w.commit().expect("c2");
w.append(&build_simple_batch(20, 1)).expect("a3");
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(r.manifest_id(), 3);
assert_eq!(r.n_superfiles(), 3);
assert_eq!(r.n_docs_total(), 6);
}
#[test]
fn merge_ranges_coalesces_overlapping_and_adjacent_drops_empty() {
let input = vec![
(100u64, 10u64), (0, 0), (10, 10), (15, 10), (25, 5), ];
let merged = merge_ranges(input);
assert_eq!(merged, vec![(10, 20), (100, 10)]);
}
#[test]
fn merge_ranges_empty_input_is_empty() {
assert!(merge_ranges(Vec::new()).is_empty());
}
#[test]
fn build_subsection_offsets_captures_total_size_and_fts_range() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_simple_batch(0, 8)).expect("append");
w.commit().expect("commit");
let r = st.reader().expect("reader");
let seg = &r.manifest().superfiles[0];
let store = &st.options().store;
let reader = store.reader(&seg.uri).expect("reader");
let offsets = seg
.subsection_offsets
.as_ref()
.expect("offsets captured at commit");
assert!(offsets.total_size > 0);
assert!(
offsets.fts.is_some(),
"an FTS superfile must record an FTS subsection"
);
assert!(
!offsets.fts_open_ranges.is_empty(),
"FTS open ranges should be populated for the cold-open fast path"
);
assert_eq!(reader.n_docs(), 8);
}
#[test]
fn build_subsection_offsets_on_garbage_returns_none() {
let garbage = Bytes::from_static(b"not a parquet file at all");
assert!(build_subsection_offsets(&garbage).is_none());
}
#[test]
fn append_with_vector_column_publishes_superfile() {
let dim = 16;
let st = Supertable::create(options_with_vector(dim)).expect("create");
let mut w = st.writer().expect("writer");
w.append(&build_vector_batch(0, 8, dim)).expect("append");
assert!(
w.buffered_bytes() > 0,
"buffered_bytes must account for the vector payload"
);
w.commit().expect("commit");
let r = st.reader().expect("reader");
assert_eq!(r.n_superfiles(), 1);
assert_eq!(r.n_docs_total(), 8);
}
fn storage_backed_st(dir: &TempDir) -> Supertable {
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(dir.path()).expect("provider"));
Supertable::create(options_id_title_serial().with_storage(storage)).expect("create")
}
fn row(title: &str) -> RecordBatch {
RecordBatch::try_new(
schema_id_title(),
vec![Arc::new(LargeStringArray::from(vec![title]))],
)
.expect("row batch")
}
#[test]
fn delete_tombstones_matching_row() {
use datafusion::prelude::{col, lit};
let dir = TempDir::new().expect("tempdir");
let st = storage_backed_st(&dir);
st.append(&build_simple_batch(0, 1)).expect("append");
let stats = st
.delete(col("title").eq(lit("doc 0 alpha")))
.expect("delete");
assert_eq!(stats.matched(), 1);
assert_eq!(stats.n_tombstoned(), 1);
}
#[test]
fn delete_unmatched_predicate_is_noop() {
use datafusion::prelude::{col, lit};
let dir = TempDir::new().expect("tempdir");
let st = storage_backed_st(&dir);
st.append(&build_simple_batch(0, 1)).expect("append");
let stats = st
.delete(col("title").eq(lit("no such title")))
.expect("delete");
assert_eq!(stats.matched(), 0);
assert_eq!(stats.n_tombstoned(), 0);
}
#[test]
fn update_replaces_matching_row() {
use datafusion::prelude::{col, lit};
let dir = TempDir::new().expect("tempdir");
let st = storage_backed_st(&dir);
st.append(&row("draft")).expect("append");
let stats = st
.update(col("title").eq(lit("draft")), &row("published"))
.expect("update");
assert_eq!(stats.matched(), 1);
assert_eq!(stats.n_tombstoned(), 1);
}
#[test]
fn update_cardinality_mismatch_is_rejected() {
use datafusion::prelude::{col, lit};
let dir = TempDir::new().expect("tempdir");
let st = storage_backed_st(&dir);
st.append(&row("draft")).expect("append");
let two = RecordBatch::try_new(
schema_id_title(),
vec![Arc::new(LargeStringArray::from(vec!["a", "b"]))],
)
.expect("two-row batch");
let mut w = st.writer().expect("writer");
let err = w
.update(col("title").eq(lit("draft")), two)
.expect_err("cardinality mismatch");
assert!(
matches!(
err,
MutationError::CardinalityMismatch {
matched: 1,
new_rows: 2
}
),
"{err:?}"
);
}
#[test]
fn update_without_storage_is_rejected() {
use datafusion::prelude::{col, lit};
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
let err = w
.update(col("title").eq(lit("x")), row("y"))
.expect_err("no storage");
assert!(matches!(err, MutationError::NoStorageAttached), "{err:?}");
}
#[test]
fn delete_without_storage_is_rejected() {
use datafusion::prelude::{col, lit};
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
let err = w.delete(col("title").eq(lit("x"))).expect_err("no storage");
assert!(matches!(err, MutationError::NoStorageAttached), "{err:?}");
}
#[test]
fn select_split_batch_packs_largest_first_within_budget() {
let dim = 128u32;
let counts: HashMap<u32, u64> = [(1, 1000), (2, 600), (3, 200), (4, 100), (5, 50)]
.into_iter()
.collect();
let none: HashSet<u32> = HashSet::new();
let budget =
estimate_split_resident_bytes(1000, dim) + estimate_split_resident_bytes(200, dim);
let batch = select_split_batch(&counts, &none, dim, budget, usize::MAX);
assert_eq!(
batch,
vec![1, 3],
"largest first, middle skipped over budget, sub-floor cells (4, 5) excluded"
);
}
#[test]
fn select_split_batch_always_admits_one() {
let counts: HashMap<u32, u64> = [(7, 1000), (9, 900)].into_iter().collect();
let none: HashSet<u32> = HashSet::new();
let batch = select_split_batch(&counts, &none, 128, 1, usize::MAX);
assert_eq!(batch, vec![7]);
}
#[test]
fn select_split_batch_respects_unsplittable_allowance_and_ties() {
let counts: HashMap<u32, u64> = [(3, 400), (8, 400), (1, 400), (6, 400)]
.into_iter()
.collect();
let unsplittable: HashSet<u32> = [1].into_iter().collect();
let batch = select_split_batch(&counts, &unsplittable, 128, u64::MAX, 2);
assert_eq!(
batch,
vec![3, 6],
"id-order ties, unsplittable 1 dropped, capped at 2"
);
}
#[test]
fn pending_metadata_schema_probes_both_producers() {
let repack = serde_json::to_vec(&RepackCheckpoint {
schema: REPACK_CHECKPOINT_SCHEMA,
})
.expect("encode");
assert_eq!(
pending_metadata_schema(&repack),
Some(REPACK_CHECKPOINT_SCHEMA)
);
let drain = serde_json::to_vec(&serde_json::json!({
"schema": DRAIN_CHECKPOINT_SCHEMA,
"unrelated": true
}))
.expect("encode");
assert_eq!(
pending_metadata_schema(&drain),
Some(DRAIN_CHECKPOINT_SCHEMA)
);
assert_eq!(pending_metadata_schema(b"not json"), None);
assert_eq!(pending_metadata_schema(b"{}"), None);
}
#[test]
fn split_batch_window_survives_disabled_merge_ceiling() {
const MIB: u64 = 1024 * 1024;
assert_eq!(
split_batch_window_bytes(0),
4096 * MIB,
"0 falls back to the default"
);
assert_eq!(
split_batch_window_bytes(512),
512 * MIB,
"nonzero passes through"
);
assert_eq!(
split_batch_window_bytes(u64::MAX),
u64::MAX,
"saturates instead of overflowing"
);
}
#[test]
fn buffered_bytes_grows_then_resets_on_commit() {
let st = Supertable::create(options_id_title_serial()).expect("create");
let mut w = st.writer().expect("writer");
assert_eq!(w.buffered_bytes(), 0);
w.append(&build_simple_batch(0, 4)).expect("append");
assert!(w.buffered_bytes() > 0, "buffer cost recorded");
assert_eq!(w.buffered_batches(), 1);
w.commit().expect("commit");
assert_eq!(w.buffered_bytes(), 0, "buffer drained on commit");
assert_eq!(w.buffered_batches(), 0);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn put_superfile_replace_creates_then_overwrites() {
let directory = TempDir::new().expect("tempdir");
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let path = "superfiles/replace-me.sf";
let first = Bytes::from_static(b"first-body-contents");
put_superfile_replace(&storage, path, first.clone())
.await
.expect("first put creates");
let (read_first, _) = storage.get(path).await.expect("read after create");
assert_eq!(read_first, first, "created object holds the first bytes");
let second = Bytes::from_static(b"second-body-different-length");
put_superfile_replace(&storage, path, second.clone())
.await
.expect("second put overwrites");
let (read_second, _) = storage.get(path).await.expect("read after overwrite");
assert_eq!(read_second, second, "overwrite installs the new bytes");
assert_ne!(
read_second, read_first,
"object content actually changed between writes"
);
}
const DELETE_LEASE_WAL_ID: i128 = 0x0DE1_5EA5;
const DELETE_LEASE_TARGET_ID: i128 = 42;
const DELETE_LEASE_PEER_OWNER: i128 = 0x0BAD_0BAD;
fn delete_lease_test_table(directory: &TempDir) -> Supertable {
let storage: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
Supertable::create(default_supertable_options().with_storage(storage)).expect("create")
}
fn delete_lease_test_entry() -> PendingDeleteEntry {
PendingDeleteEntry {
wal_id: WalId(DELETE_LEASE_WAL_ID),
target_ids: vec![DELETE_LEASE_TARGET_ID],
}
}
#[test]
fn delete_wal_doc_is_born_leased_by_this_handle() {
let directory = TempDir::new().expect("tempdir");
let table = delete_lease_test_table(&directory);
let writer = table.writer().expect("writer");
let now = Utc::now();
let doc = writer.delete_wal_doc(&delete_lease_test_entry(), now);
assert_eq!(doc.op_kind, OpKind::Delete);
assert_eq!(doc.state, WalState::Intent);
assert_eq!(
doc.created_at, now,
"created_at must come from the passed clock reading, not a second sample"
);
let lease = doc
.lease
.expect("a delete WAL must be born leased, not left unowned until a later acquire");
assert_eq!(
lease.owner,
table.handle_id(),
"the lease must name the handle that will drive the tombstone phase"
);
assert_eq!(
lease.acquired_at, now,
"acquired_at must share created_at's clock reading"
);
assert_eq!(
lease.expires_at,
now + ChronoDuration::from_std(DEFAULT_LEASE_DURATION)
.expect("default lease duration converts"),
"the lease must run a full default duration from the same reading"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn peer_sweep_skips_a_freshly_created_delete_wal() {
let directory = TempDir::new().expect("tempdir");
let table = delete_lease_test_table(&directory);
let storage = table
.inner()
.options
.storage
.as_ref()
.expect("storage attached")
.clone();
let writer = table.writer().expect("writer");
let wal_store = WalStore::new(storage);
let doc = writer.delete_wal_doc(&delete_lease_test_entry(), Utc::now());
let etag_before = wal_store.create(&doc).await.expect("create wal state doc");
let report = scan_and_recover(
&table,
SupertableHandleId(DELETE_LEASE_PEER_OWNER),
DEFAULT_LEASE_DURATION,
)
.await
.expect("sweep");
assert_eq!(report.n_scanned, 1, "the sweep must see the seeded WAL");
assert_eq!(
report.n_held_by_peer, 1,
"the writer's live lease must fence the sweep off this WAL"
);
assert_eq!(
report.n_tombstone_only_completed, 0,
"the sweep must not drive a delete the writer is still holding"
);
let (after, etag_after) = wal_store
.read(WalId(DELETE_LEASE_WAL_ID))
.await
.expect("read back");
assert_eq!(
etag_after, etag_before,
"etag unchanged → the sweep never wrote the state doc"
);
assert_eq!(
after.state,
WalState::Intent,
"the WAL must still be waiting for its owner's tombstone phase"
);
assert_eq!(
after.lease.expect("lease survives the sweep").owner,
table.handle_id(),
"ownership must still sit with the creating handle"
);
}
const DELETE_LEASE_SIDECAR_FAULTS: usize = 64;
const DELETE_LEASE_TOMBSTONES_SUFFIX: &str = ".tombstones";
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_failed_delete_hands_back_its_wal_lease() {
let directory = TempDir::new().expect("tempdir");
let local: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let faults = FaultStorage::wrap(local);
let storage: Arc<dyn StorageProvider> = Arc::<FaultStorage>::clone(&faults);
let table =
Supertable::create(default_supertable_options().with_storage(Arc::clone(&storage)))
.expect("create");
let mut writer = table.writer().expect("writer");
writer
.append(&build_title_batch(&["alpha", "beta"]))
.expect("append");
writer.commit().expect("commit appends");
writer
.delete(col("title").eq(lit("alpha")))
.expect("buffer delete");
faults.fail_with(
FaultKind::Precondition,
FaultOp::PutIfMatch,
DELETE_LEASE_TOMBSTONES_SUFFIX,
DELETE_LEASE_SIDECAR_FAULTS,
);
let err = writer
.commit()
.expect_err("a sidecar CAS that never lands must fail the delete");
assert!(
matches!(err, CommitError::PartialCommit { .. }),
"the failed delete must surface as a partial commit, got {err:?}"
);
assert!(
faults.fired() > 1,
"the failure must come from the injected sidecar faults, fired {}",
faults.fired()
);
faults.clear();
let wal_store = WalStore::new(storage);
let wal_ids = wal_store.list_wal_ids().await.expect("list wal ids");
assert_eq!(
wal_ids.len(),
1,
"the failed delete must leave its WAL for recovery, found {wal_ids:?}"
);
let (doc, _etag) = wal_store.read(wal_ids[0]).await.expect("read wal doc");
assert_eq!(
doc.state,
WalState::Intent,
"the WAL must still be waiting for its tombstone phase"
);
assert!(
doc.lease.is_none(),
"a failed delete must release its lease so the next sweep can take \
the WAL immediately, still held by {:?}",
doc.lease
);
}
const UPDATE_LEASE_PEER_OWNER: i128 = 0x0BAD_CAFE;
fn writer_with_buffered_update(table: &Supertable) -> SupertableWriter {
let mut writer = table.writer().expect("writer");
writer
.append(&build_title_batch(&["alpha", "beta"]))
.expect("append");
writer.commit().expect("commit appends");
writer
.update(col("title").eq(lit("alpha")), build_title_batch(&["gamma"]))
.expect("buffer update");
writer
}
#[test]
fn update_wal_doc_is_born_leased_by_this_handle() {
let directory = TempDir::new().expect("tempdir");
let table = delete_lease_test_table(&directory);
let writer = writer_with_buffered_update(&table);
let entry = writer
.pending_updates
.first()
.expect("update() must buffer an entry");
let now = Utc::now();
let doc = writer.update_wal_doc(entry, now);
assert_eq!(doc.op_kind, OpKind::Update);
assert_eq!(doc.state, WalState::Intent);
assert_eq!(
doc.created_at, now,
"created_at must come from the passed clock reading, not a second sample"
);
let lease = doc
.lease
.expect("an update WAL must be born leased, not left unowned until a later acquire");
assert_eq!(
lease.owner,
table.handle_id(),
"the lease must name the handle that will drive the pipeline"
);
assert_eq!(
lease.acquired_at, now,
"acquired_at must share created_at's clock reading"
);
assert_eq!(
lease.expires_at,
now + ChronoDuration::from_std(DEFAULT_LEASE_DURATION)
.expect("default lease duration converts"),
"the lease must run a full default duration from the same reading"
);
assert_eq!(doc.new_row_count, Some(1));
assert!(doc.preallocated_superfile_id.is_some());
assert_eq!(doc.tombstone_progress.len(), 1);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn peer_sweep_skips_a_freshly_created_update_wal() {
let directory = TempDir::new().expect("tempdir");
let table = delete_lease_test_table(&directory);
let storage = table
.inner()
.options
.storage
.as_ref()
.expect("storage attached")
.clone();
let writer = writer_with_buffered_update(&table);
let entry = writer
.pending_updates
.first()
.expect("update() must buffer an entry");
let wal_id = entry.wal_id;
let wal_store = WalStore::new(storage);
wal_store
.put_arrow(wal_id, entry.ipc_bytes.clone())
.await
.expect("put arrow payload");
let doc = writer.update_wal_doc(entry, Utc::now());
let etag_before = wal_store.create(&doc).await.expect("create wal state doc");
let report = scan_and_recover(
&table,
SupertableHandleId(UPDATE_LEASE_PEER_OWNER),
DEFAULT_LEASE_DURATION,
)
.await
.expect("sweep");
assert_eq!(report.n_scanned, 1, "the sweep must see the seeded WAL");
assert_eq!(
report.n_held_by_peer, 1,
"the writer's live lease must fence the sweep off this WAL"
);
assert_eq!(
report.n_full_pipeline_completed, 0,
"the sweep must not run the append phase for an update the writer holds"
);
let (after, etag_after) = wal_store.read(wal_id).await.expect("read back");
assert_eq!(
etag_after, etag_before,
"etag unchanged → the sweep never wrote the state doc"
);
assert_eq!(
after.state,
WalState::Intent,
"the WAL must still be waiting for its owner's append phase"
);
assert_eq!(
after.lease.expect("lease survives the sweep").owner,
table.handle_id(),
"ownership must still sit with the creating handle"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_failed_update_hands_back_its_wal_lease() {
let directory = TempDir::new().expect("tempdir");
let local: Arc<dyn StorageProvider> =
Arc::new(LocalFsStorageProvider::new(directory.path()).expect("provider"));
let faults = FaultStorage::wrap(local);
let storage: Arc<dyn StorageProvider> = Arc::<FaultStorage>::clone(&faults);
let table =
Supertable::create(default_supertable_options().with_storage(Arc::clone(&storage)))
.expect("create");
let mut writer = table.writer().expect("writer");
writer
.append(&build_title_batch(&["alpha", "beta"]))
.expect("append");
writer.commit().expect("commit appends");
writer
.update(col("title").eq(lit("alpha")), build_title_batch(&["gamma"]))
.expect("buffer update");
faults.fail_with(
FaultKind::Precondition,
FaultOp::PutIfMatch,
DELETE_LEASE_TOMBSTONES_SUFFIX,
DELETE_LEASE_SIDECAR_FAULTS,
);
let err = writer
.commit()
.expect_err("a sidecar CAS that never lands must fail the update");
assert!(
matches!(err, CommitError::PartialCommit { .. }),
"the failed update must surface as a partial commit, got {err:?}"
);
assert!(
faults.fired() > 1,
"the failure must come from the injected sidecar faults, fired {}",
faults.fired()
);
faults.clear();
let wal_store = WalStore::new(storage);
let wal_ids = wal_store.list_wal_ids().await.expect("list wal ids");
assert_eq!(
wal_ids.len(),
1,
"the failed update must leave its WAL for recovery, found {wal_ids:?}"
);
let (doc, _etag) = wal_store.read(wal_ids[0]).await.expect("read wal doc");
assert_eq!(
doc.state,
WalState::Appended,
"the append phase landed; only the tombstone phase is left"
);
assert!(
doc.lease.is_none(),
"a failed update must release its lease so the next sweep can take \
the WAL immediately, still held by {:?}",
doc.lease
);
}
}