feat: move WTI structural persistence off mutation path

Cargo.toml

@@ -18,6 +18,10 @@ categories = ["database-implementations", "data-structures", "caching"]
 default = ["wti-predictable-search"]
 perf_measurements = ["dep:performance_measurement", "dep:performance_measurement_codegen"]
 s3-support = ["dep:rusty-s3", "dep:url", "dep:reqwest", "dep:walkdir", "worktable_codegen/s3-support"]
+# Moves unique WorkTablesIndex structural CDC work out of the table mutation
+# path and into the background persistence worker. The persisted page format
+# is unchanged, so stores remain readable with or without this feature.
+logical-index-persistence = ["worktable_codegen/logical-index-persistence"]
 wti-hybrid-search = ["indexset/wt-slice-binary-search"]
 wti-predictable-search = ["indexset/custom-binary-search"]
 wti-std-search = ["indexset/std-binary-search"]

codegen/Cargo.toml

@@ -8,6 +8,7 @@ repository = "https://github.com/pathscale/WorkTable"
 
 [features]
 s3-support = []
+logical-index-persistence = []
 # Compatibility no-op retained for downstream manifests.
 versioned-row-publication = []
 

codegen/src/generators/index_backend.rs

@@ -39,7 +39,16 @@ pub(crate) fn persistent_unique_index_type(
     value: &TokenStream,
     worktables_node: Option<TokenStream>,
 ) -> syn::Result<TokenStream> {
+    // This intentionally evaluates the proc-macro crate's feature. WorkTable's
+    // public feature forwards to worktable_codegen in Cargo.toml, so the
+    // runtime types and emitted types are selected together. Emitting a cfg in
+    // the expansion would instead test the consuming package's unrelated
+    // feature namespace, which may rename or omit the dependency feature.
     match backend {
+        IndexBackend::WorktablesIndex if cfg!(feature = "logical-index-persistence") => Ok(match worktables_node {
+            Some(node) => quote! { PersistentWtiIndex<#key, #value, #node> },
+            None => quote! { PersistentWtiIndex<#key, #value> },
+        }),
         IndexBackend::Congee => Ok(quote! { PersistentCongeeIndex<#key, #value> }),
         IndexBackend::Arctic => Ok(quote! { PersistentArcticIndex<#key, #value> }),
         _ => unique_index_type(backend, key, value, worktables_node),

codegen/src/generators/persist/index/mod.rs

@@ -109,11 +109,16 @@ impl PersistGenerator {
                 let res = if idx.is_unique {
                     match idx.backend {
                         crate::common::model::IndexBackend::WorktablesIndex => {
+                            let map = if cfg!(feature = "logical-index-persistence") {
+                                quote! { PersistentWtiIndex }
+                            } else {
+                                quote! { IndexMap }
+                            };
                             if is_unsized(&t.to_string()) {
-                                quote! { #i: IndexMap::with_maximum_node_size(#const_name), }
+                                quote! { #i: #map::with_maximum_node_size(#const_name), }
                             } else {
                                 quote! {
-                                    #i: IndexMap::with_maximum_node_size(
+                                    #i: #map::with_maximum_node_size(
                                         get_index_page_size_from_data_length::<#t>(#const_name)
                                     ),
                                 }

codegen/src/generators/persist/table/impls.rs

@@ -237,10 +237,15 @@ impl PersistGenerator {
             })
             .collect::<Vec<_>>();
         let pk_types_unsized = is_unsized_vec(pk_types);
+        let wti_map = if cfg!(feature = "logical-index-persistence") {
+            quote! { PersistentWtiIndex }
+        } else {
+            quote! { IndexMap }
+        };
         let index_setup = if pk_types_unsized {
             quote! {
                 inner.primary_index = std::sync::Arc::new(PrimaryIndex {
-                    pk_map: IndexMap::<#pk_type, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name),
+                    pk_map: #wti_map::<#pk_type, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name),
                     reverse_pk_map: IndexMap::new(),
                 });
             }
@@ -249,7 +254,7 @@ impl PersistGenerator {
                 crate::common::model::IndexBackend::WorktablesIndex => quote! {
                     let size = get_index_page_size_from_data_length::<#pk_type>(#const_name);
                     inner.primary_index = std::sync::Arc::new(PrimaryIndex {
-                        pk_map: IndexMap::<_, OffsetEqLink<#const_name>>::with_maximum_node_size(size),
+                        pk_map: #wti_map::<_, OffsetEqLink<#const_name>>::with_maximum_node_size(size),
                         reverse_pk_map: IndexMap::new(),
                     });
                 },

codegen/src/generators/persist/table/mod.rs

@@ -89,6 +89,14 @@ impl PersistGenerator {
                 #[derive(Debug, PersistTable)]
                 #[table(pk_unsized, pk_upstream)]
             },
+            (true, crate::common::model::IndexBackend::WorktablesIndex)
+                if cfg!(feature = "logical-index-persistence") =>
+            {
+                quote! {
+                    #[derive(Debug, PersistTable)]
+                    #[table(pk_unsized, pk_wti_logical)]
+                }
+            }
             (true, _) => quote! {
                 #[derive(Debug, PersistTable)]
                 #[table(pk_unsized)]
@@ -105,6 +113,14 @@ impl PersistGenerator {
                 #[derive(Debug, PersistTable)]
                 #[table(pk_congee)]
             },
+            (false, crate::common::model::IndexBackend::WorktablesIndex)
+                if cfg!(feature = "logical-index-persistence") =>
+            {
+                quote! {
+                    #[derive(Debug, PersistTable)]
+                    #[table(pk_wti_logical)]
+                }
+            }
             (false, crate::common::model::IndexBackend::WorktablesIndex) => quote! {
                 #[derive(Debug, PersistTable)]
             },

codegen/src/persist_index/generator.rs

@@ -24,6 +24,7 @@ pub(super) struct IndexLayout {
     is_unique: bool,
     uses_upstream: bool,
     pub(super) art_backend: Option<ArtBackend>,
+    pub(super) logical_wti: bool,
 }
 
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
@@ -46,12 +47,13 @@ pub(super) fn index_layout(field: &Field) -> syn::Result<IndexLayout> {
         .ok_or_else(|| syn::Error::new_spanned(&field.ty, "index type path cannot be empty"))?
         .ident
         .clone();
-    let (is_unique, uses_upstream, art_backend) = match type_ident.to_string().as_str() {
-        "IndexMap" | "TreeIndex" => (true, false, None),
-        "UpstreamIndexMap" => (true, true, None),
-        "IndexMultiMap" | "TreeMultiIndex" => (false, false, None),
-        "PersistentArcticIndex" => (true, false, Some(ArtBackend::Arctic)),
-        "PersistentCongeeIndex" => (true, false, Some(ArtBackend::Congee)),
+    let (is_unique, uses_upstream, art_backend, logical_wti) = match type_ident.to_string().as_str() {
+        "IndexMap" | "TreeIndex" => (true, false, None, false),
+        "PersistentWtiIndex" => (true, false, None, true),
+        "UpstreamIndexMap" => (true, true, None, false),
+        "IndexMultiMap" | "TreeMultiIndex" => (false, false, None, false),
+        "PersistentArcticIndex" => (true, false, Some(ArtBackend::Arctic), false),
+        "PersistentCongeeIndex" => (true, false, Some(ArtBackend::Congee), false),
         _ => {
             return Err(syn::Error::new_spanned(
                 &field.ty,
@@ -64,6 +66,7 @@ pub(super) fn index_layout(field: &Field) -> syn::Result<IndexLayout> {
         is_unique,
         uses_upstream,
         art_backend,
+        logical_wti,
     })
 }
 

codegen/src/persist_index/space/index.rs

@@ -25,6 +25,12 @@ impl Generator {
                     Some(ArtBackend::Congee) => quote! {
                         #i: SpaceCongeeIndex<#t, { #inner_const_name as u32}>,
                     },
+                    None if layout.logical_wti && is_unsized(&t.to_string()) => quote! {
+                        #i: SpaceLogicalIndexUnsized<#t, { #inner_const_name as u32}>,
+                    },
+                    None if layout.logical_wti => quote! {
+                        #i: SpaceLogicalIndex<#t, { #inner_const_name as u32}>,
+                    },
                     None if is_unsized(&t.to_string()) => quote! {
                         #i: SpaceIndexUnsized<#t, { #inner_const_name as u32}>,
                     },
@@ -79,6 +85,12 @@ impl Generator {
                     Some(ArtBackend::Congee) => quote! {
                         #i: SpaceCongeeIndex::secondary_from_table_files_path(path, #literal_name, version).await?,
                     },
+                    None if layout.logical_wti && is_unsized(&t.to_string()) => quote! {
+                        #i: SpaceLogicalIndexUnsized::secondary_from_table_files_path(path, #literal_name, version).await?,
+                    },
+                    None if layout.logical_wti => quote! {
+                        #i: SpaceLogicalIndex::secondary_from_table_files_path(path, #literal_name, version).await?,
+                    },
                     None if is_unsized(&t.to_string()) => quote! {
                         #i: SpaceIndexUnsized::secondary_from_table_files_path(path, #literal_name, version).await?,
                     },

codegen/src/persist_table/generator/mod.rs

@@ -14,6 +14,7 @@ pub struct PersistTableAttributes {
     pub pk_upstream: bool,
     pub pk_arctic: bool,
     pub pk_congee: bool,
+    pub pk_wti_logical: bool,
     pub row_schema: Vec<(String, String)>,
     pub primary_key_fields: Vec<String>,
     pub secondary_index_types: Vec<(String, String)>,

codegen/src/persist_table/generator/space.rs

@@ -36,10 +36,18 @@ impl Generator {
         let space_secondary_indexes = name_generator.get_space_secondary_index_ident();
         let space_secondary_indexes_events = name_generator.get_space_secondary_index_events_ident();
         let avt_index_ident = name_generator.get_available_indexes_ident();
-        let space_index_type = if self.attributes.pk_unsized {
+        let space_index_type = if self.attributes.pk_unsized && self.attributes.pk_wti_logical {
+            quote! {
+                SpaceLogicalIndexUnsized<#primary_key_type, { #inner_const_name as u32 }>,
+            }
+        } else if self.attributes.pk_unsized {
             quote! {
                 SpaceIndexUnsized<#primary_key_type, { #inner_const_name as u32 }>,
             }
+        } else if self.attributes.pk_wti_logical {
+            quote! {
+                SpaceLogicalIndex<#primary_key_type, { #inner_const_name as u32 }>,
+            }
         } else if self.attributes.pk_arctic {
             quote! {
                 SpaceArcticIndex<#primary_key_type, { #inner_const_name as u32 }>,

codegen/src/persist_table/generator/space_file/mod.rs

@@ -143,8 +143,13 @@ impl Generator {
 
         let primary_index_init = if self.attributes.pk_unsized {
             let pk_ident = &self.pk_ident;
+            let map_type = if self.attributes.pk_wti_logical {
+                quote! { PersistentWtiIndex }
+            } else {
+                quote! { IndexMap }
+            };
             quote! {
-                let pk_map = IndexMap::<#pk_ident, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name);
+                let pk_map = #map_type::<#pk_ident, OffsetEqLink<#const_name>, UnsizedNode<_>>::with_maximum_node_size(#const_name);
                 for page in self.primary_index.1 {
                     let node = page
                         .inner
@@ -175,7 +180,9 @@ impl Generator {
                 let primary_index = PrimaryIndex { pk_map, reverse_pk_map };
             }
         } else {
-            let map_type = if self.pk_upstream {
+            let map_type = if self.attributes.pk_wti_logical {
+                quote! { PersistentWtiIndex }
+            } else if self.pk_upstream {
                 quote! { UpstreamIndexMap }
             } else {
                 quote! { IndexMap }

codegen/src/persist_table/parser.rs

@@ -32,6 +32,7 @@ impl Parser {
             pk_upstream: false,
             pk_arctic: false,
             pk_congee: false,
+            pk_wti_logical: false,
             row_schema: vec![],
             primary_key_fields: vec![],
             secondary_index_types: vec![],
@@ -60,6 +61,10 @@ impl Parser {
                         res.pk_congee = true;
                         return Ok(());
                     }
+                    if meta.path.is_ident("pk_wti_logical") {
+                        res.pk_wti_logical = true;
+                        return Ok(());
+                    }
                     if meta.path.is_ident("row_schema") {
                         meta.parse_nested_meta(|field| {
                             let name = field

codegen/src/worktable/mod.rs

@@ -206,9 +206,42 @@ mod tests {
             indexes: {
                 value_idx: value unique,
             },
-        });
+        })
+        .unwrap()
+        .to_string();
+
+        if cfg!(feature = "logical-index-persistence") {
+            assert!(output.contains("PersistentWtiIndex"));
+        } else {
+            assert!(output.contains("IndexMap"));
+            assert!(!output.contains("PersistentWtiIndex"));
+        }
+    }
 
-        assert!(output.is_ok());
+    #[cfg(feature = "logical-index-persistence")]
+    #[test]
+    fn logical_persistence_wraps_only_default_wti_backends() {
+        let output = expand(quote! {
+            name: LogicalDefaultBackend,
+            persist: true,
+            columns: {
+                id: u64 primary_key autoincrement,
+                wti_value: u64,
+                congee_value: u64,
+                arctic_value: u64,
+            },
+            indexes: {
+                wti_idx: wti_value unique,
+                congee_idx: congee_value unique using congee,
+                arctic_idx: arctic_value unique using arctic,
+            },
+        })
+        .unwrap()
+        .to_string();
+
+        assert!(output.contains("PersistentWtiIndex"));
+        assert!(output.contains("PersistentCongeeIndex"));
+        assert!(output.contains("PersistentArcticIndex"));
     }
 
     #[test]