WiP persistence.

Author: gz

Cargo.toml

@@ -8,6 +8,7 @@ default = ["with-serde"]
 with-serde = ["serde"]
 with-csv = ["csv"]
 with-nexmark = ["rand", "clap", "cached"]
+persistence = []
 
 [dependencies]
 anyhow = "1.0.57"

src/operator/aggregate/mod.rs

@@ -21,7 +21,7 @@ use crate::{
     time::Timestamp,
     trace::{
         cursor::{Cursor, CursorGroup},
-        spine_fueled::Spine,
+        Spine,
         Batch, BatchReader, Builder,
     },
     NumEntries, OrdIndexedZSet, OrdZSet,

src/operator/communication/shard.rs

@@ -7,7 +7,7 @@
 use crate::{
     circuit::GlobalNodeId,
     circuit_cache_key,
-    trace::{cursor::Cursor, spine_fueled::Spine, Batch, BatchReader, Builder, Trace},
+    trace::{cursor::Cursor, Spine, Batch, BatchReader, Builder, Trace},
     Circuit, Runtime, Stream,
 };
 use std::{hash::Hash, panic::Location};

src/operator/join.rs

@@ -8,7 +8,7 @@ use crate::{
     },
     time::Timestamp,
     trace::{
-        cursor::Cursor as TraceCursor, spine_fueled::Spine, Batch, BatchReader, Batcher, Builder,
+        cursor::Cursor as TraceCursor, Spine, Batch, BatchReader, Batcher, Builder,
         Trace, TraceReader,
     },
     OrdIndexedZSet, OrdZSet,

src/operator/recursive.rs

@@ -4,7 +4,7 @@ use crate::{
     algebra::{ZRingValue, ZSet},
     circuit::{schedule::Error as SchedulerError, Circuit, Stream},
     operator::DelayedFeedback,
-    trace::spine_fueled::Spine,
+    trace::Spine,
 };
 use impl_trait_for_tuples::impl_for_tuples;
 use size_of::SizeOf;

src/operator/time_series/window.rs

@@ -6,7 +6,7 @@ use crate::{
         operator_traits::{Operator, TernaryOperator},
         Circuit, OwnershipPreference, Scope, Stream,
     },
-    trace::{cursor::Cursor, ord::OrdZSet, spine_fueled::Spine, Batch, BatchReader},
+    trace::{cursor::Cursor, ord::OrdZSet, Spine, Batch, BatchReader},
 };
 use size_of::SizeOf;
 use std::{borrow::Cow, cmp::max, marker::PhantomData};

src/operator/trace.rs

@@ -4,7 +4,7 @@ use crate::{
         Circuit, ExportId, ExportStream, GlobalNodeId, OwnershipPreference, Scope, Stream,
     },
     circuit_cache_key,
-    trace::{cursor::Cursor, spine_fueled::Spine, Batch, BatchReader, Builder, Trace, TraceReader},
+    trace::{cursor::Cursor, Spine, Batch, BatchReader, Builder, Trace, TraceReader},
     NumEntries, Timestamp,
 };
 use size_of::{HumanBytes, SizeOf};

src/trace/mod.rs

@@ -17,6 +17,10 @@ pub mod ord;
 pub mod persistent;
 pub mod spine_fueled;
 
+#[cfg(not(feature = "persistence"))]
+pub use spine_fueled::Spine as Spine;
+#[cfg(feature = "persistence")]
+pub use persistent::PersistentTrace as Spine;
 pub use cursor::Cursor;
 
 use crate::{

src/trace/ord/mod.rs

@@ -34,7 +34,7 @@ pub use key_batch::OrdKeyBatch;
 pub use val_batch::OrdValBatch;
 pub use zset_batch::OrdZSet;
 
-use crate::trace::spine_fueled::Spine;
+use crate::trace::Spine;
 
 /// A trace implementation using a spine of ordered lists.
 pub type OrdValSpine<K, V, T, R, O = usize> = Spine<OrdValBatch<K, V, T, R, O>>;