feat: Initial journalling implementation

Cargo.toml

@@ -19,6 +19,7 @@ arrow-buffer = { version = "47.0.0" }
 arrow-cast = { version = "47.0.0" }
 arrow-csv = { version = "47.0.0" }
 arrow-data = { version = "47.0.0" }
+arrow-ipc = { version = "47.0.0" }
 arrow-json = { version = "47.0.0" }
 arrow-ord = { version = "47.0.0" }
 arrow-schema = { version = "47.0.0", features = ["serde"] }

crates/sparrow-batch/Cargo.toml

@@ -10,6 +10,7 @@ Defines the main struct for wrapping RecordBatches in execution.
 """
 
 [features]
+default = []
 testing = ["proptest"]
 
 [dependencies]

crates/sparrow-batch/src/batch.rs

@@ -17,7 +17,7 @@
 #[derive(Clone, PartialEq, Debug)]
 pub struct Batch {
     /// The data associated with the batch.
-    pub(crate) data: Option<BatchInfo>,
+    pub data: Option<BatchInfo>,
 
     /// An indication that the batch stream has completed up to the given time.
     /// Any rows in future batches on this stream must have a time strictly
@@ -475,6 +475,35 @@
             RowTime::from_timestamp_ns(up_to_time),
         )
     }
+
+    /// Creates a batch with the given times and key hashes, and `null` data.
+    #[cfg(any(test, feature = "testing"))]
+    pub fn null_from(
+        time: impl Into<TimestampNanosecondArray>,
+        subsort: impl Into<arrow_array::UInt64Array>,
+        key_hash: impl Into<arrow_array::UInt64Array>,
+        up_to_time: i64,
+    ) -> Self {
+        use arrow_array::NullArray;
+
+        let time: TimestampNanosecondArray = time.into();
+        let subsort: UInt64Array = subsort.into();
+        let key_hash: UInt64Array = key_hash.into();
+
+        let time: ArrayRef = Arc::new(time);
+        let subsort: ArrayRef = Arc::new(subsort);
+        let key_hash: ArrayRef = Arc::new(key_hash);
+
+        let data = Arc::new(NullArray::new(time.len()));
+
+        Batch::new_with_data(
+            data,
+            time,
+            subsort,
+            key_hash,
+            RowTime::from_timestamp_ns(up_to_time),
+        )
+    }
 }
 
 #[cfg(debug_assertions)]
@@ -602,13 +631,14 @@
 }
 
 #[derive(Clone, Debug)]
-pub(crate) struct BatchInfo {
-    pub(crate) data: ArrayRef,
-    pub(crate) time: ArrayRef,
-    pub(crate) subsort: ArrayRef,
-    pub(crate) key_hash: ArrayRef,
-    min_present_time: RowTime,
-    max_present_time: RowTime,
+#[non_exhaustive]
+pub struct BatchInfo {
+    pub data: ArrayRef,
+    pub time: ArrayRef,
+    pub subsort: ArrayRef,
+    pub key_hash: ArrayRef,
+    pub min_present_time: RowTime,
+    pub max_present_time: RowTime,
 }
 
 impl PartialEq for BatchInfo {
@@ -709,6 +739,10 @@
     pub(crate) fn key_hash(&self) -> &UInt64Array {
         self.key_hash.as_primitive()
     }
+
+    pub fn len(&self) -> usize {
+        self.time.len()
+    }
 }
 
 #[cfg(any(test, feature = "testing"))]

crates/sparrow-batch/src/row_time.rs

@@ -1,4 +1,8 @@
+use std::str::FromStr;
+
+use arrow::compute::kernels::cast_utils::string_to_timestamp_nanos;
 use arrow_array::temporal_conversions::timestamp_ns_to_datetime;
+use error_stack::{IntoReport, ResultExt};
 
 /// Wrapper around the time of a row.
 ///
@@ -26,6 +30,15 @@ impl RowTime {
     pub fn pred(&self) -> Self {
         Self(self.0 - 1)
     }
+
+    pub fn bytes(&self) -> [u8; 8] {
+        self.0.to_be_bytes()
+    }
+
+    pub fn from_bytes(bytes: [u8; 8]) -> Self {
+        let time = i64::from_be_bytes(bytes);
+        Self(time)
+    }
 }
 
 impl From<RowTime> for i64 {
@@ -39,3 +52,26 @@ impl From<i64> for RowTime {
         RowTime(value)
     }
 }
+
+impl From<RowTime> for String {
+    fn from(val: RowTime) -> Self {
+        format!("{val}")
+    }
+}
+
+#[derive(derive_more::Display, Debug)]
+#[display(fmt = "invalid row time: {_0}")]
+pub struct ParseError(String);
+
+impl error_stack::Context for ParseError {}
+
+impl FromStr for RowTime {
+    type Err = error_stack::Report<ParseError>;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        string_to_timestamp_nanos(s)
+            .into_report()
+            .change_context_lazy(|| ParseError(s.to_string()))
+            .map(Self)
+    }
+}

crates/sparrow-journal/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "sparrow-journal"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+publish = false
+description = """
+Durable journaling of batches.
+"""
+
+[dependencies]
+arrow-array.workspace = true
+arrow-buffer.workspace = true
+arrow-ipc.workspace = true
+arrow-schema.workspace = true
+derive_more.workspace = true
+error-stack.workspace = true
+futures.workspace = true
+itertools.workspace = true
+okaywal = "0.3.0"
+parking_lot.workspace = true
+sparrow-batch = { path = "../sparrow-batch" }
+tracing.workspace = true
+
+[dev-dependencies]
+sparrow-batch = { path = "../sparrow-batch", features = ["testing"] }
+sparrow-testing = { path = "../sparrow-testing" }
+tempfile.workspace = true
+
+[lib]
+doctest = false