feat(wal): select immutable manifests through the catalog quorum

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

@@ -50,6 +50,7 @@ pub struct Limits {
     pub max_active_plans: u64,
     pub max_plan_source_bytes: u64,
     pub max_descriptor_bytes: u64,
+    pub max_manifest_records: u64,
     pub max_record_bytes: u64,
     pub max_records: u64,
     pub max_producers: u64,
@@ -64,6 +65,7 @@ impl Default for Limits {
             max_active_plans: 128,
             max_plan_source_bytes: 64 * 1024 * 1024,
             max_descriptor_bytes: 256 * 1024,
+            max_manifest_records: 4096,
             max_record_bytes: 8 * 1024 * 1024,
             max_records: 100_000,
             max_producers: 16_384,
@@ -81,6 +83,7 @@ impl Limits {
                 && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes)
                 && (1..=publication::MAX_DESCRIPTOR_BYTES as u64)
                     .contains(&self.max_descriptor_bytes)
+                && (1..=16_384).contains(&self.max_manifest_records)
                 && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
                 && (1..=500_000).contains(&self.max_records)
                 && (1..=100_000).contains(&self.max_producers)
@@ -90,7 +93,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v5",
+            "lance-context/replicated-append/v6",
             self,
         ))?)
     }
@@ -414,6 +417,28 @@ impl Node {
         self.propose(publication::control_group(&self.layout), request, batch)
     }
 
+    /// Adopt an immutable starting manifest after the caller has verified it
+    /// and excluded unmanaged writers/cleaners for this dataset incarnation.
+    pub fn bootstrap_manifest(
+        &mut self,
+        request: u64,
+        bootstrap: publication::BootstrapManifest,
+    ) -> Result<()> {
+        let batch = bootstrap.batch(&self.layout)?;
+        self.propose(publication::control_group(&self.layout), request, batch)
+    }
+
+    /// Select the unique manifest path for the exact immutable prepared plan.
+    /// An enqueue is not publication; wait for durable quorum completion.
+    pub fn reserve_manifest(
+        &mut self,
+        request: u64,
+        reservation: publication::ReserveManifest,
+    ) -> Result<()> {
+        let batch = reservation.batch(&self.layout)?;
+        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 = 5;
-const MAGIC: &[u8; 8] = b"LCWALCP5";
+const FORMAT: u32 = 6;
+const MAGIC: &[u8; 8] = b"LCWALCP6";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -87,6 +87,8 @@ fn metadata_limit(limits: &Limits) -> 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)
+        + limits.max_manifest_records * 2048
+        + limits.max_tables * 256
         + 1024
 }
 

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

@@ -7,8 +7,13 @@ use super::*;
 use crate::{ingest_contract::IngestContract, record::RecordKind};
 use bincode::Options;
 use bytes::Bytes;
+mod manifest;
 mod plan;
 mod prepared;
+pub use manifest::{
+    BootstrapManifest, ManifestIndex, ManifestObject, ManifestRecord, ReserveManifest,
+    MAX_MANIFEST_VERSIONS,
+};
 pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS};
 pub use prepared::{Descriptors, PrepareRequest, Prepared, MAX_DESCRIPTOR_BYTES};
 #[cfg(test)]
@@ -17,8 +22,10 @@ mod tests;
 pub(super) const MAX_ACQUIRE_BYTES: usize = 8192;
 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 + prepared::MAX_PREPARE_BYTES;
+pub(super) const MAX_TABLE_BYTES: usize = MAX_BASE_TABLE_BYTES
+    + plan::MAX_PLAN_BYTES
+    + prepared::MAX_PREPARE_BYTES
+    + manifest::MAX_MANIFEST_INDEX_BYTES;
 fn codec(limit: usize) -> impl Options {
     bincode::DefaultOptions::new()
         .with_fixint_encoding()
@@ -67,27 +74,34 @@ enum Command {
     Acquire(Box<Acquire>),
     Plan(Box<PlanRequest>),
     Prepare(Box<PrepareRequest>),
+    BootstrapManifest(Box<BootstrapManifest>),
+    ReserveManifest(Box<ReserveManifest>),
 }
 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),
