feat(wal): persist immutable prepared Lance descriptors

.github/workflows/rust-test.yml

@@ -84,6 +84,8 @@ jobs:
         run: |
           cargo clippy -p lance-context-wal --features replication --all-targets -- -D warnings
           cargo test -p lance-context-wal --features replication --lib --test replication_process
+      - name: Prepared WAL descriptors with real Lance files and process recovery
+        run: cargo test -p lance-context-wal --features lance-integration --test replication_process lance_publication
       - name: WAL resource observation in a container cgroup namespace
         run: |
           cargo test -p lance-context-wal --features runner --test resource --no-run --message-format=json > "$RUNNER_TEMP/wal-resource-build.jsonl"

crates/lance-context-wal/Cargo.toml

@@ -7,6 +7,8 @@ publish = false
 description = "Independent Rust WAL experiments owned by lance-context"
 
 [features]
+# Real Lance materialization/recovery qualification; normal WAL replicas do not link Lance.
+lance-integration = ["replication", "ipc", "arrow-contract", "dep:lance", "dep:lance-table", "dep:lance-io", "dep:object_store", "dep:async-trait", "dep:futures", "dep:serde_json"]
 client-journal = ["dep:anyhow", "dep:raft-engine", "dep:bytes", "dep:serde", "dep:serde_json", "dep:tokio", "tokio/sync"]
 replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "dep:ring", "uuid/v4", "serde/rc"]
 raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"]
@@ -21,6 +23,12 @@ ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "de
 lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
+lance = { version = "13.0.0", optional = true }
+lance-table = { version = "13.0.0", optional = true }
+lance-io = { version = "13.0.0", optional = true }
+object_store = { version = "0.14.2", optional = true }
+async-trait = { version = "0.1.92", optional = true }
+futures = { version = "0.3", optional = true }
 ring = { version = "0.17", optional = true }
 bincode = { version = "1.3", optional = true }
 slog = { version = "2", optional = true }

crates/lance-context-wal/src/replication.rs

@@ -49,6 +49,7 @@ pub struct Limits {
     pub max_tables: u64,
     pub max_active_plans: u64,
     pub max_plan_source_bytes: u64,
+    pub max_descriptor_bytes: u64,
     pub max_record_bytes: u64,
     pub max_records: u64,
     pub max_producers: u64,
@@ -62,6 +63,7 @@ impl Default for Limits {
             max_tables: 1024,
             max_active_plans: 128,
             max_plan_source_bytes: 64 * 1024 * 1024,
+            max_descriptor_bytes: 256 * 1024,
             max_record_bytes: 8 * 1024 * 1024,
             max_records: 100_000,
             max_producers: 16_384,
@@ -77,6 +79,8 @@ impl Limits {
                 && self.catalog_key.is_none_or(|key| key != [0; 32])
                 && (1..=256).contains(&self.max_active_plans)
                 && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes)
+                && (1..=publication::MAX_DESCRIPTOR_BYTES as u64)
+                    .contains(&self.max_descriptor_bytes)
                 && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
                 && (1..=500_000).contains(&self.max_records)
                 && (1..=100_000).contains(&self.max_producers)
@@ -86,7 +90,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v4",
+            "lance-context/replicated-append/v5",
             self,
         ))?)
     }
@@ -399,6 +403,17 @@ impl Node {
         self.propose(publication::control_group(&self.layout), request, batch)
     }
 
