src/warehouse/feature_dep_edges.rs
//! Iceberg writer + reader for the **feature ACTIVATION graph**
//! (`feature_dep_edges`) — one row per `Cargo.toml` edge saying *which feature
//! enables which dependency or which other feature*.
//!
//! WHY IT EXISTS. `cargo metadata` cannot answer this. Its `features` map is
//! already flattened and resolved, so by the time cargo hands the data over, the
//! conditional structure is gone: you can see *that* a dependency is in the
//! resolved graph, not *which* feature would have to be switched off to drop it,
//! nor whether several routes reach it. Above all, the **weak** form `pkg?/feat`
//! — enable `feat` on `pkg` **only if `pkg` was already enabled by some other
//! route** — is the one edge that creates an AND-group, and it survives nowhere
//! in cargo's output. Getting `weak-dep-feature` rows into the warehouse is the
//! entire point of this table.
//!
//! **ONE ROW TYPE (L5).** The row struct, the `EnableKind`/`DepKind` enums and
//! their `as_str()` codecs all come from the [`nornir_feature_graph`] leaf, which
//! is also the PARSER that produces them. This module deliberately does not
//! define its own mirror of any of them. It did at first — a POD struct with
//! `kind: String` next to hand-written `"weak-dep-feature"` constants — and that
//! copy had *already* drifted before it ever shipped: it spelled a weak edge's
//! payload `serde?/derive`, while the parser emits `serde/derive` and carries the
//! `?` in `kind`. The leaf's own [`nornir_feature_graph::universe_from_rows`]
//! splits `enables` on `/` to recover the dep name, so those rows would have
//! yielded a dependency literally named `serde?`. Nothing would have errored.
//! Sharing the type deletes that whole class of bug by construction.
//!
//! Write path: [`append_feature_dep_edges`] appends one Iceberg snapshot per
//! parse; all rows share one `snapshot_id` + one `ts_micros`.
//!
//! Read path: [`query_feature_dep_edges`] reads the LATEST snapshot back, scoped
//! to a repo. It **errors rather than returning an empty `Vec`** when the table
//! has never been written for the requested scope — see the honesty note on that
//! function; an empty read here would be indistinguishable from the genuinely
//! meaningful answer "nothing pulls this dependency".
//!
//! Schema: [`super::iceberg_schema::feature_dep_edges`] — additive (ships via the
//! standard schema-evolution path; old binaries keep reading).
use std::sync::Arc;
use anyhow::{Context, Result, anyhow};
use arrow::array::{Array, BooleanArray, RecordBatch, StringArray, TimestampMicrosecondArray};
use chrono::Utc;
use iceberg::arrow::schema_to_arrow_schema;
use super::iceberg::{IcebergWarehouse, TABLE_FEATURE_DEP_EDGES, append_batch};
/// The row type, the two kind enums and their string codecs — re-exported from
/// the parser leaf so this table has exactly ONE definition of its own shape.
/// `EnableKind::as_str()` yields `dep` | `feature` | `dep-feature` |
/// `weak-dep-feature` | `implicit-dep`; `DepKind::as_str()` yields `normal` |
/// `build` | `dev`. Those are the `kind` / `dep_kind` column domains, and they
/// are the *same* function the writer below calls.
pub use nornir_feature_graph::{DepKind, EnableKind, FeatureDepEdgeRow};
// Column order matches `iceberg_schema::feature_dep_edges`. Positional access is
// valid because the readers below do an UNPROJECTED scan (no `.select`).
const COL_SNAPSHOT_ID: usize = 0;
const COL_TS_MICROS: usize = 1;
const COL_REPO: usize = 2;
const COL_CRATE_NAME: usize = 3;
const COL_FEATURE: usize = 4;
const COL_ENABLES: usize = 5;
const COL_KIND: usize = 6;
const COL_OPTIONAL: usize = 7;
const COL_TARGET_CFG: usize = 8;
const COL_DEP_KIND: usize = 9;
/// True when `row` is pulled with NO feature behind it (`feature == ""`), i.e.
/// the ⊥ "cannot be removed by any feature toggle" case. The empty string is a
/// MEANINGFUL value here, not a missing one.
pub fn is_unconditional(row: &FeatureDepEdgeRow) -> bool {
row.feature.is_empty()
}
/// True for the `pkg?/feat` WEAK form — the only kind that creates an AND-group
/// (it activates only if `pkg` is enabled by another route).
pub fn is_weak(row: &FeatureDepEdgeRow) -> bool {
row.kind == EnableKind::WeakDepFeature
}
/// Append `rows` under ONE logical snapshot: every row gets `snapshot_id` and
/// **one** `ts_micros`, whatever mix of repos the slice holds. A no-op on an empty
/// slice (nothing is written), which is exactly why the reader below must be able
/// to tell "never written" from "written and empty".
///
/// **`snapshot_id` is the CALLER's**, and that is load-bearing: a populate parses
/// every member and must write them all under ONE id, because the reader returns
/// the latest snapshot only. One id per member would make a multi-member
/// populate readable as "the last member is the whole workspace". See
/// `monitor` step 7d, and `two_members_in_one_snapshot_both_read_back`.
///
/// ## Why this groups by `repo` itself
///
/// The table is PARTITIONED by `repo`, and skade enforces one-partition-per-append
/// (`batch spans multiple 'repo' partitions ('alpha' vs 'beta')`). So a whole
/// workspace cannot go in a single `RecordBatch`. The obvious fix at the call site
/// — loop the members and append each — is a trap: the natural way to write that
/// loop mints a fresh id per iteration, which is exactly the bug the single-id
/// contract exists to prevent, and it fails SILENTLY (the reader returns the last
/// member and calls it the workspace).
///
/// So the split lives HERE, below the contract: the caller hands over every row
/// and one id, and this fn does one Iceberg append per `repo` partition while
/// stamping the SAME `snapshot_id` and the SAME `ts_micros` on all of them. The
/// caller cannot get the invariant wrong because it never sees the seam (L5 —
/// fix by construction, not by a guard watching two call sites agree).
///
/// Uses [`IcebergWarehouse::load_table_for_append`] (additive schema evolution +
/// the torn-metadata self-heal) and [`append_batch`], the same path
/// [`super::dep_graph::record_dep_graph`] takes. Like that writer, the optional
/// column is guarded on the LOADED table's schema rather than assumed present, so
/// a `feature_dep_edges` created by an older binary (no `target_cfg`) still takes
/// the append instead of failing on an arity mismatch.
pub async fn append_feature_dep_edges(
wh: &IcebergWarehouse,
snapshot_id: &str,
rows: &[FeatureDepEdgeRow],
) -> Result<()> {
if rows.is_empty() {
return Ok(());
}
// ONE timestamp for the whole call, shared across every partition's append —
// so "one populate" is one `(snapshot_id, ts_micros)` pair no matter how many
// repos it spanned, and the reader's `max(ts)` cannot split it.
let ts = Utc::now().timestamp_micros();
// Group by the partition column, preserving row order within each group.
let mut by_repo: std::collections::BTreeMap<&str, Vec<&FeatureDepEdgeRow>> =
std::collections::BTreeMap::new();
for r in rows {
by_repo.entry(r.repo.as_str()).or_default().push(r);
}
for (_repo, group) in by_repo {
let table = wh
.load_table_for_append(
TABLE_FEATURE_DEP_EDGES,
super::iceberg_schema::feature_dep_edges()?,
&["repo"],
)
.await?;
let schema = table.metadata().current_schema().clone();
let arrow_schema = Arc::new(schema_to_arrow_schema(&schema)?);
let mut cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec![snapshot_id; group.len()])),
Arc::new(
TimestampMicrosecondArray::from(vec![ts; group.len()]).with_timezone("+00:00"),
),
Arc::new(StringArray::from(
group.iter().map(|r| r.repo.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
group
.iter()
.map(|r| r.crate_name.clone())
.collect::<Vec<_>>(),
)),
// NB: `feature` is written VERBATIM — an empty string is the meaningful
// "unconditional" value, never coerced to NULL or a placeholder.
Arc::new(StringArray::from(
group.iter().map(|r| r.feature.clone()).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
group.iter().map(|r| r.enables.clone()).collect::<Vec<_>>(),
)),
// The codec is the LEAF's own `as_str()`, not a local string literal —
// so the stored value is by construction one the reader's `parse()`
// accepts.
Arc::new(StringArray::from(
group
.iter()
.map(|r| r.kind.as_str().to_string())
.collect::<Vec<_>>(),
)),
Arc::new(BooleanArray::from(
group.iter().map(|r| r.optional).collect::<Vec<_>>(),
)),
];
// `target_cfg` is the one OPTIONAL column: present-guard it the way
// `record_dep_graph` guards its later-added columns, and carry NULL through
// as NULL (a null-vs-empty-string mix-up would make "no cfg" look like
// `cfg()`).
if schema.field_by_name("target_cfg").is_some() {
cols.push(Arc::new(StringArray::from(
group
.iter()
.map(|r| r.target_cfg.clone())
.collect::<Vec<Option<String>>>(),
)));
}
if schema.field_by_name("dep_kind").is_some() {
cols.push(Arc::new(StringArray::from(
group
.iter()
.map(|r| r.dep_kind.as_str().to_string())
.collect::<Vec<_>>(),
)));
}
let batch = RecordBatch::try_new(arrow_schema, cols)?;
append_batch(wh.catalog(), table, batch).await?;
}
Ok(())
}
/// Read the feature-activation edges of the LATEST snapshot back, optionally
/// scoped to one `repo`. Rows come back sorted (the leaf's derived `Ord`).
///
/// **This errors instead of returning `Ok(vec![])` when the scope has never been
/// written.** That is deliberate and load-bearing. The questions this table
/// answers are negative ones — "which features pull `openssl`?", "can this
/// dependency be removed?" — so an empty `Vec` is itself a meaningful answer:
/// *nothing pulls it, it can be dropped*. If a never-populated table also
/// returned an empty `Vec`, "we never parsed the manifests" would be reported to
/// the user as "this dependency is unused", which is the exact confusion this
/// table exists to kill. The error names the warehouse root it read so the caller
/// can see WHICH warehouse was empty. (The leaf takes the same line one level up:
/// `nornir_feature_graph::universe_from_rows` refuses an empty row slice.)
///
/// It also errors on an UNCLASSIFIABLE `kind`/`dep_kind` string rather than
/// guessing a default. A row the reader cannot classify is a producer/consumer
/// disagreement, and defaulting it would silently turn (say) an unknown weak edge
/// into an unconditional one — the exact misreport this table exists to prevent.
pub async fn query_feature_dep_edges_async(
wh: &IcebergWarehouse,
repo: Option<&str>,
) -> Result<Vec<FeatureDepEdgeRow>> {
let batches: Vec<RecordBatch> = match repo {
// Push the repo predicate into the scan so the `repo` partition prunes at
// plan time. No projection ⇒ positional downcast stays valid.
Some(r) => {
super::iceberg::load_and_read_filtered(
wh,
TABLE_FEATURE_DEP_EDGES,
&skade::ScanFilter::eq("repo", r),
&[],
)
.await?
}
None => super::iceberg::load_and_read_all(wh, TABLE_FEATURE_DEP_EDGES).await?,
};
// Latest snapshot in scope (max ts_micros), so repeated parses of the same
// repo don't fold every historical generation together.
let mut latest: Option<(String, i64)> = None;
for b in &batches {
let snaps = col_str(b, COL_SNAPSHOT_ID)?;
let ts = col_ts(b, COL_TS_MICROS)?;
let repos = col_str(b, COL_REPO)?;
for i in 0..b.num_rows() {
// Residual guard: pushdown prunes at file granularity, not per row.
if repo.is_some_and(|r| repos.value(i) != r) {
continue;
}
let t = ts.value(i);
if latest.as_ref().is_none_or(|(_, lt)| t > *lt) {
latest = Some((snaps.value(i).to_string(), t));
}
}
}
// THE NON-VACUOUS ANCHOR: no snapshot in scope ⇒ the table was never written
// here. Refuse to answer rather than answer "nothing".
let Some((snap, _)) = latest else {
let scope = match repo {
Some(r) => format!("repo `{r}`"),
None => "any repo".to_string(),
};
return Err(anyhow!(
"`{TABLE_FEATURE_DEP_EDGES}` has never been populated for {scope} in the warehouse at \
`{}` — refusing to return an empty result, which would read as \
\"no feature pulls this dependency\" when the truth is \"the manifests were never \
parsed\". Run the Cargo.toml feature-graph scan for this repo first.",
wh.root().display()
));
};
let mut out: Vec<FeatureDepEdgeRow> = Vec::new();
for b in &batches {
let snaps = col_str(b, COL_SNAPSHOT_ID)?;
let repos = col_str(b, COL_REPO)?;
let crate_name = col_str(b, COL_CRATE_NAME)?;
let feature = col_str(b, COL_FEATURE)?;
let enables = col_str(b, COL_ENABLES)?;
let kind = col_str(b, COL_KIND)?;
let optional = col_bool(b, COL_OPTIONAL)?;
let target_cfg = col_str(b, COL_TARGET_CFG)?;
let dep_kind = col_str(b, COL_DEP_KIND)?;
for i in 0..b.num_rows() {
if snaps.value(i) != snap {
continue;
}
if repo.is_some_and(|r| repos.value(i) != r) {
continue;
}
let raw_kind = kind.value(i);
let raw_dep_kind = dep_kind.value(i);
out.push(FeatureDepEdgeRow {
repo: repos.value(i).to_string(),
crate_name: crate_name.value(i).to_string(),
feature: feature.value(i).to_string(),
enables: enables.value(i).to_string(),
kind: EnableKind::parse(raw_kind).ok_or_else(|| {
anyhow!(
"`{TABLE_FEATURE_DEP_EDGES}` row (repo={}, crate={}, feature={:?}, \
enables={:?}) has an unclassifiable `kind` = {raw_kind:?}; legal values \
are dep | feature | dep-feature | weak-dep-feature | implicit-dep. \
Refusing to guess: defaulting an unknown kind would silently turn a \
conditional (weak) activation into an unconditional one.",
repos.value(i),
crate_name.value(i),
feature.value(i),
enables.value(i),
)
})?,
optional: optional.value(i),
// NULL stays `None`; an empty string would be a DIFFERENT (and
// wrong) answer — `cfg('')` rather than "no target predicate".
target_cfg: target_cfg
.is_valid(i)
.then(|| target_cfg.value(i).to_string()),
dep_kind: DepKind::parse(raw_dep_kind).ok_or_else(|| {
anyhow!(
"`{TABLE_FEATURE_DEP_EDGES}` row (repo={}, crate={}, enables={:?}) has an \
unclassifiable `dep_kind` = {raw_dep_kind:?}; legal values are \
normal | build | dev.",
repos.value(i),
crate_name.value(i),
enables.value(i),
)
})?,
});
}
}
out.sort();
out.dedup();
Ok(out)
}
/// Sync wrapper over [`query_feature_dep_edges_async`], on the warehouse's own
/// runtime (the local convention — cf. `IcebergWarehouse::record_dep_graph`).
pub fn query_feature_dep_edges(
wh: &IcebergWarehouse,
repo: Option<&str>,
) -> Result<Vec<FeatureDepEdgeRow>> {
wh.block_on(query_feature_dep_edges_async(wh, repo))
}
impl IcebergWarehouse {
/// Sync wrapper over [`append_feature_dep_edges`] on this warehouse's private
/// runtime — the method form matching `record_dep_graph`.
pub fn append_feature_dep_edges(
&self,
snapshot_id: &str,
rows: &[FeatureDepEdgeRow],
) -> Result<()> {
self.block_on(append_feature_dep_edges(self, snapshot_id, rows))
}
/// Sync wrapper over [`query_feature_dep_edges_async`]. Errors when the table
/// was never populated for `repo` — see that function's honesty note.
pub fn query_feature_dep_edges(&self, repo: Option<&str>) -> Result<Vec<FeatureDepEdgeRow>> {
self.block_on(query_feature_dep_edges_async(self, repo))
}
}
fn col_str<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a StringArray> {
b.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.with_context(|| format!("feature_dep_edges col {idx} is not StringArray"))
}
fn col_bool<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a BooleanArray> {
b.column(idx)
.as_any()
.downcast_ref::<BooleanArray>()
.with_context(|| format!("feature_dep_edges col {idx} is not BooleanArray"))
}
fn col_ts<'a>(b: &'a RecordBatch, idx: usize) -> Result<&'a TimestampMicrosecondArray> {
b.column(idx)
.as_any()
.downcast_ref::<TimestampMicrosecondArray>()
.with_context(|| format!("feature_dep_edges col {idx} is not TimestampMicrosecondArray"))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::warehouse::iceberg::IcebergWarehouse;
#[allow(clippy::too_many_arguments)]
fn edge_in(
repo: &str,
crate_name: &str,
feature: &str,
enables: &str,
kind: EnableKind,
optional: bool,
target_cfg: Option<&str>,
dep_kind: DepKind,
) -> FeatureDepEdgeRow {
FeatureDepEdgeRow {
repo: repo.into(),
crate_name: crate_name.into(),
feature: feature.into(),
enables: enables.into(),
kind,
optional,
target_cfg: target_cfg.map(str::to_string),
dep_kind,
}
}
fn edge(
crate_name: &str,
feature: &str,
enables: &str,
kind: EnableKind,
optional: bool,
target_cfg: Option<&str>,
dep_kind: DepKind,
) -> FeatureDepEdgeRow {
edge_in(
"korp", crate_name, feature, enables, kind, optional, target_cfg, dep_kind,
)
}
/// LAW 2 round-trip: append a handful of edges — including the
/// `weak-dep-feature` (`pkg?/feat`) row this table exists for and the
/// `feature = ""` UNCONDITIONAL row — then read them back and assert **every
/// one of the eight payload columns** of every row, by value. Not a count,
/// not `is_ok()`: a row count cannot tell `dep` from `weak-dep-feature`, and
/// `weak` vs non-weak is precisely the distinction that decides whether the
/// activation is an AND-group.
#[test]
fn round_trip_preserves_every_payload_column_exactly() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let rows = vec![
// THE reason this table exists: `serde?/derive` is weak — it only
// fires if `serde` is enabled by another route. NOTE the payload is
// `serde/derive`, WITHOUT the `?`: the weakness lives in `kind`, and
// the leaf's `universe_from_rows` splits this on `/` to recover the
// dep name, so a `?` here would invent a dep called `serde?`.
edge(
"korp-core",
"json",
"serde/derive",
EnableKind::WeakDepFeature,
true,
None,
DepKind::Normal,
),
// UNCONDITIONAL: no feature behind it ⇒ `anyhow` cannot be removed by
// any feature toggle. `feature` is deliberately the empty string.
edge(
"korp-core",
"",
"anyhow",
EnableKind::Dep,
false,
None,
DepKind::Normal,
),
// Target-gated build dep — exercises the OPTIONAL `target_cfg` column
// carrying a real value alongside the NULLs above.
edge(
"korp-core",
"windows-shim",
"winapi",
EnableKind::Dep,
true,
Some("cfg(windows)"),
DepKind::Build,
),
// Feature → feature, and a strong dep-feature, and a dev-only edge.
edge(
"korp-core",
"full",
"json",
EnableKind::Feature,
false,
None,
DepKind::Normal,
),
edge(
"korp-core",
"full",
"tokio/rt-multi-thread",
EnableKind::DepFeature,
false,
None,
DepKind::Normal,
),
edge(
"korp-core",
"",
"criterion",
EnableKind::ImplicitDep,
true,
Some("cfg(unix)"),
DepKind::Dev,
),
];
wh.append_feature_dep_edges("snap-round-trip", &rows)
.unwrap();
let back = query_feature_dep_edges(&wh, Some("korp")).unwrap();
// Exact set equality on the FULL payload of every row (all 8 columns:
// repo, crate_name, feature, enables, kind, optional, target_cfg,
// dep_kind — the leaf's derived `PartialEq` covers each).
let mut expect = rows.clone();
expect.sort();
assert_eq!(back, expect, "every payload column round-trips by value");
// Field-by-field on the two rows whose exact values are load-bearing, so a
// failure names the column rather than dumping two vecs.
let weak = back
.iter()
.find(|r| r.enables == "serde/derive")
.expect("the weak-dep-feature row survived the round-trip");
assert_eq!(weak.repo, "korp");
assert_eq!(weak.crate_name, "korp-core");
assert_eq!(weak.feature, "json");
assert_eq!(
weak.kind,
EnableKind::WeakDepFeature,
"the AND-group kind must survive VERBATIM — `dep-feature` here would \
silently turn a conditional activation into an unconditional one"
);
assert_eq!(
weak.kind.as_str(),
"weak-dep-feature",
"and its stored spelling"
);
assert!(
!weak.kind.activates_dep(),
"the weak form does NOT pull the dep in"
);
assert!(weak.optional, "serde is an optional dep");
assert_eq!(
weak.target_cfg, None,
"no target predicate ⇒ NULL, not \"\""
);
assert_eq!(weak.dep_kind, DepKind::Normal);
assert!(is_weak(weak));
assert!(!is_unconditional(weak));
let uncond = back
.iter()
.find(|r| r.enables == "anyhow")
.expect("the unconditional row survived the round-trip");
assert_eq!(uncond.repo, "korp");
assert_eq!(uncond.crate_name, "korp-core");
assert_eq!(
uncond.feature, "",
"the EMPTY feature is the ⊥ \"cannot be removed\" marker and must not \
be coerced to NULL or a placeholder"
);
assert_eq!(uncond.enables, "anyhow");
assert_eq!(uncond.kind, EnableKind::Dep);
assert!(!uncond.optional);
assert_eq!(uncond.target_cfg, None);
assert_eq!(uncond.dep_kind, DepKind::Normal);
assert!(is_unconditional(uncond));
assert!(!is_weak(uncond));
// The one row with a REAL target_cfg — proves the optional column carries
// a value through, not just NULLs (an all-NULL column would pass a
// "target_cfg is None" check while being completely dead).
let win = back
.iter()
.find(|r| r.enables == "winapi")
.expect("winapi row");
assert_eq!(win.target_cfg.as_deref(), Some("cfg(windows)"));
assert_eq!(win.dep_kind, DepKind::Build);
assert_eq!(win.feature, "windows-shim");
assert!(win.optional);
let dev = back
.iter()
.find(|r| r.enables == "criterion")
.expect("criterion row");
assert_eq!(
dev.dep_kind,
DepKind::Dev,
"dev-deps stay labelled so `-e no-dev` can drop them"
);
assert_eq!(dev.kind, EnableKind::ImplicitDep);
assert_eq!(dev.target_cfg.as_deref(), Some("cfg(unix)"));
assert_eq!(dev.feature, "");
// Every stored `kind`/`dep_kind` must be a value the LEAF can classify —
// the writer/reader codec agreement, checked on real round-tripped data.
for r in &back {
assert_eq!(EnableKind::parse(r.kind.as_str()), Some(r.kind), "{r:?}");
assert_eq!(
DepKind::parse(r.dep_kind.as_str()),
Some(r.dep_kind),
"{r:?}"
);
}
}
/// LAW 2 anchor: an UNPOPULATED table must be an `Err` naming the warehouse
/// path, never `Ok(vec![])`. An empty vec here would read as "nothing pulls
/// this dependency" — the answer that gets a still-needed dependency deleted.
#[test]
fn unpopulated_table_errors_and_names_the_warehouse_path() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let err = query_feature_dep_edges(&wh, Some("korp"))
.expect_err("an unpopulated feature_dep_edges must ERROR, not report zero edges")
.to_string();
assert!(
err.contains("feature_dep_edges"),
"error names the table: {err}"
);
assert!(
err.contains("korp"),
"error names the requested repo: {err}"
);
assert!(
err.contains(&dir.path().display().to_string()),
"error must name the warehouse path it read ({}) so the caller can see \
WHICH warehouse was empty: {err}",
dir.path().display()
);
// Same for the unscoped read.
let err_all = query_feature_dep_edges(&wh, None)
.expect_err("unscoped read of an unpopulated table must ERROR too")
.to_string();
assert!(
err_all.contains(&dir.path().display().to_string()),
"{err_all}"
);
// And it must go GREEN the moment real rows land — otherwise the guard is
// just "always red" and proves nothing.
wh.append_feature_dep_edges(
"snap-now-populated",
&[edge(
"korp-core",
"",
"anyhow",
EnableKind::Dep,
false,
None,
DepKind::Normal,
)],
)
.unwrap();
let ok =
query_feature_dep_edges(&wh, Some("korp")).expect("a populated table reads back fine");
assert_eq!(ok.len(), 1);
assert_eq!(ok[0].enables, "anyhow");
// A repo that was never written STILL errors even though the table now has
// rows — the scope is what matters, not the table's global row count.
let other = query_feature_dep_edges(&wh, Some("no-such-repo"))
.expect_err("an unwritten repo scope must error even on a non-empty table")
.to_string();
assert!(other.contains("no-such-repo"), "{other}");
}
/// A second parse of the same repo must NOT fold both generations together —
/// the reader takes the latest snapshot only. (The same append-only
/// double-count `SNAPSHOT_TABLES` guards against on the lens side.)
#[test]
fn second_parse_supersedes_the_first_instead_of_doubling() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
wh.append_feature_dep_edges(
"snap-1",
&[edge(
"c",
"old",
"legacy-dep",
EnableKind::Dep,
true,
None,
DepKind::Normal,
)],
)
.unwrap();
// Distinct wall-clock micros so `max(ts_micros)` is unambiguous.
std::thread::sleep(std::time::Duration::from_millis(2));
wh.append_feature_dep_edges(
"snap-2",
&[edge(
"c",
"new",
"fresh-dep",
EnableKind::Dep,
true,
None,
DepKind::Normal,
)],
)
.unwrap();
let back = query_feature_dep_edges(&wh, Some("korp")).unwrap();
assert_eq!(
back.len(),
1,
"latest snapshot only, not both generations: {back:?}"
);
assert_eq!(back[0].enables, "fresh-dep");
assert_eq!(back[0].feature, "new");
}
/// **THE MULTI-MEMBER SNAPSHOT TRAP.** A populate parses N members and must
/// write them under ONE snapshot id. If the producer gave each member its own
/// id, this reader — which by design keeps only the newest snapshot — would
/// return the LAST member's rows and silently drop every other member. The
/// workspace would then read as "only `beta` has any dependencies", which is a
/// confident, plausible, completely wrong answer.
///
/// This is the read-side guard on that contract: two members, one id, both
/// must come back — scoped per repo AND in the unscoped whole-workspace read.
#[test]
fn two_members_in_one_snapshot_both_read_back() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let rows = vec![
edge_in(
"alpha",
"alpha-core",
"",
"anyhow",
EnableKind::Dep,
false,
None,
DepKind::Normal,
),
edge_in(
"alpha",
"alpha-core",
"tls",
"openssl/vendored",
EnableKind::WeakDepFeature,
true,
None,
DepKind::Normal,
),
edge_in(
"beta",
"beta-core",
"",
"serde",
EnableKind::Dep,
false,
None,
DepKind::Normal,
),
];
// ONE id for the whole run — exactly what `monitor` step 7d does.
wh.append_feature_dep_edges("one-populate", &rows).unwrap();
let alpha = query_feature_dep_edges(&wh, Some("alpha")).unwrap();
assert_eq!(
alpha.len(),
2,
"alpha's rows survived a multi-member populate: {alpha:?}"
);
assert!(alpha.iter().all(|r| r.repo == "alpha"));
assert!(
alpha.iter().any(|r| r.kind == EnableKind::WeakDepFeature),
"alpha's weak edge is not the last member's, and must not be dropped: {alpha:?}"
);
let beta = query_feature_dep_edges(&wh, Some("beta")).unwrap();
assert_eq!(beta.len(), 1, "beta's rows survived too: {beta:?}");
assert_eq!(beta[0].enables, "serde");
// The whole-workspace read sees BOTH members, not just the last one.
let all = query_feature_dep_edges(&wh, None).unwrap();
let repos: std::collections::BTreeSet<&str> = all.iter().map(|r| r.repo.as_str()).collect();
assert_eq!(
repos,
["alpha", "beta"].into_iter().collect(),
"an unscoped read must return every member of the populate, not the last \
one written: {all:?}"
);
assert_eq!(all.len(), 3);
}
/// The reader refuses a `kind` string it cannot classify instead of guessing.
/// Written directly through Arrow (the typed writer cannot produce one), which
/// is exactly the drift this guards: a future producer — or another language's
/// writer — putting an unknown token in the column.
#[test]
fn unclassifiable_kind_is_an_error_not_a_guess() {
let dir = tempfile::tempdir().unwrap();
let wh = IcebergWarehouse::open(dir.path()).unwrap();
let table = wh
.block_on(wh.load_table_for_append(
TABLE_FEATURE_DEP_EDGES,
crate::warehouse::iceberg_schema::feature_dep_edges().unwrap(),
&["repo"],
))
.unwrap();
let arrow_schema =
Arc::new(schema_to_arrow_schema(table.metadata().current_schema()).unwrap());
let cols: Vec<Arc<dyn Array>> = vec![
Arc::new(StringArray::from(vec!["snap-bogus"])),
Arc::new(
TimestampMicrosecondArray::from(vec![Utc::now().timestamp_micros()])
.with_timezone("+00:00"),
),
Arc::new(StringArray::from(vec!["korp"])),
Arc::new(StringArray::from(vec!["korp-core"])),
Arc::new(StringArray::from(vec!["json"])),
Arc::new(StringArray::from(vec!["serde/derive"])),
// Not one of the five legal kinds.
Arc::new(StringArray::from(vec!["sort-of-weak-ish"])),
Arc::new(BooleanArray::from(vec![true])),
Arc::new(StringArray::from(vec![None::<String>])),
Arc::new(StringArray::from(vec!["normal"])),
];
let batch = RecordBatch::try_new(arrow_schema, cols).unwrap();
wh.block_on(crate::warehouse::iceberg::append_batch(
wh.catalog(),
table,
batch,
))
.unwrap();
let err = query_feature_dep_edges(&wh, Some("korp"))
.expect_err("an unclassifiable kind must ERROR, not silently default")
.to_string();
assert!(
err.contains("sort-of-weak-ish"),
"error quotes the bad value: {err}"
);
assert!(
err.contains("weak-dep-feature"),
"error lists the legal values: {err}"
);
}
}