Update violation fix

Cargo.toml

@@ -3,7 +3,7 @@ members = ["codegen", "examples", "performance_measurement", "performance_measur
 
 [package]
 name = "worktable"
-version = "0.7.0"
+version = "0.7.1"
 edition = "2024"
 authors = ["Handy-caT"]
 license = "MIT"
@@ -16,7 +16,7 @@ perf_measurements = ["dep:performance_measurement", "dep:performance_measurement
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-worktable_codegen = { path = "codegen", version = "0.7.0" }
+worktable_codegen = { path = "codegen", version = "0.7.1" }
 
 eyre = "0.6.12"
 derive_more = { version = "1.0.0", features = ["from", "error", "display", "into"] }
@@ -27,14 +27,14 @@ lockfree = { version = "0.5.1" }
 fastrand = "2.3.0"
 futures = "0.3.30"
 uuid = { version = "1.10.0", features = ["v4", "v7"] }
-data_bucket = "0.2.9"
+data_bucket = "0.2.10"
 # data_bucket = { git = "https://github.com/pathscale/DataBucket", branch = "page_cdc_correction", version = "0.2.7" }
-# data_bucket = { path = "../DataBucket", version = "0.2.7" }
+# data_bucket = { path = "../DataBucket", version = "0.2.9" }
 performance_measurement_codegen = { path = "performance_measurement/codegen", version = "0.1.0", optional = true }
 performance_measurement = { path = "performance_measurement", version = "0.1.0", optional = true }
 # indexset = { version = "0.12.3", features = ["concurrent", "cdc", "multimap"] }
-# indexset = { path = "../indexset", version = "0.12.3", features = ["concurrent", "cdc", "multimap"] }
-indexset = { package = "wt-indexset", version = "0.12.5", features = ["concurrent", "cdc", "multimap"] }
+# indexset = { package = "wt-indexset", path = "../indexset", version = "0.12.5", features = ["concurrent", "cdc", "multimap"] }
+indexset = { package = "wt-indexset", version = "0.12.6", features = ["concurrent", "cdc", "multimap"] }
 convert_case = "0.6.0"
 ordered-float = "5.0.0"
 parking_lot = "0.12.3"

codegen/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "worktable_codegen"
-version = "0.7.0"
+version = "0.7.1"
 edition = "2024"
 license = "MIT"
 description = "WorkTable codegeneration crate"

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

@@ -199,6 +199,7 @@ impl Generator {
                     indexes,
                     pk_gen: PrimaryKeyGeneratorState::from_state(self.data_info.inner.pk_gen_state),
                     lock_map: LockMap::default(),
+                    update_state: IndexMap::default(),
                     table_name: "",
                     pk_phantom: std::marker::PhantomData,
                 };

codegen/src/worktable/generator/index/cdc.rs

@@ -1,4 +1,4 @@
-use crate::name_generator::WorktableNameGenerator;
+use crate::name_generator::{is_float, WorktableNameGenerator};
 use crate::worktable::generator::queries::r#type::map_to_uppercase;
 use crate::worktable::generator::Generator;
 use convert_case::{Case, Casing};
@@ -17,15 +17,16 @@ impl Generator {
         let save_row_cdc = self.gen_save_row_cdc_index_fn();
         let reinsert_row_cdc = self.gen_reinsert_row_cdc_index_fn();
         let delete_row_cdc = self.gen_delete_row_cdc_index_fn();
-        let process_diff_cdc = self.gen_process_diff_cdc_index_fn();
+        let process_difference_insert_cdc = self.gen_process_difference_insert_cdc_index_fn();
+        let process_difference_remove_cdc = self.gen_process_difference_remove_cdc_index_fn();
 
         quote! {
             impl TableSecondaryIndexCdc<#row_type_ident, #available_types_ident, #events_ident, #available_index_ident> for #index_type_ident {
                 #reinsert_row_cdc
                 #save_row_cdc
                 #delete_row_cdc
-
-                #process_diff_cdc
+                #process_difference_insert_cdc
+                #process_difference_remove_cdc
             }
         }
     }
@@ -49,15 +50,15 @@ impl Generator {
                 let index_variant: TokenStream = camel_case_name.parse().unwrap();
 
                 quote! {
-                    let (exists, events) = self.#index_field_name.insert_cdc(row.#i.clone(), link);
-                    if let Some(link) = exists {
-                        self.#index_field_name.insert_cdc(row.#i, link);
+                    let #index_field_name = if let Some(events) = self.#index_field_name.insert_checked_cdc(row.#i.clone(), link) {
+                        events.into_iter().map(|ev| ev.into()).collect()
+                    } else {
                         return Err(IndexError::AlreadyExists {
                             at: #available_index_ident::#index_variant,
                             inserted_already: inserted_indexes.clone(),
                         });
-                    }
-                    let #index_field_name = events.into_iter().map(|ev| ev.into()).collect();
+                    };
+                    inserted_indexes.push(#available_index_ident::#index_variant);
                 }
             })
             .collect::<Vec<_>>();