+    /// Persist one immutable descriptor set for an exact pending plan. The
+    /// caller must separately validate source pins and coordinate Lance commits.
+    pub fn prepare_publication(
+        &mut self,
+        request: u64,
+        prepared: publication::PrepareRequest,
+    ) -> Result<()> {
+        let batch = prepared.batch(&self.layout, &self.limits)?;
+        self.propose(publication::control_group(&self.layout), request, batch)
+    }
+
     /// Verify a catalog quorum-read attestation and persist the exact complete
     /// table prefix in its source group. Completion, never enqueue, pins it.
     pub fn pin_source(&mut self, request: u64, permit: source::Permit) -> Result<()> {

crates/lance-context-wal/src/replication/checkpoint.rs

@@ -14,8 +14,8 @@ use std::{
     path::PathBuf,
 };
 
-const FORMAT: u32 = 4;
-const MAGIC: &[u8; 8] = b"LCWALCP4";
+const FORMAT: u32 = 5;
+const MAGIC: &[u8; 8] = b"LCWALCP5";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -84,7 +84,8 @@ fn metadata_limit(limits: &Limits) -> u64 {
     // limits.identity() validates all policy maxima before this is used.
     (limits.max_records + limits.max_producers) * 268
         + limits.max_tables * (publication::MAX_BASE_TABLE_BYTES as u64 + 4)
-        + limits.max_tables.min(limits.max_active_plans) * publication::MAX_COMMAND_BYTES as u64
+        + limits.max_tables.min(limits.max_active_plans)
+            * (128 * 1024 + limits.max_descriptor_bytes + 1024)
         + limits.max_tables.min(limits.max_active_plans) * (source::MAX_PIN_BYTES as u64 + 4)
         + 1024
 }

crates/lance-context-wal/src/replication/publication.rs

@@ -1,20 +1,24 @@
 //! Replicated table bindings and publisher epochs. These commands run through
 //! the same durable Raft path as appends; a local view is not an active lease.
-//! Immutable publication plans survive takeover. Source pins, manifest outcomes
-//! and retirement remain separate work. No Lance manifest is changed here.
+//! Immutable plans and prepared descriptors survive takeover. Source pins live
+//! in the source module; manifest outcomes and retirement remain separate work.
+//! No Lance manifest is changed here.
 use super::*;
 use crate::{ingest_contract::IngestContract, record::RecordKind};
 use bincode::Options;
 use bytes::Bytes;
 mod plan;
+mod prepared;
 pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS};
+pub use prepared::{Descriptors, PrepareRequest, Prepared, MAX_DESCRIPTOR_BYTES};
 #[cfg(test)]
 mod tests;
 
 pub(super) const MAX_ACQUIRE_BYTES: usize = 8192;
-pub(super) const MAX_COMMAND_BYTES: usize = plan::MAX_PLAN_BYTES;
+pub(super) const MAX_COMMAND_BYTES: usize = prepared::MAX_PREPARE_BYTES;
 pub(super) const MAX_BASE_TABLE_BYTES: usize = MAX_ACQUIRE_BYTES + 2048;
