crates/mimir-ingest/src/readfast.rs
//! The READ-FAST zone: per repo, one INTERVAL table per source kind —
//! `code_nodes_v` / `code_edges_v` / `code_spans_v`, partitioned by repo —
//! one row per node/edge/span VERSION with `[valid_from_seq, valid_to_seq)`
//! (open = `valid_to_seq IS NULL`), derived from the event log by the fold
//! and refreshed per repo ([`EventLog::materialize`]). The anchor (the seq a
//! repo's interval rows are complete to) is a property on each interval
//! table; a read ([`EventLog::readfast_rows`]) takes the interval rows live
//! at `min(upto, anchor)` plus the event TAIL after the anchor (a `seq`
//! range scan, immune to compaction's replace snapshots) — the same open
//! rows THE fold hands out, without scanning the repo's whole history.
//!
//! The landing zone (the event tables) stays the truth: the interval tables
//! are derived, replaced whole per repo, and never consulted for a repo
//! whose anchor is missing (the read falls back to the fold). Identities are
//! STABLE (mimir-schema 5): the interval rows carry the stored columns with
//! their real versions, and both sides of the tail merge hash the same
//! stable identity the delta keys by.
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Instant;
use anyhow::{Context, Result, bail};
use skade::ScanFilter;
use skade::arrow_array::{Array, ArrayRef, Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use crate::delta::{Kind, OpenSet, unhex};
use skade::iceberg::Catalog;
use crate::log::{EventLog, Evolves, FoldScope, FoldedRows, IdHash, InFlight, ensure, i64_col, identity, identity_into, on, str_col};
use crate::schema::{COL_KEY_DUP, COL_KRATE, COL_OP, COL_PAYLOAD_HASH, COL_REPO, COL_SEQ, OP_PLUS, qualified};
/// The interval tables, one per source kind.
pub const CODE_NODES_V: &str = "code_nodes_v";
pub const CODE_EDGES_V: &str = "code_edges_v";
pub const CODE_SPANS_V: &str = "code_spans_v";
/// A version's interval: `[valid_from_seq, valid_to_seq)` in the repo's
/// consumer ordinals; `valid_to_seq` null = the version is open.
pub const COL_VALID_FROM_SEQ: &str = "valid_from_seq";
pub const COL_VALID_TO_SEQ: &str = "valid_to_seq";
/// Table property `mimir.readfast.anchor.<repo>` = the seq that repo's
/// interval rows are complete to.
const PROP_ANCHOR: &str = "mimir.readfast.anchor.";
/// What [`EventLog::materialize`] did for one repo.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MaterializeReport {
pub repo: String,
/// The seq the interval rows are complete to after the run.
pub anchor_seq: i64,
/// Interval rows written (open + closed), every kind.
pub rows: u64,
pub open_rows: u64,
pub closed_rows: u64,
/// The anchor was at the repo's tip already: nothing was written.
pub skipped: bool,
/// At least one kind ADVANCED its zone (open rows + the tail) instead
/// of refolding the repo's history.
pub incremental: bool,
pub ms: u64,
}
impl std::fmt::Display for MaterializeReport {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
if self.skipped {
return write!(f, "readfast {}: anchor already at seq {}, nothing written ({} ms)", self.repo, self.anchor_seq, self.ms);
}
write!(
f,
"readfast {}: {} interval row(s) ({} open, {} closed) anchored at seq {} in {} ms{}",
self.repo,
self.rows,
self.open_rows,
self.closed_rows,
self.anchor_seq,
self.ms,
if self.incremental { " (incremental)" } else { "" }
)
}
}
/// What [`EventLog::materialize_all`] did: one entry per consumed repo
/// stream — written, already at its tip (skipped), or failed with why.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MaterializeSweep {
/// The streams whose zones were (re)written, in the offsets' order.
pub written: Vec<MaterializeReport>,
/// Streams whose anchor was at the tip already: nothing to do.
pub skipped: usize,
/// `(repo, error)` — the streams that failed. The sweep goes on past a
/// failure: the rest still get their zones.
pub failed: Vec<(String, String)>,
}
fn interval_table_name(kind: Kind) -> Option<&'static str> {
match kind {
Kind::Nodes => Some(CODE_NODES_V),
Kind::Edges => Some(CODE_EDGES_V),
Kind::Spans => Some(CODE_SPANS_V),
_ => None,
}
}
/// The interval table's schema: the identity prefix, the interval, and the
/// payload columns (the event's own `seq` / `sha` / `op` have no row here —
/// `valid_from_sha` / `valid_to_sha` are payload columns and ride along).
fn interval_schema(spec: &crate::schema::PayloadSpec) -> Schema {
let mut fields = vec![
Field::new(COL_REPO, DataType::Utf8, false),
Field::new(COL_KRATE, DataType::Utf8, false),
Field::new(COL_PAYLOAD_HASH, DataType::Utf8, true),
Field::new(COL_KEY_DUP, DataType::Utf8, true),
Field::new(COL_VALID_FROM_SEQ, DataType::Int64, false),
Field::new(COL_VALID_TO_SEQ, DataType::Int64, true),
];
fields.extend(spec.schema.fields().iter().map(|f| Field::new(f.name(), crate::schema::storage_type(f.data_type()), true)));
Schema::new(fields).with_metadata(spec.schema.metadata().clone())
}
impl EventLog {
/// Rebuild the READ-FAST zone for `repo`: fold its events into interval
/// rows (every version, open and closed), replace the repo's partition of
/// each interval table in one commit per table, and move the anchor.
/// Idempotent: an anchor already at the repo's tip skips the work.
pub async fn materialize(&self, repo: &str) -> Result<MaterializeReport> {
if self.is_pinned() {
bail!("a pinned read handle cannot materialize the read-fast zone");
}
let log = self.clone();
let repo = repo.to_string();
on(self.shared.rt(), async move { materialize(log, &repo).await }).await
}
/// Rebuild the READ-FAST zone for EVERY consumed repo stream — the
/// maintainer's one sweep (the server's `lake_readfast` job body). A
/// stream already anchored at its tip is counted, not rewritten; a
/// stream that FAILS is recorded with why and NEVER stops the rest (one
/// bad stream used to starve every stream behind it in the offsets'
/// order). Only the offsets read itself can fail the sweep whole.
pub async fn materialize_all(&self) -> Result<MaterializeSweep> {
if self.is_pinned() {
bail!("a pinned read handle cannot materialize the read-fast zone");
}
let log = self.clone();
on(self.shared.rt(), async move {
let mut sweep = MaterializeSweep::default();
for repo in log.offsets().await?.into_keys() {
match materialize(log.clone(), &repo).await {
Ok(r) if r.skipped => sweep.skipped += 1,
Ok(r) => sweep.written.push(r),
Err(e) => sweep.failed.push((repo, format!("{e:#}"))),
}
}
Ok(sweep)
})
.await
}
/// THE fold's answer from the zone where it exists: every repo's open
/// set — an anchored source (kind, repo) read as interval rows + tail,
/// everything else (repos without a zone, the binary stream) by the
/// event fold. The consumer boots on this; [`EventLog::fold`] stays the
/// event-only truth the zone is tested against.
pub async fn fold_fast(&self) -> Result<crate::delta::OpenSet> {
let log = self.clone();
on(self.shared.rt(), async move { fold_fast(log).await }).await
}
/// The seq `repo`'s interval rows are complete to: the LOWEST per-kind
/// anchor (a crash between kinds leaves them apart; each kind's read
/// uses its own). `None` = no zone for the repo yet.
pub async fn readfast_anchor(&self, repo: &str) -> Result<Option<i64>> {
let log = self.clone();
let repo = repo.to_string();
on(self.shared.rt(), async move {
let mut min: Option<i64> = None;
for kind in Kind::SOURCE {
match anchor_of(&log, kind, &repo).await? {
Some(a) => min = Some(min.map_or(a, |m: i64| m.min(a))),
None => return Ok(None),
}
}
Ok(min)
})
.await
}
/// Every interval row of `repo` for `kind` (open and closed), all
/// columns — inspection and tests.
pub async fn interval_rows(&self, kind: Kind, repo: &str) -> Result<Vec<RecordBatch>> {
let name = interval_table_name(kind).context("a source kind")?;
let log = self.clone();
let repo = repo.to_string();
on(self.shared.rt(), async move {
let t = log.shared.wh.table(&qualified(name)).await?;
let cols = skade::arrow_schema_of(t.inner())?;
let cols: Vec<&str> = cols.fields().iter().map(|f| f.name().as_str()).collect();
let rows = skade::read_filtered(t.inner(), &ScanFilter::eq(COL_REPO, repo.as_str()), &cols).await?;
Ok(rows)
})
.await
}
/// The open rows of `repo` as of `upto` (`None` = the tip), from the
/// READ-FAST zone: the interval rows live at `min(upto, anchor)` plus the
/// event tail `(anchor, upto]` — the fold's answer without the history
/// scan. A repo without an anchor falls back to [`EventLog::fold_rows`].
pub async fn readfast_rows(&self, kind: Kind, repo: &str, upto: Option<i64>, cols: &[&str]) -> Result<FoldedRows> {
let log = self.clone();
let repo = repo.to_string();
let cols: Vec<String> = cols.iter().map(|c| c.to_string()).collect();
on(self.shared.rt(), async move {
let anchor = match interval_table_name(kind) {
Some(_) => anchor_of(&log, kind, &repo).await?,
None => None,
};
let Some(anchor) = anchor else {
let upto = upto.unwrap_or(i64::MAX);
let scope = FoldScope { repos: [(repo.clone(), upto)].into(), universe: false };
let cols: Vec<&str> = cols.iter().map(String::as_str).collect();
return log.fold_rows(kind, &scope, &cols).await;
};
readfast_read(&log, kind, &repo, anchor, upto, &cols).await
})
.await
}
}
/// The open `mir_block` payloads of one repo ([`EventLog::mir_block_payloads`]).
#[derive(Debug, Clone, Default)]
pub struct MirPayloads {
/// STABLE block moniker ([`mimir_schema::moniker::stable`]: a `-` event
/// names its identity version-stripped) → its `ty` payload.
pub rows: HashMap<String, String>,
/// Payload bytes held.
pub bytes: usize,
/// `cap` stopped the read: later blocks are absent.
pub capped: bool,
/// Read from the READ-FAST zone (+ its tail), not the event history.
pub from_zone: bool,
}
impl MirPayloads {
fn put(&mut self, id: &str, ty: &str, cap: usize) {
let id = mimir_schema::moniker::stable(id);
let id = id.as_ref();
if let Some(old) = self.rows.get(id) {
self.bytes -= old.len();
} else if self.bytes + ty.len() > cap {
self.capped = true;
return;
}
self.bytes += ty.len();
self.rows.insert(id.to_string(), ty.to_string());
}
fn drop_id(&mut self, id: &str) {
if let Some(old) = self.rows.remove(mimir_schema::moniker::stable(id).as_ref()) {
self.bytes -= old.len();
}
}
}
impl EventLog {
/// **The MIR blocks beside the room**: `repo`'s open `mir_block` rows at
/// `upto` (`None` = the head) as block moniker → payload, at most `cap`
/// payload bytes. The room drops these rows (most of its bytes, for one
/// lens); the lens that needs them reads one repo, streamed: the zone's
/// interval rows live at the cut (only `mir_block` rows kept) and the
/// event tail after the anchor, or — no zone — the repo's events.
pub async fn mir_block_payloads(&self, repo: &str, upto: Option<i64>, cap: usize) -> Result<MirPayloads> {
let log = self.clone();
let repo = repo.to_string();
on(self.shared.rt(), async move { mir_block_payloads(&log, &repo, upto, cap).await }).await
}
}
async fn mir_block_payloads(log: &EventLog, repo: &str, upto: Option<i64>, cap: usize) -> Result<MirPayloads> {
use mimir_schema::kind::MIR_BLOCK;
let at = upto.unwrap_or(i64::MAX);
let mut out = MirPayloads::default();
let anchor = anchor_of(log, Kind::Nodes, repo).await?;
let mut failed: Option<anyhow::Error> = None;
if let Some(anchor) = anchor {
out.from_zone = true;
let cut = at.min(anchor);
let t = log.shared.wh.table(&qualified(CODE_NODES_V)).await?;
let proj = crate::log::scannable(t.inner(), &["id", "kind", "ty", COL_VALID_FROM_SEQ, COL_VALID_TO_SEQ])?;
skade::read_filtered_each(t.inner(), Some(&ScanFilter::eq(COL_REPO, repo)), &proj, |b| {
let step = (|| -> Result<()> {
let (id, kind, ty) = (str_col(&b, "id")?, str_col(&b, "kind")?, str_col(&b, "ty")?);
let (from, to) = (i64_col(&b, COL_VALID_FROM_SEQ)?, i64_col(&b, COL_VALID_TO_SEQ)?);
for i in 0..b.num_rows() {
if kind.is_null(i) || kind.value(i) != MIR_BLOCK || ty.is_null(i) || id.is_null(i) {
continue;
}
if from.value(i) > cut || (!to.is_null(i) && to.value(i) <= cut) {
continue;
}
out.put(id.value(i), ty.value(i), cap);
}
Ok(())
})();
match step {
Ok(()) => true,
Err(e) => {
failed = Some(e);
false
}
}
})
.await?;
if let Some(e) = failed.take() {
return Err(e);
}
if at <= anchor {
return Ok(out);
}
}
// the events: the tail after the anchor, or the repo's whole history;
// only what can touch a block is held — `+ mir_block` rows and every `-`
let t = log.tbl(EventLog::table_name(Kind::Nodes)).await?;
// the repo's partition only (the source tables are partitioned by repo)
let filter = ScanFilter::eq(COL_REPO, repo);
let after = anchor.unwrap_or(i64::MIN);
let proj = crate::log::scannable(&t, &[COL_REPO, COL_SEQ, COL_OP, "id", "kind", "ty"])?;
// (seq, scan order, plus, id, payload)
let mut events: Vec<(i64, u64, bool, String, Option<String>)> = Vec::new();
let mut n = 0u64;
skade::read_filtered_each(&t, Some(&filter), &proj, |b| {
let step = (|| -> Result<()> {
let (r, seq, op) = (str_col(&b, COL_REPO)?, i64_col(&b, COL_SEQ)?, str_col(&b, COL_OP)?);
let (id, kind, ty) = (str_col(&b, "id")?, str_col(&b, "kind")?, str_col(&b, "ty")?);
for i in 0..b.num_rows() {
n += 1;
if r.value(i) != repo || seq.value(i) > at || seq.value(i) <= after || id.is_null(i) {
continue;
}
let plus = op.value(i) == OP_PLUS;
let block = !kind.is_null(i) && kind.value(i) == MIR_BLOCK;
if plus && !block {
continue;
}
let payload = (plus && !ty.is_null(i)).then(|| ty.value(i).to_string());
events.push((seq.value(i), n, plus, id.value(i).to_string(), payload));
}
Ok(())
})();
match step {
Ok(()) => true,
Err(e) => {
failed = Some(e);
false
}
}
})
.await?;
if let Some(e) = failed.take() {
return Err(e);
}
events.sort_unstable_by_key(|e| (e.0, e.1));
for (_, _, plus, id, payload) in events {
match (plus, payload) {
(true, Some(p)) => out.put(&id, &p, cap),
_ => out.drop_id(&id),
}
}
Ok(out)
}
pub(crate) async fn anchor_of(log: &EventLog, kind: Kind, repo: &str) -> Result<Option<i64>> {
// a kind without an interval table (the binary stream) has no zone
let Some(name) = interval_table_name(kind) else { return Ok(None) };
let ident = log.shared.wh.table_ident(&qualified(name))?;
if !log.shared.wh.catalog().table_exists(&ident).await? {
return Ok(None);
}
let t = log.shared.wh.table(&qualified(name)).await?;
Ok(t.properties().get(&format!("{PROP_ANCHOR}{repo}")).and_then(|s| s.parse::<i64>().ok()))
}
/// One identity's hash: `repo · krate · identity` (the fold's recipe).
fn id_hash(stable: bool, repo: &str, krate: &str, keys: &[StringArray], dup: &StringArray, i: usize, buf: &mut Vec<u8>) -> u128 {
buf.clear();
buf.extend_from_slice(repo.as_bytes());
buf.push(0x1e);
buf.extend_from_slice(krate.as_bytes());
buf.push(0x1e);
identity_into(stable, keys, dup, i, buf);
twox_hash::XxHash3_128::oneshot(buf)
}
async fn materialize(log: EventLog, repo: &str) -> Result<MaterializeReport> {
let t0 = Instant::now();
let mut report = MaterializeReport { repo: repo.to_string(), ..Default::default() };
// the tip: the consumer's offset; a repo never consumed has no zone
let tip = match log.offsets().await?.get(repo) {
Some(o) => o.seq,
None => {
report.skipped = true;
report.ms = t0.elapsed().as_millis() as u64;
return Ok(report);
}
};
report.anchor_seq = tip;
let mut all_anchored = true;
for kind in Kind::SOURCE {
if anchor_of(&log, kind, repo).await? != Some(tip) {
all_anchored = false;
}
}
if all_anchored {
report.skipped = true;
report.ms = t0.elapsed().as_millis() as u64;
return Ok(report);
}
let shared = log.shared.clone();
for kind in Kind::SOURCE {
let spec = log.spec(kind);
let name = interval_table_name(kind).expect("a source kind");
let schema = interval_schema(spec);
ensure(&shared.wh, name, &schema, &[], &[COL_REPO], Evolves::Payload { spec }).await?;
let mut t = shared.wh.table(&qualified(name)).await?;
t.set_write_props(log.write_props().clone());
// INCREMENTAL where the zone allows it: the repo's open rows live in
// files of their own (closed rows never share a file with them), so
// a consume's tail closes and opens versions over the OPEN rows only
// — the closed history stays where it is and the window's closed
// versions are appended in a file of their own. A first build, a
// file mixing open and closed rows (an older zone, a compaction) or
// too many closed files refolds the repo's events whole.
let anchor = anchor_of(&log, kind, repo).await?;
if anchor == Some(tip) {
continue;
}
let files = partition_files_classified(&t, repo).await?;
let mixed = files.iter().any(|f| f.1 == FileRows::Mixed);
let closed_files = files.iter().filter(|f| f.1 == FileRows::Closed).count();
let mut payload_cols: Vec<&str> = spec.schema.fields().iter().map(|f| f.name().as_str()).collect();
payload_cols.push(COL_PAYLOAD_HASH);
let (sources, open_v, closed_v, remove) = match anchor {
Some(a) if a < tip && !mixed && closed_files < MAX_CLOSED_FILES => {
report.incremental = true;
// the zone's open rows (streamed: the closed rows are dropped as read)
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
let readable = crate::log::scannable(t.inner(), &cols)?;
let mut zone: Vec<RecordBatch> = Vec::new();
let mut failed: Option<anyhow::Error> = None;
skade::read_filtered_each(t.inner(), Some(&ScanFilter::eq(COL_REPO, repo)), &readable, |b| {
let step = (|| -> Result<()> {
let to = b.column_by_name(COL_VALID_TO_SEQ).context("valid_to_seq")?;
let mask = skade::arrow_array::BooleanArray::from_iter((0..to.len()).map(|i| Some(to.is_null(i))));
let open = arrow_select::filter::filter_record_batch(&b, &mask)?;
if open.num_rows() > 0 {
zone.extend(crate::log::null_filled(t.inner(), vec![open], Some(&cols))?);
}
Ok(())
})();
match step {
Ok(()) => true,
Err(e) => {
failed = Some(e);
false
}
}
})
.await?;
if let Some(e) = failed {
return Err(e);
}
// the tail: this repo's events after the anchor
let tail = repo_tail(&log, kind, repo, a, tip, &payload_cols).await?;
let (open_v, closed_v) = advance_versions(spec.stable_keys, repo, &zone, &tail.batches, &spec.key_columns)?;
let mut sources = zone;
sources.extend(tail.batches);
let remove: Vec<String> = files.iter().filter(|f| f.1 == FileRows::Open).map(|f| f.0.clone()).collect();
(sources, open_v, closed_v, remove)
}
_ => {
// the fold, keeping every version: scan the repo's events whole
// (the payload hash rides along — the scan's prefix leaves it out)
let scope = FoldScope { repos: [(repo.to_string(), i64::MAX)].into(), universe: false };
let scanned = log.scan_rows(kind, &scope, &payload_cols).await?;
let versions = fold_versions(spec.stable_keys, repo, &scanned.batches, &spec.key_columns)?;
let (open_v, closed_v): (Vec<Version>, Vec<Version>) = versions.into_iter().partition(|v| v.to.is_none());
(scanned.batches, open_v, closed_v, files.into_iter().map(|f| f.0).collect())
}
};
let (open_n, closed_n) = (open_v.len(), closed_v.len());
let open_b = version_batches(&schema, &sources, &open_v)?;
let closed_b = version_batches(&schema, &sources, &closed_v)?;
drop(sources);
let rows: u64 = open_b.iter().chain(&closed_b).map(|b| b.num_rows() as u64).sum();
if rows != (open_n + closed_n) as u64 {
bail!("readfast {repo} {name}: folded {} version(s), built {rows} row(s) — nothing committed", open_n + closed_n);
}
// open and closed rows in files of their own (the next increment's
// contract), encoded outside the commit lock
let groups: Vec<Vec<RecordBatch>> = [open_b, closed_b].into_iter().filter(|g| !g.is_empty()).collect();
let staged = if groups.is_empty() && remove.is_empty() { None } else { Some(t.write_rewrite_files(groups, None).await?) };
// replace the repo's open files (or its partition) + move the anchor, under the commit lock
{
let _commit = shared.commit_lock.lock().await;
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
if let Some(staged) = staged {
t.commit_overwrite(&remove, staged).await.with_context(|| format!("readfast: replace {name} for {repo}"))?;
}
t.set_properties([(format!("{PROP_ANCHOR}{repo}"), tip.to_string())])
.await
.with_context(|| format!("readfast: anchor {name} for {repo}"))?;
}
report.rows += rows;
report.open_rows += open_n as u64;
report.closed_rows += closed_n as u64;
}
report.ms = t0.elapsed().as_millis() as u64;
Ok(report)
}
/// [`EventLog::fold_fast`]: the zone's open rows applied as `+` events, the
/// rest (repos without an anchor for the kind, the binary stream) folded
/// from the event tables.
async fn fold_fast(log: EventLog) -> Result<OpenSet> {
let repos: Vec<String> = log.offsets().await?.into_keys().collect();
let mut set = OpenSet::default();
let hash_col = [COL_PAYLOAD_HASH.to_string()];
for kind in Kind::ALL {
if interval_table_name(kind).is_none() {
log.fold_kind_into(&mut set, kind, None, None, None).await?;
continue;
}
let spec = log.spec(kind);
let mut slow: Vec<String> = Vec::new();
for repo in &repos {
let Some(a) = anchor_of(&log, kind, repo).await? else {
slow.push(repo.clone());
continue;
};
let fr = readfast_read(&log, kind, repo, a, None, &hash_col).await?;
let mut at = 0usize;
for (bi, b) in fr.batches.iter().enumerate() {
let (krate, dup, hash) = (str_col(b, COL_KRATE)?, str_col(b, COL_KEY_DUP)?, str_col(b, COL_PAYLOAD_HASH)?);
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
while at < fr.open.len() && fr.open[at].0 == bi as u32 {
let i = fr.open[at].1 as usize;
at += 1;
if hash.is_null(i) {
bail!("readfast {repo}: an open row of {} has no payload_hash", EventLog::table_name(kind));
}
let key = identity(spec.stable_keys, &keys, &dup, i);
set.apply_event(kind, repo, OP_PLUS, krate.value(i), key, Some(unhex(hash.value(i))?))?;
}
}
}
log.fold_kind_into(&mut set, kind, None, None, Some(&slow)).await?;
}
set.shrink();
Ok(set)
}
/// One version of one identity: the source row of the `+` event that opened
/// it, and its interval.
struct Version {
batch: u32,
row: u32,
from: i64,
to: Option<i64>,
}
/// THE fold, keeping every version: per identity in (seq) order, a `+` with
/// another payload closes the open version and opens a new one; a `-` closes
/// it. What remains open at the end has `to = None`.
fn fold_versions(stable: bool, repo: &str, batches: &[RecordBatch], key_columns: &[String]) -> Result<Vec<Version>> {
let mut decoded = Vec::with_capacity(batches.len());
for b in batches {
let keys: Vec<StringArray> = key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
decoded.push((str_col(b, COL_REPO)?, i64_col(b, COL_SEQ)?, str_col(b, COL_OP)?, str_col(b, COL_KRATE)?, str_col(b, COL_PAYLOAD_HASH)?, str_col(b, COL_KEY_DUP)?, keys));
}
let mut order: Vec<(i64, u32, u32)> = Vec::new();
for (bi, (r, seq, ..)) in decoded.iter().enumerate() {
for i in 0..r.len() {
if r.value(i) == repo {
order.push((seq.value(i), bi as u32, i as u32));
}
}
}
order.sort_unstable();
let mut open: HashMap<u128, Version, IdHash> = HashMap::with_capacity_and_hasher(order.len(), IdHash);
let mut versions: Vec<Version> = Vec::new();
let mut buf: Vec<u8> = Vec::with_capacity(256);
for (seq, bi, ri) in order {
let (_, _, op, krate, hash, dup, keys) = &decoded[bi as usize];
let i = ri as usize;
let id = id_hash(stable, repo, krate.value(i), keys, dup, i, &mut buf);
if op.value(i) == OP_PLUS {
let next = Version { batch: bi, row: ri, from: seq, to: None };
match open.entry(id) {
std::collections::hash_map::Entry::Occupied(mut o) => {
let same = {
let v = o.get();
let (.., h0, _, _) = &decoded[v.batch as usize];
!hash.is_null(i) && !h0.is_null(v.row as usize) && h0.value(v.row as usize) == hash.value(i)
};
if !same {
let mut closed = o.insert(next);
closed.to = Some(seq);
versions.push(closed);
}
}
std::collections::hash_map::Entry::Vacant(v) => {
v.insert(next);
}
}
} else if let Some(mut closed) = open.remove(&id) {
closed.to = Some(seq);
versions.push(closed);
}
}
versions.extend(open.into_values());
Ok(versions)
}
/// Past this many closed-only files in a repo's partition the next
/// materialize refolds the repo whole (one closed file per increment).
const MAX_CLOSED_FILES: usize = 48;
/// THE fold over the zone's OPEN rows (`zone`, interval batches) and the
/// event `tail` after its anchor: `(open, closed)` versions, batch indices
/// into `zone ++ tail`. `closed` = only the versions this tail closed (the
/// zone's closed history is not read). Same rules as [`fold_versions`].
fn advance_versions(stable: bool, repo: &str, zone: &[RecordBatch], tail: &[RecordBatch], key_columns: &[String]) -> Result<(Vec<Version>, Vec<Version>)> {
let mut open: HashMap<u128, Version, IdHash> = HashMap::with_hasher(IdHash);
let mut buf: Vec<u8> = Vec::with_capacity(256);
let mut hashes: Vec<StringArray> = Vec::with_capacity(zone.len() + tail.len());
for (bi, b) in zone.iter().enumerate() {
let keys: Vec<StringArray> = key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
let (krate, dup, from) = (str_col(b, COL_KRATE)?, str_col(b, COL_KEY_DUP)?, i64_col(b, COL_VALID_FROM_SEQ)?);
hashes.push(str_col(b, COL_PAYLOAD_HASH)?);
for i in 0..b.num_rows() {
let id = id_hash(stable, repo, krate.value(i), &keys, &dup, i, &mut buf);
open.insert(id, Version { batch: bi as u32, row: i as u32, from: from.value(i), to: None });
}
}
let shift = zone.len() as u32;
let mut decoded = Vec::with_capacity(tail.len());
for b in tail {
let keys: Vec<StringArray> = key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
hashes.push(str_col(b, COL_PAYLOAD_HASH)?);
decoded.push((str_col(b, COL_REPO)?, i64_col(b, COL_SEQ)?, str_col(b, COL_OP)?, str_col(b, COL_KRATE)?, str_col(b, COL_KEY_DUP)?, keys));
}
let mut order: Vec<(i64, u32, u32)> = Vec::new();
for (bi, (r, seq, ..)) in decoded.iter().enumerate() {
for i in 0..r.len() {
if r.value(i) == repo {
order.push((seq.value(i), bi as u32, i as u32));
}
}
}
order.sort_unstable();
let mut closed: Vec<Version> = Vec::new();
for (seq, bi, ri) in order {
let (_, _, op, krate, dup, keys) = &decoded[bi as usize];
let i = ri as usize;
let id = id_hash(stable, repo, krate.value(i), keys, dup, i, &mut buf);
let hash = &hashes[(shift + bi) as usize];
if op.value(i) == OP_PLUS {
let next = Version { batch: shift + bi, row: ri, from: seq, to: None };
match open.entry(id) {
std::collections::hash_map::Entry::Occupied(mut o) => {
let same = {
let v = o.get();
let h0 = &hashes[v.batch as usize];
!hash.is_null(i) && !h0.is_null(v.row as usize) && h0.value(v.row as usize) == hash.value(i)
};
if !same {
let mut c = o.insert(next);
c.to = Some(seq);
closed.push(c);
}
}
std::collections::hash_map::Entry::Vacant(v) => {
v.insert(next);
}
}
} else if let Some(mut c) = open.remove(&id) {
c.to = Some(seq);
closed.push(c);
}
}
Ok((open.into_values().collect(), closed))
}
/// What a data file of a repo's partition holds, by its `valid_to_seq`
/// null count: open rows only, closed rows only, or both (stats missing
/// count as both).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FileRows {
Open,
Closed,
Mixed,
}
/// The data files of the current snapshot that hold `repo`'s partition
/// (the interval tables are born partitioned by repo), each with its
/// [`FileRows`].
async fn partition_files_classified(t: &skade::Table, repo: &str) -> Result<Vec<(String, FileRows)>> {
use skade::iceberg::spec::{Literal, ManifestContentType, ManifestStatus, PrimitiveLiteral};
let ice = t.inner();
let meta = ice.metadata();
let Some(s) = meta.current_snapshot() else { return Ok(Vec::new()) };
let to_id = meta.current_schema().field_by_name(COL_VALID_TO_SEQ).map(|f| f.id);
let mut out = Vec::new();
for mf in ice.manifest_list_reader(s).load().await?.entries() {
if mf.content != ManifestContentType::Data {
continue;
}
for e in ice.manifest_reader().read(mf).await?.entries() {
if e.status() == ManifestStatus::Deleted {
continue;
}
let df = e.data_file();
let of_repo = df.partition().fields().first().is_some_and(|f| matches!(f, Some(Literal::Primitive(PrimitiveLiteral::String(v))) if v == repo));
if !of_repo {
continue;
}
let nulls = to_id.and_then(|id| df.null_value_counts().get(&id).copied());
let rows = match nulls {
Some(n) if n == df.record_count() => FileRows::Open,
Some(0) => FileRows::Closed,
_ => FileRows::Mixed,
};
out.push((e.file_path().to_string(), rows));
}
}
Ok(out)
}
/// The interval rows: per source batch, the versions it carries, its columns
/// taken by row index, the interval columns appended — in `schema`'s order.
fn version_batches(schema: &Schema, batches: &[RecordBatch], versions: &[Version]) -> Result<Vec<RecordBatch>> {
let mut by_batch: HashMap<u32, Vec<&Version>> = HashMap::new();
for v in versions {
by_batch.entry(v.batch).or_default().push(v);
}
let mut out = Vec::with_capacity(by_batch.len());
for (bi, vs) in by_batch {
let src = &batches[bi as usize];
let idx = Int64Array::from_iter_values(vs.iter().map(|v| v.row as i64));
let mut columns: Vec<ArrayRef> = Vec::with_capacity(schema.fields().len());
for f in schema.fields() {
let col: ArrayRef = match f.name().as_str() {
COL_VALID_FROM_SEQ => Arc::new(Int64Array::from_iter_values(vs.iter().map(|v| v.from))),
COL_VALID_TO_SEQ => Arc::new(Int64Array::from_iter(vs.iter().map(|v| v.to))),
name => {
let c = src.column_by_name(name).with_context(|| format!("readfast: column `{name}` missing from the scan"))?;
arrow_select::take::take(c, &idx, None).with_context(|| format!("readfast: take `{name}`"))?
}
};
columns.push(col);
}
out.push(RecordBatch::try_new(Arc::new(schema.clone()), columns)?);
}
Ok(out)
}
/// **One repo's events in `(after, upto]`**, streamed from its partition of
/// the SOURCE table (the event tables are partitioned by repo): only those
/// rows are held. The seq-range scan it replaces read every repo's events in
/// the range from the source AND the ring table, collected whole — a repo
/// anchored at a low seq pulled most of the lake (another repo's seqs run
/// far higher), once per repo per room fold: the 8-12 GB spikes of a live
/// ring fold (dwarf 2026-10-23, OOM at 33.8 GB).
pub(crate) async fn repo_tail(log: &EventLog, kind: Kind, repo: &str, after: i64, upto: i64, cols: &[&str]) -> Result<FoldedRows> {
let proj = log.scan_projection(kind, cols);
let t = log.tbl(EventLog::table_name(kind)).await?;
let readable = crate::log::scannable(&t, &proj)?;
let mut kept: Vec<RecordBatch> = Vec::new();
let mut failed: Option<anyhow::Error> = None;
skade::read_filtered_each(&t, Some(&ScanFilter::eq(COL_REPO, repo)), &readable, |b| {
let step = (|| -> Result<()> {
let seq = i64_col(&b, COL_SEQ)?;
let mask = skade::arrow_array::BooleanArray::from_iter((0..b.num_rows()).map(|i| Some(seq.value(i) > after && seq.value(i) <= upto)));
if mask.true_count() > 0 {
kept.push(arrow_select::filter::filter_record_batch(&b, &mask)?);
}
Ok(())
})();
match step {
Ok(()) => true,
Err(e) => {
failed = Some(e);
false
}
}
})
.await?;
if let Some(e) = failed {
return Err(e);
}
Ok(FoldedRows { batches: crate::log::null_filled(&t, kept, Some(&proj))?, ..Default::default() })
}
/// The read: interval rows live at `min(upto, anchor)` + the tail
/// `(anchor, upto]`, merged per identity (the tail wins).
pub(crate) async fn readfast_read(log: &EventLog, kind: Kind, repo: &str, anchor: i64, upto: Option<i64>, cols: &[String]) -> Result<FoldedRows> {
Ok(readfast_read_tracked(log, kind, repo, anchor, upto, cols).await?.0)
}
/// [`readfast_read`], also returning per OPEN row its identity hash and the
/// seq it holds from (an interval row holds from `valid_from_seq`; a tail row
/// from its event's seq) — aligned with `FoldedRows::open` — so a TRACKED
/// fold can seed its identity map from the zone instead of scanning the
/// whole history. Identities whose last event is a close are absent, which
/// is delta-safe: any later event is newer than the close by construction.
pub(crate) async fn readfast_read_tracked(
log: &EventLog,
kind: Kind,
repo: &str,
anchor: i64,
upto: Option<i64>,
cols: &[String],
) -> Result<(FoldedRows, Vec<(u128, i64)>)> {
let spec = log.spec(kind);
let name = interval_table_name(kind).expect("anchored kinds are source kinds");
let at = upto.unwrap_or(i64::MAX);
let cut = at.min(anchor);
// the interval side: the repo's partition, projected like a fold scan
let t = log.shared.wh.table(&qualified(name)).await?;
let mut proj: Vec<&str> = vec![COL_REPO, COL_KRATE, COL_PAYLOAD_HASH, COL_KEY_DUP, COL_VALID_FROM_SEQ, COL_VALID_TO_SEQ];
for c in spec.key_columns.iter().map(String::as_str).chain(cols.iter().map(String::as_str)) {
if !proj.contains(&c) {
proj.push(c);
}
}
let readable = crate::log::scannable(t.inner(), &proj)?;
// streamed, the versions NOT live at the cut dropped as read (a repo's
// closed history is no part of a read at one seq)
let mut live: Vec<RecordBatch> = Vec::new();
let mut failed: Option<anyhow::Error> = None;
skade::read_filtered_each(t.inner(), Some(&ScanFilter::eq(COL_REPO, repo)), &readable, |b| {
let step = (|| -> Result<()> {
let (from, to) = (i64_col(&b, COL_VALID_FROM_SEQ)?, i64_col(&b, COL_VALID_TO_SEQ)?);
let mask = skade::arrow_array::BooleanArray::from_iter((0..b.num_rows()).map(|i| Some(from.value(i) <= cut && (to.is_null(i) || to.value(i) > cut))));
if mask.true_count() > 0 {
live.push(arrow_select::filter::filter_record_batch(&b, &mask)?);
}
Ok(())
})();
match step {
Ok(()) => true,
Err(e) => {
failed = Some(e);
false
}
}
})
.await?;
if let Some(e) = failed {
return Err(e);
}
let interval = crate::log::null_filled(t.inner(), live, Some(&proj))?;
// the tail: events after the anchor, a seq-range scan (file pruning by
// the column's bounds; other repos' rows in range are skipped below)
let mut out = FoldedRows::default();
let mut tail_touched: HashMap<u128, (), IdHash> = HashMap::with_hasher(IdHash);
let mut tail = FoldedRows::default();
if at > anchor {
let cols_ref: Vec<&str> = cols.iter().map(String::as_str).collect();
tail = repo_tail(log, kind, repo, anchor, at, &cols_ref).await?;
let scope = FoldScope { repos: [(repo.to_string(), at)].into(), universe: false };
log.fold_scanned(kind, &scope, &mut tail)?;
let mut buf: Vec<u8> = Vec::with_capacity(256);
for b in &tail.batches {
let (r, krate, dup) = (str_col(b, COL_REPO)?, str_col(b, COL_KRATE)?, str_col(b, COL_KEY_DUP)?);
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
for i in 0..b.num_rows() {
if r.value(i) == repo {
tail_touched.insert(id_hash(spec.stable_keys, repo, krate.value(i), &keys, &dup, i, &mut buf), ());
}
}
}
}
// the interval rows live at `cut`, minus every identity the tail touched
let mut ids: Vec<(u128, i64)> = Vec::new();
let mut buf: Vec<u8> = Vec::with_capacity(256);
for b in interval {
let bi = out.batches.len() as u32;
let (krate, dup) = (str_col(&b, COL_KRATE)?, str_col(&b, COL_KEY_DUP)?);
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(&b, k)).collect::<Result<_>>()?;
let from = i64_col(&b, COL_VALID_FROM_SEQ)?;
let to = i64_col(&b, COL_VALID_TO_SEQ)?;
for i in 0..b.num_rows() {
if from.value(i) > cut || (!to.is_null(i) && to.value(i) <= cut) {
continue;
}
let id = id_hash(spec.stable_keys, repo, krate.value(i), &keys, &dup, i, &mut buf);
if !tail_touched.is_empty() && tail_touched.contains_key(&id) {
continue;
}
out.open.push((bi, i as u32));
ids.push((id, from.value(i)));
}
out.batches.push(b);
}
// the tail's own open rows join behind the interval batches; each carries
// its event's identity and seq for the track
let shift = out.batches.len() as u32;
let mut at = 0usize;
for (bi, b) in tail.batches.iter().enumerate() {
if at >= tail.open.len() || tail.open[at].0 != bi as u32 {
continue;
}
let (krate, dup) = (str_col(b, COL_KRATE)?, str_col(b, COL_KEY_DUP)?);
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
let seq = i64_col(b, COL_SEQ)?;
while at < tail.open.len() && tail.open[at].0 == bi as u32 {
let ri = tail.open[at].1;
out.open.push((bi as u32 + shift, ri));
ids.push((id_hash(spec.stable_keys, repo, krate.value(ri as usize), &keys, &dup, ri as usize, &mut buf), seq.value(ri as usize)));
at += 1;
}
}
out.batches.extend(tail.batches);
out.events_read = tail.events_read;
// open sorts; ids ride along, staying aligned
let mut paired: Vec<((u32, u32), (u128, i64))> = out.open.iter().copied().zip(ids).collect();
paired.sort_unstable_by_key(|p| p.0);
out.open = paired.iter().map(|p| p.0).collect();
let ids = paired.into_iter().map(|p| p.1).collect();
Ok((out, ids))
}
/// What changed in one repo between two seqs ([`EventLog::interval_diff`]):
/// per identity, its cover row at `from` and at `to`, classified. The row
/// references point into `batches` (interval batches first, then event
/// batches): `added` and `changed` reference the `to`-cover, `removed` the
/// `from`-cover.
#[derive(Debug, Clone, Default)]
pub struct IntervalDiff {
pub batches: Vec<RecordBatch>,
pub added: Vec<(u32, u32)>,
pub removed: Vec<(u32, u32)>,
/// The `to`-cover of every changed identity …
pub changed: Vec<(u32, u32)>,
/// … and, in the same order, its `from`-cover (the BEFORE row).
pub changed_from: Vec<(u32, u32)>,
/// Events read past the anchor (0 when the window lies inside it).
pub events_read: u64,
}
impl EventLog {
/// The commit diff `(from, to]` of one repo and source kind, answered
/// from the READ-FAST zone: range predicates over the interval columns
/// classify every version whose validity touches the window, and the
/// event tail past the anchor completes it — never two folds. A repo
/// without a zone is answered from its events alone (the partition-pruned
/// scan). `added` = present at `to`, not at `from`; `removed` the other
/// way; `changed` = present at both under another payload hash (a change
/// REVERTED inside the window is no change).
pub async fn interval_diff(&self, kind: Kind, repo: &str, from: i64, to: i64) -> Result<IntervalDiff> {
if from >= to {
bail!("an empty diff window: `from` {from} is not before `to` {to}");
}
if from < 0 {
bail!("`from` {from} is before the log (seq 0 = nothing consumed)");
}
let log = self.clone();
let repo = repo.to_string();
on(self.shared.rt(), async move { interval_diff(log, kind, &repo, from, to).await }).await
}
}
#[derive(Default)]
struct DiffEntry {
/// The cover at `from`: its row and payload hash.
lo: Option<((u32, u32), crate::delta::PayloadHash)>,
/// The cover at `to`.
hi: Option<((u32, u32), crate::delta::PayloadHash)>,
/// The identity's validity moved inside the window.
touched: bool,
}
async fn interval_diff(log: EventLog, kind: Kind, repo: &str, lo: i64, hi: i64) -> Result<IntervalDiff> {
let spec = log.spec(kind);
let anchor = anchor_of(&log, kind, repo).await?.unwrap_or(0);
let mut out = IntervalDiff::default();
let mut entries: HashMap<u128, DiffEntry, IdHash> = HashMap::with_hasher(IdHash);
let mut buf: Vec<u8> = Vec::with_capacity(256);
let payload_cols: Vec<&str> = spec.schema.fields().iter().map(|f| f.name().as_str()).collect();
// the interval side: one partition-pruned read; a row joins when its
// validity covers an end of the window or moves inside it
if anchor > 0 {
let (xcut, ycut) = (lo.min(anchor), hi.min(anchor));
let name = interval_table_name(kind).context("a source kind")?;
let t = log.shared.wh.table(&qualified(name)).await?;
let mut proj: Vec<&str> = vec![COL_REPO, COL_KRATE, COL_PAYLOAD_HASH, COL_KEY_DUP, COL_VALID_FROM_SEQ, COL_VALID_TO_SEQ];
for c in &payload_cols {
if !proj.contains(c) {
proj.push(c);
}
}
let readable = crate::log::scannable(t.inner(), &proj)?;
// a window inside the zone needs only the versions that START or END
// in it (an identity whose one version spans the window is unchanged
// there): streamed, every other row dropped as read — the repo's
// partition is millions of rows (nornir 5.6 M, most of them MIR
// block payloads) and a world_impact ask reads one window per repo.
// Past the anchor the tail can touch any identity live at the anchor:
// the partition is read whole.
let interval = if hi <= anchor {
let mut kept: Vec<RecordBatch> = Vec::new();
let mut failed: Option<anyhow::Error> = None;
skade::read_filtered_each(t.inner(), Some(&ScanFilter::eq(COL_REPO, repo)), &readable, |b| {
let step = (|| -> Result<()> {
let (from_c, to_c) = (i64_col(&b, COL_VALID_FROM_SEQ)?, i64_col(&b, COL_VALID_TO_SEQ)?);
let mask = skade::arrow_array::BooleanArray::from_iter((0..b.num_rows()).map(|i| {
let f = from_c.value(i);
Some((f > lo && f <= hi) || (!to_c.is_null(i) && to_c.value(i) > lo && to_c.value(i) <= hi))
}));
if mask.true_count() > 0 {
kept.push(arrow_select::filter::filter_record_batch(&b, &mask)?);
}
Ok(())
})();
match step {
Ok(()) => true,
Err(e) => {
failed = Some(e);
false
}
}
})
.await?;
if let Some(e) = failed {
return Err(e);
}
kept
} else {
skade::read_filtered(t.inner(), &ScanFilter::eq(COL_REPO, repo), &readable).await?
};
let interval = crate::log::null_filled(t.inner(), interval, Some(&proj))?;
for b in interval {
let bi = out.batches.len() as u32;
let (krate, dup, hash) = (str_col(&b, COL_KRATE)?, str_col(&b, COL_KEY_DUP)?, str_col(&b, COL_PAYLOAD_HASH)?);
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(&b, k)).collect::<Result<_>>()?;
let from_c = i64_col(&b, COL_VALID_FROM_SEQ)?;
let to_c = i64_col(&b, COL_VALID_TO_SEQ)?;
for i in 0..b.num_rows() {
let f = from_c.value(i);
let t_ = (!to_c.is_null(i)).then(|| to_c.value(i));
let present = |s: i64| f <= s && t_.is_none_or(|t| t > s);
let in_window = (f > lo && f <= hi) || t_.is_some_and(|t| t > lo && t <= hi);
if !present(xcut) && !present(ycut) && !in_window {
continue;
}
let id = id_hash(spec.stable_keys, repo, krate.value(i), &keys, &dup, i, &mut buf);
let e = entries.entry(id).or_default();
if present(xcut) {
e.lo = Some(((bi, i as u32), unhex(hash.value(i))?));
}
if present(ycut) {
e.hi = Some(((bi, i as u32), unhex(hash.value(i))?));
}
if in_window {
e.touched = true;
}
}
out.batches.push(b);
}
}
// the events past the anchor (the whole history without a zone), in seq
// order: an event at s ≤ `from` sets the from-state, every event the
// to-state, one inside the window marks the identity touched
if hi > anchor {
let mut cols = payload_cols.clone();
cols.push(COL_PAYLOAD_HASH);
let batches = if anchor > 0 {
repo_tail(&log, kind, repo, anchor, hi, &cols).await?.batches
} else {
let scope = FoldScope { repos: [(repo.to_string(), hi)].into(), universe: false };
log.scan_rows(kind, &scope, &cols).await?.batches
};
let shift = out.batches.len() as u32;
let mut decoded = Vec::with_capacity(batches.len());
for b in &batches {
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
decoded.push((str_col(b, COL_REPO)?, i64_col(b, COL_SEQ)?, str_col(b, COL_OP)?, str_col(b, COL_KRATE)?, str_col(b, COL_PAYLOAD_HASH)?, str_col(b, COL_KEY_DUP)?, keys));
}
let mut order: Vec<(i64, u32, u32)> = Vec::new();
for (bi, (r, seq, ..)) in decoded.iter().enumerate() {
for i in 0..r.len() {
if r.value(i) == repo && seq.value(i) <= hi && seq.value(i) > anchor {
order.push((seq.value(i), bi as u32, i as u32));
}
}
}
order.sort_unstable();
out.events_read = order.len() as u64;
for (s, bi, ri) in order {
let (_, _, op, krate, hash, dup, keys) = &decoded[bi as usize];
let i = ri as usize;
let id = id_hash(spec.stable_keys, repo, krate.value(i), keys, dup, i, &mut buf);
let state = if op.value(i) == OP_PLUS {
if hash.is_null(i) {
bail!("a `+` event of {} has no payload_hash", EventLog::table_name(kind));
}
Some(((shift + bi, ri), unhex(hash.value(i))?))
} else {
None
};
let e = entries.entry(id).or_default();
if s <= lo {
e.lo = state;
} else {
e.touched = true;
}
e.hi = state;
}
out.batches.extend(batches);
}
for e in entries.into_values() {
if !e.touched {
continue;
}
match (e.lo, e.hi) {
(None, Some((r, _))) => out.added.push(r),
(Some((r, _)), None) => out.removed.push(r),
(Some((rf, ha)), Some((r, hb))) if ha != hb => {
out.changed.push(r);
out.changed_from.push(rf);
}
_ => {}
}
}
out.added.sort_unstable();
out.removed.sort_unstable();
let mut both: Vec<((u32, u32), (u32, u32))> = out.changed.iter().copied().zip(out.changed_from.iter().copied()).collect();
both.sort_unstable();
(out.changed, out.changed_from) = both.into_iter().unzip();
Ok(out)
}
impl EventLog {
/// Every interval row of `repo` for `kind` (open and closed) with the
/// identity prefix, the interval and `cols` — the partition read in the
/// order an as-of read ([`EventLog::readfast_rows`]) sees it — and the
/// seq the rows are complete to. `None` = the repo has no zone for the
/// kind (its rows live in the events alone).
pub async fn interval_rows_at(&self, kind: Kind, repo: &str, cols: &[&str]) -> Result<Option<(i64, Vec<RecordBatch>)>> {
let log = self.clone();
let repo = repo.to_string();
let cols: Vec<String> = cols.iter().map(|c| c.to_string()).collect();
on(self.shared.rt(), async move {
let Some(anchor) = anchor_of(&log, kind, &repo).await? else { return Ok(None) };
let spec = log.spec(kind);
let name = interval_table_name(kind).expect("anchored kinds are source kinds");
let t = log.shared.wh.table(&qualified(name)).await?;
let mut proj: Vec<&str> = vec![COL_REPO, COL_KRATE, COL_PAYLOAD_HASH, COL_KEY_DUP, COL_VALID_FROM_SEQ, COL_VALID_TO_SEQ];
for c in spec.key_columns.iter().map(String::as_str).chain(cols.iter().map(String::as_str)) {
if !proj.contains(&c) {
proj.push(c);
}
}
let readable = crate::log::scannable(t.inner(), &proj)?;
let rows = skade::read_filtered(t.inner(), &ScanFilter::eq(COL_REPO, repo.as_str()), &readable).await?;
Ok(Some((anchor, crate::log::null_filled(t.inner(), rows, Some(&proj))?)))
})
.await
}
}