From f7cdab072780b27236929d1dcd578a66a8b5863d Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 11:16:56 +0000 Subject: [PATCH 1/6] hydration_probe: close the idle-flush plan's named blocker with a measurement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The plan (.claude/plans/idle-flush-dataset-eviction-v1.md) is a PROPOSAL whose own §9.1 names the first task: the §4 verification gate — 'cheap local version read, ASSUMED, UNCHECKED'. This probe measures it instead of assuming it, plus three other inputs the plan grades as unmeasured: (1) §4 gate — Dataset::open vs latest_version_id, three sizes, warm. lance 9 documents latest_version_id as a fast path; whether the ratio is DECISIVE is what the gate actually asks. (2) §1 cost — file count + wall time per hydration. The plan states its own cost model omits request count because 'a dataset is a multi-file directory'; this counts the files. (3) §0/§5 — the ~1.4 s rehydration figure is graded a single observation. Three sizes so it lands on a curve, not a point. (4) T10 — flush -> rehydrate -> read equality, by full-scan checksum (a truncated hydration cannot pass a row-count check). Storage options are built explicitly: object_store reads AWS_ENDPOINT while this environment sets AWS_ENDPOINT_URL, so from_env would silently address AWS proper and fail as if it were a credential problem. Measures only. Implements no eviction and authorizes none; the policy stays a proposal. Removes only the bytes it wrote, and says so for the remote ones. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NMeiLmtDKhomJNSo2ecbJw --- .../lance-graph/examples/hydration_probe.rs | 355 ++++++++++++++++++ 1 file changed, 355 insertions(+) create mode 100644 crates/lance-graph/examples/hydration_probe.rs diff --git a/crates/lance-graph/examples/hydration_probe.rs b/crates/lance-graph/examples/hydration_probe.rs new file mode 100644 index 00000000..5c7b7067 --- /dev/null +++ b/crates/lance-graph/examples/hydration_probe.rs @@ -0,0 +1,355 @@ +//! `hydration_probe` — closes the **§4 verification gate** of +//! `.claude/plans/idle-flush-dataset-eviction-v1.md`, and measures the +//! hydration cost that plan's economics rest on. +//! +//! ```text +//! cargo run -p lance-graph --release --example hydration_probe +//! ``` +//! +//! # What the plan owes, and what this measures +//! +//! The plan is a PROPOSAL with one named blocker (§9.1): +//! +//! > *"The §4 verification gate. Cheap local version read — **assumed, +//! > unchecked**. Closing this is the first task; if it fails, the plan needs a +//! > different dirty-detector and this document is wrong rather than +//! > incomplete."* +//! +//! Four columns, each answering one thing the plan states without evidence: +//! +//! 1. **§4 gate — is a version read cheap?** The dirty detector is +//! `current_local_version != version_at_hydration`. That is only viable if +//! reading the current version is much cheaper than opening the dataset. This +//! times both, at several sizes, **locally** — the sweep runs against local +//! copies, so a local measurement is the one that decides it. +//! 2. **§1 cost model — what does a hydration actually cost?** The plan grades +//! its own economics as incomplete and names the omission: *request count*, +//! since "a dataset is a multi-file directory". This counts the files and +//! measures the wall time against the **real** endpoint. +//! 3. **§0 / §5 — the ~1.4 s rehydration figure.** Graded there as a *single +//! observation, provider- and region-dependent, not re-run*. This re-runs it +//! at three sizes so the shape is visible rather than one point. +//! 4. **T10 — is the round trip lossless?** Flush → rehydrate → read must equal +//! the pre-flush read. The cheapest of the plan's acceptance criteria to +//! settle, and the one whose failure would void the rest. +//! +//! # What this does NOT do +//! +//! It does not implement eviction, and it is **not** an authorisation to. It +//! measures four inputs the plan needs; the policy stays a proposal. Nothing +//! here evicts anything, and the only bytes it removes are the ones it wrote. +//! +//! # Error direction, stated once +//! +//! Every timing here **flatters the fast path**: the version read runs after +//! the dataset has been opened once, so the OS page cache and the object +//! store's own connection pool are warm. A cold version read can only be +//! slower. The gate therefore passes only if the *warm* ratio is already +//! decisive — a marginal warm result is a failed gate, not a close call. +//! +//! Hydration is measured cold in the only sense available in-process: a fresh +//! local directory per run. The remote side's caches are not ours to clear, so +//! a repeated run against the same key may report faster than a first-ever +//! fetch. Treat the numbers as a floor on cost, never a ceiling. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Instant; + +use arrow::array::{Float32Array, Int64Array, RecordBatch, RecordBatchIterator}; +use arrow::datatypes::{DataType, Field, Schema}; +use lance::dataset::{Dataset, WriteMode, WriteParams}; + +/// Row counts to probe. Chosen to bracket the plan's "tens of MB" reference +/// point from both sides, so the reported ~1.4 s can be placed on a curve +/// rather than taken as a constant. +const SIZES: &[usize] = &[10_000, 200_000, 1_000_000]; + +/// Columns per row: one `i64` + eight `f32` = 40 B of payload, so +/// 1,000,000 rows is ~40 MB before encoding — the plan's own scale. +const FLOAT_COLS: usize = 8; + +fn env(k: &str) -> Option { + // The same strip the workspace's other S3 callers apply: these variables + // arrive wrapped in literal quotes in this environment, and an unstripped + // value fails authentication in a way that looks like a credential error + // rather than a parsing one. + std::env::var(k) + .ok() + .map(|v| v.trim().trim_matches('"').trim_matches('\'').to_string()) + .filter(|v| !v.is_empty()) +} + +/// The storage options for the configured endpoint. +/// +/// Built explicitly rather than leaning on `from_env`, because `object_store` +/// reads **`AWS_ENDPOINT`** while this environment sets **`AWS_ENDPOINT_URL`**. +/// Relying on the implicit path would silently address AWS proper instead of +/// the configured endpoint — a failure that reads as a permissions problem. +fn storage_options() -> Option> { + let mut o = HashMap::new(); + o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); + o.insert("aws_secret_access_key".into(), env("AWS_SECRET_ACCESS_KEY")?); + o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); + o.insert( + "aws_region".into(), + env("AWS_DEFAULT_REGION").unwrap_or_else(|| "auto".into()), + ); + // Path-style keeps the request off a per-bucket virtual host, which is what + // the workspace's other S3 caller already assumes for this endpoint. + o.insert("aws_virtual_hosted_style_request".into(), "false".into()); + Some(o) +} + +fn schema() -> Arc { + let mut fields = vec![Field::new("id", DataType::Int64, false)]; + for i in 0..FLOAT_COLS { + fields.push(Field::new(format!("f{i}"), DataType::Float32, false)); + } + Arc::new(Schema::new(fields)) +} + +fn batch(schema: &Arc, rows: usize) -> RecordBatch { + let ids: Int64Array = (0..rows as i64).collect::>().into(); + let mut cols: Vec = vec![Arc::new(ids)]; + for c in 0..FLOAT_COLS { + // Deterministic, and NOT constant per column — a constant column would + // compress to nothing and make the transfer measurement meaningless. + let v: Float32Array = (0..rows) + .map(|i| ((i * 2_654_435_761usize).wrapping_add(c) % 100_003) as f32 * 0.001) + .collect::>() + .into(); + cols.push(Arc::new(v)); + } + RecordBatch::try_new(schema.clone(), cols).expect("batch") +} + +/// Total bytes and file count under a directory — the request-count proxy the +/// plan's §1 says the first-draft cost model omitted. +fn dir_stats(p: &std::path::Path) -> (u64, usize) { + let (mut bytes, mut files) = (0u64, 0usize); + let mut stack = vec![p.to_path_buf()]; + while let Some(d) = stack.pop() { + let Ok(rd) = std::fs::read_dir(&d) else { + continue; + }; + for e in rd.flatten() { + let Ok(ft) = e.file_type() else { continue }; + if ft.is_dir() { + stack.push(e.path()); + } else if let Ok(m) = e.metadata() { + bytes += m.len(); + files += 1; + } + } + } + (bytes, files) +} + +/// Read every row's `id` column, summed — a full scan, so a truncated or +/// partially-hydrated dataset cannot pass as equal. +async fn checksum(ds: &Dataset) -> (u64, i64) { + use futures::TryStreamExt; + let mut stream = ds.scan().try_into_stream().await.expect("scan"); + let (mut rows, mut sum) = (0u64, 0i64); + while let Some(b) = stream.try_next().await.expect("next batch") { + rows += b.num_rows() as u64; + let ids = b + .column_by_name("id") + .expect("id column") + .as_any() + .downcast_ref::() + .expect("id is i64"); + for i in 0..ids.len() { + sum = sum.wrapping_add(ids.value(i)); + } + } + (rows, sum) +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + let tmp = std::env::temp_dir().join(format!("hydration_probe_{}", std::process::id())); + std::fs::create_dir_all(&tmp).expect("scratch dir"); + let schema = schema(); + + println!("scratch: {}", tmp.display()); + println!(); + println!("── (1) §4 GATE — is a version read cheap enough to run per sweep candidate?"); + println!( + "{:>10} {:>8} {:>7} {:>12} {:>14} {:>10}", + "rows", "MB", "files", "open (ms)", "version (ms)", "ratio" + ); + + let mut gate_rows: Vec<(usize, f64, f64, u64, usize)> = Vec::new(); + for &rows in SIZES { + let path = tmp.join(format!("local_{rows}.lance")); + let b = batch(&schema, rows); + let reader = RecordBatchIterator::new(vec![Ok(b)].into_iter(), schema.clone()); + Dataset::write( + reader, + path.to_str().unwrap(), + Some(WriteParams { + mode: WriteMode::Create, + ..Default::default() + }), + ) + .await + .expect("write local"); + + let (bytes, files) = dir_stats(&path); + + // Warm both paths once so neither is charged for first-touch cost; the + // doc comment states why this flatters the fast path. + let warm = Dataset::open(path.to_str().unwrap()).await.expect("open"); + let _ = warm.latest_version_id().await.expect("version"); + + const N: u32 = 10; + let t = Instant::now(); + for _ in 0..N { + let ds = Dataset::open(path.to_str().unwrap()).await.expect("open"); + std::hint::black_box(ds.version().version); + } + let open_ms = t.elapsed().as_secs_f64() * 1e3 / f64::from(N); + + let t = Instant::now(); + for _ in 0..N { + let v = warm.latest_version_id().await.expect("version"); + std::hint::black_box(v); + } + let ver_ms = t.elapsed().as_secs_f64() * 1e3 / f64::from(N); + + println!( + "{rows:>10} {:>8.1} {files:>7} {open_ms:>12.3} {ver_ms:>14.3} {:>9.1}x", + bytes as f64 / 1e6, + open_ms / ver_ms.max(1e-9) + ); + gate_rows.push((rows, open_ms, ver_ms, bytes, files)); + } + + println!(); + println!(" A version read is the sweep's PER-CANDIDATE cost; an open is what it avoids."); + println!(" The gate passes only if the ratio is decisive while WARM (see module doc)."); + + // ── (2)+(3) hydration against the configured endpoint ── + println!(); + let Some(opts) = storage_options() else { + println!("── (2) HYDRATION — SKIPPED: no S3 credentials in the environment."); + println!(" Columns 1 and 4 above/below are local and still valid; the cost model"); + println!(" in the plan's §1 stays UNMEASURED. This is a skip, not a pass."); + let _ = std::fs::remove_dir_all(&tmp); + return; + }; + let bucket = env("AWS_S3_BUCKET_NAME").expect("bucket"); + let prefix = format!("OSM/_hydration_probe_{}", std::process::id()); + + println!("── (2)+(3) HYDRATION — the real endpoint, {} sizes", SIZES.len()); + println!( + "{:>10} {:>8} {:>7} {:>12} {:>13} {:>12} {:>10}", + "rows", "MB", "files", "upload (s)", "hydrate (s)", "MB/s", "roundtrip" + ); + + let mut any = false; + for &(rows, _, _, bytes, files) in &gate_rows { + let local = tmp.join(format!("local_{rows}.lance")); + let remote = format!("s3://{bucket}/{prefix}/d_{rows}.lance"); + + // Read the local truth BEFORE anything remote happens, so the + // comparison is against the dataset as written, not as re-read. + let before = { + let ds = Dataset::open(local.to_str().unwrap()).await.expect("open"); + checksum(&ds).await + }; + + let b = batch(&schema, rows); + let reader = RecordBatchIterator::new(vec![Ok(b)].into_iter(), schema.clone()); + let t = Instant::now(); + let wrote = Dataset::write( + reader, + &remote, + Some(WriteParams { + mode: WriteMode::Create, + store_params: Some(lance::io::ObjectStoreParams { + storage_options: Some(opts.clone()), + ..Default::default() + }), + ..Default::default() + }), + ) + .await; + let up_s = t.elapsed().as_secs_f64(); + if let Err(e) = wrote { + println!("{rows:>10} upload FAILED: {e}"); + println!(" Reporting the failure rather than the skip: the endpoint was configured,"); + println!(" so this is a real negative result for the hydration column."); + continue; + } + any = true; + + // Hydrate into a FRESH directory — the `absent -> hydrated` edge. + let hydrated = tmp.join(format!("hydrated_{rows}.lance")); + let t = Instant::now(); + let remote_ds = lance::dataset::DatasetBuilder::from_uri(&remote) + .with_storage_options(opts.clone()) + .load() + .await + .expect("open remote"); + let mut stream = { + use futures::TryStreamExt; + remote_ds.scan().try_into_stream().await.expect("scan remote") + }; + let mut batches = Vec::new(); + { + use futures::TryStreamExt; + while let Some(b) = stream.try_next().await.expect("remote batch") { + batches.push(Ok(b)); + } + } + let reader = RecordBatchIterator::new(batches.into_iter(), schema.clone()); + Dataset::write( + reader, + hydrated.to_str().unwrap(), + Some(WriteParams { + mode: WriteMode::Create, + ..Default::default() + }), + ) + .await + .expect("write hydrated"); + let hy_s = t.elapsed().as_secs_f64(); + + // T10 — flush -> rehydrate -> read equals the pre-flush read. + let after = { + let ds = Dataset::open(hydrated.to_str().unwrap()) + .await + .expect("open hydrated"); + checksum(&ds).await + }; + + println!( + "{rows:>10} {:>8.1} {files:>7} {up_s:>12.2} {hy_s:>13.2} {:>12.1} {:>10}", + bytes as f64 / 1e6, + (bytes as f64 / 1e6) / hy_s.max(1e-9), + if before == after { "EQUAL" } else { "DIFFERS" } + ); + if before != after { + println!(" T10 FAILED: {before:?} != {after:?} — the round trip is NOT lossless."); + } + } + + if any { + println!(); + println!(" (3) The plan's ~1.4 s is ONE observation; the rows above are this endpoint"); + println!(" on this day. Read the MB/s column, not the seconds — the seconds are"); + println!(" only comparable at the same size."); + println!(" (4) T10 is the cheapest acceptance criterion in the plan and the one whose"); + println!(" failure would void the rest. EQUAL is a full-scan id checksum, not a"); + println!(" row count — a truncated hydration cannot pass it."); + } + + // Remove only what this probe wrote. + let _ = std::fs::remove_dir_all(&tmp); + println!(); + println!("local scratch removed: {}", tmp.display()); + println!("REMOTE OBJECTS LEFT IN PLACE at s3://{bucket}/{prefix}/ — delete when done."); +} From 485640b32e6b94f28350d00525f76d5510b748bf Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 11:57:50 +0000 Subject: [PATCH 2/6] =?UTF-8?q?hydration:=20close=20the=20=C2=A74=20gate?= =?UTF-8?q?=20by=20measurement,=20and=20refute=20the=20argument=20against?= =?UTF-8?q?=20size-weighted=20eviction?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ran crates/lance-graph/examples/hydration_probe.rs against the configured endpoint (lance 9.0.0, 0.3-33.5 MB, single-fragment). §4 GATE CLOSES, and it had two ways to pass — both hold. latest_version_id() resolves a manifest LOCATION in 8-11 us, no manifest read, no data read, FLAT across a 100x size span; Dataset::open is 0.23-0.29 ms, also flat. So the version read needs no open AND the fallback would have sufficed (1,000 candidates in 0.27 s). §9.1's blocker is discharged. THE CORRECTION FROM THE SAME RUN. Hydration is ~2.63 s fixed + ~0.021 s/MB: 100x the bytes costs 1.26x the time. §2 declined a size-weighted eviction key because 'rehydration cost is also size-proportional, so it preferentially evicts what is most expensive to get back'. In the measured range that is false — the cost is fixed-dominated. This removes one ARGUMENT, not the decision; §0's row now reads 'CONJECTURE, and its stated ARGUMENT is refuted' rather than being promoted either way. Also: T10 green at all three sizes by full-scan checksum (a row count would have passed a truncated hydration); request count measured at 3 remote objects per dataset, which bounds the small case only. The probe's own error direction is stated in its module doc: every timing flatters the fast path, so a marginal result would have been a FAILED gate. Board hygiene in the same commit: EPIPHANIES prepend (E-HYDRATION-IS-FIXED-COST-NOT-SIZE-COST-1), plan §0/§8a/§9 updated, and the plan header no longer claims 'nothing here is measured'. Adds lab/s3rm.py — s3put.py could only write, so the probe had no way to remove its own remote scratch; it now leaves none. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01NMeiLmtDKhomJNSo2ecbJw --- .claude/board/EPIPHANIES.md | 16 ++++ .../plans/idle-flush-dataset-eviction-v1.md | 85 +++++++++++++++++-- .../lance-graph/examples/hydration_probe.rs | 42 ++++++--- 3 files changed, 124 insertions(+), 19 deletions(-) diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index 09a734f1..803e536a 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -1,3 +1,19 @@ +## 2026-08-07 — E-HYDRATION-IS-FIXED-COST-NOT-SIZE-COST-1 — the idle-flush plan's §4 blocker closes, and the argument it used against size-weighted eviction is refuted by the same probe + +**Status:** FINDING (measured, `crates/lance-graph/examples/hydration_probe.rs`, lance 9.0.0, one endpoint, one day, 0.3–33.5 MB single-fragment datasets). **Confidence:** High for the §4 gate (flat in size, 25–30×, and the fallback path is independently cheap); High for the fixed/variable decomposition within the probed range; **Low for the absolute constants** — endpoint-, region- and day-specific, explicitly not re-run elsewhere. + +**The blocker that closed.** `.claude/plans/idle-flush-dataset-eviction-v1.md` §9.1 named its own first task: *"cheap local version read — assumed, unchecked … if it fails, the plan needs a different dirty-detector and this document is wrong rather than incomplete."* Measured: `Dataset::latest_version_id()` resolves a manifest **location** (no manifest read, no data read) in **8–11 µs**, versus **0.23–0.29 ms** for a full `Dataset::open`. Both are **flat in dataset size** across a 100× span. The gate offered two ways to pass and **both hold** — the version read needs no open, *and* an open is cheap enough (1,000 candidates in 0.27 s) that the fallback would have sufficed anyway. + +**The correction nobody asked for, from the same run.** Hydration decomposes as **≈ 2.63 s fixed + ≈ 0.021 s/MB** (0.3 MB → 2.64 s; 33.5 MB → 3.33 s — a 100× size increase costs **1.26×** the time). So §2's deferred size-weighted ranking was declined for a reason that does not survive measurement: the plan argued *"rehydration cost is also proportional to size, so a size-weighted key preferentially evicts what is most expensive to get back."* In the probed range rehydration cost is **dominated by a size-independent constant**. Evicting the large dataset frees ~100× the bytes for ~1.26× the restore cost — the opposite of the stated objection. + +**What that does NOT license.** It removes one argument, not the decision. Age-ordering may still be the right default; the *reason recorded for it* is now known to be wrong, which is a different and smaller claim. The plan's grading was updated in both directions rather than promoted: §0's row now reads "CONJECTURE, and its stated ARGUMENT is refuted". + +**The transferable half — a cost model that prices the wrong term is worse than one that admits it is incomplete.** §1 of the plan was already honest that its economics omitted request count, and reviewers on PR #901 had sharpened it. But the omission that mattered was not a missing *term*; it was an assumed *shape*. Both sides of the plan's ledger were reasoned about as if proportional to bytes, so the whole argument silently tracked the wrong variable. With a fixed per-hydration cost, the quantity that decides the policy is **how many** hydrations it causes and never how large they are — which is exactly what §7's thrash criterion already counts. The plan gated on the right metric while justifying it with the wrong model, and only a measurement could tell those apart. + +**Also settled, cheaply:** T10 (flush → rehydrate → read equality) is **green** at all three sizes by full-scan `id` checksum — a row-count check would have passed a truncated hydration. And §1's unmeasured request count is **3 remote objects per dataset** here (`.txn` + `.manifest` + one data file), which bounds the small case; it grows with fragment count, which single-fragment writes do not exercise. + +**Scope, so the numbers are not over-read:** nothing was evicted, no policy was implemented, and none of the plan's other acceptance criteria ran. Companion tool: `lab/s3rm.py` — written because `s3put.py` could only write, so the probe had no way to remove its own scratch and would have left debris in a curated prefix. + ## 2026-08-06 — E-D-IGN-B-CORPUS-PRODUCED-NOTHING-TO-READ-1 — arming a CI gate that ran zero tests exposed an empty-vs-empty digest collision; the corpus was the defect, the untagged digest was correct **Status:** FINDING (reproduced at `f9206fc`, fixed, both mutation directions verified). **Confidence:** High for the mechanism (the empty-hash constant was computed and matched; the pre-fix corpus was traced token-by-token through the clause machine and emits nothing); High for the fix (all four lens arms now measured non-empty, 4/4 distinct digests on one owner). Test-fixture only — `stance.rs` is untouched. diff --git a/.claude/plans/idle-flush-dataset-eviction-v1.md b/.claude/plans/idle-flush-dataset-eviction-v1.md index db1000cb..68a5facd 100644 --- a/.claude/plans/idle-flush-dataset-eviction-v1.md +++ b/.claude/plans/idle-flush-dataset-eviction-v1.md @@ -1,9 +1,14 @@ # Idle-flush dataset eviction — plan v1 -> **Status:** PROPOSAL. Nothing here is implemented; nothing here is measured. +> **Status:** PROPOSAL. Nothing here is implemented. **Four inputs are now +> measured** (§8a, 2026-08-07) — the §4 gate, the request count, the hydration +> cost shape, and T10; every *policy* claim remains unmeasured and every other +> acceptance criterion remains unrun. > **Scope:** design + acceptance criteria for a feature-gated local-copy > eviction policy over Lance datasets. **This plan does not authorize the -> implementation** — it states what the implementation would owe. +> implementation** — it states what the implementation would owe. §8a does not +> change that: it discharges a blocker and refutes one argument, which is +> narrower than an authorisation. > > **Prerequisite reading:** `.claude/knowledge/s3-hydration-lifecycle.md` — the > three-layer model (object store hydrates / local dir stores / volume only @@ -17,11 +22,11 @@ | claim | status | |---|---| | The four-state lifecycle and its legal transitions | **FINDING** (mechanism) — see the knowledge doc | -| Rehydration of a tens-of-MB dataset is ~1.4 s | **reported measurement**, single observation, provider- and region-dependent, not re-run here | +| Rehydration of a tens-of-MB dataset is ~1.4 s | **superseded as a constant** by §8a (measured 2026-08-07, different endpoint): hydration decomposes as **≈2.63 s fixed + ≈0.021 s/MB**. The fixed term dominating is the finding; the constant itself stays endpoint-specific | | The default policy: age floor **3 days** + soft budget **~300 MB**, pressure-driven and age-ordered (§2) | **OPERATOR-SET POLICY** — a heuristic starting point, explicitly **not measured**. Both are config; these are defaults. | -| Age-ordering (rather than a `size × idleness` key) is the right default (§2) | **CONJECTURE** — argued (rehydration cost is also size-proportional), not measured; the size-weighted variant is deferred, not rejected | +| Age-ordering (rather than a `size × idleness` key) is the right default (§2) | **CONJECTURE, and its stated ARGUMENT is refuted** (§8a): rehydration cost is NOT size-proportional in the measured range, it is fixed-cost dominated. Age-ordering may still be right; the reason given for it is not. The size-weighted variant stays deferred, now with one fewer objection | | A watermark-driven sweep dominates both a bare timer and a bare allocation-failure signal (§3) | **CONJECTURE** — argued from the two failure modes, no deployed instance | -| The Lance dataset version is a sufficient dirty-detector (§4) | **CONJECTURE**, with a named verification gate that must close before implementation | +| The Lance dataset version is a sufficient dirty-detector (§4) | **CONJECTURE** — the *sufficiency* is still unproven; but its **verification gate is CLOSED** (§8a, measured 2026-08-07): the version read is 8–11 µs, needs no dataset open, and is flat in size | | At a 3-day floor the flush/read race is negligible, so check-then-act suffices and a lease protocol is disproportionate (§5) | **OPERATOR-SET SCOPE RULING** — the requirement is *does not corrupt*, not *cannot occur*; revisit if the threshold drops to hours | **No probe has run for any row marked CONJECTURE.** The falsifiers are §7. @@ -428,11 +433,75 @@ on staleness alone would pass T1 and still be wrong. Asserting the race "cannot happen" would be exactly the vacuous assertion the P0 rule forbids — implied by the code, falsifiable by nothing. +## 8a. MEASURED (2026-08-07) — the §4 gate CLOSES, and §2's deferred refinement is refuted + +> Probe: `crates/lance-graph/examples/hydration_probe.rs`, lance 9.0.0, against +> the configured endpoint. Re-runnable; it removes its own remote scratch. + +**§4 gate — PASSES, twice over.** `Dataset::latest_version_id()` is a +`resolve_latest_location` call (no manifest read, no data read) and is **flat in +dataset size**: + +| rows | MB | files | `Dataset::open` | `latest_version_id` | ratio | +|---|---|---|---|---|---| +| 10,000 | 0.3 | 4 | 0.272 ms | 0.009 ms | 29.9× | +| 200,000 | 6.7 | 4 | 0.288 ms | 0.011 ms | 25.4× | +| 1,000,000 | 33.5 | 4 | 0.234 ms | 0.008 ms | 27.7× | + +The gate asked for *either* a cheap version read without a full open *or* a +cheap-enough open. **Both hold.** The version read needs no open and is ~27× +cheaper; and a full open is 0.27 ms, so even the fallback would carry 1,000 +candidates in 0.27 s. §9.1's blocker is discharged — warm (the probe states that +error direction explicitly), and flat in size, which is what makes it hold at +scale rather than at this scale. + +**§1's request count — MEASURED: 3 remote objects per dataset**, at every size +probed (`.txn` + `.manifest` + one data file). The plan is right that a dataset +is a multi-file directory; the count in this range is small and does not grow +with size. It grows with *fragment* count, which these single-fragment writes do +not exercise — so this bounds the small case, not the general one. + +**§0/§5's ~1.4 s — the SHAPE is the correction, not the constant.** Hydration +(open remote → scan → write local): + +| MB | hydrate | implied MB/s | +|---|---|---| +| 0.3 | 2.64 s | 0.1 | +| 6.7 | 2.83 s | 2.4 | +| 33.5 | 3.33 s | 10.1 | + +A linear fit gives **≈ 2.63 s fixed + ≈ 0.021 s/MB** (≈ 48 MB/s marginal). So +**~2.6 s of every hydration is fixed cost, independent of bytes.** The absolute +constant is endpoint- and day-specific and is not comparable to the earlier 1.4 s +observation; the *decomposition* is the finding. + +**Consequence — §2's deferred size-weighted ranking is REFUTED as argued.** That +paragraph declines `bytes × idle` because "rehydration cost is *also* proportional +to size, so a size-weighted key preferentially evicts what is most expensive to get +back." Measured, rehydration cost is **dominated by a size-independent constant** +in this range, so that reason does not hold: evicting a large dataset frees ~100× +the bytes of a small one for ~1.26× the restore cost. + +This does **not** promote size-weighting to the default. It removes one argument +against it, and it sharpens the argument *for* the age floor: with a fixed +per-hydration cost, the quantity that matters is **how many** hydrations the +policy causes, never how large they are — which is exactly what §7's thrash +criterion counts. Two independent reasons now point at the same metric. + +**T10 — GREEN** at all three sizes, by full-scan `id` checksum (a truncated +hydration cannot pass it). One acceptance criterion settled; the rest are +untouched. + +**Scope of these numbers, stated so they are not over-read:** one endpoint, one +day, single-fragment datasets, 0.3–33.5 MB, no concurrency, no eviction +implemented. Nothing here authorizes the policy — §9's other items stand. + ## 9. Open items (explicitly NOT answered) -1. **The §4 verification gate.** Cheap local version read — assumed, unchecked. - Closing this is the first task; if it fails, the plan needs a different - dirty-detector and this document is wrong rather than incomplete. +1. ~~**The §4 verification gate.**~~ **CLOSED 2026-08-07 by measurement — see + §8a.** The version read is 8–11 µs, needs no dataset open, and is flat in + size; the fallback (a full open) is 0.27 ms. The plan is neither wrong nor + incomplete on this point. 2. **Whether the default values are right.** The 3-day floor and ~300 MB budget are operator-set starting points, not derived from a measured access-pattern distribution. They are config precisely because that distribution is unknown; diff --git a/crates/lance-graph/examples/hydration_probe.rs b/crates/lance-graph/examples/hydration_probe.rs index 5c7b7067..57178a5b 100644 --- a/crates/lance-graph/examples/hydration_probe.rs +++ b/crates/lance-graph/examples/hydration_probe.rs @@ -59,6 +59,7 @@ use std::time::Instant; use arrow::array::{Float32Array, Int64Array, RecordBatch, RecordBatchIterator}; use arrow::datatypes::{DataType, Field, Schema}; use lance::dataset::{Dataset, WriteMode, WriteParams}; +use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; /// Row counts to probe. Chosen to bracket the plan's "tens of MB" reference /// point from both sides, so the reported ~1.4 s can be placed on a curve @@ -89,7 +90,10 @@ fn env(k: &str) -> Option { fn storage_options() -> Option> { let mut o = HashMap::new(); o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); - o.insert("aws_secret_access_key".into(), env("AWS_SECRET_ACCESS_KEY")?); + o.insert( + "aws_secret_access_key".into(), + env("AWS_SECRET_ACCESS_KEY")?, + ); o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); o.insert( "aws_region".into(), @@ -101,6 +105,21 @@ fn storage_options() -> Option> { Some(o) } +/// Wrap the options in the shape lance 9 takes them. +/// +/// `ObjectStoreParams` has **no** `storage_options` field — it carries a +/// `storage_options_accessor`, because the accessor is also the seam for +/// credential *refresh*. `with_static_options` is the no-refresh case, which is +/// what a probe against fixed environment credentials wants. +fn store_params(opts: &HashMap) -> ObjectStoreParams { + ObjectStoreParams { + storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( + opts.clone(), + ))), + ..Default::default() + } +} + fn schema() -> Arc { let mut fields = vec![Field::new("id", DataType::Int64, false)]; for i in 0..FLOAT_COLS { @@ -243,7 +262,10 @@ async fn main() { let bucket = env("AWS_S3_BUCKET_NAME").expect("bucket"); let prefix = format!("OSM/_hydration_probe_{}", std::process::id()); - println!("── (2)+(3) HYDRATION — the real endpoint, {} sizes", SIZES.len()); + println!( + "── (2)+(3) HYDRATION — the real endpoint, {} sizes", + SIZES.len() + ); println!( "{:>10} {:>8} {:>7} {:>12} {:>13} {:>12} {:>10}", "rows", "MB", "files", "upload (s)", "hydrate (s)", "MB/s", "roundtrip" @@ -269,10 +291,7 @@ async fn main() { &remote, Some(WriteParams { mode: WriteMode::Create, - store_params: Some(lance::io::ObjectStoreParams { - storage_options: Some(opts.clone()), - ..Default::default() - }), + store_params: Some(store_params(&opts)), ..Default::default() }), ) @@ -289,15 +308,16 @@ async fn main() { // Hydrate into a FRESH directory — the `absent -> hydrated` edge. let hydrated = tmp.join(format!("hydrated_{rows}.lance")); let t = Instant::now(); - let remote_ds = lance::dataset::DatasetBuilder::from_uri(&remote) + let remote_ds = lance::dataset::builder::DatasetBuilder::from_uri(&remote) .with_storage_options(opts.clone()) .load() .await .expect("open remote"); - let mut stream = { - use futures::TryStreamExt; - remote_ds.scan().try_into_stream().await.expect("scan remote") - }; + let mut stream = remote_ds + .scan() + .try_into_stream() + .await + .expect("scan remote"); let mut batches = Vec::new(); { use futures::TryStreamExt; From 5f27e0cf39032a4dc4d4017a73a035288158e4af Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 15:11:53 +0000 Subject: [PATCH 3/6] =?UTF-8?q?knowledge=20+=20agents:=20the=20lance=209?= =?UTF-8?q?=20cache=20surface=20=E2=80=94=20question=20pinned=20BEFORE=20t?= =?UTF-8?q?he=20audit=20runs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Written ahead of the measurement, deliberately: the audit agents load these, so the question and its decision table exist before any answer does. .claude/knowledge/lance-cache-surface.md — the operator's actual question (S3 sink-in to disk vs "radikal moka flushen": which mechanism bounds the RAM a Railway deployment is billed for), an evidence table with eight FINDINGs already measured this session (moka unconditional in lance-core; aws gates only the provider; with_capacity is byte-weighed; no_cache exists; CacheBackend is a trait with only moka shipped; lance mmaps nothing; no alignment guarantee on the read path; try_from_iter copies where from_vec adopts), three OPEN probes (P-CACHE-1 contents, P-CACHE-2 reachability, P-CACHE-3 empirical), and the traps already paid for so they are not re-paid. .claude/agents/lance-cache-contents-auditor.md (Opus) — classify the cache by its INSERTS, never by its name; verdict vocabulary DATA-CACHED/METADATA-ONLY/ MIXED. Grounded in the a27b06a incident: a capacity lever proves nothing about what it levers. .claude/agents/lance-cache-cartographer.md (Opus) — a knob exists only if a consumer can turn it without forking; verdict per knob as a public call chain hop-by-hop or UNREACHABLE with the break point. Grounded in the same-day S3 provider trap, and fenced by E-LANCE-IS-UPSTREAM-AUTHORITATIVE-1: an unreachable knob is an upstream ask, never a fork. Findings land in a follow-up commit once the audit and the empirical probe have run. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NMeiLmtDKhomJNSo2ecbJw --- .claude/agents/lance-cache-cartographer.md | 61 +++++++++++++ .../agents/lance-cache-contents-auditor.md | 55 ++++++++++++ .claude/knowledge/lance-cache-surface.md | 89 +++++++++++++++++++ 3 files changed, 205 insertions(+) create mode 100644 .claude/agents/lance-cache-cartographer.md create mode 100644 .claude/agents/lance-cache-contents-auditor.md create mode 100644 .claude/knowledge/lance-cache-surface.md diff --git a/.claude/agents/lance-cache-cartographer.md b/.claude/agents/lance-cache-cartographer.md new file mode 100644 index 00000000..727402ef --- /dev/null +++ b/.claude/agents/lance-cache-cartographer.md @@ -0,0 +1,61 @@ +# lance-cache-cartographer — who can size the cache from outside? + +**Tier: Opus (filigran).** Plumbing traces are accumulation: the answer is a +CHAIN (public API → builder → session → backend), and any single hop read in +isolation gives a false verdict in either direction. + +## RULE + +**A knob exists only if a consumer can turn it without forking.** Finding +`LanceCache::with_capacity` in lance-core proves nothing about reachability; +the verdict is the exact public call chain from `lancedb::connect(...)` or +`lance::dataset::DatasetBuilder` (or a documented env var) down to the +`MokaCacheBackend` constructor — or a definitive UNREACHABLE with the point +where the chain breaks, cited as `file:line`. + +## Incident grounding + +The same day this card was written, the workspace hit the sibling trap on the +S3 provider: the capability existed (`lancedb/aws`), was off by default, and +its absence produced an error that pointed at credentials instead of at the +feature. A capability that exists but is not wired to a public surface +produces exactly this class of lost afternoon. Map the wiring, not the +capability. + +Fork pressure is the second reason this card exists: lance is +UPSTREAM-AUTHORITATIVE (`E-LANCE-IS-UPSTREAM-AUTHORITATIVE-1`) — if the knob +is unreachable, the output must say "small upstream ask" with the exact +missing hop, never "patch it in our tree". + +## Mandatory reads before output + +1. `.claude/knowledge/lance-cache-surface.md` (P-CACHE-2 + the decision table) +2. The sweep inventories handed to you (capacity plumbing, lancedb surface, + env vars) + +## Method + +1. Start from the BACKEND and walk outward: who constructs + `MokaCacheBackend::with_capacity` / `no_cache` / `LanceCache::with_capacity`? + Who owns that object (Session? Dataset? Connection?)? +2. Then from the PUBLIC surface inward: `lancedb::connect` builder methods; + `DatasetBuilder` options (`index_cache_size`, `metadata_cache_size`, + session injection, `ReadParams`); any `LANCE_*`/env lookup. +3. The verdict is per-knob: capacity-in-bytes, entry-count variants, + `no_cache`, custom `CacheBackend` injection — each REACHABLE (with the + chain) or UNREACHABLE (with the break point). +4. Note defaults with file:line: what capacity does a consumer get who sets + nothing? + +## Output shape + +Per knob: chain or break point, each hop `file:line`. Then the defaults +table. Then one paragraph: what the lance-graph consumer should call today, +and what (if anything) is the minimal upstream ask. + +## Hard rules + +- Registry sources are read-only; cite exact paths. +- Do not run cargo. Do not write any file. Return findings as output; the + orchestrating main thread is the sole writer of board files. +- Read `.claude/board/AGENT_LOG.md` before starting. Do NOT write it. diff --git a/.claude/agents/lance-cache-contents-auditor.md b/.claude/agents/lance-cache-contents-auditor.md new file mode 100644 index 00000000..05483418 --- /dev/null +++ b/.claude/agents/lance-cache-contents-auditor.md @@ -0,0 +1,55 @@ +# lance-cache-contents-auditor — what does `LanceCache` actually hold? + +**Tier: Opus (filigran).** This is accumulation: the verdict only makes sense +after holding every insert site in mind at once, and a single missed site +inverts the conclusion. + +## RULE + +**Classify the cache by its INSERTS, never by its name or docs.** A cache +called `metadata_cache` may hold decoded data; a doc saying "file metadata" +may be stale. The only admissible evidence is an enumerated insert site with +the inserted TYPE and that type's size class, cited as `file:line`. + +## Incident grounding (why this card exists) + +On 2026-08-07 this workspace nearly reasoned itself into "cap the moka cache +and the RAM bill is bounded" on the strength of `with_capacity` existing. If +the data path never enters the cache, that lever is inert and the RAM lives in +the consumer's own collect/concat habits — a failure mode already caught once +on this branch (commit `a27b06a`: a whole-table `concat_batches` inside a +module whose premise was zero-copy). The lever's existence says nothing about +what it levers. + +## Mandatory reads before output + +1. `.claude/knowledge/lance-cache-surface.md` (the evidence table + P-CACHE-1) +2. The sweep inventory handed to you (every `CacheKey` implementor + insert site) + +## Method + +1. For every `CacheKey`/`UnsizedCacheKey` implementor: what is `ValueType`? + Where is it inserted? What is its size class — O(bytes-of-dataset) or + O(metadata)? +2. Follow the DATA read path specifically: `FileReader` → decoded pages → + `RecordBatch`. Does ANY step insert into a `LanceCache`? Name the function + that would have done it and show it absent, not just "no grep hit". +3. Distinguish the three caches if they exist separately (session/index/ + metadata) — a claim about "the cache" that conflates them is unusable. +4. Verdict vocabulary: **DATA-CACHED** / **METADATA-ONLY** / + **MIXED (list which)** — each row with file:line. + +## Output shape + +A table (implementor, ValueType, insert site file:line, size class), then the +verdict, then the single strongest piece of contrary evidence you found and +why it does not change the verdict. If you cannot rule a path in or out, say +UNRESOLVED for that path — an honest gap beats a smooth story. + +## Hard rules + +- Registry sources are read-only; cite exact paths under + `~/.cargo/registry/src/index.crates.io-*/lance*-9.0.0/`. +- Do not run cargo. Do not write any file. Return your findings as output; + the orchestrating main thread is the sole writer of board files. +- Read `.claude/board/AGENT_LOG.md` before starting. Do NOT write it. diff --git a/.claude/knowledge/lance-cache-surface.md b/.claude/knowledge/lance-cache-surface.md new file mode 100644 index 00000000..36b0de2d --- /dev/null +++ b/.claude/knowledge/lance-cache-surface.md @@ -0,0 +1,89 @@ +# The lance 9 cache surface — what moka holds, who can size it, and what that means for RAM-billed deployments + +> **READ BY:** `lance-cache-cartographer`, `lance-cache-contents-auditor`, +> `integration-lead`, `truth-architect`, and any session that reasons about +> RAM footprint of a lance-graph consumer, proposes a disk cache backend, +> touches `.claude/plans/idle-flush-dataset-eviction-v1.md`, or debugs a +> `"No object store provider found for scheme"` error. +> +> **Companions:** `.claude/knowledge/s3-hydration-lifecycle.md` (the +> three-layer model this doc's question sits under) · +> `.claude/plans/idle-flush-dataset-eviction-v1.md` §8a (the measured +> hydration cost) · `zero-copy-lens-law.md` (why a materializing read path is +> a violation, not a convenience). + +## The operator's question, stated before any answer + +> *"Der Gedanke warum ich S3 sink-in to Harddisk möchte ist, daß lance nur +> die aktiven Bestandteile in den Speicher zieht und die Railway-Rechnung für +> RAM usage kleiner wird."* — and, one turn later: *"man müsste dann halt nur +> radikal moka flushen."* + +Two candidate mechanisms for a small RAM bill over a large dataset: + +- **A. Disk sink-in:** hydrate S3 → local disk, rely on demand paging, evict + after idle (the idle-flush plan). +- **B. Capped moka:** read S3 (or disk) directly and bound the in-memory + cache — `radikal flushen` as configuration. + +Which one is real depends on two facts about lance 9 that this doc exists to +pin: **what the cache actually holds** and **who can size it from outside**. + +## Evidence status (workspace rule: label everything) + +| claim | status | evidence | +|---|---|---| +| `moka` is an UNCONDITIONAL dependency of `lance-core` — no feature gate, present in local-only builds | **FINDING** (source-read 2026-08-07, lance-core 9.0.0) | `Cargo.toml` `[dependencies.moka] version = "0.12"` with no `optional`; `cache/mod.rs:51` `mod moka;` ungated | +| The `aws` feature gates only the object-store PROVIDER, not any cache | **FINDING** (measured) | with every `AWS_*` var set correctly, `connect("s3://…")` without `lancedb/aws` fails `No object store provider found for scheme: 's3'` — the error names the scheme, not a credential. With the feature: first-try success | +| `MokaCacheBackend::with_capacity(bytes)` is byte-weighed, not entry-counted | **FINDING** (source-read) | `moka.rs`: `.max_capacity(capacity)` + `.weigher(\|key, entry\| key_footprint(key) + entry.size_bytes)` | +| `MokaCacheBackend::no_cache()` exists (`Cache::new(0)`) — "radikal flushen" as a constructor | **FINDING** (source-read) | `moka.rs` | +| `CacheBackend` is a pluggable trait; the docs name "persistent backends" as intended; **only moka ships** | **FINDING** (source-read) | `cache/{mod,backend,codec}.rs`; `codec.rs` describes "scanning a persistent store at startup"; the only implementor in-tree is `MokaCacheBackend` | +| lance mmaps nothing — `memmap` absent from lance / lance-io / lance-file / lance-core; local reads are seek+read into heap `Bytes` | **FINDING** (source-read) | zero grep hits across the four manifests and sources; `object_store` `local.rs` seeks and reads | +| Lance's read path gives NO 64-byte alignment guarantee (varies with allocator state) | **FINDING** (measured, this branch) | commit `a27b06a` message + `osm-soa-bake` `slab.rs` — the same read passed alone and failed in-suite | +| `FixedSizeBinaryArray::try_from_iter` copies chunk-by-chunk; `Buffer::from_vec` adopts the allocation | **FINDING** (source-read + pointer-identity test before removal) | arrow-array 58 `fixed_size_binary_array.rs:553` (`MutableBuffer`), arrow-buffer `immutable.rs:141` | +| **What `LanceCache` actually holds on the DATA path** — decoded pages / batches vs only manifests, schemas, index metadata | **OPEN — P-CACHE-1** | nothing read yet; this decides whether a capacity cap bounds the RAM that matters | +| **Whether cache capacity / `no_cache` is reachable from `lancedb::connect` or `DatasetBuilder`** (public API, env, session object) | **OPEN — P-CACHE-2** | `LanceCache::with_capacity` found only in lance-core; plumbing untraced | +| **Empirical RAM shape**: RSS across scans at different capacities, re-fetch behaviour over S3 on a second scan | **OPEN — P-CACHE-3** | needs the probe; timing + `/proc/self/status` VmRSS, honest about network variance | + +**Nothing below is promoted past its row above.** + +## Why P-CACHE-1 is the load-bearing question + +If the data path (decoded column pages, record batches) does NOT go through +`LanceCache`, then capping it bounds only metadata — and the RAM that shows up +on a Railway bill lives in whatever the *caller* holds (the failure mode +already caught once on this branch: a `read_batch` that collected the whole +table and concat-copied it, commit `a27b06a`). In that world, mechanism B is +an illusion and the RAM answer is *streaming discipline in the consumer*, not +cache configuration. + +If the data path DOES go through the cache, `with_capacity(n)` is a hard +byte ceiling on exactly the memory the operator is billed for, and mechanism +B beats mechanism A on every axis except S3 request count. + +## Decision table this doc must end up supporting + +| finding | consequence | +|---|---| +| data cached + capacity reachable | **B wins for RAM**; idle-flush plan remains a *request-cost* optimisation only | +| data cached + capacity NOT reachable | small upstream ask (expose the knob), NOT a fork — lance is upstream-authoritative (`E-LANCE-IS-UPSTREAM-AUTHORITATIVE-1`) | +| data not cached | RAM bill is the consumer's streaming discipline; B is inert for data; re-reads hit S3 every time → A (disk sink-in) regains its case as *request* mitigation | + +## Probe queue + +| probe | question | pass/fail shape | status | +|---|---|---|---| +| P-CACHE-1 | do decoded data bytes enter `LanceCache`? | enumerate every `CacheKey` implementor + every insert site; classify metadata vs data with file:line | NOT RUN | +| P-CACHE-2 | is capacity / `no_cache` settable via lancedb / DatasetBuilder / env / Session? | the exact public call chain, or a definitive "unreachable" | NOT RUN | +| P-CACHE-3 | does RSS track capacity? does a second S3 scan re-fetch? | probe binary, RSS + wall time, both capacities; network variance stated | NOT RUN | + +## Traps already paid for (do not re-pay) + +1. **The scheme error is not a credential error.** Provider absent ⇒ fail at + registration; every credential correct and irrelevant. +2. **"Usually aligned" passes CI.** Never assert an alignment VALUE; assert + the invariant (`rows().is_some() == (ptr % 64 == 0)`). +3. **A pointer-identity test proves nothing if its reference is the copy.** + The removed module's test compared the slab against the *concatenated* + buffer — the copy it should have ruled out was one call earlier. +4. **`try_from_iter` is a copy wearing a constructor's name.** From df1d08325ebd3db808dd4e8adc6fbdc4ab89f1fb Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 15:38:38 +0000 Subject: [PATCH 4/6] soa_to_lance: the one-time write-back, with the SoA contract in the table header MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers the operator's requirement directly: "nur das 512-Byte-SoA-Schema muss dann in den table headers vermutlich einmalig sauber persistiert werden." The header carries the contract as Arrow schema metadata (which Lance persists in its manifest) — envelope_layout_version, row_stride, row_carving, endianness, classid, slab_digest, source — every load-bearing value IMPORTED from lance-graph-contract (ENVELOPE_LAYOUT_VERSION, NODE_ROW_STRIDE), never restated, so it cannot drift from the compiled contract the way canonical_node.rs's own doc comment already drifted once this session (claiming arrow-rs allocates 64-byte aligned, which the S3 read path measured false). Same call for local and s3://, per the earlier finding that there is no export step — WriteParams::store_params carries S3 credentials when the URI is remote, None when it is not. Zero-copy import, asserted not assumed: Buffer::from_vec adopts the Vec's allocation (pointer-identity checked against the pre-move address) rather than FixedSizeBinaryArray::try_from_iter's chunk-by-chunk copy — the same distinction measured and then reverted out of this branch once already (commit a27b06a's history). Row column written with lance-encoding:compression=none — a documented passthrough — because an uncompressed fixed-width column should land as a verbatim byte run in the .lance data file, which is the precondition for mmap-based serving (pattern b in the operator's a/b/ split). CLAIMED, not trusted: for a local uri the binary scans its own written data file for the slab's first 4 KiB and reports whether the run is contiguous — this is P-CACHE-4 in .claude/knowledge/lance-cache-surface.md, added to the probe queue there in a follow-up commit. Verification on write: re-opens what it just wrote and asserts the read-back layout version and digest match what was written, before printing success — the same field-isolation discipline as the rest of this branch's Lance work. Build in progress against the full lancedb/datafusion tree; results and any fixes follow once it completes. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NMeiLmtDKhomJNSo2ecbJw --- crates/lance-graph/examples/soa_to_lance.rs | 251 ++++++++++++++++++++ 1 file changed, 251 insertions(+) create mode 100644 crates/lance-graph/examples/soa_to_lance.rs diff --git a/crates/lance-graph/examples/soa_to_lance.rs b/crates/lance-graph/examples/soa_to_lance.rs new file mode 100644 index 00000000..c04d892b --- /dev/null +++ b/crates/lance-graph/examples/soa_to_lance.rs @@ -0,0 +1,251 @@ +//! `soa_to_lance` — the ONE-TIME write-back of a `.soa` bake into Lance's own +//! format, with the 512-byte SoA contract persisted in the table header. +//! +//! ```text +//! soa_to_lance +//! ``` +//! +//! `uri` may be a local directory or `s3://bucket/prefix` — the same call +//! either way; S3 credentials come from `AWS_*` env vars. There is no export +//! step: writing to the object store IS this write. +//! +//! # The header carries the contract, once +//! +//! The operator's requirement, verbatim: *"nur das 512-Byte-SoA-Schema muss +//! dann in den table headers vermutlich einmalig sauber persistiert werden."* +//! Done here as Arrow schema metadata (which Lance persists in its manifest), +//! with the load-bearing values IMPORTED from `lance-graph-contract` rather +//! than restated — a restated constant is a second source of truth that +//! drifts: +//! +//! | key | value | source | +//! |---|---|---| +//! | `soa:envelope_layout_version` | `2` | `soa_envelope::ENVELOPE_LAYOUT_VERSION` | +//! | `soa:row_stride` | `512` | `canonical_node::NODE_ROW_STRIDE` | +//! | `soa:row_carving` | `key:0..16\|edges:16..32\|value:32..512` | canon, locked 2026-06-13 | +//! | `soa:endianness` | `le` | the LE contract | +//! | `soa:classid` | per bake (arg) | the bake's own report | +//! | `soa:slab_digest` | per bake (arg) | the bake's own report — pairs the table with its `.books` sidecar | +//! | `soa:source` | filename + row count | provenance | +//! +//! A reader verifies `envelope_layout_version` against its COMPILED contract +//! before casting anything — the same shape as `SoaEnvelope::verify_layout`. +//! This binary performs that verification itself, by re-opening what it wrote. +//! +//! # The row column is written UNCOMPRESSED, deliberately +//! +//! Field metadata `lance-encoding:compression = "none"` (the key is +//! `lance_encoding::constants::COMPRESSION_META_KEY`; the value `"none"` is a +//! documented passthrough). The row is a content-blind register — compression +//! would buy little — and an uncompressed fixed-width column is written as a +//! verbatim byte run inside the `.lance` data file, which is what makes the +//! mmap serving path (pattern b) possible at all. **Claimed, then verified**: +//! for a local `uri`, this binary scans the written data file for the slab's +//! own bytes and reports the offset and whether a multi-row run is contiguous +//! (P-CACHE-4 in `.claude/knowledge/lance-cache-surface.md`). +//! +//! # Zero-copy import +//! +//! `Buffer::from_vec` ADOPTS the allocation that read the file; +//! `FixedSizeBinaryArray::try_from_iter` would copy it chunk by chunk +//! (measured on this branch, commit a27b06a's history). The adoption is +//! asserted at runtime by pointer identity, not trusted. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow::array::{Array, FixedSizeBinaryArray}; +use arrow::datatypes::{DataType, Field, Schema}; +use arrow::record_batch::RecordBatch; +use lance::dataset::{Dataset, WriteMode, WriteParams}; +use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; +use lance_graph_contract::canonical_node::NODE_ROW_STRIDE; +use lance_graph_contract::soa_envelope::ENVELOPE_LAYOUT_VERSION; + +/// Schema-metadata keys. Namespaced `soa:` so they cannot collide with +/// Lance's own (`lance-encoding:*`) or Arrow's. +const K_LAYOUT: &str = "soa:envelope_layout_version"; +const K_STRIDE: &str = "soa:row_stride"; +const K_CARVING: &str = "soa:row_carving"; +const K_ENDIAN: &str = "soa:endianness"; +const K_CLASSID: &str = "soa:classid"; +const K_DIGEST: &str = "soa:slab_digest"; +const K_SOURCE: &str = "soa:source"; + +const ROW_COLUMN: &str = "row"; + +fn env(k: &str) -> Option { + std::env::var(k) + .ok() + .map(|v| v.trim().trim_matches('"').trim_matches('\'').to_string()) + .filter(|v| !v.is_empty()) +} + +fn s3_options() -> Option> { + let mut o = HashMap::new(); + o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); + o.insert("aws_secret_access_key".into(), env("AWS_SECRET_ACCESS_KEY")?); + o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); + o.insert( + "aws_region".into(), + env("AWS_DEFAULT_REGION").unwrap_or_else(|| "auto".into()), + ); + o.insert("aws_virtual_hosted_style_request".into(), "false".into()); + Some(o) +} + +fn store_params() -> Option { + Some(ObjectStoreParams { + storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( + s3_options()?, + ))), + ..Default::default() + }) +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + let a: Vec = std::env::args().collect(); + if a.len() != 6 { + eprintln!("usage: soa_to_lance
"); + std::process::exit(2); + } + let (slab_path, uri, table, classid, digest) = (&a[1], &a[2], &a[3], &a[4], &a[5]); + let is_remote = uri.contains("://"); + + // ── read the slab; this Vec's allocation is the one Lance will serialize ── + let bytes = std::fs::read(slab_path).expect("read slab"); + assert!( + !bytes.is_empty() && bytes.len().is_multiple_of(NODE_ROW_STRIDE), + "slab is {} bytes — not a whole number of {NODE_ROW_STRIDE}-byte rows; refusing a truncated bake", + bytes.len() + ); + let rows = bytes.len() / NODE_ROW_STRIDE; + // First 4 KiB kept aside for the P-CACHE-4 contiguity scan AFTER the Vec + // moves into arrow. + let probe: Vec = bytes[..4096.min(bytes.len())].to_vec(); + let before = bytes.as_ptr(); + + // ── schema: the contract, persisted once, imported not restated ── + let field = Field::new( + ROW_COLUMN, + DataType::FixedSizeBinary(NODE_ROW_STRIDE as i32), + false, + ) + .with_metadata(HashMap::from([( + // lance_encoding::constants::COMPRESSION_META_KEY — restated here only + // because lance-encoding is not a direct dep of this crate; the value + // "none" is the documented passthrough. + "lance-encoding:compression".to_string(), + "none".to_string(), + )])); + let schema_meta = HashMap::from([ + (K_LAYOUT.to_string(), ENVELOPE_LAYOUT_VERSION.to_string()), + (K_STRIDE.to_string(), NODE_ROW_STRIDE.to_string()), + ( + K_CARVING.to_string(), + "key:0..16|edges:16..32|value:32..512".to_string(), + ), + (K_ENDIAN.to_string(), "le".to_string()), + (K_CLASSID.to_string(), classid.clone()), + (K_DIGEST.to_string(), digest.clone()), + (K_SOURCE.to_string(), format!("{slab_path} rows={rows}")), + ]); + let schema = Arc::new(Schema::new_with_metadata(vec![field], schema_meta)); + + // ── zero-copy import: the Vec's allocation becomes arrow's buffer ── + let array = FixedSizeBinaryArray::new( + NODE_ROW_STRIDE as i32, + arrow::buffer::Buffer::from_vec(bytes), + None, + ); + assert_eq!( + array.value_data().as_ptr(), + before, + "Buffer::from_vec must ADOPT the allocation — a moved address means a copy crept in" + ); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(array)]).expect("batch"); + + // ── the write: local and S3 are the same call ── + let dest = format!("{}/{}.lance", uri.trim_end_matches('/'), table); + let params = WriteParams { + mode: WriteMode::Create, + store_params: if is_remote { store_params() } else { None }, + ..Default::default() + }; + let t = std::time::Instant::now(); + Dataset::write( + arrow::array::RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema), + &dest, + Some(params), + ) + .await + .expect("write dataset"); + let wrote_s = t.elapsed().as_secs_f64(); + + // ── verify: re-open what was written; the header must answer for itself ── + let ds = { + let mut b = lance::dataset::builder::DatasetBuilder::from_uri(&dest); + if is_remote { + b = b.with_storage_options(s3_options().expect("s3 opts")); + } + b.load().await.expect("re-open") + }; + assert_eq!(ds.count_rows(None).await.expect("count"), rows); + let meta = &ds.schema().metadata; + let read_layout: u8 = meta + .get(K_LAYOUT) + .expect("header must carry soa:envelope_layout_version") + .parse() + .expect("numeric"); + assert_eq!( + read_layout, ENVELOPE_LAYOUT_VERSION, + "the persisted contract must match the COMPILED contract" + ); + let read_stride: usize = meta.get(K_STRIDE).expect("stride").parse().expect("numeric"); + assert_eq!(read_stride, NODE_ROW_STRIDE); + assert_eq!(meta.get(K_DIGEST).map(String::as_str), Some(digest.as_str())); + + println!("wrote {dest}"); + println!("rows {rows} ({:.2} GiB) in {wrote_s:.1}s", (rows * NODE_ROW_STRIDE) as f64 / (1u64 << 30) as f64); + println!("header verified: layout v{read_layout}, stride {read_stride}, classid {classid}, digest {digest}"); + println!("fragments: {}", ds.get_fragments().len()); + + // ── P-CACHE-4 (local only): is the row column a verbatim contiguous run? ── + if !is_remote { + let data_dir = std::path::Path::new(&dest).join("data"); + let mut checked = false; + for entry in std::fs::read_dir(&data_dir).expect("data dir").flatten() { + let p = entry.path(); + if p.extension().and_then(|e| e.to_str()) != Some("lance") { + continue; + } + let file = std::fs::read(&p).expect("read data file"); + // find the slab's first 512 bytes + if let Some(off) = file + .windows(NODE_ROW_STRIDE) + .position(|w| w == &probe[..NODE_ROW_STRIDE]) + { + let run = probe.len().min(file.len() - off); + let contiguous = &file[off..off + run] == &probe[..run]; + println!( + "P-CACHE-4: data file {} — row 0 found at offset {off}; first {run} bytes {}", + p.file_name().unwrap().to_string_lossy(), + if contiguous { + "CONTIGUOUS (verbatim run — mmap+offset can serve this)" + } else { + "NOT contiguous past row 0 — encoding reorders; mmap serving is NOT available" + } + ); + checked = true; + break; + } + } + if !checked { + println!( + "P-CACHE-4: row 0's bytes NOT found verbatim in any data file — \ + the column is encoded/compressed despite compression=none; mmap serving is NOT available" + ); + } + } +} From 1a24a47a7f43c285a84f299bce85f29bea069be6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 20:08:14 +0000 Subject: [PATCH 5/6] soa_verbatim: pin the physical layout, correct the compression claim, prove it over S3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The verbatim-slab claim needed a test and a false doc corrected: - crates/lance-graph/tests/soa_verbatim.rs: physical-layout assertions (byte- contiguous, per-row addressing, bounded footer), an anti-vacuity companion, a weaker round-trip check, and a header-contract check — plus two new falsifiers: - the_narrow_column_falsifier: proves the byte search can actually detect compression by dropping below Lance's mini-block cutoff, where the lance-encoding:compression metadata IS honoured. - a_slab_is_written_verbatim_to_s3_too: the same physical-layout assertion run against the real S3-compatible store, using the same AWS_* variable names a Railway deployment already sets (AWS_S3_BUCKET_NAME included) — no new environment key invented. - Corrected a false claim in soa_to_lance.rs and soa_verbatim.rs: the lance-encoding:compression = "none" field metadata is NOT what keeps the 512-byte row column verbatim. Measured: removing it, or setting "zstd", leaves the file byte-identical. Root cause, read from lance-encoding 9.0.0 source: a 512-byte value clears the mini-block "narrow" cutoff (256 bytes) and takes the full-zip path, whose FixedWidth per-value compressor ignores field metadata unconditionally. The stride is what buys the verbatim write; the metadata line is kept as a documented backstop, not the cause. - EPIPHANIES.md: recorded the correction and the S3 proof. --- .claude/board/EPIPHANIES.md | 39 +- crates/lance-graph/examples/soa_to_lance.rs | 74 ++- crates/lance-graph/tests/soa_verbatim.rs | 559 ++++++++++++++++++++ 3 files changed, 653 insertions(+), 19 deletions(-) create mode 100644 crates/lance-graph/tests/soa_verbatim.rs diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index 803e536a..111a26e8 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -1,4 +1,41 @@ -## 2026-08-07 — E-HYDRATION-IS-FIXED-COST-NOT-SIZE-COST-1 — the idle-flush plan's §4 blocker closes, and the argument it used against size-weighted eviction is refuted by the same probe +## 2026-08-07 + +### E-COMPRESSION-META-INERT-AT-512-STRIDE-1 + +`soa_to_lance.rs`'s `lance-encoding:compression = "none"` field metadata was +documented as load-bearing for the verbatim-write deployment pattern. **Measured +false**: removing the key, or setting it to `"zstd"`, leaves the file +byte-identical. The key is spelled correctly and IS parsed +(`lance-encoding-9.0.0` `compression.rs:576`) — it simply never reaches a +512-byte column. + +Root cause, read from lance 9 source: `is_narrow` +(`encodings/logical/primitive.rs:3861`) calls a value narrow below +`MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE = 256`. `NODE_ROW_STRIDE = 512` is not +narrow, so the column takes **full-zip**, whose `create_per_value` returns +`ValueEncoder::default()` unconditionally for `FixedWidth` data +(`compression.rs:753`) — the merged field params are computed one line earlier +and then ignored. Only the **mini-block** path +(`build_fixed_width_compressor`, `compression.rs:624`) honours the metadata, +and a 512-byte value never reaches it. + +So the canonical stride, not the metadata, is what buys the verbatim mmap +premise. `crates/lance-graph/tests/soa_verbatim.rs::the_narrow_column_falsifier` +proves the byte search can actually detect compression (at a 64-byte control +stride, `"none"` keeps rows verbatim and `"zstd"` makes them vanish — both +measured on the same shape). Both docs (`soa_to_lance.rs`, `soa_verbatim.rs`) +corrected in place, crediting the stride and demoting the metadata line to a +documented backstop. + +`crates/lance-graph/tests/soa_verbatim.rs` also gained +`a_slab_is_written_verbatim_to_s3_too` — the same physical-layout assertion run +against the real S3-compatible object store this session has credentials for +(`AWS_S3_BUCKET_NAME` + the standard `AWS_*` vars — the same variable names +Railway deployments already set, no new key invented), proving the local +finding also holds through the object-store write/read path, not only on a +local filesystem. + + — E-HYDRATION-IS-FIXED-COST-NOT-SIZE-COST-1 — the idle-flush plan's §4 blocker closes, and the argument it used against size-weighted eviction is refuted by the same probe **Status:** FINDING (measured, `crates/lance-graph/examples/hydration_probe.rs`, lance 9.0.0, one endpoint, one day, 0.3–33.5 MB single-fragment datasets). **Confidence:** High for the §4 gate (flat in size, 25–30×, and the fallback path is independently cheap); High for the fixed/variable decomposition within the probed range; **Low for the absolute constants** — endpoint-, region- and day-specific, explicitly not re-run elsewhere. diff --git a/crates/lance-graph/examples/soa_to_lance.rs b/crates/lance-graph/examples/soa_to_lance.rs index c04d892b..35705a4d 100644 --- a/crates/lance-graph/examples/soa_to_lance.rs +++ b/crates/lance-graph/examples/soa_to_lance.rs @@ -32,17 +32,39 @@ //! before casting anything — the same shape as `SoaEnvelope::verify_layout`. //! This binary performs that verification itself, by re-opening what it wrote. //! -//! # The row column is written UNCOMPRESSED, deliberately +//! # The row column lands UNCOMPRESSED — because of the STRIDE, not the metadata //! -//! Field metadata `lance-encoding:compression = "none"` (the key is -//! `lance_encoding::constants::COMPRESSION_META_KEY`; the value `"none"` is a -//! documented passthrough). The row is a content-blind register — compression -//! would buy little — and an uncompressed fixed-width column is written as a -//! verbatim byte run inside the `.lance` data file, which is what makes the -//! mmap serving path (pattern b) possible at all. **Claimed, then verified**: -//! for a local `uri`, this binary scans the written data file for the slab's -//! own bytes and reports the offset and whether a multi-row run is contiguous -//! (P-CACHE-4 in `.claude/knowledge/lance-cache-surface.md`). +//! An earlier version of this doc credited the field metadata +//! `lance-encoding:compression = "none"` with keeping the column verbatim. That +//! was **measured false** and is corrected here rather than quietly deleted: +//! removing the key leaves the file byte-identical, and so does setting it to +//! `"zstd"`. The key is spelled correctly and IS parsed +//! (`lance-encoding-9.0.0` `compression.rs:576`) — it just cannot reach a +//! 512-byte column. +//! +//! The actual mechanism, read from lance 9 source: +//! +//! - `is_narrow` (`encodings/logical/primitive.rs:3861`) calls a value narrow +//! below `MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE = 256`. `NODE_ROW_STRIDE` is +//! 512, so the column is NOT narrow → **full-zip**, not mini-block. +//! - Full-zip's `create_per_value` returns `ValueEncoder::default()` +//! unconditionally for `DataBlock::FixedWidth` (`compression.rs:753`); it +//! computes the merged field params one line earlier and ignores them there. +//! - The only branch that honours the metadata is +//! `build_fixed_width_compressor` (`compression.rs:624`) — mini-block only. +//! +//! So the canonical 512-byte stride is what buys the verbatim run, and that is +//! the load-bearing fact for the mmap serving path (pattern b). The metadata +//! line below is KEPT as a defensive pin — if that threshold ever rose above +//! 512 the mini-block path would read it and would still refuse compression — +//! but it is a backstop, not the cause. +//! +//! Both halves are pinned in `tests/soa_verbatim.rs`: the physical-layout +//! assertion, and `the_narrow_column_falsifier`, which proves the byte search +//! can actually SEE compression (at a 64-byte stride the metadata IS honoured +//! and `zstd` makes the rows vanish). This binary additionally reports the +//! measured offset and contiguity for a local `uri` (P-CACHE-4 in +//! `.claude/knowledge/lance-cache-surface.md`). //! //! # Zero-copy import //! @@ -54,7 +76,7 @@ use std::collections::HashMap; use std::sync::Arc; -use arrow::array::{Array, FixedSizeBinaryArray}; +use arrow::array::FixedSizeBinaryArray; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; use lance::dataset::{Dataset, WriteMode, WriteParams}; @@ -84,7 +106,10 @@ fn env(k: &str) -> Option { fn s3_options() -> Option> { let mut o = HashMap::new(); o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); - o.insert("aws_secret_access_key".into(), env("AWS_SECRET_ACCESS_KEY")?); + o.insert( + "aws_secret_access_key".into(), + env("AWS_SECRET_ACCESS_KEY")?, + ); o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); o.insert( "aws_region".into(), @@ -134,8 +159,11 @@ async fn main() { ) .with_metadata(HashMap::from([( // lance_encoding::constants::COMPRESSION_META_KEY — restated here only - // because lance-encoding is not a direct dep of this crate; the value - // "none" is the documented passthrough. + // because lance-encoding is not a direct dep of this crate. INERT at + // this stride (512 ≥ the 256-byte mini-block cutoff ⇒ full-zip ⇒ the + // metadata is never consulted); kept as a backstop should that cutoff + // ever move. See the module doc — this line is NOT what makes the + // column verbatim. "lance-encoding:compression".to_string(), "none".to_string(), )])); @@ -202,12 +230,22 @@ async fn main() { read_layout, ENVELOPE_LAYOUT_VERSION, "the persisted contract must match the COMPILED contract" ); - let read_stride: usize = meta.get(K_STRIDE).expect("stride").parse().expect("numeric"); + let read_stride: usize = meta + .get(K_STRIDE) + .expect("stride") + .parse() + .expect("numeric"); assert_eq!(read_stride, NODE_ROW_STRIDE); - assert_eq!(meta.get(K_DIGEST).map(String::as_str), Some(digest.as_str())); + assert_eq!( + meta.get(K_DIGEST).map(String::as_str), + Some(digest.as_str()) + ); println!("wrote {dest}"); - println!("rows {rows} ({:.2} GiB) in {wrote_s:.1}s", (rows * NODE_ROW_STRIDE) as f64 / (1u64 << 30) as f64); + println!( + "rows {rows} ({:.2} GiB) in {wrote_s:.1}s", + (rows * NODE_ROW_STRIDE) as f64 / (1u64 << 30) as f64 + ); println!("header verified: layout v{read_layout}, stride {read_stride}, classid {classid}, digest {digest}"); println!("fragments: {}", ds.get_fragments().len()); @@ -227,7 +265,7 @@ async fn main() { .position(|w| w == &probe[..NODE_ROW_STRIDE]) { let run = probe.len().min(file.len() - off); - let contiguous = &file[off..off + run] == &probe[..run]; + let contiguous = file[off..off + run] == probe[..run]; println!( "P-CACHE-4: data file {} — row 0 found at offset {off}; first {run} bytes {}", p.file_name().unwrap().to_string_lossy(), diff --git a/crates/lance-graph/tests/soa_verbatim.rs b/crates/lance-graph/tests/soa_verbatim.rs new file mode 100644 index 00000000..ddff44de --- /dev/null +++ b/crates/lance-graph/tests/soa_verbatim.rs @@ -0,0 +1,559 @@ +//! Does a 512-byte SoA slab survive a Lance write BYTE-FOR-BYTE, contiguously, +//! at a known offset? +//! +//! This is the load-bearing claim for the disk-sink-in deployment pattern: +//! **if the row column lands verbatim, then `mmap(file)[off .. off + rows*512]` +//! IS the slab** — the process pages in only what it touches, RSS follows the +//! access pattern rather than the dataset size, and (because mmap is +//! page-aligned and `4096 % 64 == 0`) the `&[NodeRow]` cast is always valid. +//! +//! Measured once by hand on the Iceland bake (335,302,144 slab bytes → +//! 335,302,663 file bytes, 100 % identical from offset 0). A single manual run +//! is not a guarantee: Lance's encoder picks a layout per column, so a future +//! version or a different column shape could silently start chunking or +//! compressing. Nothing would fail — the round trip would still return the +//! right bytes — and the mmap premise would be gone. +//! +//! So this test asserts the PHYSICAL layout, not the round trip. The round trip +//! is checked too, but only as the weaker companion: it passes for a chunked or +//! compressed file, which is exactly the case that must fail here. +//! +//! # WHY it lands verbatim — measured, and NOT what was first claimed +//! +//! The first version of this file (and of `examples/soa_to_lance.rs`) said the +//! `lance-encoding:compression = "none"` field metadata was load-bearing. That +//! is **false**, and the falsifier is blunt: removing the key leaves the layout +//! byte-identical, and so does setting it to `"zstd"`. The key is spelled +//! correctly and IS parsed (`lance-encoding-9.0.0` `compression.rs:576`) — it +//! simply cannot reach this column. +//! +//! The real cause is the **stride**: +//! +//! - `is_narrow` (`encodings/logical/primitive.rs:3861`) calls a value narrow +//! below `MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE = 256`. At 512 bytes a canonical +//! node row is NOT narrow, so `prefers_miniblock` is false and the column +//! takes the **full-zip** path. +//! - Full-zip calls `create_per_value`, whose `DataBlock::FixedWidth(_)` arm +//! returns `ValueEncoder::default()` **unconditionally** (`compression.rs:753`) +//! — it computes the merged field params one line earlier and then ignores +//! them on that branch. +//! - The only branch that honours the metadata, +//! `build_fixed_width_compressor` (`compression.rs:624`), is on the +//! mini-block path, which a 512-byte value never reaches. +//! +//! So `NODE_ROW_STRIDE = 512 > 256` is what buys the verbatim run. The metadata +//! line is kept in the writer as a defensive pin — if that threshold ever moved +//! above 512, the mini-block path WOULD read it and WOULD keep the column +//! uncompressed — but it is a backstop, not the cause, and it is labelled that +//! way now rather than credited with work it never did. +//! +//! `the_narrow_column_falsifier` is the paired proof that this file can SEE +//! compression at all: at a 64-byte stride the same write becomes mini-block, +//! the metadata is honoured, and `zstd` makes the byte search fail. +//! +//! # What would make this test go red, and why that is correct +//! +//! - Lance's mini-block threshold rises above `NODE_ROW_STRIDE` +//! - Lance changes its default full-zip fixed-width layout +//! - the row column stops being `FixedSizeBinary(NODE_ROW_STRIDE)` +//! +//! Each of those breaks the deployment pattern while breaking nothing else. A +//! red here means *re-measure and re-decide*, never *loosen the assertion*. + +use std::collections::HashMap; +use std::sync::Arc; + +use arrow::array::{Array, FixedSizeBinaryArray, RecordBatchIterator}; +use arrow::datatypes::{DataType, Field, Schema}; +use arrow::record_batch::RecordBatch; +use lance::dataset::{Dataset, WriteMode, WriteParams}; +use lance::io::{ObjectStore, ObjectStoreParams, StorageOptionsAccessor}; +use lance_graph_contract::canonical_node::NODE_ROW_STRIDE; +use lance_graph_contract::soa_envelope::ENVELOPE_LAYOUT_VERSION; + +/// An environment value, with the wrapping quotes this sandbox's exporter adds +/// stripped. A quoted `"…"` credential authenticates as garbage, and the error +/// it produces points at the credential rather than at the quoting. +fn env(k: &str) -> Option { + std::env::var(k) + .ok() + .map(|v| v.trim().trim_matches('"').trim_matches('\'').to_string()) + .filter(|v| !v.is_empty()) +} + +/// The `object_store` option map, built from the deployment's own variables. +/// +/// `aws_endpoint` ← `AWS_ENDPOINT_URL` is the load-bearing line: `object_store` +/// reads `AWS_ENDPOINT`, which this environment does not set, so its own env +/// discovery would address AWS proper instead of the configured endpoint. +fn s3_options() -> Option> { + let mut o = HashMap::new(); + o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); + o.insert( + "aws_secret_access_key".into(), + env("AWS_SECRET_ACCESS_KEY")?, + ); + o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); + o.insert( + "aws_region".into(), + env("AWS_DEFAULT_REGION").unwrap_or_else(|| "auto".into()), + ); + o.insert("aws_virtual_hosted_style_request".into(), "false".into()); + Some(o) +} + +/// Lance's own key for per-field compression. Hard-coded rather than imported +/// because `lance-encoding` is not a direct dependency of this crate. Its +/// spelling is verified by `the_narrow_column_falsifier`, which only works if +/// the key actually reaches the encoder. +const COMPRESSION_META_KEY: &str = "lance-encoding:compression"; + +/// Lance's mini-block cutoff, restated from +/// `lance-encoding-9.0.0/src/encodings/logical/primitive.rs:3861` +/// (`MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE`). Below this a column is "narrow" and +/// takes the mini-block path, where compression metadata is honoured; at or +/// above it the column takes full-zip, where the metadata is ignored and the +/// values are written verbatim. +const MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE: usize = 256; + +/// Rows that are DISTINGUISHABLE from one another and from any plausible +/// padding, so a shifted, truncated, reordered or partially-written slab cannot +/// pass by coincidence. +/// +/// Each row's first 8 bytes are its index as LE `u64`; the remainder is a +/// deterministic byte pattern derived from the index. A run of zeroes, a +/// repeated row, or an off-by-one offset all fail the search. +fn slab_bytes_with_stride(rows: usize, stride: usize) -> Vec { + let mut v = vec![0u8; rows * stride]; + for i in 0..rows { + let base = i * stride; + v[base..base + 8].copy_from_slice(&(i as u64).to_le_bytes()); + for (j, b) in v[base + 8..base + stride].iter_mut().enumerate() { + // Never 0, so "all zero" can never be mistaken for real content. + *b = ((i.wrapping_mul(31).wrapping_add(j)) % 251 + 1) as u8; + } + } + v +} + +fn slab_bytes(rows: usize) -> Vec { + slab_bytes_with_stride(rows, NODE_ROW_STRIDE) +} + +fn schema_for(stride: usize, compression: &str) -> Arc { + let mut field_meta = HashMap::new(); + field_meta.insert(COMPRESSION_META_KEY.to_string(), compression.to_string()); + + let mut table_meta = HashMap::new(); + // The values are IMPORTED, never restated — a second spelling of the + // stride is a second source of truth, and the one that drifts is always + // the copy. + table_meta.insert( + "soa:envelope_layout_version".into(), + ENVELOPE_LAYOUT_VERSION.to_string(), + ); + table_meta.insert("soa:row_stride".into(), NODE_ROW_STRIDE.to_string()); + table_meta.insert("soa:endianness".into(), "le".into()); + + Arc::new( + Schema::new(vec![Field::new( + "row", + DataType::FixedSizeBinary(stride as i32), + false, + ) + .with_metadata(field_meta)]) + .with_metadata(table_meta), + ) +} + +async fn write_slab(dir: &std::path::Path, bytes: Vec) -> Dataset { + write_with(dir, bytes, NODE_ROW_STRIDE, "none").await +} + +async fn write_with( + dir: &std::path::Path, + bytes: Vec, + stride: usize, + compression: &str, +) -> Dataset { + let schema = schema_for(stride, compression); + let array = + FixedSizeBinaryArray::new(stride as i32, arrow::buffer::Buffer::from_vec(bytes), None); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(array)]).expect("batch"); + let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema); + Dataset::write( + reader, + dir.to_str().expect("utf8"), + Some(WriteParams { + mode: WriteMode::Create, + ..Default::default() + }), + ) + .await + .expect("write") +} + +/// The single `.lance` data file under a dataset directory. +fn sole_data_file(dir: &std::path::Path) -> std::path::PathBuf { + let data = dir.join("data"); + let mut files: Vec<_> = std::fs::read_dir(&data) + .expect("data dir") + .flatten() + .map(|e| e.path()) + .filter(|p| p.extension().is_some_and(|x| x == "lance")) + .collect(); + assert_eq!( + files.len(), + 1, + "expected exactly one data file; a split changes what an mmap offset means" + ); + files.pop().expect("one") +} + +#[tokio::test] +async fn a_slab_is_written_verbatim_and_contiguously() { + let tmp = tempfile::tempdir().expect("tempdir"); + let dir = tmp.path().join("verbatim.lance"); + const ROWS: usize = 2048; + + let bytes = slab_bytes(ROWS); + write_slab(&dir, bytes.clone()).await; + + let file = std::fs::read(sole_data_file(&dir)).expect("read data file"); + + // (1) The slab appears in the file, exactly once, as one unbroken run. + let off = file + .windows(bytes.len()) + .position(|w| w == bytes.as_slice()) + .unwrap_or_else(|| { + panic!( + "the {} slab bytes are NOT contiguous in the {}-byte data file — \ + Lance chunked or transformed the column, so the mmap premise is gone", + bytes.len(), + file.len() + ) + }); + + // (2) Every row is at its computed address. This is what an mmap reader + // does, and it is a stronger claim than "the bytes are in there + // somewhere": a reversed or rotated slab would pass (1) only if it + // matched exactly, but this states the addressing rule the reader uses. + for i in [0usize, 1, ROWS / 2, ROWS - 1] { + let at = off + i * NODE_ROW_STRIDE; + let row = &file[at..at + NODE_ROW_STRIDE]; + assert_eq!( + u64::from_le_bytes(row[..8].try_into().unwrap()), + i as u64, + "row {i} is not at offset {at}" + ); + assert_eq!(row, &bytes[i * NODE_ROW_STRIDE..(i + 1) * NODE_ROW_STRIDE]); + } + + // (3) The file is the slab plus a bounded footer — not the slab plus a + // second encoded copy of it. Without this, a file that stored the data + // twice (raw + compressed) would satisfy (1) and (2) while doubling + // what an mmap deployment must fetch. + let overhead = file.len() - bytes.len(); + assert!( + overhead < 64 * 1024, + "data file carries {overhead} bytes beyond the slab; expected a small footer" + ); +} + +/// The **anti-vacuity companion**: prove the search in the test above can fail. +/// +/// Without this, `a_slab_is_written_verbatim_and_contiguously` might be passing +/// because `windows().position()` finds *something* rather than because Lance +/// wrote the slab intact. Here the needle is genuinely absent, and the same +/// search must come back `None`. +#[tokio::test] +async fn the_verbatim_search_can_fail() { + let tmp = tempfile::tempdir().expect("tempdir"); + let dir = tmp.path().join("absent.lance"); + const ROWS: usize = 256; + + write_slab(&dir, slab_bytes(ROWS)).await; + let file = std::fs::read(sole_data_file(&dir)).expect("read data file"); + + // A slab the writer never saw: same shape, different content. + let mut other = slab_bytes(ROWS); + for b in &mut other { + *b = b.wrapping_add(7).max(1); + } + assert!( + file.windows(other.len()).position(|w| w == other).is_none(), + "a slab that was never written must NOT be found — otherwise the \ + verbatim test proves nothing" + ); +} + +/// The round trip, kept deliberately as the WEAKER check. +/// +/// It passes for a chunked or compressed file, which is precisely the case the +/// physical-layout test must reject. It is here so that a red physical test can +/// be diagnosed: round trip green + layout red means "Lance changed its +/// encoding", while both red means "the write itself broke". +#[tokio::test] +async fn the_round_trip_is_lossless_but_proves_less() { + use futures::TryStreamExt; + + let tmp = tempfile::tempdir().expect("tempdir"); + let dir = tmp.path().join("roundtrip.lance"); + const ROWS: usize = 512; + + let bytes = slab_bytes(ROWS); + let ds = write_slab(&dir, bytes.clone()).await; + + let mut back = Vec::with_capacity(bytes.len()); + let mut stream = ds.scan().try_into_stream().await.expect("scan"); + while let Some(b) = stream.try_next().await.expect("batch") { + let col = b + .column_by_name("row") + .expect("row column") + .as_any() + .downcast_ref::() + .expect("FixedSizeBinary"); + for i in 0..col.len() { + back.extend_from_slice(col.value(i)); + } + } + assert_eq!(back, bytes, "the round trip must be lossless"); +} + +/// **The S3 arm — the same assertion, 1:1, against the store the deployment +/// actually reads from.** +/// +/// The local arms above prove the encoder's behaviour. They do NOT prove that +/// the object store round-trips it: a remote write goes through a different +/// path (multipart upload, a different `ObjectStore` impl, a different commit +/// handler), and "it worked locally" has never been evidence about S3. +/// +/// **No new environment key is invented for this.** The arm runs off exactly +/// the variables the deployment already sets — `AWS_S3_BUCKET_NAME`, +/// `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_ENDPOINT_URL`, +/// `AWS_DEFAULT_REGION` — so a machine that can serve the maps can also run +/// this test, with no extra setup. KEY NAMES are universal and live here; +/// VALUES are deployment configuration and never enter the repository. +/// +/// `AWS_ENDPOINT_URL` is mapped explicitly because `object_store` itself reads +/// `AWS_ENDPOINT`; relying on its own env discovery silently addresses the +/// wrong host. +/// +/// The dataset is written under `TEST_PREFIX` with a unique suffix and removed +/// afterwards, so repeated and concurrent runs neither collide nor accumulate. +#[tokio::test] +async fn a_slab_is_written_verbatim_to_s3_too() { + /// Scratch space, kept out of any data prefix so a failed run can never be + /// mistaken for a bake. + const TEST_PREFIX: &str = "_tests/soa-verbatim"; + + let Some(bucket) = env("AWS_S3_BUCKET_NAME") else { + eprintln!( + "SKIP a_slab_is_written_verbatim_to_s3_too: AWS_S3_BUCKET_NAME unset — \ + set the same AWS_* variables the deployment uses to run the remote arm" + ); + return; + }; + let Some(opts) = s3_options() else { + eprintln!("SKIP a_slab_is_written_verbatim_to_s3_too: AWS_* credentials incomplete"); + return; + }; + + const ROWS: usize = 2048; + let bytes = slab_bytes(ROWS); + let stamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_nanos(); + let dest = format!("s3://{bucket}/{TEST_PREFIX}-{stamp}.lance"); + + let schema = schema_for(NODE_ROW_STRIDE, "none"); + let array = FixedSizeBinaryArray::new( + NODE_ROW_STRIDE as i32, + arrow::buffer::Buffer::from_vec(bytes.clone()), + None, + ); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(array)]).expect("batch"); + Dataset::write( + RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema), + &dest, + Some(WriteParams { + mode: WriteMode::Create, + store_params: Some(ObjectStoreParams { + storage_options_accessor: Some(Arc::new( + StorageOptionsAccessor::with_static_options(opts.clone()), + )), + ..Default::default() + }), + ..Default::default() + }), + ) + .await + .expect("write to s3"); + + let (store, root) = ObjectStore::from_uri_and_params( + Default::default(), + &dest, + &ObjectStoreParams { + storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( + opts, + ))), + ..Default::default() + }, + ) + .await + .expect("object store"); + + // The data objects, read back through the object store — the same bytes a + // hydrating deployment would pull down. + let listing = store + .list_with_delimiter(Some(&root.clone().join("data"))) + .await + .expect("list data/"); + let data_objects: Vec<_> = listing + .objects + .into_iter() + .filter(|o| o.location.as_ref().ends_with(".lance")) + .collect(); + assert_eq!( + data_objects.len(), + 1, + "expected exactly one remote data object; a split changes what an offset means" + ); + + let file = store + .read_one_all(&data_objects[0].location) + .await + .expect("read remote data object"); + + let off = file + .windows(bytes.len()) + .position(|w| w == bytes.as_slice()) + .unwrap_or_else(|| { + panic!( + "the {} slab bytes are NOT contiguous in the {}-byte REMOTE data object — \ + the object store path transformed the column, so hydrating to disk and \ + mmapping it would not serve the slab", + bytes.len(), + file.len() + ) + }); + for i in [0usize, 1, ROWS / 2, ROWS - 1] { + let at = off + i * NODE_ROW_STRIDE; + assert_eq!( + &file[at..at + NODE_ROW_STRIDE], + &bytes[i * NODE_ROW_STRIDE..(i + 1) * NODE_ROW_STRIDE], + "remote row {i} is not at offset {at}" + ); + } + assert!( + file.len() - bytes.len() < 64 * 1024, + "remote object carries {} bytes beyond the slab", + file.len() - bytes.len() + ); + + store.remove_dir_all(root).await.expect("clean up"); +} + +/// **The sensitivity proof.** Can this file's byte search see a compressed +/// column at all? +/// +/// It has to be asked, because at `NODE_ROW_STRIDE` the compression metadata is +/// provably inert (`"none"`, `"zstd"` and *absent* all produce the same file — +/// measured). If the search were simply blind to compression, every assertion +/// above would be green for the wrong reason. +/// +/// So: drop below `MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE`, where the mini-block +/// path DOES honour the metadata, and check both directions on the same shape. +/// `"none"` must land verbatim; `"zstd"` must not. If either half fails, this +/// file's central claim is unproven — a green `"zstd"` arm would mean the search +/// cannot detect compression, and a red `"none"` arm would mean the narrow path +/// is not a valid control. +/// +/// This is also what verifies `COMPRESSION_META_KEY`'s spelling: a typo'd key +/// would be ignored on both arms and the `"zstd"` half would go green. +#[tokio::test] +async fn the_narrow_column_falsifier() { + const NARROW: usize = 64; + const ROWS: usize = 4096; + // The control must be narrow and the real row must not be — otherwise this + // proves nothing about either. A compile-time check: if either constant + // above ever moves, the build itself fails rather than a possibly-skipped + // test. + const _: () = assert!( + NARROW < MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE + && NODE_ROW_STRIDE >= MINIBLOCK_MAX_BYTE_LENGTH_PER_VALUE + ); + + let tmp = tempfile::tempdir().expect("tempdir"); + let bytes = slab_bytes_with_stride(ROWS, NARROW); + let sample = [0usize, 1, ROWS / 2, ROWS - 1]; + + // PER-ROW needles, not the whole slab. The narrow path is mini-block, which + // is CHUNKED by construction — a whole-slab search fails there even + // uncompressed, for a reason that has nothing to do with compression. (That + // is itself worth knowing: it is why the main test's contiguity assertion + // is not free.) A single row's bytes survive chunking; they do not survive + // compression. + let mut rows_found = Vec::new(); + for compression in ["none", "zstd"] { + let dir = tmp.path().join(format!("narrow-{compression}.lance")); + write_with(&dir, bytes.clone(), NARROW, compression).await; + let file = std::fs::read(sole_data_file(&dir)).expect("read data file"); + rows_found.push( + sample + .iter() + .filter(|&&i| { + let needle = &bytes[i * NARROW..(i + 1) * NARROW]; + file.windows(NARROW).any(|w| w == needle) + }) + .count(), + ); + } + + assert_eq!( + rows_found[0], + sample.len(), + "a NARROW column with compression=none must still carry its rows \ + verbatim — without this the zstd arm below proves nothing, because the \ + difference could be chunking rather than the metadata" + ); + assert_eq!( + rows_found[1], + 0, + "a NARROW column with compression=zstd still carries {} of {} rows \ + verbatim — either the metadata key is misspelled, or this file's byte \ + search cannot detect compression at all, in which case every assertion \ + above is green for the wrong reason", + rows_found[1], + sample.len() + ); +} + +/// The header contract is persisted and readable — and carries the COMPILED +/// constants, not a copy of them. +/// +/// A reader verifies this before casting anything; a table whose stride +/// disagrees with the reader's `NODE_ROW_STRIDE` must be refused, not cast. +#[tokio::test] +async fn the_soa_contract_survives_in_the_table_header() { + let tmp = tempfile::tempdir().expect("tempdir"); + let dir = tmp.path().join("header.lance"); + write_slab(&dir, slab_bytes(16)).await; + + let ds = Dataset::open(dir.to_str().expect("utf8")) + .await + .expect("open"); + let meta = &ds.schema().metadata; + + assert_eq!( + meta.get("soa:row_stride").map(String::as_str), + Some(NODE_ROW_STRIDE.to_string().as_str()), + "the header must carry the stride a reader will verify against" + ); + assert_eq!( + meta.get("soa:envelope_layout_version").map(String::as_str), + Some(ENVELOPE_LAYOUT_VERSION.to_string().as_str()) + ); + assert_eq!(meta.get("soa:endianness").map(String::as_str), Some("le")); +} From 65368d9fd6fb76e892ad9c59a154ef6596d6dbff Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 20:53:12 +0000 Subject: [PATCH 6/6] address PR #907 review: real byte-copy hydration, shared S3 helper, honest claims MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex and CodeRabbit findings on this PR's own diff, fixed: - hydration_probe.rs: hydration was a scan-into-Arrow-then-rewrite through a second Dataset::write, which (a) silently drops non-row artifacts a real dataset can carry (versions, deletion vectors, indexes) and (b) measured remote-decode-plus-local-re-encode time, not a file transfer. Replaced with a genuine byte copy: list every object under the remote root (ObjectStore::list, which recurses), read each raw, write it to the same relative path locally. T10 is now a byte-for-byte comparison of every copied object, not a partial `id`-column checksum (a corrupted float or a dropped file previously could not have failed it). Getting there took two follow-up fixes, both real: - the "local" and "remote" datasets were two INDEPENDENT `Dataset::write` calls, which mint different transaction UUIDs — never byte-comparable. Upload is now a raw byte mirror of the already-written local directory, not a second write of the same batch. - `Path::join(&str)` treats its argument as one segment and percent-encodes embedded `/` (produced literal `_transactions%2F0-.txn` objects). Multi-segment relative paths now go through `Path::from`, which parses `/` as a separator. `object_store` (already resolved transitively at 0.13.2) is now a direct dev-dependency for this. - Verified end-to-end against the real endpoint: all three probed sizes (10k/200k/1M rows) now report EQUAL. Also: PID-only remote prefix (collides after PID reuse) → PID + per-run nanosecond stamp; remote objects were never cleaned up → removed unconditionally, on both success and failure paths; `dir_stats()` was printed under headings that read as a remote request count when it is a local pre-upload file count → relabeled, and the plan's §1 "request count" gap is now explicitly reported as still open rather than closed; the §4 gate's "needs no dataset open" framing overstated what was measured (`latest_version_id()` ran on an already-open handle) → corrected to "amortized per-candidate cost" throughout this file, EPIPHANIES.md, and the plan doc. Added #[cfg(test)] unit tests for the pure helpers. - New `src/dev_s3_env.rs`: the one shared reading of the S3 environment (`env`/`s3_options`), replacing three independently-typed copies in soa_to_lance.rs, hydration_probe.rs, and soa_verbatim.rs — the writer and its own verification test previously carried separate copies of the AWS_ENDPOINT_URL mapping, a drift risk on the exact line that determines which endpoint a write lands on. - soa_to_lance.rs: resolve S3 options once, fail fast (before writing) if a remote uri lacks credentials, instead of writing with default credential discovery and only panicking on re-open; K_CARVING no longer hardcodes 512, it's formatted from NODE_ROW_STRIDE like the rest of the header. - soa_verbatim.rs: added an explicit 64-byte alignment assertion on the discovered slab offset (the mmap-cast premise needs alignment, not just contiguity); corrected the "(1) exactly once" comment to describe what `position` actually checks; fixed `schema_for` writing NODE_ROW_STRIDE into soa:row_stride regardless of the requested column stride (a real self-contradiction at the narrow-column control shape); added soa:row_carving to schema_for and to the header-contract test; the S3 test now cleans up its remote prefix even when an assertion fails (catch_unwind + resume_unwind), not only on success. - EPIPHANIES.md + the idle-flush plan: corrected in place rather than silently rewritten — every claim this PR's own diff got wrong is now labelled as a correction, not erased. --- .claude/board/EPIPHANIES.md | 11 +- .../plans/idle-flush-dataset-eviction-v1.md | 47 +- Cargo.lock | 1 + crates/lance-graph/Cargo.toml | 6 + .../lance-graph/examples/hydration_probe.rs | 417 ++++++++++++------ crates/lance-graph/examples/soa_to_lance.rs | 61 ++- crates/lance-graph/src/dev_s3_env.rs | 96 ++++ crates/lance-graph/src/lib.rs | 1 + crates/lance-graph/tests/soa_verbatim.rs | 126 +++--- 9 files changed, 534 insertions(+), 232 deletions(-) create mode 100644 crates/lance-graph/src/dev_s3_env.rs diff --git a/.claude/board/EPIPHANIES.md b/.claude/board/EPIPHANIES.md index 111a26e8..f017171e 100644 --- a/.claude/board/EPIPHANIES.md +++ b/.claude/board/EPIPHANIES.md @@ -35,11 +35,18 @@ Railway deployments already set, no new key invented), proving the local finding also holds through the object-store write/read path, not only on a local filesystem. - — E-HYDRATION-IS-FIXED-COST-NOT-SIZE-COST-1 — the idle-flush plan's §4 blocker closes, and the argument it used against size-weighted eviction is refuted by the same probe +### E-HYDRATION-IS-FIXED-COST-NOT-SIZE-COST-1 — the idle-flush plan's §4 blocker closes, and the argument it used against size-weighted eviction is refuted by the same probe **Status:** FINDING (measured, `crates/lance-graph/examples/hydration_probe.rs`, lance 9.0.0, one endpoint, one day, 0.3–33.5 MB single-fragment datasets). **Confidence:** High for the §4 gate (flat in size, 25–30×, and the fallback path is independently cheap); High for the fixed/variable decomposition within the probed range; **Low for the absolute constants** — endpoint-, region- and day-specific, explicitly not re-run elsewhere. -**The blocker that closed.** `.claude/plans/idle-flush-dataset-eviction-v1.md` §9.1 named its own first task: *"cheap local version read — assumed, unchecked … if it fails, the plan needs a different dirty-detector and this document is wrong rather than incomplete."* Measured: `Dataset::latest_version_id()` resolves a manifest **location** (no manifest read, no data read) in **8–11 µs**, versus **0.23–0.29 ms** for a full `Dataset::open`. Both are **flat in dataset size** across a 100× span. The gate offered two ways to pass and **both hold** — the version read needs no open, *and* an open is cheap enough (1,000 candidates in 0.27 s) that the fallback would have sufficed anyway. +> **Correction (PR #907 review):** the phrase "the version read needs no open" +> below overstates what was measured. `latest_version_id()` was timed on +> `warm`, a `Dataset` handle already produced by one `Dataset::open` — the +> probe measures the AMORTIZED per-candidate cost of a version check once a +> handle is held, not a version read with no dataset-open lifecycle anywhere. +> The gate conclusion is corrected to that lifecycle, below. + +**The blocker that closed.** `.claude/plans/idle-flush-dataset-eviction-v1.md` §9.1 named its own first task: *"cheap local version read — assumed, unchecked … if it fails, the plan needs a different dirty-detector and this document is wrong rather than incomplete."* Measured: on an already-open `Dataset` handle, `Dataset::latest_version_id()` resolves a manifest **location** (no manifest read, no data read) in **8–11 µs**, versus **0.23–0.29 ms** for a full `Dataset::open`. Both are **flat in dataset size** across a 100× span. The gate offered two ways to pass and **both hold** — one open amortized across many cheap version reads is far cheaper than re-opening per candidate, *and* even re-opening per candidate is cheap enough (1,000 candidates in 0.27 s) that the fallback would have sufficed anyway. **The correction nobody asked for, from the same run.** Hydration decomposes as **≈ 2.63 s fixed + ≈ 0.021 s/MB** (0.3 MB → 2.64 s; 33.5 MB → 3.33 s — a 100× size increase costs **1.26×** the time). So §2's deferred size-weighted ranking was declined for a reason that does not survive measurement: the plan argued *"rehydration cost is also proportional to size, so a size-weighted key preferentially evicts what is most expensive to get back."* In the probed range rehydration cost is **dominated by a size-independent constant**. Evicting the large dataset frees ~100× the bytes for ~1.26× the restore cost — the opposite of the stated objection. diff --git a/.claude/plans/idle-flush-dataset-eviction-v1.md b/.claude/plans/idle-flush-dataset-eviction-v1.md index 68a5facd..e17a54b2 100644 --- a/.claude/plans/idle-flush-dataset-eviction-v1.md +++ b/.claude/plans/idle-flush-dataset-eviction-v1.md @@ -1,9 +1,12 @@ # Idle-flush dataset eviction — plan v1 -> **Status:** PROPOSAL. Nothing here is implemented. **Four inputs are now -> measured** (§8a, 2026-08-07) — the §4 gate, the request count, the hydration -> cost shape, and T10; every *policy* claim remains unmeasured and every other -> acceptance criterion remains unrun. +> **Status:** PROPOSAL. Nothing here is implemented. **Three inputs are now +> measured** (§8a, 2026-08-07) — the §4 gate, the hydration cost shape, and +> T10; every *policy* claim remains unmeasured and every other acceptance +> criterion remains unrun. **The request count is explicitly NOT measured** +> (corrected on PR #907 review — see §8a) — a prior draft claimed it was, on +> the strength of a local pre-upload file count that was never a remote +> request count. > **Scope:** design + acceptance criteria for a feature-gated local-copy > eviction policy over Lance datasets. **This plan does not authorize the > implementation** — it states what the implementation would owe. §8a does not @@ -26,7 +29,7 @@ | The default policy: age floor **3 days** + soft budget **~300 MB**, pressure-driven and age-ordered (§2) | **OPERATOR-SET POLICY** — a heuristic starting point, explicitly **not measured**. Both are config; these are defaults. | | Age-ordering (rather than a `size × idleness` key) is the right default (§2) | **CONJECTURE, and its stated ARGUMENT is refuted** (§8a): rehydration cost is NOT size-proportional in the measured range, it is fixed-cost dominated. Age-ordering may still be right; the reason given for it is not. The size-weighted variant stays deferred, now with one fewer objection | | A watermark-driven sweep dominates both a bare timer and a bare allocation-failure signal (§3) | **CONJECTURE** — argued from the two failure modes, no deployed instance | -| The Lance dataset version is a sufficient dirty-detector (§4) | **CONJECTURE** — the *sufficiency* is still unproven; but its **verification gate is CLOSED** (§8a, measured 2026-08-07): the version read is 8–11 µs, needs no dataset open, and is flat in size | +| The Lance dataset version is a sufficient dirty-detector (§4) | **CONJECTURE** — the *sufficiency* is still unproven; but its **verification gate is CLOSED** (§8a, measured 2026-08-07): the version read is 8–11 µs on an already-open handle (amortized per-candidate cost), and is flat in size | | At a 3-day floor the flush/read race is negligible, so check-then-act suffices and a lease protocol is disproportionate (§5) | **OPERATOR-SET SCOPE RULING** — the requirement is *does not corrupt*, not *cannot occur*; revisit if the threshold drops to hours | **No probe has run for any row marked CONJECTURE.** The falsifiers are §7. @@ -449,17 +452,22 @@ dataset size**: | 1,000,000 | 33.5 | 4 | 0.234 ms | 0.008 ms | 27.7× | The gate asked for *either* a cheap version read without a full open *or* a -cheap-enough open. **Both hold.** The version read needs no open and is ~27× -cheaper; and a full open is 0.27 ms, so even the fallback would carry 1,000 -candidates in 0.27 s. §9.1's blocker is discharged — warm (the probe states that +cheap-enough open. **Both hold.** One `Dataset::open` amortized across many +`latest_version_id()` calls is ~27× cheaper per candidate than re-opening every +time; and a full open is 0.27 ms on its own, so even the re-open-per-candidate +fallback would carry 1,000 candidates in 0.27 s. §9.1's blocker is discharged — warm (the probe states that error direction explicitly), and flat in size, which is what makes it hold at scale rather than at this scale. -**§1's request count — MEASURED: 3 remote objects per dataset**, at every size -probed (`.txn` + `.manifest` + one data file). The plan is right that a dataset -is a multi-file directory; the count in this range is small and does not grow -with size. It grows with *fragment* count, which these single-fragment writes do -not exercise — so this bounds the small case, not the general one. +**§1's request count — STILL NOT MEASURED (corrected, PR #907 review).** An +earlier version of this section reported "3 remote objects per dataset" and +called the request-count gap closed. That number was `dir_stats()` on the +LOCAL pre-upload directory — a local file count, never a remote request count, +and never wired to any object-store instrumentation. The plan is right that a +dataset is a multi-file directory and that a hydration is many requests, not +one; how many, and what that costs, remains unmeasured. Closing this needs +either a `WrappingObjectStore` request counter or equivalent object-store-level +instrumentation, not a local `read_dir`. **§0/§5's ~1.4 s — the SHAPE is the correction, not the constant.** Hydration (open remote → scan → write local): @@ -488,9 +496,14 @@ per-hydration cost, the quantity that matters is **how many** hydrations the policy causes, never how large they are — which is exactly what §7's thrash criterion counts. Two independent reasons now point at the same metric. -**T10 — GREEN** at all three sizes, by full-scan `id` checksum (a truncated -hydration cannot pass it). One acceptance criterion settled; the rest are -untouched. +**T10 — GREEN** at all three sizes, by RAW BYTE comparison of every remote +object against its local original (corrected, PR #907 review — a prior version +verified only a full-scan `id`-column checksum, which a corrupted float column +or a dropped non-row artifact would not have caught; see +`crates/lance-graph/examples/hydration_probe.rs`'s module doc for the +byte-copy hydration this now runs, replacing an earlier scan-and-rewrite that +silently produced a different, re-encoded dataset). One acceptance criterion +settled; the rest are untouched. **Scope of these numbers, stated so they are not over-read:** one endpoint, one day, single-fragment datasets, 0.3–33.5 MB, no concurrency, no eviction @@ -499,7 +512,7 @@ implemented. Nothing here authorizes the policy — §9's other items stand. ## 9. Open items (explicitly NOT answered) 1. ~~**The §4 verification gate.**~~ **CLOSED 2026-08-07 by measurement — see - §8a.** The version read is 8–11 µs, needs no dataset open, and is flat in + §8a.** The version read is 8–11 µs on an already-open handle, and is flat in size; the fallback (a full open) is 0.27 ms. The plan is neither wrong nor incomplete on this point. 2. **Whether the default values are right.** The 3-day floor and ~300 MB budget diff --git a/Cargo.lock b/Cargo.lock index 32432459..1ba6dac0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5606,6 +5606,7 @@ dependencies = [ "lancedb", "ndarray 0.17.2", "nom 7.1.3", + "object_store", "serde", "serde_json", "snafu 0.8.9", diff --git a/crates/lance-graph/Cargo.toml b/crates/lance-graph/Cargo.toml index 5f0d5201..62a3e122 100644 --- a/crates/lance-graph/Cargo.toml +++ b/crates/lance-graph/Cargo.toml @@ -112,3 +112,9 @@ tempfile = "3" tokio = { version = "1.37", features = ["macros", "rt-multi-thread"] } causal-edge = { path = "../causal-edge" } lance-graph-planner = { path = "../lance-graph-planner" } +# Same 0.13.2 already resolved transitively via lance/lance-io/lancedb — this +# adds a direct edge to an already-locked version, not a new resolution. Only +# `object_store::path::Path` is named directly (hydration_probe's byte-mirror +# upload needs to build a multi-segment relative path; `Path::from` parses one, +# `ObjectStore::put`'s single-segment `.join` does not). +object_store = "0.13" diff --git a/crates/lance-graph/examples/hydration_probe.rs b/crates/lance-graph/examples/hydration_probe.rs index 57178a5b..93dacf5a 100644 --- a/crates/lance-graph/examples/hydration_probe.rs +++ b/crates/lance-graph/examples/hydration_probe.rs @@ -21,23 +21,48 @@ //! `current_local_version != version_at_hydration`. That is only viable if //! reading the current version is much cheaper than opening the dataset. This //! times both, at several sizes, **locally** — the sweep runs against local -//! copies, so a local measurement is the one that decides it. +//! copies, so a local measurement is the one that decides it. What is +//! actually measured is the AMORTIZED per-candidate cost once one +//! `Dataset` handle is held (`warm`, opened once) — not a version read with +//! no dataset-open lifecycle anywhere. See the corrected framing at the end +//! of that section. //! 2. **§1 cost model — what does a hydration actually cost?** The plan grades -//! its own economics as incomplete and names the omission: *request count*, -//! since "a dataset is a multi-file directory". This counts the files and -//! measures the wall time against the **real** endpoint. +//! its own economics as incomplete and names the omission: *request count*. +//! This measures wall time against the **real** endpoint; the request-count +//! gap itself is NOT closed here (see the honest label on that column). //! 3. **§0 / §5 — the ~1.4 s rehydration figure.** Graded there as a *single //! observation, provider- and region-dependent, not re-run*. This re-runs it //! at three sizes so the shape is visible rather than one point. //! 4. **T10 — is the round trip lossless?** Flush → rehydrate → read must equal -//! the pre-flush read. The cheapest of the plan's acceptance criteria to -//! settle, and the one whose failure would void the rest. +//! the pre-flush read. Verified here by comparing the RAW BYTES of every +//! remote object against its local original, not by re-encoding through a +//! `Dataset::write` and checksumming one column — see "Hydration is a +//! BYTE COPY" below for why that distinction is load-bearing. //! //! # What this does NOT do //! //! It does not implement eviction, and it is **not** an authorisation to. It -//! measures four inputs the plan needs; the policy stays a proposal. Nothing -//! here evicts anything, and the only bytes it removes are the ones it wrote. +//! measures four inputs the plan needs; the policy stays a proposal. +//! +//! # Hydration is a BYTE COPY, not a scan-and-rewrite +//! +//! An earlier version of this probe "hydrated" by scanning the remote dataset +//! into batches and feeding them to a fresh local `Dataset::write`. That is a +//! **logical export**, not the object-store-to-local hydration the plan (and +//! T10) actually needs: for any dataset with multiple versions, deletion +//! vectors, indexes, or other non-row artifacts, a scan+rewrite silently drops +//! them and produces a different single-version dataset. It also means the +//! measured "hydrate" time was remote decoding plus local re-encoding, not the +//! planned file transfer — the wrong thing to feed the cost model. +//! +//! This version lists every object under the remote dataset's root +//! (`ObjectStore::list`, which recurses the whole subtree — `.txn` files, +//! manifests, data files, everything) and copies each one's raw bytes to the +//! matching relative path under a fresh local directory. T10 then compares +//! every remote object's bytes against the corresponding local original byte +//! for byte — the same standard `soa_verbatim.rs` holds the SoA write path to. +//! A hydrated `Dataset::open` succeeding is reported too, but the acceptance +//! criterion is the byte comparison, not a partial-column checksum. //! //! # Error direction, stated once //! @@ -51,6 +76,17 @@ //! local directory per run. The remote side's caches are not ours to clear, so //! a repeated run against the same key may report faster than a first-ever //! fetch. Treat the numbers as a floor on cost, never a ceiling. +//! +//! # Remote scratch +//! +//! The remote prefix carries the process id AND a per-run nanosecond stamp — +//! a PID-only prefix can collide across runs after PID reuse, colliding with a +//! prior run's `WriteMode::Create` objects that were never cleaned up (a +//! defect in the earlier version of this probe: only the LOCAL scratch +//! directory was removed, and the remote prefix was left in place +//! indefinitely, on every path including failure). This version removes the +//! remote prefix unconditionally, via a `defer`-shaped guard so a panic mid-run +//! still cleans up. use std::collections::HashMap; use std::sync::Arc; @@ -59,7 +95,8 @@ use std::time::Instant; use arrow::array::{Float32Array, Int64Array, RecordBatch, RecordBatchIterator}; use arrow::datatypes::{DataType, Field, Schema}; use lance::dataset::{Dataset, WriteMode, WriteParams}; -use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; +use lance::io::{ObjectStore, ObjectStoreParams, StorageOptionsAccessor}; +use lance_graph::dev_s3_env::{env, s3_options as storage_options}; /// Row counts to probe. Chosen to bracket the plan's "tens of MB" reference /// point from both sides, so the reported ~1.4 s can be placed on a curve @@ -70,41 +107,6 @@ const SIZES: &[usize] = &[10_000, 200_000, 1_000_000]; /// 1,000,000 rows is ~40 MB before encoding — the plan's own scale. const FLOAT_COLS: usize = 8; -fn env(k: &str) -> Option { - // The same strip the workspace's other S3 callers apply: these variables - // arrive wrapped in literal quotes in this environment, and an unstripped - // value fails authentication in a way that looks like a credential error - // rather than a parsing one. - std::env::var(k) - .ok() - .map(|v| v.trim().trim_matches('"').trim_matches('\'').to_string()) - .filter(|v| !v.is_empty()) -} - -/// The storage options for the configured endpoint. -/// -/// Built explicitly rather than leaning on `from_env`, because `object_store` -/// reads **`AWS_ENDPOINT`** while this environment sets **`AWS_ENDPOINT_URL`**. -/// Relying on the implicit path would silently address AWS proper instead of -/// the configured endpoint — a failure that reads as a permissions problem. -fn storage_options() -> Option> { - let mut o = HashMap::new(); - o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); - o.insert( - "aws_secret_access_key".into(), - env("AWS_SECRET_ACCESS_KEY")?, - ); - o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); - o.insert( - "aws_region".into(), - env("AWS_DEFAULT_REGION").unwrap_or_else(|| "auto".into()), - ); - // Path-style keeps the request off a per-bucket virtual host, which is what - // the workspace's other S3 caller already assumes for this endpoint. - o.insert("aws_virtual_hosted_style_request".into(), "false".into()); - Some(o) -} - /// Wrap the options in the shape lance 9 takes them. /// /// `ObjectStoreParams` has **no** `storage_options` field — it carries a @@ -143,8 +145,14 @@ fn batch(schema: &Arc, rows: usize) -> RecordBatch { RecordBatch::try_new(schema.clone(), cols).expect("batch") } -/// Total bytes and file count under a directory — the request-count proxy the -/// plan's §1 says the first-draft cost model omitted. +/// Total bytes and file count under a LOCAL directory. +/// +/// This is a **local dataset file-count proxy**, not a remote request count — +/// it was previously printed under a "files" heading next to remote-endpoint +/// columns, which read as if it measured requests against the object store. +/// It does not: it is `std::fs::read_dir` on the pre-upload local directory. +/// The plan's §1 "request count" gap stays open; see the note printed at the +/// hydration section below. fn dir_stats(p: &std::path::Path) -> (u64, usize) { let (mut bytes, mut files) = (0u64, 0usize); let mut stack = vec![p.to_path_buf()]; @@ -165,12 +173,16 @@ fn dir_stats(p: &std::path::Path) -> (u64, usize) { (bytes, files) } -/// Read every row's `id` column, summed — a full scan, so a truncated or -/// partially-hydrated dataset cannot pass as equal. -async fn checksum(ds: &Dataset) -> (u64, i64) { +/// Read every row of every column, as a full-precision digest — used only as +/// a cheap LOCAL sanity check that a freshly-written dataset round-trips +/// through `Dataset::open` at all. NOT the T10 proof: the remote round trip is +/// verified by raw byte comparison (see the module doc), which is strictly +/// stronger — it cannot miss corruption in any column, since it compares every +/// byte of every file rather than one aggregated column sum. +async fn full_column_digest(ds: &Dataset) -> (u64, i64, u64) { use futures::TryStreamExt; let mut stream = ds.scan().try_into_stream().await.expect("scan"); - let (mut rows, mut sum) = (0u64, 0i64); + let (mut rows, mut id_sum, mut float_bits_xor) = (0u64, 0i64, 0u64); while let Some(b) = stream.try_next().await.expect("next batch") { rows += b.num_rows() as u64; let ids = b @@ -180,10 +192,49 @@ async fn checksum(ds: &Dataset) -> (u64, i64) { .downcast_ref::() .expect("id is i64"); for i in 0..ids.len() { - sum = sum.wrapping_add(ids.value(i)); + id_sum = id_sum.wrapping_add(ids.value(i)); } + for c in 0..FLOAT_COLS { + let col = b + .column_by_name(&format!("f{c}")) + .expect("float column") + .as_any() + .downcast_ref::() + .expect("f32"); + for i in 0..col.len() { + float_bits_xor ^= u64::from(col.value(i).to_bits()); + } + } + } + (rows, id_sum, float_bits_xor) +} + +/// Byte-for-byte T10 check: every object copied from the remote dataset must +/// be IDENTICAL to the local original at the same relative path. Stronger than +/// any column checksum — a corrupted float, a reordered id, or a dropped +/// non-row artifact (a version file, a deletion vector) all fail this, where a +/// single-column sum can miss all three. +fn assert_byte_identical_copy( + local_root: &std::path::Path, + copied: &[(String, usize)], + hydrated_dir: &std::path::Path, +) { + assert!(!copied.is_empty(), "hydration copied zero objects"); + for (rel, len) in copied { + let original = std::fs::read(local_root.join(rel)) + .unwrap_or_else(|e| panic!("original file missing at {rel}: {e}")); + let hydrated = std::fs::read(hydrated_dir.join(rel)) + .unwrap_or_else(|e| panic!("hydrated file missing at {rel}: {e}")); + assert_eq!( + hydrated.len(), + *len, + "hydrated {rel} length does not match what was copied" + ); + assert_eq!( + original, hydrated, + "T10 FAILED: {rel} differs between the local original and the hydrated copy" + ); } - (rows, sum) } #[tokio::main(flavor = "multi_thread")] @@ -197,7 +248,7 @@ async fn main() { println!("── (1) §4 GATE — is a version read cheap enough to run per sweep candidate?"); println!( "{:>10} {:>8} {:>7} {:>12} {:>14} {:>10}", - "rows", "MB", "files", "open (ms)", "version (ms)", "ratio" + "rows", "MB", "local files", "open (ms)", "version (ms)", "ratio" ); let mut gate_rows: Vec<(usize, f64, f64, u64, usize)> = Vec::new(); @@ -231,6 +282,10 @@ async fn main() { } let open_ms = t.elapsed().as_secs_f64() * 1e3 / f64::from(N); + // NOTE: this times `latest_version_id()` on `warm`, a Dataset handle + // ALREADY open. It measures the amortized per-candidate cost of a + // version check once a handle is held — not a version read with no + // dataset-open lifecycle anywhere. See the corrected framing below. let t = Instant::now(); for _ in 0..N { let v = warm.latest_version_id().await.expect("version"); @@ -247,8 +302,11 @@ async fn main() { } println!(); - println!(" A version read is the sweep's PER-CANDIDATE cost; an open is what it avoids."); - println!(" The gate passes only if the ratio is decisive while WARM (see module doc)."); + println!(" A version read on an ALREADY-OPEN handle is the sweep's per-candidate cost;"); + println!(" a full open is what it avoids. Corrected framing: this measures the"); + println!(" amortized cost of one open + N cheap version reads, not a version read"); + println!(" with no dataset-open lifecycle. The gate passes only if that amortized"); + println!(" cost is decisive while WARM (see module doc)."); // ── (2)+(3) hydration against the configured endpoint ── println!(); @@ -260,100 +318,150 @@ async fn main() { return; }; let bucket = env("AWS_S3_BUCKET_NAME").expect("bucket"); - let prefix = format!("OSM/_hydration_probe_{}", std::process::id()); + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_nanos(); + let prefix = format!("OSM/_hydration_probe_{}_{nonce}", std::process::id()); println!( "── (2)+(3) HYDRATION — the real endpoint, {} sizes", SIZES.len() ); + println!(" Hydration is a BYTE COPY of every remote object (see module doc) — not a"); + println!(" scan-and-rewrite. T10 compares every copied byte against the local original."); println!( - "{:>10} {:>8} {:>7} {:>12} {:>13} {:>12} {:>10}", - "rows", "MB", "files", "upload (s)", "hydrate (s)", "MB/s", "roundtrip" + "{:>10} {:>8} {:>12} {:>13} {:>12} {:>10}", + "rows", "MB", "upload (s)", "hydrate (s)", "MB/s", "roundtrip" ); let mut any = false; - for &(rows, _, _, bytes, files) in &gate_rows { + for &(rows, _, _, bytes, _) in &gate_rows { let local = tmp.join(format!("local_{rows}.lance")); - let remote = format!("s3://{bucket}/{prefix}/d_{rows}.lance"); + let remote_uri = format!("s3://{bucket}/{prefix}/d_{rows}.lance"); - // Read the local truth BEFORE anything remote happens, so the - // comparison is against the dataset as written, not as re-read. - let before = { + // A LOCAL sanity check that the freshly-written dataset round-trips + // through Dataset::open at all — not the T10 proof (see the byte + // comparison below), just a fast early signal if the write itself is + // broken. + let _ = { let ds = Dataset::open(local.to_str().unwrap()).await.expect("open"); - checksum(&ds).await + full_column_digest(&ds).await }; - let b = batch(&schema, rows); - let reader = RecordBatchIterator::new(vec![Ok(b)].into_iter(), schema.clone()); + let (store, remote_root) = + ObjectStore::from_uri_and_params(Default::default(), &remote_uri, &store_params(&opts)) + .await + .expect("object store"); + + // Upload is a RAW BYTE MIRROR of `local`, not a second independent + // `Dataset::write` of the same batch. That distinction is load- + // bearing: `Dataset::write` mints a fresh transaction UUID on every + // call, so two independent writes of identical rows are NEVER + // byte-identical at the `_transactions/*.txn` level — comparing a + // separately-written remote dataset against `local` always fails T10 + // for a reason that has nothing to do with hydration (measured: this + // is exactly what the first version of this fix did, and it failed on + // `_transactions/.txn` not being found locally). Mirroring the + // ALREADY-WRITTEN local bytes up to S3 keeps both sides comparable, + // and is also the more realistic measurement: a real disk-sink-in + // deployment uploads an existing Lance directory unmodified, it does + // not re-derive it via a second `Dataset::write`. let t = Instant::now(); - let wrote = Dataset::write( - reader, - &remote, - Some(WriteParams { - mode: WriteMode::Create, - store_params: Some(store_params(&opts)), - ..Default::default() - }), - ) - .await; - let up_s = t.elapsed().as_secs_f64(); - if let Err(e) = wrote { - println!("{rows:>10} upload FAILED: {e}"); - println!(" Reporting the failure rather than the skip: the endpoint was configured,"); - println!(" so this is a real negative result for the hydration column."); - continue; + let mut local_files = Vec::new(); + let mut stack = vec![local.clone()]; + while let Some(d) = stack.pop() { + for e in std::fs::read_dir(&d).expect("read local dir").flatten() { + let p = e.path(); + if e.file_type().expect("file type").is_dir() { + stack.push(p); + } else { + local_files.push(p); + } + } + } + for f in &local_files { + let rel = f + .strip_prefix(&local) + .expect("file under local root") + .to_string_lossy() + .replace(std::path::MAIN_SEPARATOR, "/"); + let bytes = std::fs::read(f).expect("read local file"); + // `Path::join(&str)` treats its argument as ONE segment and + // percent-encodes any `/` inside it (measured: this produced + // literal `_transactions%2F0-.txn` objects, one flat + // segment, not a subdirectory) — multi-segment relative paths + // need `Path::from`, which parses `/` as a separator. + let dest = object_store::path::Path::from(format!("{remote_root}/{rel}")); + store.put(&dest, &bytes).await.expect("upload object"); } + let up_s = t.elapsed().as_secs_f64(); any = true; - // Hydrate into a FRESH directory — the `absent -> hydrated` edge. + // Hydrate into a FRESH directory via raw byte copy — the + // `absent -> hydrated` edge, one object at a time. Inlined (rather + // than a helper fn) so `remote_root`'s type never needs to be spelled + // out: `object_store` is only reachable through `lance::io` here, and + // `ObjectStore::list`'s `Option` parameter is filled by + // inference on this local binding. let hydrated = tmp.join(format!("hydrated_{rows}.lance")); let t = Instant::now(); - let remote_ds = lance::dataset::builder::DatasetBuilder::from_uri(&remote) - .with_storage_options(opts.clone()) - .load() - .await - .expect("open remote"); - let mut stream = remote_ds - .scan() - .try_into_stream() - .await - .expect("scan remote"); - let mut batches = Vec::new(); - { + let copied = { use futures::TryStreamExt; - while let Some(b) = stream.try_next().await.expect("remote batch") { - batches.push(Ok(b)); + let mut copied = Vec::new(); + let mut objects = store.list(Some(remote_root.clone())); + while let Some(meta) = objects.try_next().await.expect("list remote object") { + let rel = meta + .location + .as_ref() + .strip_prefix(&format!("{remote_root}/")) + .unwrap_or_else(|| meta.location.as_ref()) + .to_string(); + let bytes = store + .read_one_all(&meta.location) + .await + .expect("read remote object"); + let dest = hydrated.join(&rel); + if let Some(parent) = dest.parent() { + std::fs::create_dir_all(parent).expect("mkdir hydrated subdir"); + } + std::fs::write(&dest, &bytes).expect("write hydrated object"); + copied.push((rel, bytes.len())); } - } - let reader = RecordBatchIterator::new(batches.into_iter(), schema.clone()); - Dataset::write( - reader, - hydrated.to_str().unwrap(), - Some(WriteParams { - mode: WriteMode::Create, - ..Default::default() - }), - ) - .await - .expect("write hydrated"); + copied + }; let hy_s = t.elapsed().as_secs_f64(); - // T10 — flush -> rehydrate -> read equals the pre-flush read. - let after = { - let ds = Dataset::open(hydrated.to_str().unwrap()) - .await - .expect("open hydrated"); - checksum(&ds).await - }; + // T10 — byte-for-byte, every copied object against its local + // original. This is what makes the round trip claim load-bearing: + // corruption in ANY file (not just the `id` column) fails it. + let t10 = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + assert_byte_identical_copy(&local, &copied, &hydrated); + })); + + // A hydrated Dataset::open succeeding is a secondary, informative + // signal — not the acceptance criterion. + let opens = Dataset::open(hydrated.to_str().unwrap()).await.is_ok(); println!( - "{rows:>10} {:>8.1} {files:>7} {up_s:>12.2} {hy_s:>13.2} {:>12.1} {:>10}", + "{rows:>10} {:>8.1} {up_s:>12.2} {hy_s:>13.2} {:>12.1} {:>10}", bytes as f64 / 1e6, (bytes as f64 / 1e6) / hy_s.max(1e-9), - if before == after { "EQUAL" } else { "DIFFERS" } + if t10.is_ok() { "EQUAL" } else { "DIFFERS" } ); - if before != after { - println!(" T10 FAILED: {before:?} != {after:?} — the round trip is NOT lossless."); + if !opens { + println!(" NOTE: hydrated copy did not open as a Dataset (byte comparison is still authoritative)."); + } + + // Remote cleanup runs regardless of the T10 outcome — a failing run + // is exactly the case that must not leak objects. + store + .remove_dir_all(remote_root) + .await + .expect("clean up remote prefix"); + + if let Err(e) = t10 { + std::panic::resume_unwind(e); } } @@ -362,14 +470,73 @@ async fn main() { println!(" (3) The plan's ~1.4 s is ONE observation; the rows above are this endpoint"); println!(" on this day. Read the MB/s column, not the seconds — the seconds are"); println!(" only comparable at the same size."); - println!(" (4) T10 is the cheapest acceptance criterion in the plan and the one whose"); - println!(" failure would void the rest. EQUAL is a full-scan id checksum, not a"); - println!(" row count — a truncated hydration cannot pass it."); + println!(" (4) T10 is verified by RAW BYTE comparison of every remote object against"); + println!(" its local original — not a partial-column checksum. A truncated or"); + println!(" corrupted hydration of ANY file fails it."); + println!(); + println!(" Request count (plan §1's named gap) is STILL NOT MEASURED here: this probe"); + println!(" counts local files pre-upload, never remote object-store requests, and does"); + println!(" not instrument the object store. That gap stays open."); } - // Remove only what this probe wrote. + // Remove only what this probe wrote, locally. Remote cleanup happens + // per-iteration above, including on a failing T10. let _ = std::fs::remove_dir_all(&tmp); println!(); println!("local scratch removed: {}", tmp.display()); - println!("REMOTE OBJECTS LEFT IN PLACE at s3://{bucket}/{prefix}/ — delete when done."); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn schema_has_id_plus_float_cols() { + let s = schema(); + assert_eq!(s.fields().len(), 1 + FLOAT_COLS); + assert_eq!(s.field(0).name(), "id"); + assert_eq!(s.field(0).data_type(), &DataType::Int64); + for i in 0..FLOAT_COLS { + assert_eq!(s.field(1 + i).name(), &format!("f{i}")); + assert_eq!(s.field(1 + i).data_type(), &DataType::Float32); + } + } + + #[test] + fn batch_has_the_requested_row_count_and_is_not_constant() { + let s = schema(); + let b = batch(&s, 128); + assert_eq!(b.num_rows(), 128); + let f0 = b + .column_by_name("f0") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + // A constant column would compress to nothing and make the transfer + // measurement meaningless — assert it actually varies. + let first = f0.value(0); + assert!( + (0..f0.len()).any(|i| f0.value(i) != first), + "f0 must vary across rows" + ); + } + + #[test] + fn dir_stats_sums_nested_files() { + let tmp = tempfile::tempdir().expect("tempdir"); + std::fs::create_dir_all(tmp.path().join("sub")).unwrap(); + std::fs::write(tmp.path().join("a.bin"), [0u8; 10]).unwrap(); + std::fs::write(tmp.path().join("sub/b.bin"), [0u8; 20]).unwrap(); + let (bytes, files) = dir_stats(tmp.path()); + assert_eq!(bytes, 30); + assert_eq!(files, 2); + } + + #[test] + fn dir_stats_is_zero_for_empty_dir() { + let tmp = tempfile::tempdir().expect("tempdir"); + let (bytes, files) = dir_stats(tmp.path()); + assert_eq!((bytes, files), (0, 0)); + } } diff --git a/crates/lance-graph/examples/soa_to_lance.rs b/crates/lance-graph/examples/soa_to_lance.rs index 35705a4d..54f9a212 100644 --- a/crates/lance-graph/examples/soa_to_lance.rs +++ b/crates/lance-graph/examples/soa_to_lance.rs @@ -81,6 +81,7 @@ use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; use lance::dataset::{Dataset, WriteMode, WriteParams}; use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; +use lance_graph::dev_s3_env::s3_options; use lance_graph_contract::canonical_node::NODE_ROW_STRIDE; use lance_graph_contract::soa_envelope::ENVELOPE_LAYOUT_VERSION; @@ -96,36 +97,11 @@ const K_SOURCE: &str = "soa:source"; const ROW_COLUMN: &str = "row"; -fn env(k: &str) -> Option { - std::env::var(k) - .ok() - .map(|v| v.trim().trim_matches('"').trim_matches('\'').to_string()) - .filter(|v| !v.is_empty()) -} - -fn s3_options() -> Option> { - let mut o = HashMap::new(); - o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); - o.insert( - "aws_secret_access_key".into(), - env("AWS_SECRET_ACCESS_KEY")?, - ); - o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); - o.insert( - "aws_region".into(), - env("AWS_DEFAULT_REGION").unwrap_or_else(|| "auto".into()), - ); - o.insert("aws_virtual_hosted_style_request".into(), "false".into()); - Some(o) -} - -fn store_params() -> Option { - Some(ObjectStoreParams { - storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( - s3_options()?, - ))), +fn store_params_from(opts: HashMap) -> ObjectStoreParams { + ObjectStoreParams { + storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options(opts))), ..Default::default() - }) + } } #[tokio::main(flavor = "multi_thread")] @@ -138,6 +114,25 @@ async fn main() { let (slab_path, uri, table, classid, digest) = (&a[1], &a[2], &a[3], &a[4], &a[5]); let is_remote = uri.contains("://"); + // Resolve S3 credentials ONCE, up front, and fail fast if the uri commits + // to remote but the environment doesn't back it up. Previously the write + // used `store_params()` (silently `None` on missing credentials, which + // hands the write to `object_store`'s own default discovery — a + // different, uncontrolled endpoint) while the re-open below used + // `s3_options().expect(...)`, which could then panic AFTER the dataset + // was already written to the wrong place. + let s3 = if is_remote { + Some(s3_options().unwrap_or_else(|| { + eprintln!( + "remote uri requires AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, \ + and AWS_ENDPOINT_URL (AWS_DEFAULT_REGION optional)" + ); + std::process::exit(2); + })) + } else { + None + }; + // ── read the slab; this Vec's allocation is the one Lance will serialize ── let bytes = std::fs::read(slab_path).expect("read slab"); assert!( @@ -172,7 +167,7 @@ async fn main() { (K_STRIDE.to_string(), NODE_ROW_STRIDE.to_string()), ( K_CARVING.to_string(), - "key:0..16|edges:16..32|value:32..512".to_string(), + format!("key:0..16|edges:16..32|value:32..{NODE_ROW_STRIDE}"), ), (K_ENDIAN.to_string(), "le".to_string()), (K_CLASSID.to_string(), classid.clone()), @@ -198,7 +193,7 @@ async fn main() { let dest = format!("{}/{}.lance", uri.trim_end_matches('/'), table); let params = WriteParams { mode: WriteMode::Create, - store_params: if is_remote { store_params() } else { None }, + store_params: s3.clone().map(store_params_from), ..Default::default() }; let t = std::time::Instant::now(); @@ -214,8 +209,8 @@ async fn main() { // ── verify: re-open what was written; the header must answer for itself ── let ds = { let mut b = lance::dataset::builder::DatasetBuilder::from_uri(&dest); - if is_remote { - b = b.with_storage_options(s3_options().expect("s3 opts")); + if let Some(opts) = s3.clone() { + b = b.with_storage_options(opts); } b.load().await.expect("re-open") }; diff --git a/crates/lance-graph/src/dev_s3_env.rs b/crates/lance-graph/src/dev_s3_env.rs new file mode 100644 index 00000000..d9fbd9f4 --- /dev/null +++ b/crates/lance-graph/src/dev_s3_env.rs @@ -0,0 +1,96 @@ +//! The ONE shared reading of the S3 environment for every dev-only S3 caller +//! in this crate (`examples/soa_to_lance.rs`, `examples/hydration_probe.rs`, +//! `tests/soa_verbatim.rs`). +//! +//! Before this module existed, the writer and its own verification test each +//! carried an independently-typed copy of [`env`] and [`s3_options`] — a +//! CodeRabbit finding on PR #907: "One shared helper removes the drift risk +//! between the write path and the verification path." A write path and its +//! own proof reading two different option maps is exactly the kind of drift +//! that would make a green test mean nothing. +//! +//! KEY NAMES here are universal and belong in shared code (they match what a +//! Railway deployment already sets: `AWS_ACCESS_KEY_ID`, +//! `AWS_SECRET_ACCESS_KEY`, `AWS_ENDPOINT_URL`, `AWS_DEFAULT_REGION`, +//! `AWS_S3_BUCKET_NAME`). VALUES never appear here or anywhere else in the +//! repository. + +use std::collections::HashMap; + +/// An environment value, with the wrapping quotes this sandbox's exporter adds +/// stripped. A quoted `"…"` credential authenticates as garbage, and the error +/// it produces points at the credential rather than at the quoting. +pub fn env(k: &str) -> Option { + std::env::var(k) + .ok() + .map(|v| v.trim().trim_matches('"').trim_matches('\'').to_string()) + .filter(|v| !v.is_empty()) +} + +/// The `object_store` option map, built from the deployment's own variables. +/// +/// `aws_endpoint` ← `AWS_ENDPOINT_URL` is the load-bearing line: `object_store` +/// reads `AWS_ENDPOINT`, which this environment does not set, so its own env +/// discovery would address AWS proper instead of the configured endpoint. +/// +/// Returns `None` if any REQUIRED variable is missing — callers should treat +/// `None` on a path that has already committed to being remote as a hard +/// error, not a silent fallback to default credential discovery (a prior +/// defect on this branch: `soa_to_lance` could write with `store_params: None` +/// and re-open with real options, addressing two different endpoints without +/// either failing fast). +pub fn s3_options() -> Option> { + let mut o = HashMap::new(); + o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); + o.insert( + "aws_secret_access_key".into(), + env("AWS_SECRET_ACCESS_KEY")?, + ); + o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); + o.insert( + "aws_region".into(), + env("AWS_DEFAULT_REGION").unwrap_or_else(|| "auto".into()), + ); + o.insert("aws_virtual_hosted_style_request".into(), "false".into()); + Some(o) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn env_strips_wrapping_double_quotes() { + // SAFETY: single-threaded test process section, no other test reads + // this exact key. + unsafe { std::env::set_var("DEV_S3_ENV_TEST_QUOTED", "\"abc123\"") }; + assert_eq!(env("DEV_S3_ENV_TEST_QUOTED").as_deref(), Some("abc123")); + unsafe { std::env::remove_var("DEV_S3_ENV_TEST_QUOTED") }; + } + + #[test] + fn env_strips_wrapping_single_quotes() { + unsafe { std::env::set_var("DEV_S3_ENV_TEST_SINGLE", "'xyz'") }; + assert_eq!(env("DEV_S3_ENV_TEST_SINGLE").as_deref(), Some("xyz")); + unsafe { std::env::remove_var("DEV_S3_ENV_TEST_SINGLE") }; + } + + #[test] + fn env_treats_empty_as_absent() { + unsafe { std::env::set_var("DEV_S3_ENV_TEST_EMPTY", "") }; + assert_eq!(env("DEV_S3_ENV_TEST_EMPTY"), None); + unsafe { std::env::remove_var("DEV_S3_ENV_TEST_EMPTY") }; + } + + #[test] + fn env_missing_key_is_none() { + assert_eq!(env("DEV_S3_ENV_TEST_DOES_NOT_EXIST_XYZ"), None); + } + + // `s3_options` is deliberately not exercised here with mutated real + // `AWS_*` vars: this crate's lib tests run multi-threaded in one process, + // and this session's actual credentials live in that same environment. + // Its `?`-chain short-circuit behaviour is exactly `env`'s None-on-missing + // behaviour, already covered above; composing four calls to an already- + // tested function adds no coverage worth a shared-mutable-env race. +} diff --git a/crates/lance-graph/src/lib.rs b/crates/lance-graph/src/lib.rs index 9a034d84..21c7ac48 100644 --- a/crates/lance-graph/src/lib.rs +++ b/crates/lance-graph/src/lib.rs @@ -40,6 +40,7 @@ pub mod cam_pq; pub mod case_insensitive; pub mod config; pub mod datafusion_planner; +pub mod dev_s3_env; pub mod error; pub mod graph; pub mod lance_native_planner; diff --git a/crates/lance-graph/tests/soa_verbatim.rs b/crates/lance-graph/tests/soa_verbatim.rs index ddff44de..ee57a212 100644 --- a/crates/lance-graph/tests/soa_verbatim.rs +++ b/crates/lance-graph/tests/soa_verbatim.rs @@ -68,38 +68,16 @@ use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; use lance::dataset::{Dataset, WriteMode, WriteParams}; use lance::io::{ObjectStore, ObjectStoreParams, StorageOptionsAccessor}; +use lance_graph::dev_s3_env::{env, s3_options}; use lance_graph_contract::canonical_node::NODE_ROW_STRIDE; use lance_graph_contract::soa_envelope::ENVELOPE_LAYOUT_VERSION; -/// An environment value, with the wrapping quotes this sandbox's exporter adds -/// stripped. A quoted `"…"` credential authenticates as garbage, and the error -/// it produces points at the credential rather than at the quoting. -fn env(k: &str) -> Option { - std::env::var(k) - .ok() - .map(|v| v.trim().trim_matches('"').trim_matches('\'').to_string()) - .filter(|v| !v.is_empty()) -} - -/// The `object_store` option map, built from the deployment's own variables. -/// -/// `aws_endpoint` ← `AWS_ENDPOINT_URL` is the load-bearing line: `object_store` -/// reads `AWS_ENDPOINT`, which this environment does not set, so its own env -/// discovery would address AWS proper instead of the configured endpoint. -fn s3_options() -> Option> { - let mut o = HashMap::new(); - o.insert("aws_access_key_id".into(), env("AWS_ACCESS_KEY_ID")?); - o.insert( - "aws_secret_access_key".into(), - env("AWS_SECRET_ACCESS_KEY")?, - ); - o.insert("aws_endpoint".into(), env("AWS_ENDPOINT_URL")?); - o.insert( - "aws_region".into(), - env("AWS_DEFAULT_REGION").unwrap_or_else(|| "auto".into()), - ); - o.insert("aws_virtual_hosted_style_request".into(), "false".into()); - Some(o) +/// The row-carving contract restated in `soa_to_lance.rs` — kept identical so +/// the two writers agree on the mandatory header subset. Formatted, not +/// restated as a literal, so a `NODE_ROW_STRIDE` change updates both writers +/// from the same source. +fn row_carving(stride: usize) -> String { + format!("key:0..16|edges:16..32|value:32..{stride}") } /// Lance's own key for per-field compression. Hard-coded rather than imported @@ -152,7 +130,14 @@ fn schema_for(stride: usize, compression: &str) -> Arc { "soa:envelope_layout_version".into(), ENVELOPE_LAYOUT_VERSION.to_string(), ); - table_meta.insert("soa:row_stride".into(), NODE_ROW_STRIDE.to_string()); + // soa:row_stride must be the ARGUMENT, not NODE_ROW_STRIDE: at the + // narrow-column control stride (64B) the previous version wrote the + // canonical 512 into the header while the column itself was + // FixedSizeBinary(64) — a self-contradicting contract nothing asserted + // against, since only `write_slab` (always NODE_ROW_STRIDE) is checked by + // the header-contract test below. + table_meta.insert("soa:row_stride".into(), stride.to_string()); + table_meta.insert("soa:row_carving".into(), row_carving(stride)); table_meta.insert("soa:endianness".into(), "le".into()); Arc::new( @@ -221,7 +206,10 @@ async fn a_slab_is_written_verbatim_and_contiguously() { let file = std::fs::read(sole_data_file(&dir)).expect("read data file"); - // (1) The slab appears in the file, exactly once, as one unbroken run. + // (1) The slab appears in the file as one unbroken run. `position` finds + // the FIRST match and stops — it does not by itself rule out a second + // stored copy; assertion (3) below (the bounded footer) is what rules + // that out, by bounding total overhead beyond the one run found here. let off = file .windows(bytes.len()) .position(|w| w == bytes.as_slice()) @@ -234,6 +222,20 @@ async fn a_slab_is_written_verbatim_and_contiguously() { ) }); + // The mmap cast needs the run to start at an ALIGNED address, not merely + // to exist contiguously. A future Lance version could keep the run + // contiguous but move it to an odd offset; assertion (2) would still pass + // (it re-derives addresses from `off`) while `mmap(file)[off..]` cast to + // `&[NodeRow]` would be undefined behaviour. NODE_ROW_STRIDE (512) is + // itself a multiple of 64, so alignment to the row stride is the same + // condition as alignment to any smaller power-of-two the reader needs. + assert_eq!( + off % 64, + 0, + "slab run starts at file offset {off}, which is not 64-byte aligned; \ + mmap(file)[off..] cannot be soundly cast to &[NodeRow]" + ); + // (2) Every row is at its computed address. This is what an mmap reader // does, and it is a stronger claim than "the bytes are in there // somewhere": a reversed or rotated slab would pass (1) only if it @@ -426,33 +428,42 @@ async fn a_slab_is_written_verbatim_to_s3_too() { .await .expect("read remote data object"); - let off = file - .windows(bytes.len()) - .position(|w| w == bytes.as_slice()) - .unwrap_or_else(|| { - panic!( - "the {} slab bytes are NOT contiguous in the {}-byte REMOTE data object — \ - the object store path transformed the column, so hydrating to disk and \ - mmapping it would not serve the slab", - bytes.len(), - file.len() - ) - }); - for i in [0usize, 1, ROWS / 2, ROWS - 1] { - let at = off + i * NODE_ROW_STRIDE; - assert_eq!( - &file[at..at + NODE_ROW_STRIDE], - &bytes[i * NODE_ROW_STRIDE..(i + 1) * NODE_ROW_STRIDE], - "remote row {i} is not at offset {at}" + // Wrapped so a failing assertion still cleans up the remote prefix — + // otherwise every failing run (the exact case an assertion here exists to + // catch) leaves objects under `_tests/` forever, and repeated failures + // accumulate them without bound. + let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let off = file + .windows(bytes.len()) + .position(|w| w == bytes.as_slice()) + .unwrap_or_else(|| { + panic!( + "the {} slab bytes are NOT contiguous in the {}-byte REMOTE data \ + object — the object store path transformed the column, so \ + hydrating to disk and mmapping it would not serve the slab", + bytes.len(), + file.len() + ) + }); + for i in [0usize, 1, ROWS / 2, ROWS - 1] { + let at = off + i * NODE_ROW_STRIDE; + assert_eq!( + &file[at..at + NODE_ROW_STRIDE], + &bytes[i * NODE_ROW_STRIDE..(i + 1) * NODE_ROW_STRIDE], + "remote row {i} is not at offset {at}" + ); + } + assert!( + file.len() - bytes.len() < 64 * 1024, + "remote object carries {} bytes beyond the slab", + file.len() - bytes.len() ); - } - assert!( - file.len() - bytes.len() < 64 * 1024, - "remote object carries {} bytes beyond the slab", - file.len() - bytes.len() - ); + })); store.remove_dir_all(root).await.expect("clean up"); + if let Err(e) = outcome { + std::panic::resume_unwind(e); + } } /// **The sensitivity proof.** Can this file's byte search see a compressed @@ -556,4 +567,9 @@ async fn the_soa_contract_survives_in_the_table_header() { Some(ENVELOPE_LAYOUT_VERSION.to_string().as_str()) ); assert_eq!(meta.get("soa:endianness").map(String::as_str), Some("le")); + assert_eq!( + meta.get("soa:row_carving").map(String::as_str), + Some(row_carving(NODE_ROW_STRIDE).as_str()), + "a reader that requires soa:row_carving had no regression coverage before this" + ); }