feat: Add paths module to influxdb3_write

influxdb3_write/src/lib.rs

@@ -7,6 +7,7 @@
 //! to be persisted. A new open segment will be created and new writes will be written to that segment.
 
 pub mod catalog;
+pub mod paths;
 pub mod persister;
 pub mod wal;
 pub mod write_buffer;

influxdb3_write/src/paths.rs

@@ -0,0 +1,128 @@
+use crate::SegmentId;
+use std::convert::AsRef;
+use std::ops::Deref;
+use std::path::Path;
+use std::path::PathBuf;
+
+/// File extension for catalog files
+const CATALOG_FILE_EXTENSION: &str = "json";
+
+/// File extension for parquet files
+const PARQUET_FILE_EXTENSION: &str = "parquet";
+
+/// File extension for segment files
+const SEGMENT_FILE_EXTENSION: &str = "wal";
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CatalogFilePath(PathBuf);
+
+impl CatalogFilePath {
+    pub fn new(prefix: impl Into<PathBuf>, sequence_number: u64) -> Self {
+        let mut path = prefix.into();
+        path.push("catalogs");
+        path.push(format!("{sequence_number:010}"));
+        path.set_extension(CATALOG_FILE_EXTENSION);
+        Self(path)
+    }
+}
+
+impl Deref for CatalogFilePath {
+    type Target = Path;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl AsRef<Path> for CatalogFilePath {
+    fn as_ref(&self) -> &Path {
+        &self.0
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ParquetFilePath(PathBuf);
+
+impl ParquetFilePath {
+    pub fn new(
+        prefix: impl Into<PathBuf>,
+        db_name: &str,
+        table_name: &str,
+        year: u16,
+        month: u8,
+        day: u8,
+        file_number: usize,
+    ) -> Self {
+        let mut path = prefix.into();
+        path.push("dbs");
+        path.push(db_name);
+        path.push(table_name);
+        path.push(format!("{year}-{month:02}-{day:02}"));
+        path.push(format!("{file_number:010}"));
+        path.set_extension(PARQUET_FILE_EXTENSION);
+        Self(path)
+    }
+}
+
+impl Deref for ParquetFilePath {
+    type Target = Path;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl AsRef<Path> for ParquetFilePath {
+    fn as_ref(&self) -> &Path {
+        &self.0
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SegmentFilePath(PathBuf);
+
+impl SegmentFilePath {
+    pub fn new(prefix: impl Into<PathBuf>, segment_id: SegmentId) -> Self {
+        let mut path = prefix.into();
+        path.push("segments");
+        path.push(format!("{:010}", segment_id.0));
+        path.set_extension(SEGMENT_FILE_EXTENSION);
+        Self(path)
+    }
+}
+
+impl Deref for SegmentFilePath {
+    type Target = Path;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl AsRef<Path> for SegmentFilePath {
+    fn as_ref(&self) -> &Path {
+        &self.0
+    }
+}
+
+#[test]
+fn catalog_file_path_new() {
+    assert_eq!(
+        *CatalogFilePath::new("prefix/dir", 0),
+        PathBuf::from("prefix/dir/catalogs/0000000000.json").as_ref()
+    );
+}
+#[test]
+fn parquet_file_path_new() {
+    assert_eq!(
+        *ParquetFilePath::new("prefix/dir", "my_db", "my_table", 2038, 1, 19, 0),
+        PathBuf::from("prefix/dir/dbs/my_db/my_table/2038-01-19/0000000000.parquet").as_ref()
+    );
+}
+#[test]
+fn segment_file_path_new() {
+    assert_eq!(
+        *SegmentFilePath::new("prefix/dir", SegmentId::new(0)),
+        PathBuf::from("prefix/dir/segments/0000000000.wal").as_ref()
+    );
+}

influxdb3_write/src/wal.rs

@@ -1,6 +1,7 @@
 //! This is the implementation of the `Wal` that the buffer uses to make buffered data durable
 //! on disk.
 
+use crate::paths::SegmentFilePath;
 use crate::{
     SegmentFile, SegmentId, SegmentIdBytes, SequenceNumber, Wal, WalOp, WalOpBatch,
     WalSegmentReader, WalSegmentWriter,
@@ -24,9 +25,6 @@ use thiserror::Error;
 type FileTypeIdentifier = [u8; 8];
 const FILE_TYPE_IDENTIFIER: &[u8] = b"idb3.001";
 
-/// File extension for segment files
-const SEGMENT_FILE_EXTENSION: &str = "wal";
-
 #[derive(Debug, Error)]
 pub enum Error {
     #[error("io error: {source}")]
@@ -49,7 +47,7 @@ pub enum Error {
     #[error("invalid segment file {segment_id:?} at {path:?}: {reason}")]
     InvalidSegmentFile {
         segment_id: SegmentId,
-        path: PathBuf,
+        path: SegmentFilePath,
         reason: String,
     },
 
@@ -94,7 +92,7 @@ impl WalImpl {
     pub fn new(path: impl Into<PathBuf>) -> Result<Self> {
         let root = path.into();
         info!(wal_dir=?root, "Ensuring WAL directory exists");
-        std::fs::create_dir_all(&root)?;
+        std::fs::create_dir_all(&root.join("segments"))?;
 
         // ensure the directory creation is actually fsync'd so that when we create files there
         // we don't lose them (see: https://www.usenix.org/system/files/conference/osdi14/osdi14-paper-pillai.pdf)
@@ -117,7 +115,7 @@ impl WalImpl {
     }
 
     fn segment_files(&self) -> Result<Vec<SegmentFile>> {
-        let dir = std::fs::read_dir(&self.root)?;
+        let dir = std::fs::read_dir(&self.root.join("segments"))?;
 
         let mut segment_files = Vec::new();
 
@@ -161,7 +159,7 @@ impl WalImpl {
     }
 
     fn delete_wal_segment(&self, segment_id: SegmentId) -> Result<()> {
-        let path = build_segment_path(self.root.clone(), segment_id);
+        let path = SegmentFilePath::new(self.root.clone(), segment_id);
         std::fs::remove_file(path)?;
         Ok(())
     }
@@ -197,7 +195,7 @@ pub struct WalSegmentWriterImpl {
 
 impl WalSegmentWriterImpl {
     pub fn new_or_open(root: PathBuf, segment_id: SegmentId) -> Result<Self> {
-        let path = build_segment_path(root, segment_id);
+        let path = SegmentFilePath::new(root, segment_id);
 
         // if there's already a file there, validate its header and pull the sequence number from the last entry
         if path.exists() {
@@ -225,6 +223,15 @@ impl WalSegmentWriterImpl {
             }
         }
 
+        let parent = path
+            .parent()
+            .expect("A SegmentFilePath should have a parent directory");
+
+        // Make sure that the segments directory with the prefix exists
+        if !parent.exists() {
+            std::fs::create_dir_all(parent)?;
+        }
+
         // it's a new file, initialize it with the header and get ready to start writing
         let mut f = OpenOptions::new().write(true).create(true).open(&path)?;
 
@@ -325,7 +332,7 @@ pub struct WalSegmentReaderImpl {
 
 impl WalSegmentReaderImpl {
     pub fn new(root: impl Into<PathBuf>, segment_id: SegmentId) -> Result<Self> {
-        let path = build_segment_path(root, segment_id);
+        let path = SegmentFilePath::new(root, segment_id);
         let f = BufReader::new(File::open(path.clone())?);
 
         let mut reader = Self { f, segment_id };
@@ -359,7 +366,7 @@ impl WalSegmentReaderImpl {
     }
 
     fn read_segment_file_info_if_exists(
-        path: PathBuf,
+        path: SegmentFilePath,
         segment_id: SegmentId,
     ) -> Result<Option<ExistingSegmentFileInfo>> {
         let f = match File::open(path.clone()) {
@@ -549,13 +556,6 @@ where
     }
 }
 
-fn build_segment_path(dir: impl Into<PathBuf>, id: SegmentId) -> PathBuf {
-    let mut path = dir.into();
-    path.push(format!("{:010}", id.0));
-    path.set_extension(SEGMENT_FILE_EXTENSION);
-    path
-}
-
 fn segment_id_from_file_name(name: &str) -> Result<SegmentId> {
     let id = name
         .parse::<u32>()