Improved OCEL Import and Memory Usage + Misc Bug Fixes

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## Unreleased
+
+- New `ReadableOCEL` and `AppendableOCEL` traits; OCEL exporters (CSV, XML, SQL, JSON) and the JSON importer are generic over them
+- `SlimLinkedOCEL` implements `AppendableOCEL`, with auto-declare types, auto-grow attributes, and value coercion via new `OCELAttributeValue::try_coerce_to`; misordered streams are buffered and resolved on `finalize`
+- `import_ocel_json_into` and `import_ocel_xml_into` stream JSON / XML directly into any `AppendableOCEL` (e.g., `SlimLinkedOCEL`) without materializing an `OCEL` first; `SlimLinkedOCEL::import_from_*` uses the streaming paths automatically
+- `ReadableOCEL::iter_events_of_type` / `iter_objects_of_type` (default filters; `SlimLinkedOCEL` overrides via per-type indices, avoiding a full scan per type in SQL export)
+- New `OCELAttributeType::as_type_str` returns `&'static str` (cheaper than `to_type_string`)
+- `EventIndex` / `ObjectIndex` are now `u32`-backed; `into_inner` returns `u32` (**Breaking**)
+- `SlimLinkedOCEL`, `SlimOCELEvent`, `SlimOCELObject` no longer derive `Deserialize` (still serialize); construct via `Importable` or `AppendableOCEL` (**Breaking**)
+- `SlimOCELEvent::relationships` and `SlimOCELObject::relationships` are now `Vec<(QualifierIdx, ObjectIndex)>` (was `Vec<(String, ObjectIndex)>`); resolve qualifier strings via `SlimLinkedOCEL::qualifier_str` (**Breaking**)
+- CSV exporter streams rows instead of buffering all rows; tracks `(time, object_id)` pairs in a `HashSet` to skip redundant object-attribute rows
+- `SlimLinkedOCEL::from_ocel` now goes through `AppendableOCEL`, so it shares attribute coercion, schema-grow, and duplicate-id detection with the streaming import paths. Behavior changes: events/objects with attribute names not in the declared type schema grow the schema (before they were silently dropped); duplicate event/object ids are skipped with a warning (before they were kept but unreachable); references to undeclared types auto-create the type (before they were dropped with warning)
+- `try_json_to_ocel` returning `Result<OCEL, serde_json::Error>` added alongside the existing panicking `json_to_ocel`
+- `SlimLinkedOCEL::get_o2o_rev_obs_of_obtype` / `get_e2o_rev_evs_of_evtype`: qualifier-optional reverse-relation getters filtered by type
+- New public object-centric analysis functions (also exposed as bindings): per-event sojourn and synchronization times with optional `top_k` (`analysis::object_centric::oc_performance`), E2O `(event_type, object_type)` counts and `source -> target` conversion rate (`analysis::object_centric::oc_statistics`), and per-object-type directly-follows graph and activity-trace variants (`discovery::object_centric::dfg` / `variants`)
+- Fix SQL export/import of floats and timestamps: floats are written as `DOUBLE PRECISION` (full f64 precision) and timestamps as naive UTC (avoids a double-applied timezone offset); import maps `DOUBLE` / `DOUBLE PRECISION` columns back to float, so round-trips no longer drop float attributes
+- New direct dependency on `hashbrown` for the slim per-id hash tables
+
 ## 0.5.6
 
 - Translate a `ProcessTree` into a `PetriNet`:

macros_process_mining/src/lib.rs