+            Self::BootstrapManifest(p) => (p.authority.table, p.authority.owner, p.authority.epoch),
+            Self::ReserveManifest(p) => (p.authority.table, p.authority.owner, p.authority.epoch),
         }
     }
     fn byte_limit(&self) -> usize {
         match self {
             Self::Acquire(_) => MAX_ACQUIRE_BYTES,
             Self::Plan(_) => plan::MAX_PLAN_BYTES,
             Self::Prepare(_) => prepared::MAX_PREPARE_BYTES,
+            Self::BootstrapManifest(_) | Self::ReserveManifest(_) => {
+                manifest::MAX_MANIFEST_COMMAND_BYTES
+            }
         }
     }
     fn batch(self, layout: &Layout) -> Result<Batch> {
         let (table, owner, sequence) = self.identity();
         let limit = self.byte_limit();
         let envelope = Envelope {
-            version: 3,
+            version: 4,
             cluster: layout.cluster,
             layout: layout.fingerprint()?,
             group: control_group(layout),
@@ -132,6 +146,7 @@ impl Authority {
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Table {
+    pub manifests: Option<Arc<ManifestIndex>>,
     pub destination: Destination,
     pub ingest: IngestContract,
     pub authority: Authority,
@@ -151,13 +166,17 @@ pub enum Rejection {
     SourcePosition,
     PlanIdentity,
     PreparedConflict,
+    ManifestIdentity,
+    ManifestPredecessor,
 }
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Reply {
     Acquired(Authority),
     Planned { index: u64 },
     Rejected(Rejection),
     Prepared { index: u64 },
+    ManifestReserved(Arc<ManifestRecord>),
+    ManifestTaken(Arc<ManifestRecord>),
 }
 
 #[derive(Serialize, Deserialize)]
@@ -191,7 +210,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 == 3, "unsupported publisher command");
+    ensure!(envelope.version == 4, "unsupported publisher command");
     ensure!(
         bytes.len() <= envelope.command.byte_limit(),
         "publisher command type byte limit"
@@ -234,6 +253,8 @@ pub(super) fn validate_record(
         ),
         Command::Plan(p) => p.validate(layout, limits)?,
         Command::Prepare(p) => p.validate(layout, limits)?,
+        Command::BootstrapManifest(p) => p.validate(layout)?,
+        Command::ReserveManifest(p) => p.validate(layout)?,
     }
     Ok(())
 }
@@ -247,6 +268,12 @@ impl Catalog {
         let a = match e.command {
             Command::Plan(p) => return self.plan(*p, index, limits),
             Command::Prepare(request) => return self.prepare(*request, index),
+            Command::BootstrapManifest(request) => {
+                return self.bootstrap_manifest(*request, index, limits)
+            }
+            Command::ReserveManifest(request) => {
+                return self.reserve_manifest(*request, index, limits)
+            }
             Command::Acquire(a) => *a,
         };
         let old = self.tables.get(&a.table);
@@ -286,9 +313,11 @@ impl Catalog {
             grant_index: index,
         };
         let progress = old.map(|table| table.progress.clone()).unwrap_or_default();
+        let manifests = old.and_then(|table| table.manifests.clone());
         self.tables.insert(
             a.table,
             Table {
+                manifests,
                 destination: a.destination,
                 ingest: a.ingest,
                 authority: authority.clone(),
@@ -299,6 +328,10 @@ impl Catalog {
         Reply::Acquired(authority)
     }
     pub fn validate(&self, layout: &Layout, group: u64, cut: u64, limits: &Limits) -> Result<()> {
+        ensure!(
+            self.manifest_records() as u64 <= limits.max_manifest_records,
+            "manifest record capacity"
+        );
         ensure!(
             self.tables.len() as u64 <= limits.max_tables,
             "publisher catalog exceeds limit"
@@ -320,6 +353,9 @@ impl Catalog {
         for (id, table) in &self.tables {
             validate_binding(&table.destination, &table.ingest)?;
             table.progress.validate(layout, table, cut, limits)?;
+            if let Some(manifests) = &table.manifests {
+                manifests.validate(table, cut)?;
+            }
             let a = &table.authority;
             ensure!(
                 !id.is_nil()