crates/mimir-ingest/src/log.rs
//! The event log in skade: the `mimir_v2` tables (three event tables per
//! producer stream + bookkeeping), the one atomic commit per (repo, sha), and
//! the fold that derives state (the open set, valid_from / valid_to) from the
//! events.
use std::collections::{BTreeMap, HashSet};
use std::future::Future;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, RwLock};
use std::time::{SystemTime, UNIX_EPOCH};
use anyhow::{Context, Result, bail};
use skade::arrow_array::{
Array, ArrayRef, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray,
};
use skade::arrow_schema::{DataType, Field, Schema};
use skade::iceberg::Catalog;
use skade::iceberg::table::Table as IceTable;
use skade::{ScanFilter, Warehouse, WriteProps};
use crate::delta::{Kind, OpenSet, PayloadHash, Stream, push_stable, stable_into, unhex};
use crate::schema::{
OP_PLUS,
BACKFILL_RUNS, BIN_EDGE_EVENTS, BIN_INGEST_RUNS, BIN_NODE_EVENTS, BIN_OFFSETS, backfill_runs_schema, bin_offsets_schema, CODE_EDGE_EVENTS, CODE_NODE_EVENTS, CODE_SPAN_EVENTS, CODE_SYMBOL_EVENTS,
COL_KEY_DUP, COL_KRATE, COL_OP, COL_PAYLOAD_HASH, COL_REPO, COL_REV, COL_SEQ, CONSUMER_OFFSETS, FEED_RUNS, INGEST_RUNS, KEY_NULL,
KEY_SEP, PROP_ADDED, PROP_RING, PROP_SCHEMA, PayloadSpec, RING_SPLIT, Specs, UNIVERSE_EDGE_EVENTS, UNIVERSE_MANIFEST, UNIVERSE_NODE_EVENTS, UNIVERSE_REPO, UNIVERSE_SPAN_EVENTS, bin_ingest_runs_schema, consumer_offsets_schema,
feed_runs_schema, ingest_runs_schema, qualified, universe_manifest_schema,
};
/// A repo's consumer offset: the last first-parent commit consumed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Offset {
pub branch: String,
pub sha: String,
pub seq: i64,
}
/// One ledger row of `ingest_runs`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunRow {
pub repo: String,
pub seq: i64,
pub sha: String,
pub status: String,
pub crates_dumped: i64,
pub rows_plus: i64,
pub rows_minus: i64,
/// Keys the dump repeated with different payloads (kept, disambiguated).
pub key_collisions: i64,
/// Position events (`code_span_events`), apart from the semantic ones.
pub span_plus: i64,
pub span_minus: i64,
/// Owner ids that needed a feature digest / hash to stay apart.
pub owner_disambiguations: i64,
pub ms: i64,
pub error: Option<String>,
}
/// One ledger row of `consumer_feed_runs`: a non-graph feed (git heat, sbom, code
/// index, embeddings …) run for one consumed (repo, sha).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FeedRow {
pub repo: String,
/// The consumer ordinal of `sha`.
pub seq: i64,
pub sha: String,
/// The feed's name (its key in the ledger).
pub feed: String,
/// `ok` · `error` · `skipped` (could not run here; `detail` says why).
pub status: String,
/// What it wrote (files scanned, components, vectors …).
pub rows: i64,
pub ms: i64,
pub detail: Option<String>,
}
/// One row of `backfill_runs`: a member's history asked for from a commit, or
/// the word that the consumer got from there to the tip.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BackfillRow {
pub repo: String,
/// The first commit of the history asked for.
pub from_sha: String,
/// The member's consumer ordinal when it was asked (0: never consumed).
pub base_seq: i64,
/// The member's tip when it was asked.
pub target_sha: String,
/// First-parent commits from `from_sha` to that tip.
pub total: i64,
/// `requested` · `done` (`crate::schema::STATUS_REQUESTED` / `STATUS_DONE`).
pub status: String,
pub detail: Option<String>,
}
impl BackfillRow {
/// Asked for and not yet walked to the tip.
pub fn is_open(&self) -> bool {
self.status == crate::schema::STATUS_REQUESTED
}
}
/// One ledger row of `bin_ingest_runs` (the binary stream's ledger).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BinRunRow {
pub repo: String,
/// The consumer ordinal of `sha` (the source stream's seq).
pub seq: i64,
/// The binary release ordinal of the repo.
pub rev: i64,
pub sha: String,
/// `name@target-triple:profile`.
pub artifact: String,
pub status: String,
pub rows_plus: i64,
pub rows_minus: i64,
pub key_collisions: i64,
pub ms: i64,
/// mimir-binary's join ledger (JSON).
pub ledger: Option<String>,
}
/// An artifact's offset in the binary stream (`bin_offsets`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BinOffset {
pub rev: i64,
pub seq: i64,
pub sha: String,
}
/// One new `universe_manifest` row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct UniverseRow {
pub krate: String,
pub toolchain: String,
pub rows: i64,
pub rustc_release: Option<String>,
}
/// The snapshot the open-set cache was derived from: the current snapshot id
/// of each event table ([`Kind::ALL`] order). Any commit that adds events
/// moves it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Stamp(pub [Option<i64>; 6]);
/// An identity whose last event closed it holds this slot.
pub(crate) const CLOSED: u32 = u32::MAX;
/// What a tracked fold keeps so a DELTA can advance it later
/// ([`EventLog::fold_stream_tracked`] / [`EventLog::fold_stream_delta`]):
/// per identity its latest event and the slot the caller's records hold it
/// at, plus the table snapshots the fold read (where the next delta starts).
pub struct FoldTrack {
/// identity → its latest event: (seq, rev, slot or [`CLOSED`]).
pub(crate) last: std::collections::HashMap<u128, (i64, i32, u32), IdHash>,
/// The caller's next record slot.
pub(crate) next: u32,
/// Event table → its snapshot id when last read.
pub(crate) read_at: std::collections::BTreeMap<&'static str, i64>,
}
impl FoldTrack {
/// The live slots, sorted — the caller's open set.
pub fn open(&self) -> Vec<u32> {
let mut open: Vec<u32> = self.last.values().filter_map(|e| (e.2 != CLOSED).then_some(e.2)).collect();
open.sort_unstable();
open
}
/// Identities tracked (open and closed) — the garbage bound: records
/// minus the open set is what a periodic fresh fold reclaims.
pub fn len(&self) -> usize {
self.last.len()
}
/// No identity tracked at all.
pub fn is_empty(&self) -> bool {
self.last.is_empty()
}
/// The caller stored records OUTSIDE the tracked fold (the binary
/// stream's rows beside a source fold): the next delta's slots start
/// after them.
pub fn skip_to(&mut self, next: u32) {
debug_assert!(next >= self.next, "slots never go back");
self.next = next;
}
}
/// What [`EventLog::fold_stream_delta`] did.
#[derive(Debug)]
pub enum FoldDelta {
/// The delta applied: the new open set and the events it read.
Advanced { open: Vec<u32>, events: u64 },
/// The window holds a commit a delta cannot express — fold afresh.
Full,
}
/// Everything one release commits — in ONE skade atomic release. A consumer
/// step (repo, sha) carries the offset + ledger row and, with the binary
/// layer, the binary stream's events + ledger rows too; a binary-only
/// re-ingest carries only the binary part; the toolchain dump only universe
/// events + manifest rows.
pub(crate) struct CommitWrite {
/// What the release is about (repo or `@universe`, sha) — for messages.
pub repo: String,
pub sha: String,
pub offset: Option<Offset>,
pub run: Option<RunRow>,
pub bin_runs: Vec<BinRunRow>,
/// Event batches per table; empty = no snapshot. Each table at most once.
pub events: Vec<(Kind, Vec<RecordBatch>)>,
/// The ring's event batches, per ring table ([`EventLog::ring_table_name`]):
/// a split lake's `@universe` rows. Empty on a legacy lake (its ring rows
/// ride `events`).
pub ring_events: Vec<(Kind, Vec<RecordBatch>)>,
/// Newly dumped universe crates.
pub universe: Vec<UniverseRow>,
}
/// The Mímir v2 change-event log.
///
/// A HANDLE: cheap to clone, every clone shares the one open lake. The handle
/// model (wiring map R13):
///
/// * **Many readers.** Every read method takes `&self`; any number of clones
/// read concurrently, from any thread, while a release is being written.
/// [`EventLog::pin`] gives a reader ONE committed state of the whole lake
/// (every table as of the same release), so a fold that takes seconds
/// never sees half a release.
/// * **One writer.** Releases go through the [`LogWriter`] seat
/// ([`EventLog::writer`]); a second claim while one is held is an error,
/// not a queue. The consumer holds the seat for its lifetime.
/// * **One process.** The lake is a skade warehouse: its redb catalog is
/// file-locked, so a second PROCESS cannot open a lake this one holds —
/// [`EventLog::open_lake`] says so by name ([`LakeHeld`]). Other processes
/// read through the holder (the server's RPCs); there is no cross-process
/// read path, and no copy-aside snapshot is taken behind the holder's back.
/// * **One runtime, the lake's own.** An Iceberg table handle captures the
/// tokio runtime it was built on and runs its scans there; the catalog
/// caches handles. A handle built on a caller's runtime would die with it
/// (or stall while that runtime's one thread sits in a `cargo check`), and
/// every later reader would inherit it from the cache. So the log owns a
/// small multi-thread runtime: every table load and every release is run
/// on it, whichever runtime — or plain thread ([`EventLog::block_on`]) —
/// the caller is on.
#[derive(Clone)]
pub struct EventLog {
pub(crate) shared: Arc<Shared>,
/// `Some` = a read handle bound to one committed state ([`EventLog::pin`]).
pin: Option<Arc<Pinned>>,
}
/// What every handle of one open lake shares.
pub(crate) struct Shared {
pub(crate) wh: Warehouse,
specs: Specs,
props: WriteProps,
/// (crate, toolchain) pairs in `universe_manifest`: read once at open,
/// extended by every release this process commits.
universe: RwLock<HashSet<(String, String)>>,
/// repo → its last binary release (`rev`), read once from `bin_offsets`
/// and moved by every release this process commits.
revs: Mutex<std::collections::HashMap<String, i64>>,
/// The writer seat ([`LogWriter`]).
pub(crate) seat: AtomicBool,
/// The ring lives in its own tables and the source tables are written per
/// repo (`mimir.ring = split`, read at open, set by the split).
pub(crate) ring_split: AtomicBool,
/// Odd while a catalog commit is in flight, bumped again when it is done:
/// a reader whose table loads saw no movement holds one committed state
/// without asking the catalog twice.
pub(crate) commits: AtomicU64,
/// The catalog's pointer generation when the release in flight began: a
/// reader that still sees this generation while `commits` is odd is
/// looking at the state before that release, whole.
pub(crate) release_base: AtomicU64,
/// ONE catalog commit at a time: a release, a ledger row, the
/// compaction's replace. `commits` / `release_base` describe one commit
/// in flight, so two must never overlap.
pub(crate) commit_lock: tokio::sync::Mutex<()>,
/// The event tables' UUIDs (minted at creation, fixed for the handle).
identity: OnceLock<String>,
/// The lake directory this log was opened from ([`EventLog::open_lake`]);
/// `None` for a warehouse handed in. A job that writes or deletes files
/// by a table's `location` refuses a table that lies outside it.
pub(crate) root: Option<std::path::PathBuf>,
/// The lake's runtime (see [`EventLog`]). `None` only while dropping.
rt: Option<tokio::runtime::Runtime>,
}
impl Shared {
pub(crate) fn rt(&self) -> &tokio::runtime::Runtime {
self.rt.as_ref().expect("the lake's runtime lives as long as its handles")
}
}
impl Drop for Shared {
fn drop(&mut self) {
// the last handle may go away inside an async task, where a runtime
// must not block on its own shutdown
if let Some(rt) = self.rt.take() {
rt.shutdown_background();
}
}
}
/// The lake's runtime: enough workers that a reader's scan and the writer's
/// release never queue behind each other.
fn lake_runtime() -> Result<tokio::runtime::Runtime> {
tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.thread_name("mimir-lake")
.enable_all()
.build()
.context("start the lake's runtime")
}
/// Run `fut` on `rt` from wherever the caller is polled.
pub(crate) async fn on<T: Send + 'static>(rt: &tokio::runtime::Runtime, fut: impl Future<Output = Result<T>> + Send + 'static) -> Result<T> {
rt.spawn(fut).await.context("a task on the lake's runtime was dropped")?
}
/// One committed state of the lake: every table as loaded between two releases.
struct Pinned {
tables: BTreeMap<&'static str, IceTable>,
}
/// Every table of the lake ([`EventLog::pin`] loads exactly these), the
/// unqualified names (`mimir_v2.<name>` in the catalog).
pub const TABLE_NAMES: [&str; 16] = [
UNIVERSE_NODE_EVENTS,
UNIVERSE_EDGE_EVENTS,
UNIVERSE_SPAN_EVENTS,
CODE_NODE_EVENTS,
CODE_EDGE_EVENTS,
CODE_SPAN_EVENTS,
BIN_NODE_EVENTS,
BIN_EDGE_EVENTS,
CODE_SYMBOL_EVENTS,
CONSUMER_OFFSETS,
INGEST_RUNS,
BIN_INGEST_RUNS,
BIN_OFFSETS,
UNIVERSE_MANIFEST,
FEED_RUNS,
BACKFILL_RUNS,
];
pub(crate) const TABLES: [&str; 16] = TABLE_NAMES;
/// The lake is held by another process (redb's file lock on its catalog).
/// Not an IO failure: the holder is the reader to ask.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LakeHeld {
pub lake: std::path::PathBuf,
}
impl std::fmt::Display for LakeHeld {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"the lake {} is already open — by another process (a lake is single-process: read it through that process, the server's RPCs) or by another handle in this one",
self.lake.display()
)
}
}
impl std::error::Error for LakeHeld {}
/// The lake's tables differ from this build's schema in a way that is not
/// evolved in place. What IS evolved, on open and without a word: a table that
/// only LACKS nullable columns gets them added (Iceberg schema evolution; the
/// rows written before read them as null). Everything else — a column this
/// build does not have (dropped or renamed here, or the lake was written by a
/// newer build), a column of another type, another sort order — is refused:
/// the lake is derived data, delete its directory and the consumer rebuilds it
/// from each repo's tip.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LakeSchemaMismatch {
/// The first table that differs (`mimir_v2.<table>`).
pub table: String,
pub have: Vec<String>,
pub want: Vec<String>,
pub why: Unevolvable,
}
/// Why a lake's table cannot be brought to this build's schema in place.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Unevolvable {
/// The table has columns this build does not (and maybe lacks others).
Columns,
/// A column this build needs is missing and cannot be added: it is not
/// nullable, or it is part of the key.
Required { column: String },
/// A column both sides have is of another type.
Retyped { column: String, have: String, want: String },
/// The lake was written at a newer `mimir-schema` than this build's.
Newer { lake: String, build: String },
/// The lake was written before [`IDENTITY_SCHEMA`]: its rows key and own
/// by VERSIONED names, this build by stable ones — not a column to add.
/// [`EventLog::open_lake`] answers this one itself: the lake (derived
/// data) is wiped and the consumer re-reads from each repo's tip.
Older { lake: String, build: String },
/// The table's sort order is another one.
SortOrder,
}
impl std::fmt::Display for LakeSchemaMismatch {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let only = |a: &[String], b: &[String]| a.iter().filter(|c| !b.contains(c)).cloned().collect::<Vec<_>>().join(", ");
write!(f, "this lake was written with another schema: {} ", self.table)?;
match &self.why {
Unevolvable::Columns | Unevolvable::SortOrder => {
write!(f, "has [{}] this build does not, and lacks [{}]", only(&self.have, &self.want), only(&self.want, &self.have))?
}
Unevolvable::Required { column } => write!(f, "lacks `{column}`, which cannot be added to rows that exist")?,
Unevolvable::Retyped { column, have, want } => write!(f, "has `{column}` as {have}, this build as {want}")?,
Unevolvable::Newer { lake, build } => write!(f, "is at mimir-schema {lake}, this build at {build}")?,
Unevolvable::Older { lake, build } => {
write!(f, "is at mimir-schema {lake}, before the stable identities of {build} — its rows key by versioned names")?
}
}
f.write_str(" — only added nullable columns are evolved in place; a lake that differs otherwise is not migrated: delete the lake directory and the consumer rebuilds it from each repo's tip")
}
}
impl std::error::Error for LakeSchemaMismatch {}
/// The log already has its one writer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WriterSeatTaken;
impl std::fmt::Display for WriterSeatTaken {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("this log already has its one writer (a consumer holds the seat)")
}
}
impl std::error::Error for WriterSeatTaken {}
impl EventLog {
/// Open (creating where missing) the tables in namespace `mimir_v2`: the
/// six event tables (three per stream) + offsets, both ledgers and the
/// universe manifest. An existing table that only lacks nullable columns
/// of the expected schema is EVOLVED in place (the columns are added, the
/// rows written before read them as null); any other difference is the
/// typed [`LakeSchemaMismatch`], not a silent append into the wrong shape.
/// Both edge event tables declare an Iceberg sort order on `from` (their
/// rows are written sorted by `from` too).
pub async fn open(wh: Warehouse, specs: Specs) -> Result<Self> {
Self::open_on(lake_runtime()?, wh, specs).await
}
async fn open_on(rt: tokio::runtime::Runtime, wh: Warehouse, specs: Specs) -> Result<Self> {
Self::open_rooted(rt, wh, specs, None).await
}
async fn open_rooted(rt: tokio::runtime::Runtime, wh: Warehouse, specs: Specs, root: Option<std::path::PathBuf>) -> Result<Self> {
Self::open_mode(rt, wh, specs, root, false).await
}
/// Test hook: open (create) a lake in the LEGACY shape — ring rows in
/// the source tables, no partition — the shape every lake written before
/// the split has, so the split and the compaction that follows can be
/// exercised on a fixture.
#[doc(hidden)]
pub async fn open_legacy_for_test(wh: Warehouse, specs: Specs) -> Result<Self> {
Self::open_mode(lake_runtime()?, wh, specs, None, true).await
}
async fn open_mode(rt: tokio::runtime::Runtime, wh: Warehouse, mut specs: Specs, root: Option<std::path::PathBuf>, legacy: bool) -> Result<Self> {
// table handles are created (and cached by the catalog) here: on the
// lake's runtime like every later one
let opened = on(&rt, async move {
// the mode before anything is created: a lake with a node table
// that says `split`, or a lake born now (split unless legacy)
let node_ident = wh.table_ident(&qualified(CODE_NODE_EVENTS))?;
let existed = wh.catalog().table_exists(&node_ident).await?;
let split = if existed {
wh.table(&qualified(CODE_NODE_EVENTS)).await?.properties().get(PROP_RING).map(String::as_str) == Some(RING_SPLIT)
} else {
!legacy
};
for kind in Kind::ALL {
let sort: &[&str] = if matches!(kind, Kind::Edges | Kind::BinEdges) { &[EDGE_SORT_COLUMN] } else { &[] };
let spec = specs.get_mut(kind);
let schema = spec.event_schema_with(kind.stream() == Stream::Binary);
// a source table born split is partitioned by repo from its first file
let partition: &[&str] = if split && kind.stream() == Stream::Source { &[COL_REPO] } else { &[] };
let added = ensure(&wh, Self::table_name(kind), &schema, sort, partition, Evolves::Payload { spec: &*spec }).await?;
// the lake's own hash rule: what was added in place counts
// only where it has a value
spec.added = added.into_iter().filter(|c| spec.schema.field_with_name(c).is_ok()).collect();
if let Some(ring) = Self::ring_table_name(kind) {
ensure(&wh, ring, &schema, sort, &[], Evolves::Payload { spec: &*spec }).await?;
}
}
if split && !existed {
let mut t = wh.table(&qualified(CODE_NODE_EVENTS)).await?;
t.set_properties([(PROP_RING, RING_SPLIT)]).await.context("stamp the lake split")?;
}
ensure(&wh, CONSUMER_OFFSETS, &consumer_offsets_schema(), &[], &[], Evolves::Nullable).await?;
ensure(&wh, INGEST_RUNS, &ingest_runs_schema(), &[], &[], Evolves::Nullable).await?;
ensure(&wh, BIN_INGEST_RUNS, &bin_ingest_runs_schema(), &[], &[], Evolves::Nullable).await?;
ensure(&wh, BIN_OFFSETS, &bin_offsets_schema(), &[], &[], Evolves::Nullable).await?;
ensure(&wh, UNIVERSE_MANIFEST, &universe_manifest_schema(), &[], &[], Evolves::Nullable).await?;
ensure(&wh, FEED_RUNS, &feed_runs_schema(), &[], &[], Evolves::Nullable).await?;
ensure(&wh, BACKFILL_RUNS, &backfill_runs_schema(), &[], &[], Evolves::Nullable).await?;
Ok((wh, specs, split))
})
.await;
let (wh, specs, split) = match opened {
Ok(x) => x,
Err(e) => {
// a refused lake: a runtime must not be dropped in an async task
rt.shutdown_background();
return Err(e);
}
};
let probe = Self {
shared: Arc::new(Shared {
wh,
specs,
props: write_props()?,
universe: RwLock::new(HashSet::new()),
revs: Mutex::new(Default::default()),
seat: AtomicBool::new(false),
ring_split: AtomicBool::new(split),
commits: AtomicU64::new(0),
release_base: AtomicU64::new(0),
commit_lock: tokio::sync::Mutex::new(()),
identity: OnceLock::new(),
root,
rt: Some(rt),
}),
pin: None,
};
let universe = probe.read_universe().await?;
*probe.shared.universe.write().unwrap_or_else(|e| e.into_inner()) = universe;
Ok(probe)
}
/// Open the lake at `dir` (created when missing) with the mimir-schema
/// payloads. A lake another process holds is the typed [`LakeHeld`]
/// (downcast the error), not a generic IO failure. A lake stamped before
/// [`IDENTITY_SCHEMA`] ([`Unevolvable::Older`]) is WIPED here — it is
/// derived data whose rows key by versioned names — and reopened fresh;
/// the consumer then re-reads every repo from its tip.
pub async fn open_lake(dir: &std::path::Path) -> Result<Self> {
match Self::open_lake_once(dir).await {
Err(e) if e.downcast_ref::<LakeSchemaMismatch>().is_some_and(|m| matches!(m.why, Unevolvable::Older { .. })) => {
std::fs::remove_dir_all(dir).with_context(|| format!("wipe the pre-stable-identity lake {}", dir.display()))?;
Self::open_lake_once(dir).await
}
r => r,
}
}
async fn open_lake_once(dir: &std::path::Path) -> Result<Self> {
std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
let rt = lake_runtime()?;
let lake = dir.to_path_buf();
let opened = rt.spawn(async move { skade::open(&lake).await }).await.context("a task on the lake's runtime was dropped");
let wh = match opened {
Ok(Ok(wh)) => wh,
Ok(Err(e)) => {
rt.shutdown_background();
return Err(if lake_is_held(dir, &e) {
anyhow::Error::new(LakeHeld { lake: dir.to_path_buf() })
} else {
anyhow::Error::new(e).context(format!("open the lake {}", dir.display()))
});
}
Err(e) => {
rt.shutdown_background();
return Err(e);
}
};
let root = std::fs::canonicalize(dir).unwrap_or_else(|_| dir.to_path_buf());
Self::open_rooted(rt, wh, Specs::mimir()?, Some(root)).await
}
/// [`EventLog::open_lake`] for a caller with no async context. Callable
/// from any thread (it does the open on a thread of its own).
pub fn open_lake_blocking(dir: &std::path::Path) -> Result<Self> {
std::thread::scope(|s| {
s.spawn(|| {
// only polls the outer future: every table handle is built on
// the lake's runtime inside `open_lake`
tokio::runtime::Builder::new_current_thread().enable_all().build().context("start the open")?.block_on(Self::open_lake(dir))
})
.join()
.unwrap_or_else(|p| std::panic::resume_unwind(p))
})
}
/// Run `fut` to completion on the lake's runtime, blocking the calling
/// thread — the entry for callers with no async context (a plain thread,
/// a blocking-pool thread). Never call it from an async task.
pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
self.shared.rt().block_on(fut)
}
/// The identity of THIS log: the Iceberg table UUIDs of every event
/// table (minted when the table was created). A wiped and re-created lake
/// has a different identity — whatever was derived for the old one (the
/// dumper's mirror) must not be trusted for it.
pub async fn identity(&self) -> Result<String> {
if let Some(id) = self.shared.identity.get() {
return Ok(id.clone());
}
let mut ids = Vec::new();
for kind in Kind::ALL {
ids.push(self.tbl(Self::table_name(kind)).await?.metadata().uuid().to_string());
}
Ok(self.shared.identity.get_or_init(|| ids.join(",")).clone())
}
/// What a dumper's kept dumps (its mirror) belong to: THIS log at THIS
/// payload schema. A lake evolved in place keeps its identity, but a dump
/// taken before the evolution says nothing about the new columns — a
/// dumper attached under the new schema finds another stamp, discards
/// what it kept and dumps every unit again, once.
pub async fn dump_identity(&self) -> Result<String> {
let id = self.identity().await?;
Ok(match self.spec(Kind::Nodes).version() {
Some(v) => format!("{id}@{v}"),
None => id,
})
}
/// A READ handle bound to ONE committed state of the whole lake: every
/// read through it (offsets, ledgers, folds, the stamp) sees the tables as
/// of the same release, however many releases land meanwhile. Never waits
/// for the writer — a release in flight is simply not in the state yet.
/// Pinning a pinned handle returns the same state.
pub async fn pin(&self) -> Result<EventLog> {
if self.pin.is_some() {
return Ok(self.clone());
}
let catalog = self.shared.wh.catalog();
for look in 0..PIN_LOOKS {
let (s1, g1) = (self.shared.commits.load(Ordering::Acquire), catalog.pointer_generation());
let tables = self.load_tables().await?;
let (s2, g2) = (self.shared.commits.load(Ordering::Acquire), catalog.pointer_generation());
let whole = s1 == s2
&& if s1 % 2 == 0 {
// no release began or ended while the tables loaded
true
} else {
// one release is being written the whole time, and none of
// its table pointers is published yet: the state before it
g1 == g2 && g1 == self.shared.release_base.load(Ordering::Acquire)
};
if whole {
return Ok(EventLog { shared: self.shared.clone(), pin: Some(Arc::new(Pinned { tables })) });
}
// a release is publishing its pointers (microseconds): look again
if look < 8 {
tokio::task::yield_now().await;
} else {
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
}
}
bail!("no committed state of the lake held still for {PIN_LOOKS} looks")
}
/// Is this a pinned read handle ([`EventLog::pin`])?
pub fn is_pinned(&self) -> bool {
self.pin.is_some()
}
/// Claim the writer seat. One writer per log: a second claim while the
/// first [`LogWriter`] lives is [`WriterSeatTaken`]. A pinned handle is a
/// reader and cannot claim it.
pub fn writer(&self) -> Result<LogWriter> {
if self.pin.is_some() {
bail!("a pinned read handle cannot write the log");
}
if self.shared.seat.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire).is_err() {
return Err(anyhow::Error::new(WriterSeatTaken));
}
Ok(LogWriter { log: self.clone() })
}
/// Does a writer hold the seat right now?
pub fn has_writer(&self) -> bool {
self.shared.seat.load(Ordering::Acquire)
}
/// What `dump` — the dump of a tree that is not a commit — WOULD change
/// in `repo` ([`crate::consumer::DryDelta`]), against the repo's open state
/// folded from this handle. Read-only, and it needs no consumer: any
/// reader may ask, also while the consumer is mid-commit (pin first for
/// "as of the last release").
pub async fn dry_delta(&self, repo: &str, dump: &crate::dump::CommitDump) -> Result<crate::consumer::DryDelta> {
let open = self.fold(Some(repo), None).await?;
crate::consumer::dry_delta(self, &open, repo, dump)
}
/// The `universe_manifest` snapshot (moves when the ring is written).
pub async fn manifest_snapshot(&self) -> Result<Option<i64>> {
Ok(snapshot_id(&self.tbl(UNIVERSE_MANIFEST).await?))
}
/// Table `name` (unqualified): the pinned state's, or the catalog's now.
pub(crate) async fn tbl(&self, name: &'static str) -> Result<IceTable> {
if let Some(p) = &self.pin {
return p.tables.get(name).cloned().with_context(|| format!("`{name}` is not a table of the lake"));
}
let shared = self.shared.clone();
on(self.shared.rt(), async move { Ok(shared.wh.table(&qualified(name)).await?.inner().clone()) }).await
}
async fn load_tables(&self) -> Result<BTreeMap<&'static str, IceTable>> {
let shared = self.shared.clone();
on(self.shared.rt(), async move {
let mut out = BTreeMap::new();
for name in TABLES {
out.insert(name, shared.wh.table(&qualified(name)).await?.inner().clone());
}
Ok(out)
})
.await
}
/// The event table schema of `kind` (binary-stream tables carry `rev`).
pub fn event_schema(&self, kind: Kind) -> Schema {
self.spec(kind).event_schema_with(kind.stream() == Stream::Binary)
}
/// The lake itself. Reads through it see the catalog NOW, not a pinned
/// handle's state — and a table handle loaded through it is built on the
/// CALLER's runtime: load tables through the log's own methods instead.
pub fn warehouse(&self) -> &Warehouse {
&self.shared.wh
}
/// The Parquet knobs the log writes with (zstd; see `write_props`).
pub fn write_props(&self) -> &WriteProps {
&self.shared.props
}
pub fn spec(&self, kind: Kind) -> &PayloadSpec {
let specs = &self.shared.specs;
match kind {
Kind::Nodes => &specs.nodes,
Kind::Edges => &specs.edges,
Kind::Spans => &specs.spans,
Kind::BinNodes => &specs.bin_nodes,
Kind::BinEdges => &specs.bin_edges,
Kind::Symbols => &specs.symbols,
}
}
/// The event table of `kind`.
pub fn table_name(kind: Kind) -> &'static str {
match kind {
Kind::Nodes => CODE_NODE_EVENTS,
Kind::Edges => CODE_EDGE_EVENTS,
Kind::Spans => CODE_SPAN_EVENTS,
Kind::BinNodes => BIN_NODE_EVENTS,
Kind::BinEdges => BIN_EDGE_EVENTS,
Kind::Symbols => CODE_SYMBOL_EVENTS,
}
}
/// The ring table of a source `kind` (`None` for the binary stream).
pub fn ring_table_name(kind: Kind) -> Option<&'static str> {
match kind {
Kind::Nodes => Some(UNIVERSE_NODE_EVENTS),
Kind::Edges => Some(UNIVERSE_EDGE_EVENTS),
Kind::Spans => Some(UNIVERSE_SPAN_EVENTS),
Kind::BinNodes | Kind::BinEdges | Kind::Symbols => None,
}
}
/// Does the ring live in its own tables (and are the source tables
/// written per repo)? `false` = the legacy shape, until
/// [`EventLog::split_ring`] has copied, verified and switched.
pub fn ring_split(&self) -> bool {
self.shared.ring_split.load(Ordering::Acquire)
}
pub(crate) fn set_ring_split(&self) {
self.shared.ring_split.store(true, Ordering::Release);
}
/// The table a scan of `kind` for `repo` reads: the ring table for
/// `@universe` on a split lake, the source table otherwise.
pub(crate) fn table_for(&self, kind: Kind, repo: &str) -> &'static str {
match Self::ring_table_name(kind) {
Some(ring) if repo == UNIVERSE_REPO && self.ring_split() => ring,
_ => Self::table_name(kind),
}
}
/// The ring table of `kind` when the ring lives there (a split lake).
fn ring_tbl_name(&self, kind: Kind) -> Option<&'static str> {
Self::ring_table_name(kind).filter(|_| self.ring_split())
}
/// The binary stream's ledger rows of `repo`, in (rev, artifact) order.
pub async fn bin_runs(&self, repo: &str) -> Result<Vec<BinRunRow>> {
let t = self.tbl(BIN_INGEST_RUNS).await?;
let batches = skade::read_filtered(&t, &ScanFilter::eq("repo", repo), &[]).await?;
let mut rows = Vec::new();
for b in &batches {
let (r, seq, rev, sha) = (str_col(b, "repo")?, i64_col(b, "seq")?, i64_col(b, "rev")?, str_col(b, "sha")?);
let (art, st, led) = (str_col(b, "artifact")?, str_col(b, "status")?, str_col(b, "ledger")?);
let (rp, rm, kc, ms) = (i64_col(b, "rows_plus")?, i64_col(b, "rows_minus")?, i64_col(b, "key_collisions")?, i64_col(b, "ms")?);
for i in 0..b.num_rows() {
if r.value(i) != repo {
continue;
}
rows.push(BinRunRow {
repo: r.value(i).into(),
seq: seq.value(i),
rev: rev.value(i),
sha: sha.value(i).into(),
artifact: art.value(i).into(),
status: st.value(i).into(),
rows_plus: rp.value(i),
rows_minus: rm.value(i),
key_collisions: kc.value(i),
ms: ms.value(i),
ledger: (!led.is_null(i)).then(|| led.value(i).to_string()),
});
}
}
rows.sort_by(|a, b| (a.rev, &a.artifact).cmp(&(b.rev, &b.artifact)));
Ok(rows)
}
/// Per artifact of `repo`: its offset (the row with the highest rev).
pub async fn bin_offsets(&self, repo: &str) -> Result<BTreeMap<String, BinOffset>> {
let t = self.tbl(BIN_OFFSETS).await?;
let batches = skade::read_filtered(&t, &ScanFilter::eq("repo", repo), &["repo", "artifact", "rev", "seq", "sha"]).await?;
let mut out: BTreeMap<String, BinOffset> = BTreeMap::new();
for b in &batches {
let (r, a, rev, seq, sha) = (str_col(b, "repo")?, str_col(b, "artifact")?, i64_col(b, "rev")?, i64_col(b, "seq")?, str_col(b, "sha")?);
for i in 0..b.num_rows() {
if r.value(i) != repo {
continue;
}
let o = BinOffset { rev: rev.value(i), seq: seq.value(i), sha: sha.value(i).to_string() };
match out.get(a.value(i)) {
Some(prev) if prev.rev >= o.rev => {}
_ => {
out.insert(a.value(i).to_string(), o);
}
}
}
}
Ok(out)
}
/// Is `krate` already in the universe manifest under `toolchain`?
pub fn universe_has(&self, krate: &str, toolchain: &str) -> bool {
let u = self.shared.universe.read().unwrap_or_else(|e| e.into_inner());
u.contains(&(krate.to_string(), toolchain.to_string()))
}
/// The repo's offset: its `consumer_offsets` row with the highest seq.
pub async fn offset(&self, repo: &str) -> Result<Option<Offset>> {
let t = self.tbl(CONSUMER_OFFSETS).await?;
let batches = skade::read_filtered(&t, &ScanFilter::eq("repo", repo), &["repo", "branch", "sha", "seq"]).await?;
let mut best: Option<Offset> = None;
for b in &batches {
let r = str_col(b, "repo")?;
let br = str_col(b, "branch")?;
let sha = str_col(b, "sha")?;
let seq = i64_col(b, "seq")?;
for i in 0..b.num_rows() {
if r.value(i) != repo {
continue;
}
if best.as_ref().is_none_or(|o| seq.value(i) > o.seq) {
best = Some(Offset { branch: br.value(i).into(), sha: sha.value(i).into(), seq: seq.value(i) });
}
}
}
Ok(best)
}
/// The ledger rows of `repo`, in commit order (seq, then write order).
pub async fn runs(&self, repo: &str) -> Result<Vec<RunRow>> {
let t = self.tbl(INGEST_RUNS).await?;
let batches = skade::read_filtered(&t, &ScanFilter::eq("repo", repo), &[]).await?;
let mut rows: Vec<(i64, RunRow)> = Vec::new();
for b in &batches {
let (r, seq, sha, st) = (str_col(b, "repo")?, i64_col(b, "seq")?, str_col(b, "sha")?, str_col(b, "status")?);
let (cd, rp, rm, ms) = (i64_col(b, "crates_dumped")?, i64_col(b, "rows_plus")?, i64_col(b, "rows_minus")?, i64_col(b, "ms")?);
let kc = i64_col(b, "key_collisions")?;
let (sp, sm) = (i64_col(b, "span_plus")?, i64_col(b, "span_minus")?);
let od = i64_col(b, "owner_disambiguations")?;
let err = str_col(b, "error")?;
let ts = b
.column_by_name("ts")
.and_then(|c| c.as_any().downcast_ref::<TimestampMicrosecondArray>())
.context("ingest_runs.ts")?;
for i in 0..b.num_rows() {
if r.value(i) != repo {
continue;
}
rows.push((
ts.value(i),
RunRow {
repo: r.value(i).into(),
seq: seq.value(i),
sha: sha.value(i).into(),
status: st.value(i).into(),
crates_dumped: cd.value(i),
rows_plus: rp.value(i),
rows_minus: rm.value(i),
key_collisions: kc.value(i),
span_plus: sp.value(i),
span_minus: sm.value(i),
owner_disambiguations: od.value(i),
ms: ms.value(i),
error: (!err.is_null(i)).then(|| err.value(i).to_string()),
},
));
}
}
rows.sort_by_key(|r| (r.1.seq, r.0));
Ok(rows.into_iter().map(|(_, r)| r).collect())
}
/// The feed ledger rows of `repo`, in (seq, write) order.
pub async fn feed_runs(&self, repo: &str) -> Result<Vec<FeedRow>> {
let t = self.tbl(FEED_RUNS).await?;
let batches = skade::read_filtered(&t, &ScanFilter::eq("repo", repo), &[]).await?;
let mut rows: Vec<(i64, FeedRow)> = Vec::new();
for b in &batches {
let (r, seq, sha, feed, st) = (str_col(b, "repo")?, i64_col(b, "seq")?, str_col(b, "sha")?, str_col(b, "feed")?, str_col(b, "status")?);
let (n, ms, detail) = (i64_col(b, "rows")?, i64_col(b, "ms")?, str_col(b, "detail")?);
let ts = b.column_by_name("ts").and_then(|c| c.as_any().downcast_ref::<TimestampMicrosecondArray>()).context("consumer_feed_runs.ts")?;
for i in 0..b.num_rows() {
if r.value(i) != repo {
continue;
}
rows.push((
ts.value(i),
FeedRow {
repo: r.value(i).into(),
seq: seq.value(i),
sha: sha.value(i).into(),
feed: feed.value(i).into(),
status: st.value(i).into(),
rows: n.value(i),
ms: ms.value(i),
detail: (!detail.is_null(i)).then(|| detail.value(i).to_string()),
},
));
}
}
rows.sort_by_key(|r| (r.1.seq, r.0));
Ok(rows.into_iter().map(|(_, r)| r).collect())
}
/// Per feed of `repo`: its LAST ledger row — the feed's own offset,
/// derived from the ledger (which sha it last ran for, and how that went).
/// A feed absent here never ran for this repo.
pub async fn feed_last(&self, repo: &str) -> Result<BTreeMap<String, FeedRow>> {
let mut out = BTreeMap::new();
for r in self.feed_runs(repo).await? {
out.insert(r.feed.clone(), r);
}
Ok(out)
}
/// The backfill rows of the log (of `repo` when given), in write order.
pub async fn backfills(&self, repo: Option<&str>) -> Result<Vec<BackfillRow>> {
let t = self.tbl(BACKFILL_RUNS).await?;
let batches = match repo {
Some(r) => skade::read_filtered(&t, &ScanFilter::eq("repo", r), &[]).await?,
None => skade::read_all(&t).await?,
};
let mut rows: Vec<(i64, BackfillRow)> = Vec::new();
for b in &batches {
let (r, from, base, target) = (str_col(b, "repo")?, str_col(b, "from_sha")?, i64_col(b, "base_seq")?, str_col(b, "target_sha")?);
let (total, st, detail) = (i64_col(b, "total")?, str_col(b, "status")?, str_col(b, "detail")?);
let ts = b.column_by_name("ts").and_then(|c| c.as_any().downcast_ref::<TimestampMicrosecondArray>()).context("backfill_runs.ts")?;
for i in 0..b.num_rows() {
if repo.is_some_and(|x| x != r.value(i)) {
continue;
}
rows.push((
ts.value(i),
BackfillRow {
repo: r.value(i).into(),
from_sha: from.value(i).into(),
base_seq: base.value(i),
target_sha: target.value(i).into(),
total: total.value(i),
status: st.value(i).into(),
detail: (!detail.is_null(i)).then(|| detail.value(i).to_string()),
},
));
}
}
rows.sort_by_key(|r| r.0);
Ok(rows.into_iter().map(|(_, r)| r).collect())
}
/// Per repo: its LAST backfill row (`requested` = one is open, `done` =
/// the last one asked for was walked to the tip). A repo absent here was
/// never backfilled.
pub async fn backfill_last(&self) -> Result<BTreeMap<String, BackfillRow>> {
let mut out = BTreeMap::new();
for r in self.backfills(None).await? {
out.insert(r.repo.clone(), r);
}
Ok(out)
}
/// Current snapshot of every event table.
pub async fn stamp(&self) -> Result<Stamp> {
let mut s = [None; 6];
for kind in Kind::ALL {
s[kind.idx()] = snapshot_id(&self.tbl(Self::table_name(kind)).await?);
}
Ok(Stamp(s))
}
/// Read every event of one table (all columns), for inspection.
pub async fn events(&self, kind: Kind) -> Result<Vec<RecordBatch>> {
let t = self.tbl(Self::table_name(kind)).await?;
let mut out = null_filled(&t, skade::read_all(&t).await?, None)?;
if let Some(ring) = self.ring_tbl_name(kind) {
let r = self.tbl(ring).await?;
out.extend(null_filled(&r, skade::read_all(&r).await?, None)?);
}
Ok(out)
}
/// THE fold: state = per key, the last event in (repo, seq) order — for
/// the binary stream (repo, seq, rev) — open iff it is `+`. `repo`
/// restricts to one repo; `upto_seq` gives the state as of that consumer
/// ordinal (valid_from ≤ N < valid_to, derived). Universe events are not
/// repo state and are skipped.
pub async fn fold(&self, repo: Option<&str>, upto_seq: Option<i64>) -> Result<OpenSet> {
let mut set = OpenSet::default();
for kind in Kind::ALL {
self.fold_kind_into(&mut set, kind, repo, upto_seq, None).await?;
}
Ok(set)
}
/// One kind of [`Self::fold`], folded into `set`. `only` restricts to
/// those repos (`Some(&[])` = nothing to read — the zone covered them
/// all, see [`crate::readfast`]); `None` = every repo.
pub(crate) async fn fold_kind_into(&self, set: &mut OpenSet, kind: Kind, repo: Option<&str>, upto_seq: Option<i64>, only: Option<&[String]>) -> Result<()> {
if only.is_some_and(|w| w.is_empty()) {
return Ok(());
}
{
let spec = self.spec(kind);
let binary = kind.stream() == Stream::Binary;
let mut cols: Vec<&str> = vec![COL_REPO, COL_SEQ, COL_OP, COL_KRATE, COL_PAYLOAD_HASH, COL_KEY_DUP];
if binary {
cols.push(COL_REV);
}
cols.extend(spec.key_columns.iter().map(String::as_str));
let t = self.tbl(Self::table_name(kind)).await?;
let batches = match (repo, only) {
(Some(r), _) => skade::read_filtered(&t, &ScanFilter::eq(COL_REPO, r), &cols).await?,
(None, Some(w)) => skade::read_filtered(&t, &ScanFilter::is_in(COL_REPO, w.iter().map(String::as_str)), &cols).await?,
(None, None) => skade::read_columns(&t, &cols).await?,
};
// (repo, seq, rev, batch, row): (seq, rev) order per repo is the log
// order. One key appears at most once per (repo, seq[, rev]), so
// ties do not matter. Source tables have no rev (0).
let mut order: Vec<(&str, i64, i64, usize, usize)> = Vec::new();
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)?,
if binary { Some(i64_col(b, COL_REV)?) } else { None },
str_col(b, COL_OP)?,
str_col(b, COL_KRATE)?,
str_col(b, COL_PAYLOAD_HASH)?,
str_col(b, COL_KEY_DUP)?,
keys,
));
}
for (bi, (r, seq, rev, ..)) in decoded.iter().enumerate() {
for i in 0..r.len() {
let repo_v = r.value(i);
if repo_v == UNIVERSE_REPO || repo.is_some_and(|x| x != repo_v) {
continue;
}
if upto_seq.is_some_and(|n| seq.value(i) > n) {
continue;
}
order.push((repo_v, seq.value(i), rev.as_ref().map_or(0, |r| r.value(i)), bi, i));
}
}
order.sort_by(|a, b| (a.0, a.1, a.2).cmp(&(b.0, b.1, b.2)));
for (repo_v, _, _, bi, i) in order {
let (_, _, _, op, krate, hash, dup, keys) = &decoded[bi];
let key = identity(spec.stable_keys, keys, dup, i);
let h: Option<PayloadHash> = if hash.is_null(i) { None } else { Some(unhex(hash.value(i))?) };
set.apply_event(kind, repo_v, op.value(i), krate.value(i), key, h)?;
}
}
Ok(())
}
/// Every repo's offset (`consumer_offsets`, highest seq per repo).
pub async fn offsets(&self) -> Result<BTreeMap<String, Offset>> {
let t = self.tbl(CONSUMER_OFFSETS).await?;
let mut out: BTreeMap<String, Offset> = BTreeMap::new();
for b in skade::read_columns(&t, &["repo", "branch", "sha", "seq"]).await? {
let (r, br, sha, seq) = (str_col(&b, "repo")?, str_col(&b, "branch")?, str_col(&b, "sha")?, i64_col(&b, "seq")?);
for i in 0..b.num_rows() {
if out.get(r.value(i)).is_none_or(|o| seq.value(i) > o.seq) {
out.insert(r.value(i).into(), Offset { branch: br.value(i).into(), sha: sha.value(i).into(), seq: seq.value(i) });
}
}
}
Ok(out)
}
/// THE fold ([`Self::fold`]: per identity the last event in (repo, seq
/// [, rev]) order, open iff `+`), keeping the PAYLOAD: the open rows of
/// one event table under `scope`, with the event prefix, the key columns
/// and `cols` projected. For readers that need more than identities (the
/// room: node kind / crate / origin, edge endpoints). Universe events are
/// included only when `scope.universe` (all of them: the ring is
/// append-only, dumped once per crate@version + toolchain).
pub async fn fold_rows(&self, kind: Kind, scope: &FoldScope, cols: &[&str]) -> Result<FoldedRows> {
let mut rows = self.scan_rows(kind, scope, cols).await?;
self.fold_scanned(kind, scope, &mut rows)?;
Ok(rows)
}
/// The first half of [`Self::fold_rows`]: the scan (lake IO), with
/// nothing folded yet (`open` empty). [`Self::fold_scanned`] is the
/// second, CPU only — a reader folds several tables side by side.
pub async fn scan_rows(&self, kind: Kind, scope: &FoldScope, cols: &[&str]) -> Result<FoldedRows> {
let proj = self.scan_projection(kind, cols);
let mut wanted: Vec<&str> = scope.repos.keys().map(String::as_str).collect();
if scope.universe {
wanted.push(UNIVERSE_REPO);
}
let mut out = FoldedRows::default();
if wanted.is_empty() {
return Ok(out);
}
let ring = self.ring_tbl_name(kind);
if ring.is_some() {
wanted.retain(|r| *r != UNIVERSE_REPO);
}
if !wanted.is_empty() {
let t = self.tbl(Self::table_name(kind)).await?;
let batches = skade::read_filtered(&t, &ScanFilter::is_in(COL_REPO, wanted.iter().copied()), &scannable(&t, &proj)?).await?;
out.batches = null_filled(&t, batches, Some(&proj))?;
}
if let (Some(ring), true) = (ring, scope.universe) {
let t = self.tbl(ring).await?;
let batches = skade::read_columns(&t, &scannable(&t, &proj)?).await?;
out.batches.extend(null_filled(&t, batches, Some(&proj))?);
}
Ok(out)
}
/// [`Self::scan_rows`] with any filter in place of the scope's repos —
/// for a scan trimmed at read time on another column (the ring's rows of
/// the crates a scope depends on: `krate IN (…)`). What the filter lets
/// through is folded as [`Self::fold_scanned`] folds a scope: rows of a
/// repo outside the fold's scope are skipped there.
pub async fn scan_rows_where(&self, kind: Kind, filter: &ScanFilter, cols: &[&str]) -> Result<FoldedRows> {
let proj = self.scan_projection(kind, cols);
let mut batches = Vec::new();
for name in std::iter::once(Self::table_name(kind)).chain(self.ring_tbl_name(kind)) {
let t = self.tbl(name).await?;
let read = skade::read_filtered(&t, filter, &scannable(&t, &proj)?).await?;
batches.extend(null_filled(&t, read, Some(&proj))?);
}
Ok(FoldedRows { batches, ..Default::default() })
}
/// The table a streamed scan reads: the ring table for a scan of the
/// ring (`scan.repo == @universe`) on a split lake, the source table
/// otherwise. On a legacy lake every scan reads the source table.
async fn scan_tbl(&self, kind: Kind, scan: &Scan) -> Result<IceTable> {
let name = match scan.repo.as_deref() {
Some(r) => self.table_for(kind, r),
None => Self::table_name(kind),
};
self.tbl(name).await
}
/// **Fold scans of `kind` without holding them.** Two streamed passes
/// ([`skade::read_filtered_each`]): the first reads the event prefix and
/// the key and folds them (per identity, its last event in `scope`); the
/// second reads `cols` too and hands `open` each batch with the rows of it
/// that are an identity's open (`+`, last) event — each identity once.
/// Nothing of a scan is held past its batch: what `open` keeps is the
/// caller's. For scans too big to collect (the ring view's 35 M edge rows).
/// Returns the events of the scope the fold read.
pub async fn fold_each(
&self,
kind: Kind,
scope: &FoldScope,
scans: &[Scan],
cols: &[&str],
mut open: impl FnMut(&RecordBatch, &[u32]) -> Result<()> + Send,
) -> Result<u64> {
let spec = self.spec(kind);
let binary = kind.stream() == Stream::Binary;
// identity → its last event in scope: (seq, rev, is `+`)
let mut last: std::collections::HashMap<u128, (i64, i64, bool), IdHash> = std::collections::HashMap::with_hasher(IdHash);
let mut events_read = 0u64;
let mut failed: Option<anyhow::Error> = None;
let keys_only = self.scan_projection(kind, &[]);
for scan in scans {
let t = self.scan_tbl(kind, scan).await?;
let table_schema = skade::arrow_schema_of(&t)?;
let proj = scannable(&t, &keys_only)?;
skade::read_filtered_each(&t, Some(&scan.filter), &proj, |b| {
let step = null_fill_one(&table_schema, b, &keys_only).and_then(|b| {
each_in_scope(spec, binary, scope, scan, &b, |id, at, plus, _| {
events_read += 1;
match last.entry(id) {
std::collections::hash_map::Entry::Occupied(mut o) => {
if (o.get().0, o.get().1) <= at {
o.insert((at.0, at.1, plus));
}
}
std::collections::hash_map::Entry::Vacant(v) => {
v.insert((at.0, at.1, plus));
}
}
})
});
keep_going(step, &mut failed)
})
.await?;
if let Some(e) = failed.take() {
return Err(e);
}
}
last.retain(|_, e| e.2);
let all = self.scan_projection(kind, cols);
for scan in scans {
let t = self.scan_tbl(kind, scan).await?;
let table_schema = skade::arrow_schema_of(&t)?;
let proj = scannable(&t, &all)?;
skade::read_filtered_each(&t, Some(&scan.filter), &proj, |b| {
let step = null_fill_one(&table_schema, b, &all).and_then(|b| {
let mut rows = Vec::new();
each_in_scope(spec, binary, scope, scan, &b, |id, at, _, i| {
// the open event, once (an identity is emitted and forgotten)
if last.get(&id).is_some_and(|e| (e.0, e.1) == at) {
last.remove(&id);
rows.push(i as u32);
}
})?;
if rows.is_empty() { Ok(()) } else { open(&b, &rows) }
});
keep_going(step, &mut failed)
})
.await?;
if let Some(e) = failed.take() {
return Err(e);
}
}
Ok(events_read)
}
/// **Fold scans of `kind` in ONE streamed pass, the caller keeping what
/// it derives from a row, not the row.** Every event in `scope` that is,
/// so far, its identity's latest and a `+` is handed to `emit` (a batch
/// at a time: the batch and those rows, in order); the caller keeps one
/// thing per handed row — slot `n` is the `n`-th row handed out over the
/// whole fold. A later event of the identity (a `+` again, or a `-`)
/// supersedes its slot. Returns the open slots, ascending (the order the
/// rows were handed out), and the events read.
///
/// One read where [`Self::fold_each`] needs two: a superseded row was
/// handed out for nothing, which costs little where the history is thin
/// (dwarf: 97% of the events read are open).
///
/// **The READ-FAST zone**: a repo in `scope` with an anchor
/// (`crate::readfast`) is read as interval rows + the event tail — not a
/// history scan — when the scans include the LOCAL one (`repo: None`,
/// the source tables); that local scan then narrows to the repos
/// without a zone. A zone batch carries the identity prefix and the
/// payload columns, no `seq`/`op` (nothing the room reads). Every row
/// emitted from the zone is open.
pub async fn fold_stream(
&self,
kind: Kind,
scope: &FoldScope,
scans: &[Scan],
cols: &[&str],
mut emit: impl FnMut(&RecordBatch, &[u32]) -> Result<()> + Send,
) -> Result<(Vec<u32>, u64)> {
let local = scans.iter().any(|s| s.repo.is_none());
let mut fast: Vec<(String, i64, i64)> = Vec::new();
if local {
for (r, n) in &scope.repos {
if let Some(a) = crate::readfast::anchor_of(self, kind, r).await? {
fast.push((r.clone(), *n, a));
}
}
}
if fast.is_empty() {
return self.fold_stream_events(kind, scope, scans, cols, &mut emit, 0).await;
}
let cols_owned: Vec<String> = cols.iter().map(|c| c.to_string()).collect();
let (mut open, mut events_read, mut next) = (Vec::new(), 0u64, 0u32);
for (repo, n, a) in &fast {
let fr = crate::readfast::readfast_read(self, kind, repo, *a, Some(*n), &cols_owned).await?;
let mut at = 0usize;
for (bi, b) in fr.batches.iter().enumerate() {
let mut rows: Vec<u32> = Vec::new();
while at < fr.open.len() && fr.open[at].0 == bi as u32 {
rows.push(fr.open[at].1);
at += 1;
}
if rows.is_empty() {
continue;
}
emit(b, &rows)?;
open.extend(next..next + rows.len() as u32);
next += rows.len() as u32;
}
events_read += fr.events_read;
}
// the rest: the repos without a zone (a rebuilt local scan) and the
// pass-through scans (the ring)
let slow: BTreeMap<String, i64> = scope.repos.iter().filter(|(r, _)| !fast.iter().any(|(f, ..)| f == *r)).map(|(r, n)| (r.clone(), *n)).collect();
let mut rest: Vec<Scan> = scans.iter().filter(|s| s.repo.is_some()).cloned().collect();
if !slow.is_empty() {
rest.insert(0, Scan { filter: ScanFilter::is_in(COL_REPO, slow.keys().map(String::as_str)), repo: None });
}
if !rest.is_empty() {
let slow_scope = FoldScope { repos: slow, universe: scope.universe };
let (more, read) = self.fold_stream_events(kind, &slow_scope, &rest, cols, &mut emit, next).await?;
open.extend(more);
events_read += read;
}
open.sort_unstable();
Ok((open, events_read))
}
/// [`Self::fold_stream`] over the EVENT tables alone (the zone not
/// consulted); slots start at `slot_base`.
async fn fold_stream_events(
&self,
kind: Kind,
scope: &FoldScope,
scans: &[Scan],
cols: &[&str],
emit: &mut (impl FnMut(&RecordBatch, &[u32]) -> Result<()> + Send),
slot_base: u32,
) -> Result<(Vec<u32>, u64)> {
let mut track = FoldTrack { last: std::collections::HashMap::with_hasher(IdHash), next: slot_base, read_at: std::collections::BTreeMap::new() };
let events_read = self.fold_events_core(kind, scope, scans, cols, emit, &mut track).await?;
Ok((track.open(), events_read))
}
/// [`Self::fold_stream`], keeping what a later DELTA advances: the
/// identity map and the table snapshots the fold read ([`FoldTrack`]).
/// A repo with a READ-FAST zone seeds the track FROM the zone (an open
/// interval row carries its identity columns and holds from
/// `valid_from_seq`) plus the event tail after the anchor — the whole
/// history is never re-scanned. Identities whose last event is a close
/// are absent from a zone-seeded track, which is delta-safe: any later
/// event is newer than the close by construction.
pub async fn fold_stream_tracked(
&self,
kind: Kind,
scope: &FoldScope,
scans: &[Scan],
cols: &[&str],
mut emit: impl FnMut(&RecordBatch, &[u32]) -> Result<()> + Send,
) -> Result<(Vec<u32>, u64, FoldTrack)> {
let mut track = FoldTrack { last: std::collections::HashMap::with_hasher(IdHash), next: 0, read_at: std::collections::BTreeMap::new() };
let local = scans.iter().any(|s| s.repo.is_none());
let mut fast: Vec<(String, i64, i64)> = Vec::new();
if local {
for (r, n) in &scope.repos {
if let Some(a) = crate::readfast::anchor_of(self, kind, r).await? {
fast.push((r.clone(), *n, a));
}
}
}
let mut events_read = 0u64;
if !fast.is_empty() {
// the source table's snapshot is the delta start whatever path read it
let name = Self::table_name(kind);
let t = self.tbl(name).await?;
if let Some(at) = t.metadata().current_snapshot().map(|s| s.snapshot_id()) {
track.read_at.insert(name, at);
}
let cols_owned: Vec<String> = cols.iter().map(|c| c.to_string()).collect();
for (repo, n, a) in &fast {
let (fr, ids) = crate::readfast::readfast_read_tracked(self, kind, repo, *a, Some(*n), &cols_owned).await?;
let mut at = 0usize;
for (bi, b) in fr.batches.iter().enumerate() {
let mut rows: Vec<u32> = Vec::new();
while at < fr.open.len() && fr.open[at].0 == bi as u32 {
rows.push(fr.open[at].1);
let (id, seq) = ids[at];
track.last.insert(id, (seq, 0, track.next + rows.len() as u32 - 1));
at += 1;
}
if rows.is_empty() {
continue;
}
emit(b, &rows)?;
track.next += rows.len() as u32;
}
events_read += fr.events_read;
}
}
// the rest: the repos without a zone and the pass-through scans
let slow: std::collections::BTreeMap<String, i64> = scope.repos.iter().filter(|(r, _)| !fast.iter().any(|(f, ..)| f == *r)).map(|(r, n)| (r.clone(), *n)).collect();
let mut rest: Vec<Scan> = scans.iter().filter(|s| s.repo.is_some()).cloned().collect();
if !slow.is_empty() {
rest.insert(0, Scan { filter: ScanFilter::is_in(COL_REPO, slow.keys().map(String::as_str)), repo: None });
}
if !rest.is_empty() {
let slow_scope = FoldScope { repos: slow, universe: scope.universe };
events_read += self.fold_events_core(kind, &slow_scope, &rest, cols, &mut emit, &mut track).await?;
}
Ok((track.open(), events_read, track))
}
/// **Advance a tracked fold by the lake's delta**: only the data files
/// appended to the event table since the track's fold are read
/// ([`skade::read_delta`]), their events applied over the track's
/// identities exactly as a fresh fold would apply them. The caller's
/// records only ever GROW — a superseded or closed identity's old slot
/// simply leaves the open set.
///
/// **Compaction-aware** (wave item 8): the lake compaction's `replace`
/// snapshot is stamped row-preserving by skade
/// ([`skade::COMPACTION_PROP`]) — it rewrites rows the track already
/// applied into new files without changing any row — so `read_delta`
/// SKIPS it and hands back only the files appended after it. The track
/// survives: nothing re-reads, nothing re-emits, `read_at` moves past the
/// replace, and the fold is kept across compactions.
///
/// [`FoldDelta::Full`] = the delta cannot stand in for a fresh fold: a
/// commit in the window rewrote rows in a way the delta cannot skip (an
/// UNSTAMPED replace, an overwrite, or an append the track had not read
/// that the compaction swallowed into its rewritten files), removed rows
/// (equality deletes), or a scan reads a ring table (its `krate IN
/// owners` trim is lake-side; the event-level check cannot re-apply it).
pub async fn fold_stream_delta(
&self,
kind: Kind,
scope: &FoldScope,
scans: &[Scan],
cols: &[&str],
track: &mut FoldTrack,
mut emit: impl FnMut(&RecordBatch, &[u32]) -> Result<()> + Send,
) -> Result<FoldDelta> {
let spec = self.spec(kind);
let binary = kind.stream() == Stream::Binary;
let all = self.scan_projection(kind, cols);
let mut events_read = 0u64;
for scan in scans {
if scan.repo.is_some() {
return Ok(FoldDelta::Full);
}
let name = Self::table_name(kind);
let t = self.tbl(name).await?;
let Some(to) = t.metadata().current_snapshot().map(|s| s.snapshot_id()) else { continue };
let from = track.read_at.get(name).copied();
if from == Some(to) {
continue;
}
// THROUGH a stamped row-preserving replace that swallowed the
// window's appends (the post-consume lake compaction races the
// readers and rewrites the just-appended files): the replace's
// merged files are read wholesale and rows the track already
// holds REAPPEAR — the track's apply is idempotent (a re-seen
// row re-takes its slot; the old slot leaves the open set, its
// record becomes reclaimable garbage). Live 2026-10-20 the plain
// delta refused every real commit's window this way and the ring
// room paid a 140-200 s refold per consume.
let (batches, plan) = skade::read_delta_through_replaces(&t, from, to).await?;
if plan.needs_full_reload || !plan.delete_files.is_empty() {
return Ok(FoldDelta::Full);
}
let table_schema = skade::arrow_schema_of(&t)?;
for b in batches {
let b = null_fill_one(&table_schema, b, &all)?;
let mut rows: Vec<u32> = Vec::new();
let mut bad_rev = false;
each_in_scope(spec, binary, scope, scan, &b, |id, at, plus, i| {
events_read += 1;
let rev = i32::try_from(at.1).unwrap_or_else(|_| {
bad_rev = true;
0
});
let newer = track.last.get(&id).is_none_or(|e| (e.0, e.1) <= (at.0, rev));
if !newer {
return;
}
let slot = if plus {
rows.push(i as u32);
track.next + rows.len() as u32 - 1
} else {
CLOSED
};
track.last.insert(id, (at.0, rev, slot));
})?;
if bad_rev {
bail!("{kind:?}: a release ordinal past 2^31");
}
if !rows.is_empty() {
track.next += rows.len() as u32;
emit(&b, &rows)?;
}
}
track.read_at.insert(name, to);
}
Ok(FoldDelta::Advanced { open: track.open(), events: events_read })
}
/// The shared event-scan fold: scans applied in order, `track.last`
/// deciding per identity, open rows handed to `emit` batch by batch.
async fn fold_events_core(
&self,
kind: Kind,
scope: &FoldScope,
scans: &[Scan],
cols: &[&str],
emit: &mut (impl FnMut(&RecordBatch, &[u32]) -> Result<()> + Send),
track: &mut FoldTrack,
) -> Result<u64> {
let spec = self.spec(kind);
let binary = kind.stream() == Stream::Binary;
let mut events_read = 0u64;
let mut failed: Option<anyhow::Error> = None;
let all = self.scan_projection(kind, cols);
for scan in scans {
let name = match scan.repo.as_deref() {
Some(r) => self.table_for(kind, r),
None => Self::table_name(kind),
};
let t = self.tbl(name).await?;
if let Some(at) = t.metadata().current_snapshot().map(|s| s.snapshot_id()) {
track.read_at.insert(name, at);
}
let table_schema = skade::arrow_schema_of(&t)?;
let proj = scannable(&t, &all)?;
skade::read_filtered_each(&t, Some(&scan.filter), &proj, |b| {
let step = null_fill_one(&table_schema, b, &all).and_then(|b| {
let mut rows: Vec<u32> = Vec::new();
let mut bad_rev = false;
each_in_scope(spec, binary, scope, scan, &b, |id, at, plus, i| {
events_read += 1;
let rev = i32::try_from(at.1).unwrap_or_else(|_| {
bad_rev = true;
0
});
let newer = track.last.get(&id).is_none_or(|e| (e.0, e.1) <= (at.0, rev));
if !newer {
return;
}
let slot = if plus {
rows.push(i as u32);
track.next + rows.len() as u32 - 1
} else {
CLOSED
};
track.last.insert(id, (at.0, rev, slot));
})?;
if bad_rev {
bail!("{kind:?}: a release ordinal past 2^31");
}
if rows.is_empty() {
return Ok(());
}
track.next += rows.len() as u32;
emit(&b, &rows)
});
keep_going(step, &mut failed)
})
.await?;
if let Some(e) = failed.take() {
return Err(e);
}
}
Ok(events_read)
}
/// The columns a scan of `kind` reads: the event prefix, the key, `cols`.
pub(crate) fn scan_projection<'a>(&'a self, kind: Kind, cols: &[&'a str]) -> Vec<&'a str> {
let spec = self.spec(kind);
let mut proj: Vec<&str> = vec![COL_REPO, COL_SEQ, COL_OP, COL_KRATE, COL_KEY_DUP];
if kind.stream() == Stream::Binary {
proj.push(COL_REV);
}
for c in spec.key_columns.iter().map(String::as_str).chain(cols.iter().copied()) {
if !proj.contains(&c) {
proj.push(c);
}
}
proj
}
/// The second half of [`Self::fold_rows`] over what [`Self::scan_rows`]
/// read: sets `open` and `events_read`. CPU only, no lake.
pub fn fold_scanned(&self, kind: Kind, scope: &FoldScope, out: &mut FoldedRows) -> Result<()> {
let spec = self.spec(kind);
let binary = kind.stream() == Stream::Binary;
let batches = &out.batches;
// identity (128-bit xxh3 of repo · crate · key · dup) → the last
// event: (seq, rev, batch, row, is_plus). The identity's bytes go
// into ONE reused buffer (no String per row) and the map's own hash
// is the identity's low bits (it is a hash already).
// NOT pre-sized to the row count: the map holds one entry per DISTINCT
// identity, and an event-sourced history has many events per identity —
// a rows-sized table held gigabytes of empty buckets on the dependency
// room (heaptrack: 1.66 GB). Growing re-hashes a few times and peaks
// near the real size instead.
let mut last: std::collections::HashMap<u128, (i64, i64, u32, u32, bool), IdHash> = std::collections::HashMap::with_hasher(IdHash);
let mut buf: Vec<u8> = Vec::with_capacity(256);
// the batches of a scan come per data file, mostly one repo each
let mut scope_of: Option<(&str, Option<i64>)> = None;
let mut events_read = 0u64;
for (bi, b) in batches.iter().enumerate() {
let (r, seq, op, krate, dup) = (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)?);
let rev = if binary { Some(i64_col(b, COL_REV)?) } else { None };
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
for i in 0..b.num_rows() {
let repo_v = r.value(i);
// the repo's last seq in scope (i64::MAX = the ring, when on)
let upto = match scope_of {
Some((name, upto)) if name == repo_v => upto,
_ if repo_v == UNIVERSE_REPO => {
let upto = scope.universe.then_some(i64::MAX);
scope_of = Some((UNIVERSE_REPO, upto));
upto
}
_ => match scope.repos.get_key_value(repo_v) {
Some((name, n)) => {
scope_of = Some((name.as_str(), Some(*n)));
Some(*n)
}
None => None,
},
};
if !upto.is_some_and(|n| seq.value(i) <= n) {
continue;
}
events_read += 1;
buf.clear();
buf.extend_from_slice(repo_v.as_bytes());
buf.push(0x1e);
buf.extend_from_slice(krate.value(i).as_bytes());
buf.push(0x1e);
identity_into(spec.stable_keys, &keys, &dup, i, &mut buf);
let id = twox_hash::XxHash3_128::oneshot(&buf);
let at = (seq.value(i), rev.as_ref().map_or(0, |r| r.value(i)));
let ev = (at.0, at.1, bi as u32, i as u32, op.value(i) == OP_PLUS);
match last.entry(id) {
std::collections::hash_map::Entry::Occupied(mut o) => {
if (o.get().0, o.get().1) <= at {
o.insert(ev);
}
}
std::collections::hash_map::Entry::Vacant(v) => {
v.insert(ev);
}
}
}
}
let mut open: Vec<(u32, u32)> = last.into_values().filter(|e| e.4).map(|e| (e.2, e.3)).collect();
open.sort_unstable();
out.open = open;
out.events_read = events_read;
Ok(())
}
/// The `universe_manifest` rows: (crate, toolchain, rows, rustc_release).
pub async fn universe_rows(&self) -> Result<Vec<(String, String, i64, Option<String>)>> {
let t = self.tbl(UNIVERSE_MANIFEST).await?;
let mut out = Vec::new();
for b in skade::read_all(&t).await? {
let (k, tc, rows, rel) = (str_col(&b, "crate")?, str_col(&b, "toolchain")?, i64_col(&b, "rows")?, str_col(&b, "rustc_release")?);
for i in 0..b.num_rows() {
out.push((k.value(i).to_string(), tc.value(i).to_string(), rows.value(i), (!rel.is_null(i)).then(|| rel.value(i).to_string())));
}
}
out.sort();
Ok(out)
}
async fn read_universe(&self) -> Result<HashSet<(String, String)>> {
let t = self.tbl(UNIVERSE_MANIFEST).await?;
let mut out = HashSet::new();
for b in skade::read_columns(&t, &["crate", "toolchain"]).await? {
let (k, tc) = (str_col(&b, "crate")?, str_col(&b, "toolchain")?);
for i in 0..b.num_rows() {
out.insert((k.value(i).to_string(), tc.value(i).to_string()));
}
}
Ok(out)
}
}
/// The log's ONE writer seat ([`EventLog::writer`]): every release goes
/// through it. Dropping it frees the seat. It reads like any handle
/// ([`LogWriter::log`]) — its reads see the catalog as it is now.
pub struct LogWriter {
log: EventLog,
}
impl Drop for LogWriter {
fn drop(&mut self) {
self.log.shared.seat.store(false, Ordering::Release);
}
}
impl LogWriter {
/// The log this seat writes (an unpinned handle).
pub fn log(&self) -> &EventLog {
&self.log
}
/// The next binary release ordinal of `repo` (1 for the first): from
/// `bin_offsets` (one projected read per repo and process), then kept in
/// memory — never a ledger scan.
pub async fn next_rev(&self, repo: &str) -> Result<i64> {
if let Some(r) = self.log.shared.revs.lock().unwrap_or_else(|e| e.into_inner()).get(repo) {
return Ok(r + 1);
}
let last = self.log.bin_offsets(repo).await?.values().map(|o| o.rev).max().unwrap_or(0);
self.log.shared.revs.lock().unwrap_or_else(|e| e.into_inner()).insert(repo.to_string(), last);
Ok(last + 1)
}
/// ONE skade atomic release: event tables + offset + ledger(s) (+
/// universe manifest) advance together or not at all.
pub(crate) async fn commit(&self, c: CommitWrite) -> Result<()> {
let shared = self.log.shared.clone();
on(self.log.shared.rt(), Self::release(shared, c)).await
}
async fn release(shared: Arc<Shared>, c: CommitWrite) -> Result<()> {
// The lock FIRST: a table handle loaded here is the base the atomic
// release is committed against, and the catalog refuses a base that
// moved (CatalogCommitConflicts — no retry). The compaction's replace
// commits under this same lock, so between the loads and the commit
// nothing lands. (Loaded before the lock, a replace in between failed
// the release with a conflict and stopped the consumer's run short.)
let _commit = shared.commit_lock.lock().await;
let now = now_micros();
let mut seen: HashSet<&'static str> = HashSet::new();
let mut tables = Vec::new();
for (kind, batches) in c.events {
if !seen.insert(EventLog::table_name(kind)) {
bail!("release for {} @ {} carries {} twice", c.repo, c.sha, EventLog::table_name(kind));
}
if !batches.is_empty() {
tables.push((shared.wh.table(&qualified(EventLog::table_name(kind))).await?, vec![batches]));
}
}
for (kind, batches) in c.ring_events {
let Some(ring) = EventLog::ring_table_name(kind) else { bail!("release for {} @ {}: {kind:?} has no ring table", c.repo, c.sha) };
if !seen.insert(ring) {
bail!("release for {} @ {} carries {ring} twice", c.repo, c.sha);
}
if !batches.is_empty() {
tables.push((shared.wh.table(&qualified(ring)).await?, vec![batches]));
}
}
if let Some(o) = &c.offset {
let offset = RecordBatch::try_new(
Arc::new(consumer_offsets_schema()),
vec![
str_arr([c.repo.as_str()]),
str_arr([o.branch.as_str()]),
str_arr([o.sha.as_str()]),
Arc::new(Int64Array::from(vec![o.seq])),
ts_arr(vec![now]),
],
)?;
tables.push((shared.wh.table(&qualified(CONSUMER_OFFSETS)).await?, vec![vec![offset]]));
}
if let Some(run) = &c.run {
tables.push((shared.wh.table(&qualified(INGEST_RUNS)).await?, vec![vec![run_batch(run, now)?]]));
}
if !c.bin_runs.is_empty() {
tables.push((shared.wh.table(&qualified(BIN_INGEST_RUNS)).await?, vec![vec![bin_run_batch(&c.bin_runs, now)?]]));
tables.push((shared.wh.table(&qualified(BIN_OFFSETS)).await?, vec![vec![bin_offset_batch(&c.bin_runs, now)?]]));
}
if !c.universe.is_empty() {
let n = c.universe.len();
let manifest = RecordBatch::try_new(
Arc::new(universe_manifest_schema()),
vec![
str_arr(c.universe.iter().map(|u| u.krate.as_str())),
str_arr(c.universe.iter().map(|u| u.toolchain.as_str())),
ts_arr(vec![now; n]),
Arc::new(Int64Array::from_iter_values(c.universe.iter().map(|u| u.rows))),
Arc::new(StringArray::from(c.universe.iter().map(|u| u.rustc_release.clone()).collect::<Vec<_>>())),
],
)?;
tables.push((shared.wh.table(&qualified(UNIVERSE_MANIFEST)).await?, vec![vec![manifest]]));
}
if tables.is_empty() {
bail!("release for {} @ {} writes nothing", c.repo, c.sha);
}
// an evolved table keeps the columns it was created with first
let tables = tables
.into_iter()
.map(|(t, groups)| {
let groups = groups.into_iter().map(|g| in_table_order(&t, g)).collect::<Result<Vec<_>>>()?;
Ok((t, groups))
})
.collect::<Result<Vec<_>>>()?;
let released = {
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
shared.wh.ingest_multi(tables, &shared.props).await
};
let (_, stats) = released.with_context(|| format!("atomic release for {} @ {}", c.repo, c.sha))?;
if stats.commits != 1 {
bail!("atomic release for {} @ {} made {} commits, expected exactly 1", c.repo, c.sha, stats.commits);
}
{
let mut universe = shared.universe.write().unwrap_or_else(|e| e.into_inner());
for u in c.universe {
universe.insert((u.krate, u.toolchain));
}
}
let mut revs = shared.revs.lock().unwrap_or_else(|e| e.into_inner());
for r in &c.bin_runs {
let cur = revs.entry(r.repo.clone()).or_insert(0);
*cur = (*cur).max(r.rev);
}
Ok(())
}
/// One feed's outcome for a consumed (repo, sha): ONLY its ledger row.
pub async fn record_feed(&self, row: &FeedRow) -> Result<()> {
let shared = self.log.shared.clone();
let batch = RecordBatch::try_new(
Arc::new(feed_runs_schema()),
vec![
str_arr([row.repo.as_str()]),
Arc::new(Int64Array::from(vec![row.seq])),
str_arr([row.sha.as_str()]),
str_arr([row.feed.as_str()]),
str_arr([row.status.as_str()]),
Arc::new(Int64Array::from(vec![row.rows])),
Arc::new(Int64Array::from(vec![row.ms])),
Arc::new(StringArray::from(vec![row.detail.clone()])),
ts_arr(vec![now_micros()]),
],
)?;
on(self.log.shared.rt(), async move {
let _commit = shared.commit_lock.lock().await;
let mut t = shared.wh.table(&qualified(FEED_RUNS)).await?;
let batch = in_table_order(&t, vec![batch])?;
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
t.append(&batch).await?;
Ok(())
})
.await
}
/// One backfill row ([`BackfillRow`]): ONLY that row.
pub async fn record_backfill(&self, row: &BackfillRow) -> Result<()> {
let shared = self.log.shared.clone();
let batch = RecordBatch::try_new(
Arc::new(backfill_runs_schema()),
vec![
str_arr([row.repo.as_str()]),
str_arr([row.from_sha.as_str()]),
Arc::new(Int64Array::from(vec![row.base_seq])),
str_arr([row.target_sha.as_str()]),
Arc::new(Int64Array::from(vec![row.total])),
str_arr([row.status.as_str()]),
Arc::new(StringArray::from(vec![row.detail.clone()])),
ts_arr(vec![now_micros()]),
],
)?;
on(self.log.shared.rt(), async move {
let _commit = shared.commit_lock.lock().await;
let mut t = shared.wh.table(&qualified(BACKFILL_RUNS)).await?;
let batch = in_table_order(&t, vec![batch])?;
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
t.append(&batch).await?;
Ok(())
})
.await
}
/// A failed attempt: ONLY a ledger row (offset and events untouched).
pub(crate) async fn record_error(&self, run: &RunRow) -> Result<()> {
let shared = self.log.shared.clone();
let batch = run_batch(run, now_micros())?;
on(self.log.shared.rt(), async move {
let _commit = shared.commit_lock.lock().await;
let mut t = shared.wh.table(&qualified(INGEST_RUNS)).await?;
let batch = in_table_order(&t, vec![batch])?;
shared.release_base.store(shared.wh.catalog().pointer_generation(), Ordering::Release);
let _in_flight = InFlight::begin(&shared.commits);
t.append(&batch).await?;
Ok(())
})
.await
}
}
/// Marks a catalog commit in flight on the shared counter (odd), and done
/// (even) when dropped — also when the commit failed or its future was
/// dropped, so a reader never waits on a counter stuck odd.
pub(crate) struct InFlight<'a>(&'a AtomicU64);
impl<'a> InFlight<'a> {
pub(crate) fn begin(counter: &'a AtomicU64) -> Self {
counter.fetch_add(1, Ordering::AcqRel);
Self(counter)
}
}
impl Drop for InFlight<'_> {
fn drop(&mut self) {
self.0.fetch_add(1, Ordering::AcqRel);
}
}
/// How many times [`EventLog::pin`] looks for a state that holds still (a
/// look that fails waits ~1 ms for the publishing release: ~10 s in all).
const PIN_LOOKS: usize = 10_000;
/// The table's stamp: the newest snapshot that ADDED events. A compaction's
/// `replace` snapshot moves nothing a fold sees, so the stamp looks through
/// it to the append it replaced — by the parent id, which the replace carries
/// even once that append's own metadata is expired. (A replace whose parent
/// is a replace is looked through too.)
fn snapshot_id(t: &IceTable) -> Option<i64> {
let md = t.metadata();
let mut cur = md.current_snapshot()?;
loop {
if cur.summary().operation != skade::iceberg::spec::Operation::Replace {
return Some(cur.snapshot_id());
}
let parent = cur.parent_snapshot_id()?;
match md.snapshot_by_id(parent) {
Some(p) => cur = p,
None => return Some(parent),
}
}
}
/// The Parquet knobs of every file the log writes: zstd(3) — the level a
/// release's small files are written at as they land (level 3 of the
/// largest edge file: 1.10 GB → 95 MB in 4.1 s; level 6: 87 MB in 7.8 s;
/// the compaction rewrites at 6 for the life of the lake), 256 k-row row
/// groups (finer skipping on `from` / `repo`), dictionary on.
fn write_props() -> Result<WriteProps> {
use skade::parquet::basic::{Compression, ZstdLevel};
Ok(WriteProps::new(Compression::ZSTD(ZstdLevel::try_new(3).context("zstd level")?)).row_group_size(256 * 1024))
}
/// Did opening `dir` fail because another process holds its catalog? redb
/// reports its file lock as "Database already open"; the catalog file must
/// exist for that to be the reason.
fn lake_is_held(dir: &std::path::Path, e: &skade::SkadeError) -> bool {
let msg = format!("{e:?} {e}").to_lowercase();
dir.join("catalog.redb").exists() && (msg.contains("already open") || msg.contains("cannot acquire lock") || msg.contains("databasealreadyopen"))
}
/// Which events [`EventLog::fold_rows`] reads: per repo the consumer
/// ordinal it stops at (the state as of `seq ≤ N`: valid_from ≤ N <
/// valid_to), and whether the universe ring is included. A repo not in
/// `repos` is not read at all.
#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
pub struct FoldScope {
pub repos: BTreeMap<String, i64>,
pub universe: bool,
}
/// The open rows of one event table (see [`EventLog::fold_rows`]).
#[derive(Debug, Clone, Default)]
pub struct FoldedRows {
/// The scanned batches (event prefix + key + requested columns).
pub batches: Vec<RecordBatch>,
/// (batch, row) of each open identity's last (`+`) event, sorted.
pub open: Vec<(u32, u32)>,
/// Events inside the scope (both ops), i.e. what the fold consumed.
pub events_read: u64,
}
/// The column `code_edge_events` is sorted by (CSR build = sorted scan).
pub const EDGE_SORT_COLUMN: &str = "from";
/// The `mimir-schema` version that introduced STABLE identities (version-
/// stripped keys and owners). A lake stamped before it is refused as
/// [`Unevolvable::Older`]; [`EventLog::open_lake`] wipes such a lake (it is
/// derived data) and the consumer re-reads every repo from its tip.
const IDENTITY_SCHEMA: u64 = 5;
/// What [`ensure`] may add in place to a table that lacks columns.
pub(crate) enum Evolves<'a> {
/// An event table: nullable payload columns outside the key. The table is
/// stamped with the payload's `mimir-schema` version ([`PROP_SCHEMA`]).
Payload { spec: &'a PayloadSpec },
/// A bookkeeping table: any nullable column.
Nullable,
}
/// Table `name` with `schema`: created when missing; when it is there, brought
/// to `schema` by adding the nullable columns it lacks — ONE metadata commit
/// (columns + properties), no data file touched — or refused
/// ([`LakeSchemaMismatch`]) when the difference is anything else. Returns the
/// columns added in place over the table's life ([`PROP_ADDED`]).
pub(crate) async fn ensure(wh: &Warehouse, name: &str, schema: &Schema, sort_cols: &[&str], partition_cols: &[&str], evolves: Evolves<'_>) -> Result<Vec<String>> {
use skade::iceberg::transaction::{AddColumn, ApplyTransactionAction, Transaction};
let q = qualified(name);
let ident = wh.table_ident(&q)?;
let version = match &evolves {
Evolves::Payload { spec } => spec.version(),
Evolves::Nullable => None,
};
if !wh.catalog().table_exists(&ident).await? {
let mut t = if sort_cols.is_empty() && partition_cols.is_empty() {
wh.create_table(&q, schema).await.with_context(|| format!("create {q}"))?
} else {
create_sorted(wh, &q, schema, sort_cols, partition_cols).await.with_context(|| format!("create {q}"))?
};
if let Some(v) = version {
t.set_properties([(PROP_SCHEMA, v)]).await.with_context(|| format!("stamp {q}"))?;
}
return Ok(Vec::new());
}
let t = wh.table(&q).await?;
let want_ice = skade::arrow_to_iceberg(schema)?;
let have_ice = t.inner().metadata().current_schema().clone();
let have: Vec<String> = have_ice.as_struct().fields().iter().map(|f| f.name.clone()).collect();
let want: Vec<String> = schema.fields().iter().map(|f| f.name().clone()).collect();
let refuse = |why: Unevolvable| anyhow::Error::new(LakeSchemaMismatch { table: q.clone(), have: have.clone(), want: want.clone(), why });
let stamp = t.properties().get(PROP_SCHEMA).cloned();
if let (Some(lake), Some(build)) = (&stamp, version)
&& let (Ok(l), Ok(b)) = (lake.parse::<u64>(), build.parse::<u64>())
&& l > b
{
return Err(refuse(Unevolvable::Newer { lake: lake.clone(), build: build.to_string() }));
}
// The identity shape changed at IDENTITY_SCHEMA (stable identities): a
// payload table STAMPED before it holds rows that key and own by
// versioned names, which no added column fixes. An UNSTAMPED table is
// not a pre-5 lake (every lake is stamped at creation) — it falls
// through to the column checks and their typed refusal.
if let Some(build) = version
&& build.parse::<u64>().is_ok_and(|b| b >= IDENTITY_SCHEMA)
&& stamp.as_deref().and_then(|s| s.parse::<u64>().ok()).is_some_and(|l| l < IDENTITY_SCHEMA)
{
return Err(refuse(Unevolvable::Older { lake: stamp.clone().unwrap_or_default(), build: build.to_string() }));
}
if have.iter().any(|c| !want.contains(c)) {
return Err(refuse(Unevolvable::Columns));
}
let mut missing = Vec::new();
for f in want_ice.as_struct().fields() {
match have_ice.field_by_name(&f.name) {
Some(h) if h.field_type != f.field_type => {
return Err(refuse(Unevolvable::Retyped { column: f.name.clone(), have: h.field_type.to_string(), want: f.field_type.to_string() }));
}
Some(_) => {}
None => {
let addable = !f.required
&& match &evolves {
Evolves::Payload { spec } => spec.schema.field_with_name(&f.name).is_ok() && !spec.key_columns.contains(&f.name),
Evolves::Nullable => true,
};
if !addable {
return Err(refuse(Unevolvable::Required { column: f.name.clone() }));
}
missing.push(f);
}
}
}
if !sort_cols.is_empty() && sorted_by(t.inner()) != sort_cols {
return Err(anyhow::Error::new(LakeSchemaMismatch {
table: q,
have: sorted_by(t.inner()).iter().map(|c| format!("sorted by {c}")).collect(),
want: sort_cols.iter().map(|c| format!("sorted by {c}")).collect(),
why: Unevolvable::SortOrder,
}));
}
let mut added: Vec<String> = t.properties().get(PROP_ADDED).map(|s| s.split(',').filter(|c| !c.is_empty()).map(str::to_string).collect()).unwrap_or_default();
let restamp = version.is_some_and(|v| stamp.as_deref() != Some(v));
if missing.is_empty() && !restamp {
return Ok(added);
}
// THE evolve step: the columns and the properties that describe them
// move in one commit of the table's metadata
let mut tx = Transaction::new(t.inner());
if !missing.is_empty() {
let mut action = tx.update_schema();
for f in &missing {
action = action.add_column(AddColumn::optional(&f.name, f.field_type.as_ref().clone()));
added.push(f.name.clone());
}
tx = action.apply(tx)?;
}
let mut props = tx.update_table_properties();
if !missing.is_empty() {
props = props.set(PROP_ADDED.to_string(), added.join(","));
}
if let Some(v) = version {
props = props.set(PROP_SCHEMA.to_string(), v.to_string());
}
tx = props.apply(tx)?;
tx.commit(wh.catalog().as_ref()).await.with_context(|| format!("evolve {q} in place"))?;
Ok(added)
}
/// skade's public constructors do not declare a sort order, so the table is
/// created through the Iceberg catalog API directly (same schema derivation,
/// same format version as the warehouse's other tables).
async fn create_sorted(wh: &Warehouse, q: &str, schema: &Schema, sort_cols: &[&str], partition_cols: &[&str]) -> Result<skade::Table> {
use skade::iceberg::TableCreation;
use skade::iceberg::spec::{NullOrder, SortDirection, SortField, SortOrder, Transform, UnboundPartitionField, UnboundPartitionSpec};
let ident = wh.table_ident(q)?;
// the namespace, which skade's own create_table makes on the way
if !wh.catalog().namespace_exists(ident.namespace()).await? {
wh.catalog().create_namespace(ident.namespace(), Default::default()).await.with_context(|| format!("create namespace of {q}"))?;
}
let ice = skade::arrow_to_iceberg(schema)?;
let partition_spec = if partition_cols.is_empty() {
None
} else {
let mut pb = UnboundPartitionSpec::builder();
for c in partition_cols {
let id = ice.field_id_by_name(c).with_context(|| format!("partition column `{c}` not in the schema"))?;
pb = pb.add_partition_field(UnboundPartitionField::builder().source_ids(vec![id]).name(c.to_string()).transform(Transform::Identity).build()?)?;
}
Some(pb.build())
};
let mut b = SortOrder::builder();
b.with_order_id(1);
for c in sort_cols {
let id = ice.field_id_by_name(c).with_context(|| format!("sort column `{c}` not in the schema"))?;
b.with_sort_field(
SortField::builder()
.source_id(id)
.transform(Transform::Identity)
.direction(SortDirection::Ascending)
.null_order(NullOrder::First)
.build(),
);
}
let sort_order = if sort_cols.is_empty() { None } else { Some(b.build(&ice)?) };
// the format every table of the lake is created at (skade's own)
let format_version = wh.new_table_format();
let creation = TableCreation {
name: ident.name().to_string(),
location: None,
schema: ice,
partition_spec,
sort_order,
properties: Default::default(),
format_version,
};
wh.catalog().create_table(ident.namespace(), creation).await?;
Ok(wh.table(q).await?)
}
/// The columns of a table's default sort order (empty = unsorted).
pub fn sorted_by(t: &skade::iceberg::table::Table) -> Vec<&str> {
let md = t.metadata();
let schema = md.current_schema();
md.default_sort_order()
.fields
.iter()
.filter_map(|f| schema.name_by_field_id(f.source_id))
.collect()
}
/// The columns of `cols` a scan of `t` can be asked for. Iceberg scans a table
/// at its current SNAPSHOT's schema, and an evolution takes no snapshot: until
/// the next release, a column added in place is not there to ask for —
/// although every row's value for it is known (null, [`null_filled`]).
pub(crate) fn scannable<'a>(t: &IceTable, cols: &[&'a str]) -> Result<Vec<&'a str>> {
let md = t.metadata();
let Some(snapshot) = md.current_snapshot() else {
return Ok(cols.to_vec());
};
let at = snapshot.schema(md)?;
Ok(cols.iter().copied().filter(|c| at.field_by_name(c).is_some()).collect())
}
/// One scan of [`EventLog::fold_each`]: a filter, and — when the filter lets
/// rows of other repos through (`krate IN (…)` over the ring's owners) — the
/// one repo whose rows count.
#[derive(Debug, Clone)]
pub struct Scan {
pub filter: ScanFilter,
pub repo: Option<String>,
}
/// A streamed scan's callback: an error is kept and the scan stopped.
fn keep_going(step: Result<()>, failed: &mut Option<anyhow::Error>) -> bool {
match step {
Ok(()) => true,
Err(e) => {
*failed = Some(e);
false
}
}
}
/// Every row of `b` in `scope` (and of `scan`'s repo): its identity (as the
/// fold hashes it), its (seq, rev), whether it is a `+`, its row.
fn each_in_scope(spec: &PayloadSpec, binary: bool, scope: &FoldScope, scan: &Scan, b: &RecordBatch, mut f: impl FnMut(u128, (i64, i64), bool, usize)) -> Result<()> {
let (r, seq, op, krate, dup) = (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)?);
let rev = if binary { Some(i64_col(b, COL_REV)?) } else { None };
let keys: Vec<StringArray> = spec.key_columns.iter().map(|k| str_col(b, k)).collect::<Result<_>>()?;
let mut buf: Vec<u8> = Vec::with_capacity(256);
for i in 0..b.num_rows() {
let repo_v = r.value(i);
if scan.repo.as_deref().is_some_and(|only| only != repo_v) {
continue;
}
let upto = if repo_v == UNIVERSE_REPO { scope.universe.then_some(i64::MAX) } else { scope.repos.get(repo_v).copied() };
if !upto.is_some_and(|n| seq.value(i) <= n) {
continue;
}
buf.clear();
buf.extend_from_slice(repo_v.as_bytes());
buf.push(0x1e);
buf.extend_from_slice(krate.value(i).as_bytes());
buf.push(0x1e);
identity_into(spec.stable_keys, &keys, &dup, i, &mut buf);
f(twox_hash::XxHash3_128::oneshot(&buf), (seq.value(i), rev.as_ref().map_or(0, |r| r.value(i))), op.value(i) == OP_PLUS, i);
}
Ok(())
}
/// [`null_filled`] of one batch, against the table's schema `table`.
fn null_fill_one(table: &Schema, b: RecordBatch, cols: &[&str]) -> Result<RecordBatch> {
let schema = b.schema();
let absent: Vec<&Field> = table.fields().iter().map(|f| f.as_ref()).filter(|f| cols.contains(&f.name().as_str()) && schema.index_of(f.name()).is_err()).collect();
if absent.is_empty() {
return Ok(b);
}
let mut fields: Vec<Field> = schema.fields().iter().map(|f| f.as_ref().clone()).collect();
let mut columns = b.columns().to_vec();
for f in absent {
fields.push(f.clone().with_nullable(true));
columns.push(skade::arrow_array::new_null_array(f.data_type(), b.num_rows()));
}
Ok(RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?)
}
/// `batches` (a scan of `t`) with every column of `cols` — `None` = of the
/// table — that the scan did not hand back added as nulls: a column added in
/// place, read before the first release after its evolution.
pub(crate) fn null_filled(t: &IceTable, batches: Vec<RecordBatch>, cols: Option<&[&str]>) -> Result<Vec<RecordBatch>> {
let table = skade::arrow_schema_of(t)?;
batches
.into_iter()
.map(|b| {
let schema = b.schema();
let absent: Vec<&Field> = table
.fields()
.iter()
.map(|f| f.as_ref())
.filter(|f| cols.is_none_or(|c| c.contains(&f.name().as_str())) && schema.index_of(f.name()).is_err())
.collect();
if absent.is_empty() {
return Ok(b);
}
let mut fields: Vec<Field> = schema.fields().iter().map(|f| f.as_ref().clone()).collect();
let mut columns = b.columns().to_vec();
for f in absent {
fields.push(f.clone().with_nullable(true));
columns.push(skade::arrow_array::new_null_array(f.data_type(), b.num_rows()));
}
Ok(RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)?)
})
.collect()
}
/// `batches` with their columns in `t`'s order. skade writes by POSITION, and
/// a table evolved in place has its added columns last, wherever this build's
/// schema lists them. A batch already in table order — or one that is not the
/// table's columns at all (the write refuses it) — passes through untouched.
fn in_table_order(t: &skade::Table, batches: Vec<RecordBatch>) -> Result<Vec<RecordBatch>> {
let table = t.arrow_schema()?;
let order: Vec<&str> = table.fields().iter().map(|f| f.name().as_str()).collect();
batches
.into_iter()
.map(|b| {
let schema = b.schema();
let idx: Vec<usize> = order.iter().filter_map(|c| schema.index_of(c).ok()).collect();
if idx.len() != order.len() || idx.len() != schema.fields().len() || idx.iter().enumerate().all(|(at, i)| at == *i) {
return Ok(b);
}
Ok(b.project(&idx)?)
})
.collect()
}
/// The map hasher of a key that IS a hash (a 128-bit identity): its low 64
/// bits folded with its high ones, no further mixing.
#[derive(Clone, Copy, Default)]
pub(crate) struct IdHash;
pub(crate) struct IdHasher(u64);
impl std::hash::BuildHasher for IdHash {
type Hasher = IdHasher;
fn build_hasher(&self) -> IdHasher {
IdHasher(0)
}
}
impl std::hash::Hasher for IdHasher {
fn finish(&self) -> u64 {
self.0
}
fn write(&mut self, bytes: &[u8]) {
// never used by a u128 key; a correct fallback all the same
for b in bytes {
self.0 = (self.0.rotate_left(5) ^ u64::from(*b)).wrapping_mul(0x51_7c_c1_b7_27_22_0a_95);
}
}
fn write_u128(&mut self, v: u128) {
self.0 = (v as u64) ^ ((v >> 64) as u64);
}
}
/// [`identity`]'s bytes appended to `buf` — the same string, never built.
pub(crate) fn identity_into(stable: bool, keys: &[StringArray], dup: &StringArray, i: usize, buf: &mut Vec<u8>) {
let mut ch = [0u8; 4];
for (p, k) in keys.iter().enumerate() {
if p > 0 {
buf.extend_from_slice(KEY_SEP.encode_utf8(&mut ch).as_bytes());
}
if k.is_null(i) {
buf.extend_from_slice(KEY_NULL.encode_utf8(&mut ch).as_bytes());
} else if stable {
stable_into(k.value(i), buf);
} else {
buf.extend_from_slice(k.value(i).as_bytes());
}
}
if !dup.is_null(i) {
buf.extend_from_slice(KEY_SEP.encode_utf8(&mut ch).as_bytes());
buf.extend_from_slice(dup.value(i).as_bytes());
}
}
/// The identity of row `i`: key columns joined with U+001F (null = U+0000),
/// plus the `key_dup` part when present — the same string the delta keys by.
/// `stable` ([`PayloadSpec::stable_keys`]) renders every moniker with its
/// version as `*`, since the stored columns keep their real versions.
pub(crate) fn identity(stable: bool, keys: &[StringArray], dup: &StringArray, i: usize) -> String {
let mut key = String::new();
for (p, k) in keys.iter().enumerate() {
if p > 0 {
key.push(KEY_SEP);
}
if k.is_null(i) {
key.push(KEY_NULL);
} else if stable {
push_stable(&mut key, k.value(i));
} else {
key.push_str(k.value(i));
}
}
if !dup.is_null(i) {
key.push(KEY_SEP);
key.push_str(dup.value(i));
}
key
}
fn bin_offset_batch(rows: &[BinRunRow], now: i64) -> Result<RecordBatch> {
Ok(RecordBatch::try_new(
Arc::new(bin_offsets_schema()),
vec![
str_arr(rows.iter().map(|r| r.repo.as_str())),
str_arr(rows.iter().map(|r| r.artifact.as_str())),
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.rev))),
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.seq))),
str_arr(rows.iter().map(|r| r.sha.as_str())),
ts_arr(vec![now; rows.len()]),
],
)?)
}
fn bin_run_batch(rows: &[BinRunRow], now: i64) -> Result<RecordBatch> {
let i64s = |f: fn(&BinRunRow) -> i64| -> ArrayRef { Arc::new(Int64Array::from_iter_values(rows.iter().map(f))) };
Ok(RecordBatch::try_new(
Arc::new(bin_ingest_runs_schema()),
vec![
str_arr(rows.iter().map(|r| r.repo.as_str())),
i64s(|r| r.seq),
i64s(|r| r.rev),
str_arr(rows.iter().map(|r| r.sha.as_str())),
str_arr(rows.iter().map(|r| r.artifact.as_str())),
str_arr(rows.iter().map(|r| r.status.as_str())),
i64s(|r| r.rows_plus),
i64s(|r| r.rows_minus),
i64s(|r| r.key_collisions),
i64s(|r| r.ms),
Arc::new(StringArray::from(rows.iter().map(|r| r.ledger.clone()).collect::<Vec<_>>())),
ts_arr(vec![now; rows.len()]),
],
)?)
}
fn run_batch(r: &RunRow, now: i64) -> Result<RecordBatch> {
Ok(RecordBatch::try_new(
Arc::new(ingest_runs_schema()),
vec![
str_arr([r.repo.as_str()]),
Arc::new(Int64Array::from(vec![r.seq])),
str_arr([r.sha.as_str()]),
str_arr([r.status.as_str()]),
Arc::new(Int64Array::from(vec![r.crates_dumped])),
Arc::new(Int64Array::from(vec![r.rows_plus])),
Arc::new(Int64Array::from(vec![r.rows_minus])),
Arc::new(Int64Array::from(vec![r.key_collisions])),
Arc::new(Int64Array::from(vec![r.span_plus])),
Arc::new(Int64Array::from(vec![r.span_minus])),
Arc::new(Int64Array::from(vec![r.owner_disambiguations])),
Arc::new(Int64Array::from(vec![r.ms])),
Arc::new(StringArray::from(vec![r.error.clone()])),
ts_arr(vec![now]),
],
)?)
}
fn str_arr<'a>(v: impl IntoIterator<Item = &'a str>) -> ArrayRef {
Arc::new(StringArray::from_iter_values(v))
}
fn ts_arr(v: Vec<i64>) -> ArrayRef {
Arc::new(TimestampMicrosecondArray::from(v).with_timezone("+00:00"))
}
fn now_micros() -> i64 {
SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_micros() as i64).unwrap_or(0)
}
/// Column `name` as Utf8 (cast when the reader handed back another string type).
pub(crate) fn str_col(b: &RecordBatch, name: &str) -> Result<StringArray> {
let c = b.column_by_name(name).with_context(|| format!("column `{name}` missing"))?;
let c = if c.data_type() == &DataType::Utf8 { c.clone() } else { skade::arrow_cast::cast(c, &DataType::Utf8)? };
Ok(c.as_any().downcast_ref::<StringArray>().context("utf8 downcast")?.clone())
}
pub(crate) fn i64_col(b: &RecordBatch, name: &str) -> Result<Int64Array> {
let c = b.column_by_name(name).with_context(|| format!("column `{name}` missing"))?;
Ok(c.as_any()
.downcast_ref::<Int64Array>()
.with_context(|| format!("column `{name}` is {:?}, not Int64", c.data_type()))?
.clone())
}