make MemWriter optional

Author: parmesant

src/parseable/staging/writer.rs

@@ -69,17 +69,6 @@ static ARROW_FLUSH_SIZE_LIMIT: Lazy<usize> = Lazy::new(|| {
     }
 });
 
-const ONE_PARQUET_PER_ARROW_VAR: &str = "ONE_PARQUET_PER_ARROW";
-static ONE_PARQUET_PER_ARROW: Lazy<bool> = Lazy::new(|| {
-    if let Ok(var) = std::env::var(ONE_PARQUET_PER_ARROW_VAR)
-        && let Ok(var) = var.parse::<bool>()
-    {
-        var
-    } else {
-        false
-    }
-});
-
 const ENABLE_MEMORY_STAGING_VAR: &str = "ENABLE_MEMORY_STAGING";
 pub static ENABLE_MEMORY_STAGING: Lazy<bool> = Lazy::new(|| {
     if let Ok(var) = std::env::var(ENABLE_MEMORY_STAGING_VAR)
@@ -91,13 +80,28 @@ pub static ENABLE_MEMORY_STAGING: Lazy<bool> = Lazy::new(|| {
     }
 });
 
-#[derive(Default)]
+// #[derive(Default)]
 pub struct Writer {
-    pub mem: MemWriter<4096>,
+    pub mem: Option<MemWriter<4096>>,
     pub disk: HashMap<String, DiskWriter>,
     disk_pending: HashMap<String, PendingDiskBatch>,
 }
 
+impl Default for Writer {
+    fn default() -> Self {
+        let mem = if *ENABLE_MEMORY_STAGING {
+            Some(MemWriter::default())
+        } else {
+            None
+        };
+        Self {
+            mem,
+            disk: Default::default(),
+            disk_pending: Default::default(),
+        }
+    }
+}
+
 impl Writer {
     #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn push_disk(
@@ -256,13 +260,10 @@ impl DiskWriter {
         range: TimeRange,
     ) -> Result<Self, StagingError> {
         let mut path = path.into();
-        // a rudimentary way to ensure one parquet per arrow file
-        if *ONE_PARQUET_PER_ARROW {
-            path.set_extension(Ulid::new().to_string());
-            path.add_extension(PART_FILE_EXTENSION);
-        } else {
-            path.set_extension(PART_FILE_EXTENSION);
-        }
+
+        path.set_extension(Ulid::new().to_string());
+        path.add_extension(PART_FILE_EXTENSION);
+
         let file = OpenOptions::new()
             .write(true)
             .truncate(true)

src/parseable/streams.rs

@@ -61,7 +61,7 @@ use crate::{
     metadata::{LogStreamMetadata, SchemaVersion},
     metrics,
     option::Mode,
-    parseable::{DEFAULT_TENANT, PARSEABLE, staging::writer::ENABLE_MEMORY_STAGING},
+    parseable::{DEFAULT_TENANT, PARSEABLE},
     storage::{StreamType, object_storage::to_bytes, retention::Retention},
     sync::FLUSH_AND_CONVERT_RUNTIME,
     utils::time::{Minute, TimeRange},
@@ -202,8 +202,8 @@ impl Stream {
             guard.push_disk(filename, record, file_path, range, *DISK_WRITE_BATCH_ROWS)?;
         }
 
-        if *ENABLE_MEMORY_STAGING {
-            guard.mem.push(schema_key, record)?;
+        if let Some(mem) = guard.mem.as_mut() {
+            mem.push(schema_key, record)?;
         }
 
         Ok(())
@@ -560,22 +560,23 @@ impl Stream {
         &self,
         schema: &Arc<Schema>,
     ) -> Result<Vec<RecordBatch>, StagingError> {
-        let writer = self.writer.lock().map_err(|poisoned| {
+        let mut writer = self.writer.lock().map_err(|poisoned| {
             StagingError::PoisonError(PoisonError::new(format!(
                 "Writer lock poisoned while cloning record batches for stream {} - {}",
                 self.stream_name, poisoned
             )))
         })?;
 
-        if *ENABLE_MEMORY_STAGING {
-            writer.mem.recordbatch_cloned(schema)
+        if let Some(mem) = writer.mem.as_mut() {
+            mem.recordbatch_cloned(schema)
         } else {
             Ok(Vec::new())
         }
     }
 
     pub fn clear(&self) -> Result<(), StagingError> {
-        self.writer
+        if let Some(m) = self
+            .writer
             .lock()
             .map_err(|poisoned| {
                 StagingError::PoisonError(PoisonError::new(format!(
@@ -584,7 +585,10 @@ impl Stream {
                 )))
             })?
             .mem
-            .clear();
+            .as_mut()
+        {
+            m.clear()
+        }
         Ok(())
     }
 
@@ -601,9 +605,9 @@ impl Stream {
                     self.stream_name, poisoned
                 )))
             })?;
-            // why clean Writer.MemWriter?
-            if *ENABLE_MEMORY_STAGING {
-                writer.mem.clear();
+
+            if let Some(mem) = writer.mem.as_mut() {
+                mem.clear();
             }
             writer.take_flushable_disk(forced)
         };