@@ -325,16 +325,16 @@ pub fn register_binding(args: TokenStream, item: TokenStream) -> TokenStream {
     let serialization_logic = if attrs.debug_output {
         quote! {
             let final_result = format!("{:?}", result);
-            serde_json::to_value(final_result).map_err(|e| e.to_string())
+            serde_json::to_vec(&final_result).map_err(|e| e.to_string())
         }
     } else if attrs.stringify_error {
         quote! {
             let ok_result = result.map_err(|e| e.to_string())?;
-            serde_json::to_value(ok_result).map_err(|e| e.to_string())
+            serde_json::to_vec(&ok_result).map_err(|e| e.to_string())
         }
     } else {
         quote! {
-            serde_json::to_value(result).map_err(|e| e.to_string())
+            serde_json::to_vec(&result).map_err(|e| e.to_string())
         }
     };
 
@@ -398,7 +398,7 @@ pub fn register_binding(args: TokenStream, item: TokenStream) -> TokenStream {
             quote! {
                 let id = format!("res_{}", uuid::Uuid::new_v4());
                 __state_guard.insert(id.clone(), crate::bindings::RegistryItem::#variant_ident(result));
-                serde_json::to_value(id).map_err(|e| e.to_string())
+                serde_json::to_vec(&id).map_err(|e| e.to_string())
             }
         } else {
             serialization_logic.clone()
@@ -421,7 +421,7 @@ pub fn register_binding(args: TokenStream, item: TokenStream) -> TokenStream {
             };
             let id = format!("res_{}", uuid::Uuid::new_v4());
             state_lock.add(&id, crate::bindings::RegistryItem::#variant_ident(result));
-            serde_json::to_value(id).map_err(|e| e.to_string())
+            serde_json::to_vec(&id).map_err(|e| e.to_string())
         }
     } else {
         quote! {
@@ -468,7 +468,7 @@ pub fn register_binding(args: TokenStream, item: TokenStream) -> TokenStream {
             use serde_json::Value;
             use std::sync::RwLock;
 
-            fn #wrapper_name(args: &Value, state_lock: &AppState) -> Result<Value, String> {
+            fn #wrapper_name(args: &Value, state_lock: &AppState) -> Result<Vec<u8>, String> {
                 let arg_map = args.as_object().ok_or("Args must be JSON object")?;
                 #execution_block
             }

process_mining/Cargo.toml

@@ -19,6 +19,7 @@ chrono = { version = "0.4.40", features = ["serde"] }
 duckdb = { version = "1.2.1", optional = true, features = ["chrono"]}
 flate2 = "1.1.1"
 graphviz-rust = { version = "0.9.3", optional = true }
+hashbrown = "0.15"
 itertools = { version = "0.14.0" }
 kuzu = {version = "=0.11.2", optional = true}
 nalgebra = { version = "0.33.2", optional = true }

process_mining/examples/ocel_stats.rs

@@ -1,7 +1,11 @@
-use process_mining::{Importable, OCEL};
+use process_mining::core::event_data::object_centric::linked_ocel::{
+    LinkedOCELAccess, SlimLinkedOCEL,
+};
+use process_mining::Importable;
 use std::env;
 use std::error::Error;
 use std::path::PathBuf;
+use std::time::Instant;
 
 fn main() -> Result<(), Box<dyn Error>> {
     let args: Vec<String> = env::args().collect();
@@ -12,25 +16,32 @@ fn main() -> Result<(), Box<dyn Error>> {
 
     let path = PathBuf::from(&args[1]);
     println!("Importing OCEL from {:?}", path);
+    let now = Instant::now();
+    let ocel = SlimLinkedOCEL::import_from_path(&path)?;
+    println!("Successfully imported OCEL in {:?}.", now.elapsed());
+    println!("Number of events: {}", ocel.get_all_evs().count());
+    println!("Number of objects: {}", ocel.get_all_obs().count());
 
-    let ocel = OCEL::import_from_path(&path)?;
-    println!("Successfully imported OCEL.");
-    println!("Number of events: {}", ocel.events.len());
-    println!("Number of objects: {}", ocel.objects.len());
-
-    println!(
-        "Event Types: {:?}",
-        ocel.event_types
-            .iter()
-            .map(|et| &et.name)
-            .collect::<Vec<_>>()
-    );
+    println!("Event Types: {:?}", ocel.get_ev_types().collect::<Vec<_>>());
     println!(
         "Object Types: {:?}",
-        ocel.object_types
-            .iter()
-            .map(|ot| &ot.name)
-            .collect::<Vec<_>>()
+        ocel.get_ob_types().collect::<Vec<_>>()
     );
+
+    let preview_n = 10;
+    println!("First {} events:", preview_n);
+    for ev in ocel.get_all_evs().take(preview_n) {
+        let ev_type = ocel.get_ev_type_of(ev);
+        let timestamp = ocel.get_ev_time(ev);
+        println!(
+            "Event {:?}: Type: {}, Timestamp: {}",
+            ev, ev_type, timestamp
+        );
+        let attrs = ocel.get_ev_attrs(ev);
+        for attr in attrs {
+            let val = ocel.get_ev_attr_val(ev, attr);
+            println!("  Attribute: {} = {:?}", attr, val);
+        }
+    }
     Ok(())
 }

process_mining/src/analysis/object_centric/mod.rs

@@ -1,3 +1,5 @@
 //! Object-centric Process Analysis
 
 pub mod object_attribute_changes;
+pub mod oc_performance;
+pub mod oc_statistics;

process_mining/src/analysis/object_centric/oc_performance.rs

@@ -0,0 +1,152 @@
+//! Object-centric performance analysis over [`SlimLinkedOCEL`]: per-event sojourn and
+//! synchronization times.
+
+use macros_process_mining::register_binding;
+use rayon::prelude::*;
+
+use crate::core::event_data::object_centric::linked_ocel::{
+    slim_linked_ocel::{EventIndex, ObjectIndex},
+    LinkedOCELAccess, SlimLinkedOCEL,
+};
+
+/// Each object's reverse-E2O events in `(time, id)` order.
+///
+/// Events are sorted per call, so this makes no assumption about global event ordering
+fn sorted_events_per_object(ocel: &SlimLinkedOCEL) -> Vec<Vec<EventIndex>> {
+    (0..ocel.get_num_obs() as u32)
+        .into_par_iter()
+        .map(|i| {
+            let mut evs: Vec<EventIndex> =
+                ObjectIndex::from(i).get_e2o_rev(ocel).copied().collect();
+            evs.sort_by(|a, b| {
+                a.get_time(ocel)
+                    .cmp(b.get_time(ocel))
+                    .then_with(|| ocel.get_ev_id(a).cmp(ocel.get_ev_id(b)))
+            });
+            evs
+        })
+        .collect()
+}
+
+/// The `(time, id)`-immediate predecessor of `e` on object `o`, using the
+/// per-object sorted lists from [`sorted_events_per_object`].
+/// `None` if `e` is the first event on `o`.
+#[inline]
+fn df_predecessor(
+    sorted: &[Vec<EventIndex>],
+    ocel: &SlimLinkedOCEL,
+    e: EventIndex,
+    o: ObjectIndex,
+) -> Option<EventIndex> {
+    let evs = &sorted[o.into_inner() as usize];
+    let key = (e.get_time(ocel), ocel.get_ev_id(&e));
+    let pos = evs
+        .binary_search_by(|x| (x.get_time(ocel), ocel.get_ev_id(x)).cmp(&key))
+        .ok()?;
+    pos.checked_sub(1).map(|p| evs[p])
+}
+
+/// Per-event synchronization time and the delaying object.
+///
+/// For each event with at least one directly-follows predecessor, the synchronization time is
+/// `max_predecessor_time - min_predecessor_time` in integer microseconds (the span between its
+/// earliest and latest directly-preceding event). The delaying object is the object linking the
+/// latest predecessor (ties broken by ascending object id).
+/// Returns one row `(event_id, sync_us, delaying_object_id)` per qualifying event.
+///
+/// `top_k`: if `Some(k)`, return only the `k` rows with the largest `sync_us`, ties broken by
+/// ascending event id, sorted descending. `None` returns every qualifying event.
+#[register_binding]
+pub fn locel_oc_perf_sync_per_event(
+    ocel: &SlimLinkedOCEL,
+    #[bind(default)] top_k: Option<usize>,
+) -> Vec<(String, i64, String)> {
+    let sorted = sorted_events_per_object(ocel);
+    let mut rows: Vec<(EventIndex, i64, ObjectIndex)> = (0..ocel.get_num_evs() as u32)
+        .into_par_iter()
+        .filter_map(|i| {
+            let e = EventIndex::from(i);
+            let mut min_us = i64::MAX;
+            // (latest predecessor time, its object) = the delaying edge.
+            let mut delaying: Option<(i64, ObjectIndex)> = None;
+            for &o in e.get_e2o(ocel) {
+                if let Some(p) = df_predecessor(&sorted, ocel, e, o) {
+                    let t = p.get_time(ocel).timestamp_micros();
+                    min_us = min_us.min(t);
+                    let keep = match delaying {
+                        Some((bt, bo)) => {
+                            bt > t || (bt == t && ocel.get_ob_id(&bo) <= ocel.get_ob_id(&o))
+                        }
+                        None => false,
+                    };
+                    if !keep {
+                        delaying = Some((t, o));
+                    }
+                }
+            }
+            delaying.map(|(max_us, o)| (e, max_us - min_us, o))
+        })
+        .collect();
+    if let Some(k) = top_k {
+        let cmp = |a: &(EventIndex, i64, ObjectIndex), b: &(EventIndex, i64, ObjectIndex)| {
+            b.1.cmp(&a.1)
+                .then_with(|| ocel.get_ev_id(&a.0).cmp(ocel.get_ev_id(&b.0)))
+        };
+        if k < rows.len() {
+            rows.select_nth_unstable_by(k, cmp);
+            rows.truncate(k);
+        }
+        rows.sort_unstable_by(cmp);
+    }
+    rows.into_iter()
+        .map(|(e, max_minus_min, o)| {
+            (
+                ocel.get_ev_id(&e).to_string(),
+                max_minus_min,
+                ocel.get_ob_id(&o).to_string(),
+            )
+        })
+        .collect()
+}
+
+/// Per-event sojourn time.
+///
+/// For each event with at least one directly-follows predecessor, the sojourn time is
+/// `event_time - latest_predecessor_time` in integer microseconds. Returns one row
+/// `(event_id, sojourn_us)` per qualifying event.
+///
+/// `top_k`: if `Some(k)`, return only the `k` rows with the largest `sojourn_us`, ties broken by
+/// ascending event id, sorted descending. `None` returns every qualifying event.
+#[register_binding]
+pub fn locel_oc_perf_sojourn_per_event(
+    ocel: &SlimLinkedOCEL,
+    #[bind(default)] top_k: Option<usize>,
+) -> Vec<(String, i64)> {
+    let sorted = sorted_events_per_object(ocel);
+    let mut rows: Vec<(EventIndex, i64)> = (0..ocel.get_num_evs() as u32)
+        .into_par_iter()
+        .filter_map(|i| {
+            let e = EventIndex::from(i);
+            let latest = e
+                .get_e2o(ocel)
+                .filter_map(|&o| df_predecessor(&sorted, ocel, e, o))
+                .map(|p| p.get_time(ocel).timestamp_micros())
+                .max()?;
+            Some((e, e.get_time(ocel).timestamp_micros() - latest))
+        })
+        .collect();
+    if let Some(k) = top_k {
+        let cmp = |a: &(EventIndex, i64), b: &(EventIndex, i64)| {
+            b.1.cmp(&a.1)
+                .then_with(|| ocel.get_ev_id(&a.0).cmp(ocel.get_ev_id(&b.0)))
+        };
+        if k < rows.len() {
+            rows.select_nth_unstable_by(k, cmp);
+            rows.truncate(k);
+        }
+        rows.sort_unstable_by(cmp);
+    }
+    rows.into_iter()
+        .map(|(e, sojourn_us)| (ocel.get_ev_id(&e).to_string(), sojourn_us))
+        .collect()
+}

process_mining/src/analysis/object_centric/oc_statistics.rs

@@ -0,0 +1,79 @@
+//! Descriptive statistics over object-centric event data (E2O type counts, conversion rates).
+
+use std::collections::HashMap;
+
+use macros_process_mining::register_binding;
+use rayon::prelude::*;
+
+use crate::core::event_data::object_centric::linked_ocel::{
+    slim_linked_ocel::{EventIndex, ObjectIndex},
+    LinkedOCELAccess, SlimLinkedOCEL,
+};
+
+/// Count E2O relationships per `(event_type, object_type)` pair.
+///
+/// Each entry is `(event_type, object_type, count)`. A single `(event, object)` pair connected
+/// by multiple qualifiers contributes once per qualifier. Pairs with zero relations are omitted.
+/// Result row order is unspecified.
+#[register_binding]
+pub fn locel_event_object_type_counts(ocel: &SlimLinkedOCEL) -> Vec<(String, String, i64)> {
+    let num_events = ocel.get_num_evs() as u32;
+    let counts: HashMap<(usize, usize), i64> = (0..num_events)
+        .into_par_iter()
+        .fold(HashMap::new, |mut acc, i| {
+            let ev = EventIndex::from(i).get_ev(ocel);
+            for (_q, ob) in &ev.relationships {
+                let ot = ob.get_ob(ocel).object_type;
+                *acc.entry((ev.event_type, ot)).or_insert(0) += 1;
+            }
+            acc
+        })
+        .reduce(HashMap::new, merge_sum_maps);
+    let ev_types: Vec<&str> = <SlimLinkedOCEL as LinkedOCELAccess>::get_ev_types(ocel).collect();
+    let ob_types: Vec<&str> = <SlimLinkedOCEL as LinkedOCELAccess>::get_ob_types(ocel).collect();
+    counts
+        .into_iter()
+        .map(|((e, o), c)| (ev_types[e].to_string(), ob_types[o].to_string(), c))
+        .collect()
+}
+
+/// Conversion rate from `source_type` to `target_type` via O2O, restricted to targets touched by `activity`.
+///
+/// Returns the fraction of `source_type` objects that have at least one outgoing O2O edge to a
+/// `target_type` object related (via E2O) to some event of the given event type. Returns `0.0`
+/// if no `source_type` objects exist.
+#[register_binding]
+pub fn locel_conversion_rate(
+    ocel: &SlimLinkedOCEL,
+    activity: String,
+    source_type: String,
+    target_type: String,
+) -> f64 {
+    let sources: Vec<ObjectIndex> = ocel.get_obs_of_type(&source_type).copied().collect();
+    let total = sources.len();
+    if total == 0 {
+        return 0.0;
+    }
+    let reached = sources
+        .par_iter()
+        .filter(|&&s| {
+            s.get_o2o(ocel).any(|&t| {
+                t.get_ob_type(ocel) == &target_type
+                    && t.get_e2o_rev(ocel)
+                        .any(|&e| e.get_ev_type(ocel) == &activity)
+            })
+        })
+        .count();
+    reached as f64 / total as f64
+}
+
+/// Merge `b` into `a` by summing `i64` counts for matching keys. Used as the rayon reduce step.
+fn merge_sum_maps<K: std::hash::Hash + Eq>(
+    mut a: HashMap<K, i64>,
+    b: HashMap<K, i64>,
+) -> HashMap<K, i64> {
+    for (k, v) in b {
+        *a.entry(k).or_insert(0) += v;
+    }
+    a
+}