-pub(super) const MAX_TABLE_BYTES: usize = MAX_BASE_TABLE_BYTES + plan::MAX_PLAN_BYTES;
+pub(super) const MAX_TABLE_BYTES: usize =
+    MAX_BASE_TABLE_BYTES + plan::MAX_PLAN_BYTES + prepared::MAX_PREPARE_BYTES;
 fn codec(limit: usize) -> impl Options {
     bincode::DefaultOptions::new()
         .with_fixint_encoding()
@@ -62,25 +66,28 @@ impl Acquire {
 enum Command {
     Acquire(Box<Acquire>),
     Plan(Box<PlanRequest>),
+    Prepare(Box<PrepareRequest>),
 }
 impl Command {
     fn identity(&self) -> (Uuid, Uuid, u64) {
         match self {
             Self::Acquire(a) => (a.table, a.owner, a.expected_epoch + 1),
             Self::Plan(p) => (p.authority.table, p.authority.owner, p.authority.epoch),
+            Self::Prepare(p) => (p.authority.table, p.authority.owner, p.authority.epoch),
         }
     }
     fn byte_limit(&self) -> usize {
         match self {
             Self::Acquire(_) => MAX_ACQUIRE_BYTES,
-            Self::Plan(_) => MAX_COMMAND_BYTES,
+            Self::Plan(_) => plan::MAX_PLAN_BYTES,
+            Self::Prepare(_) => prepared::MAX_PREPARE_BYTES,
         }
     }
     fn batch(self, layout: &Layout) -> Result<Batch> {
         let (table, owner, sequence) = self.identity();
         let limit = self.byte_limit();
         let envelope = Envelope {
-            version: 2,
+            version: 3,
             cluster: layout.cluster,
             layout: layout.fingerprint()?,
             group: control_group(layout),
@@ -105,6 +112,24 @@ pub struct Authority {
     pub grant_index: u64,
 }
 
+impl Authority {
+    fn validate(&self, layout: &Layout) -> Result<()> {
+        let a = self;
+        ensure!(
+            a.cluster == layout.cluster
+                && a.layout == layout.fingerprint()?
+                && a.group == control_group(layout)
+                && !a.owner.is_nil()
+                && a.epoch > 0
+                && a.epoch <= a.grant_index
+                && a.grant_index < u64::MAX,
+            "invalid publication authority scope"
+        );
+        ensure!(!a.table.is_nil(), "nil authority table");
+        Ok(())
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Table {
     pub destination: Destination,
@@ -124,12 +149,15 @@ pub enum Rejection {
     Authority,
     PendingPlan,
     SourcePosition,
+    PlanIdentity,
+    PreparedConflict,
 }
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Reply {
     Acquired(Authority),
     Planned { index: u64 },
     Rejected(Rejection),
+    Prepared { index: u64 },
 }
 
 #[derive(Serialize, Deserialize)]
@@ -163,7 +191,7 @@ pub(super) fn decode(bytes: &[u8]) -> Result<Envelope> {
         "publisher command too large"
     );
     let envelope: Envelope = codec(MAX_COMMAND_BYTES).deserialize(bytes)?;
-    ensure!(envelope.version == 2, "unsupported publisher command");
+    ensure!(envelope.version == 3, "unsupported publisher command");
     ensure!(
         bytes.len() <= envelope.command.byte_limit(),
         "publisher command type byte limit"
@@ -205,6 +233,7 @@ pub(super) fn validate_record(
             "unserviceable ingestion limit"
         ),
         Command::Plan(p) => p.validate(layout, limits)?,
+        Command::Prepare(p) => p.validate(layout, limits)?,
     }
     Ok(())
 }
@@ -217,6 +246,7 @@ impl Catalog {
     pub fn apply(&mut self, e: Envelope, index: u64, limits: &Limits) -> Reply {
         let a = match e.command {
             Command::Plan(p) => return self.plan(*p, index, limits),
+            Command::Prepare(request) => return self.prepare(*request, index),
             Command::Acquire(a) => *a,
         };
         let old = self.tables.get(&a.table);

crates/lance-context-wal/src/replication/publication/plan.rs

@@ -124,18 +124,8 @@ impl PlanRequest {
         Command::Plan(Box::new(self)).batch(layout)
     }
     pub(super) fn validate(&self, layout: &Layout, limits: &Limits) -> Result<()> {
-        let a = &self.authority;
-        ensure!(
-            a.cluster == layout.cluster
-                && a.layout == layout.fingerprint()?
-                && a.group == control_group(layout)
-                && !a.owner.is_nil()
-                && a.epoch > 0
-                && a.epoch <= a.grant_index
-                && a.grant_index < u64::MAX,
-            "invalid publication authority scope"
-        );
-        self.plan.validate(layout, a.table, limits)
+        self.authority.validate(layout)?;
+        self.plan.validate(layout, self.authority.table, limits)
     }
 }
 
@@ -152,13 +142,16 @@ pub struct Progress {
     pub minimum_read_version: u64,
     /// Cloning a table view shares the immutable metadata; it never clones bodies.
     pub pending: Option<Arc<PendingPlan>>,
+    /// Materialization result shares ownership independently of the selection.
+    pub prepared: Option<Arc<Prepared>>,
 }
 impl Default for Progress {
     fn default() -> Self {
         Self {
             checkpoint: BTreeMap::new(),
             minimum_read_version: 1,
             pending: None,
+            prepared: None,
         }
     }
 }
@@ -189,6 +182,21 @@ impl Progress {
                 }),
             "invalid publication progress"
         );
+        if let Some(prepared) = &self.prepared {
+            let pending = self
+                .pending
+                .as_ref()
+                .context("prepared result without plan")?;
+            prepared.descriptors.validate(limits.max_descriptor_bytes)?;
+            ensure!(
+                prepared.plan_index == pending.index
+                    && prepared.transaction == pending.plan.transaction
+                    && prepared.index > pending.index
+                    && prepared.index <= cut
+                    && prepared.index != table.authority.grant_index,
+                "invalid prepared checkpoint identity or index"
+            );
+        }
         if let Some(pending) = &self.pending {
             pending
                 .plan