@@ -86,33 +87,58 @@ impl Generator {
         let name_generator = WorktableNameGenerator::from_table_name(self.name.to_string());
         let row_type_ident = name_generator.get_row_type_ident();
         let events_ident = name_generator.get_space_secondary_index_events_ident();
+        let available_index_ident = name_generator.get_available_indexes_ident();
 
-        let reinsert_rows = self
+        let (insert_rows, remove_rows): (Vec<_>, Vec<_>) = self
             .columns
             .indexes
             .iter()
             .map(|(i, idx)| {
                 let index_field_name = &idx.name;
+                let camel_case_name = index_field_name
+                    .to_string()
+                    .from_case(Case::Snake)
+                    .to_case(Case::Pascal);
+                let index_variant: TokenStream = camel_case_name.parse().unwrap();
+
                 let remove = if idx.is_unique {
                     quote! {
-                        if row_new.#i != row_old.#i {
+                        if row_new.#i == row_old.#i {
+                            let events = self.#index_field_name.insert_cdc(row_new.#i.clone(), link_new).1;
+                            #index_field_name.extend(events.into_iter().map(|ev| ev.into()).collect::<Vec<_>>());
+                        } else {
                             let (_, events) = TableIndexCdc::remove_cdc(&self.#index_field_name, row_old.#i.clone(), link_old);
                             #index_field_name.extend(events.into_iter().map(|ev| ev.into()).collect::<Vec<_>>());
                         }
                     }
                 } else {
                     quote! {
+                        let events = self.#index_field_name.insert_cdc(row_new.#i.clone(), link_new).1;
+                        #index_field_name.extend(events.into_iter().map(|ev| ev.into()).collect::<Vec<_>>());
                         let (_, events) = TableIndexCdc::remove_cdc(&self.#index_field_name, row_old.#i.clone(), link_old);
                         #index_field_name.extend(events.into_iter().map(|ev| ev.into()).collect::<Vec<_>>());
                     }
                 };
-                quote! {
-                    let (_, events) = self.#index_field_name.insert_cdc(row_new.#i.clone(), link_new);
-                    let mut #index_field_name: Vec<_> = events.into_iter().map(|ev| ev.into()).collect();
-                    #remove
-                }
+                let insert = quote! {
+                    let mut #index_field_name = if row_new.#i != row_old.#i {
+                        let #index_field_name: Vec<_> = if let Some(events) = self.#index_field_name.insert_checked_cdc(row_new.#i.clone(), link_new) {
+                            events.into_iter().map(|ev| ev.into()).collect()
+                        } else {
+                            return Err(IndexError::AlreadyExists {
+                                at: #available_index_ident::#index_variant,
+                                inserted_already: inserted_indexes.clone(),
+                            });
+                        };
+                        inserted_indexes.push(#available_index_ident::#index_variant);
+
+                        #index_field_name
+                    } else {
+                        vec![]
+                    };
+                };
+                (insert, remove)
             })
-            .collect::<Vec<_>>();
+            .unzip();
         let idents = self
             .columns
             .indexes
@@ -121,8 +147,17 @@ impl Generator {
             .collect::<Vec<_>>();
 
         quote! {
-            fn reinsert_row_cdc(&self, row_old: #row_type_ident, link_old: Link, row_new: #row_type_ident, link_new: Link) -> eyre::Result<#events_ident> {
-                #(#reinsert_rows)*
+            fn reinsert_row_cdc(
+                &self,
+                row_old: #row_type_ident,
+                link_old: Link,
+                row_new: #row_type_ident,
+                link_new: Link
+            ) -> Result<#events_ident, IndexError<#available_index_ident>> {
+                let mut inserted_indexes: Vec<#available_index_ident> = vec![];
+
+                #(#insert_rows)*
+                #(#remove_rows)*
                 core::result::Result::Ok(
                     #events_ident {
                         #(#idents,)*
@@ -169,9 +204,10 @@ impl Generator {
         }
     }
 
-    fn gen_process_diff_cdc_index_fn(&self) -> TokenStream {
+    fn gen_process_difference_remove_cdc_index_fn(&self) -> TokenStream {
         let name_generator = WorktableNameGenerator::from_table_name(self.name.to_string());
         let avt_type_ident = name_generator.get_available_type_ident();
+        let available_index_ident = name_generator.get_available_indexes_ident();
         let events_ident = name_generator.get_space_secondary_index_events_ident();
 
         let process_difference_rows = self.columns.indexes.iter().map(|(i, idx)| {
@@ -182,10 +218,12 @@ impl Generator {
                 let type_str = t.to_string();
                 let variant_ident = Ident::new(&map_to_uppercase(&type_str), Span::mixed_site());
 
-                let (new_value_expr, old_value_expr) = if type_str == "String" {
-                    (quote! { new.to_string() }, quote! { old.to_string() })
+                let old_value_expr = if type_str == "String" {
+                    quote! { old.to_string() }
+                } else if is_float(type_str.as_str()) {
+                    quote! { OrderedFloat(*old) }
                 } else {
-                    (quote! { *new }, quote! { *old })
+                    quote! { *old }
                 };
 
                 quote! {
@@ -196,11 +234,79 @@ impl Generator {
                             let (_, evs) = TableIndexCdc::remove_cdc(&self.#index_field_name, key_old, link);
                             events.extend_from_slice(evs.as_ref());
                         }
+                        events
+                    } else {
+                        vec![]
+                    };
+                }
+            } else {
+                quote! {}
+            }
+        });
+        let idents = self
+            .columns
+            .indexes
+            .values()
+            .map(|idx| &idx.name)
+            .collect::<Vec<_>>();
 
+        quote! {
+            fn process_difference_remove_cdc(
+                &self,
+                link: Link,
+                difference: std::collections::HashMap<&str, Difference<#avt_type_ident>>
+            ) -> Result<#events_ident, IndexError<#available_index_ident>> {
+                #(#process_difference_rows)*
+                core::result::Result::Ok(
+                    #events_ident {
+                        #(#idents,)*
+                    }
+                )
+            }
+        }
+    }
+
+    fn gen_process_difference_insert_cdc_index_fn(&self) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_table_name(self.name.to_string());
+        let avt_type_ident = name_generator.get_available_type_ident();
+        let available_index_ident = name_generator.get_available_indexes_ident();
+        let events_ident = name_generator.get_space_secondary_index_events_ident();
+
+        let process_difference_insert_rows = self.columns.indexes.iter().map(|(i, idx)| {
+            let index_field_name = &idx.name;
+            let diff_key = Literal::string(i.to_string().as_str());
+
+            if let Some(t) = self.columns.columns_map.get(&idx.field) {
+                let type_str = t.to_string();
+                let variant_ident = Ident::new(&map_to_uppercase(&type_str), Span::mixed_site());
+                let camel_case_name = index_field_name
+                    .to_string()
+                    .from_case(Case::Snake)
+                    .to_case(Case::Pascal);
+                let index_variant: TokenStream = camel_case_name.parse().unwrap();
+
+                let new_value_expr = if type_str == "String" {
+                    quote! { new.to_string() }
+                } else if is_float(type_str.as_str()) {
+                    quote! { OrderedFloat(*new) }
+                } else {
+                    quote! { *new }
+                };
+
+                quote! {
+                    let #index_field_name = if let Some(diff) = difference.get(#diff_key) {
+                        let mut events = vec![];
                         if let #avt_type_ident::#variant_ident(new) = &diff.new {
                             let key_new = #new_value_expr;
-                            let (_, evs) = TableIndexCdc::insert_cdc(&self.#index_field_name, key_new, link);
-                            events.extend_from_slice(evs.as_ref());
+                            if let Some(evs) =  TableIndexCdc::insert_checked_cdc(&self.#index_field_name, key_new, link) {
+                                events.extend_from_slice(evs.as_ref());
+                            } else {
+                                return Err(IndexError::AlreadyExists {
+                                    at: #available_index_ident::#index_variant,
+                                    inserted_already: inserted_indexes.clone(),
+                                });
+                            }
+                            inserted_indexes.push(#available_index_ident::#index_variant);
                         }
                         events
                     } else {
@@ -219,12 +325,14 @@ impl Generator {
             .collect::<Vec<_>>();
 
         quote! {
-            fn process_difference_cdc(
+            fn process_difference_insert_cdc(
                 &self,
                 link: Link,
                 difference: std::collections::HashMap<&str, Difference<#avt_type_ident>>
-            ) -> core::result::Result<#events_ident, WorkTableError> {
-                #(#process_difference_rows)*
+            ) -> Result<#events_ident, IndexError<#available_index_ident>> {
+                let mut inserted_indexes: Vec<#available_index_ident> = vec![];
+
+                #(#process_difference_insert_rows)*
                 core::result::Result::Ok(
                     #events_ident {
                         #(#idents,)*

codegen/src/worktable/generator/index/mod.rs

@@ -164,7 +164,7 @@ impl Generator {
                 }
 
                 impl AvailableIndex for #avt_type_ident {
-                    fn to_string(&self) -> String {
+                    fn to_string_value(&self) -> String {
                         ToString::to_string(&self)
                     }
                 }