fdb maelstrom

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

src/persist-client/src/internal/metrics.rs

@@ -3012,6 +3012,10 @@ impl Consensus for MetricsConsensus {
             .inc_by(u64::cast_from(deleted));
         Ok(deleted)
     }
+
+    fn truncate_counts(&self) -> bool {
+        self.consensus.truncate_counts()
+    }
 }
 
 /// A standard set of metrics for an async task. Call [TaskMetrics::instrument_task] to instrument

src/persist/src/cfg.rs

@@ -213,7 +213,7 @@ impl BlobConfig {
 /// Config for an implementation of [Consensus].
 #[derive(Debug, Clone)]
 pub enum ConsensusConfig {
-    /// Config for [FdbConsensus].
+    /// Config for FoundationDB.
     FoundationDB(FdbConsensusConfig),
     /// Config for [PostgresConsensus].
     Postgres(PostgresConsensusConfig),

src/persist/src/foundationdb.rs

@@ -11,8 +11,8 @@
 //!
 //! We're storing the consensus data in a subspace at `/mz/consensus`. Each key maps to a subspace
 //! with the following structure:
-//! ./seqno/<key> -> <seqno>
-//! ./data/<key>/<seqno> -> <data>
+//! * `./seqno/<key> -> <seqno>`
+//! * `./data/<key>/<seqno> -> <data>`
 
 use std::io::Write;
 use std::sync::OnceLock;
@@ -304,24 +304,27 @@ impl FdbConsensus {
         trx.set(&data_seqno_key, &pack(&new.data.as_ref()));
         Ok(CaSResult::Committed)
     }
+
     async fn scan_trx(
         &self,
         trx: &Transaction,
         data_key: &Subspace,
         from: &SeqNo,
         limit: &usize,
-    ) -> Result<Vec<VersionedData>, FdbTransactError> {
+        entries: &mut Vec<VersionedData>,
+    ) -> Result<(), FdbTransactError> {
         let mut limit = *limit;
         let seqno_start = data_key.pack(&from);
         let seqno_end = data_key.pack(&SeqNo::maximum());
 
         let mut range = RangeOption::from(seqno_start..=seqno_end);
         range.limit = Some(limit);
 
-        let mut entries = Vec::new();
+        entries.clear();
 
         loop {
             let output = trx.get_range(&range, 1, false).await?;
+            entries.reserve(output.len());
             for key_value in &output {
                 let seqno = data_key.unpack(key_value.key())?;
                 let value: Vec<u8> = unpack(key_value.value())?;
@@ -342,9 +345,7 @@ impl FdbConsensus {
                 break;
             }
         }
-
-        entries.sort_by_key(|e| e.seqno);
-        Ok(entries)
+        Ok(())
     }
     async fn truncate_trx(
         &self,
@@ -462,15 +463,19 @@ impl Consensus for FdbConsensus {
         limit: usize,
     ) -> Result<Vec<VersionedData>, ExternalError> {
         let data_key = self.data.subspace(&key);
-        let ok = self
-            .db
+        let mut entries = Vec::new();
+        self.db
             .transact_boxed(
-                (&data_key, from, limit),
-                |trx, (data_key, from, limit)| self.scan_trx(trx, data_key, from, limit).boxed(),
+                (&data_key, from, limit, &mut entries),
+                |trx, (data_key, from, limit, entries)| {
+                    self.scan_trx(trx, data_key, from, limit, entries).boxed()
+                },
                 TransactOption::default(),
             )
             .await?;
-        Ok(ok)
+
+        entries.sort_by_key(|e| e.seqno);
+        Ok(entries)
     }
 
     async fn truncate(&self, key: &str, seqno: SeqNo) -> Result<usize, ExternalError> {

src/persist/src/location.rs

@@ -441,7 +441,7 @@ pub trait Consensus: std::fmt::Debug + Send + Sync {
     /// data at this key.
     async fn truncate(&self, key: &str, seqno: SeqNo) -> Result<usize, ExternalError>;
 
-    /// Returns true if [`truncate`] returns the number of versions deleted.
+    /// Returns true if [`Self::truncate`] returns the number of versions deleted.
     fn truncate_counts(&self) -> bool {
         true
     }
@@ -508,6 +508,10 @@ impl<A: Consensus + 'static> Consensus for Tasked<A> {
         )
         .await?
     }
+
+    fn truncate_counts(&self) -> bool {
+        self.0.truncate_counts()
+    }
 }
 
 /// Metadata about a particular blob stored by persist

src/persist/src/unreliable.rs

@@ -228,6 +228,10 @@ impl Consensus for UnreliableConsensus {
             .run_op("truncate", || self.consensus.truncate(key, seqno))
             .await
     }
+
+    fn truncate_counts(&self) -> bool {
+        self.consensus.truncate_counts()
+    }
 }
 
 #[cfg(test)]

test/persist/mzcompose.py

@@ -13,20 +13,29 @@
 
 import argparse
 
+from materialize import MZ_ROOT
 from materialize.mzcompose.composition import (
     Composition,
     WorkflowArgumentParser,
 )
 from materialize.mzcompose.service import Service
 from materialize.mzcompose.services.cockroach import Cockroach
+from materialize.mzcompose.services.foundationdb import FoundationDB
 from materialize.mzcompose.services.postgres import PostgresMetadata
 
 SERVICES = [
     Cockroach(setup_materialize=True, in_memory=True),
     PostgresMetadata(),
+    FoundationDB(),
     Service(
         "maelstrom-persist",
-        {"mzbuild": "maelstrom-persist", "volumes": ["./maelstrom:/store"]},
+        {
+            "mzbuild": "maelstrom-persist",
+            "volumes": [
+                "./maelstrom:/store",
+                f"{MZ_ROOT}/misc/foundationdb/fdb.cluster:/etc/foundationdb/fdb.cluster",
+            ],
+        },
     ),
 ]
 
@@ -50,7 +59,7 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
     parser.add_argument(
         "--consensus",
         type=str,
-        choices=["mem", "cockroach", "maelstrom", "postgres"],
+        choices=["mem", "cockroach", "maelstrom", "postgres", "foundationdb"],
         default="maelstrom",
     )
     parser.add_argument(
@@ -76,6 +85,17 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
             "postgres://root@postgres-metadata:26257?options=--search_path=consensus"
         )
         c.up("postgres-metadata")
+    elif args.consensus == "foundationdb":
+        consensus_uri = "foundationdb:"
+        c.up("foundationdb")
+        c.run(
+            "foundationdb",
+            "-C",
+            "/etc/foundationdb/fdb.cluster",
+            "--exec",
+            "configure new single memory",
+            entrypoint="fdbcli",
+        )
     else:
         # empty consensus uri defaults to Maelstrom consensus implementation
         consensus_uri = ""