add env vars for sort pruning and parquet creation

Author: parmesant

src/parseable/staging/writer.rs

@@ -38,6 +38,7 @@ use itertools::Itertools;
 use once_cell::sync::Lazy;
 use rand::distributions::{Alphanumeric, DistString};
 use tracing::error;
+use ulid::Ulid;
 
 use crate::{
     parseable::{ARROW_FILE_EXTENSION, PART_FILE_EXTENSION},
@@ -68,6 +69,17 @@ 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
+    }
+});
+
 #[derive(Default)]
 pub struct Writer {
     pub mem: MemWriter<4096>,
@@ -279,7 +291,15 @@ impl Drop for DiskWriter {
         }
 
         let mut arrow_path = self.path.to_owned();
-        arrow_path.set_extension(ARROW_FILE_EXTENSION);
+
+        // a rudimentary way to ensure one parquet per arrow file
+        if *ONE_PARQUET_PER_ARROW {
+            arrow_path.set_extension(Ulid::new().to_string());
+            #[allow(clippy::incompatible_msrv)]
+            arrow_path.add_extension(ARROW_FILE_EXTENSION);
+        } else {
+            arrow_path.set_extension(ARROW_FILE_EXTENSION);
+        }
 
         // If file exists, append a random string before .date to avoid overwriting
         if arrow_path.exists() {

src/parseable/streams.rs

@@ -85,6 +85,17 @@ static DISK_WRITE_BATCH_ROWS: Lazy<usize> = Lazy::new(|| {
     }
 });
 
+const SORT_FOR_METRIC_PRUNING_VAR: &str = "SORT_FOR_METRIC_PRUNING";
+static SORT_FOR_METRIC_PRUNING: Lazy<bool> = Lazy::new(|| {
+    if let Ok(var) = std::env::var(SORT_FOR_METRIC_PRUNING_VAR)
+        && let Ok(var) = var.parse::<bool>()
+    {
+        var
+    } else {
+        true
+    }
+});
+
 const INPROCESS_DIR_PREFIX: &str = "processing_";
 const METRIC_ROW_GROUP_PREP_IN_FLIGHT: usize = 1;
 
@@ -917,7 +928,7 @@ impl Stream {
             |s| s.as_str().to_string(),
         );
 
-        if sort_for_metric_pruning {
+        if sort_for_metric_pruning && *SORT_FOR_METRIC_PRUNING {
             // Buffer batches up to the row-group target, then
             // concat + sort + write as a single contiguous batch. The
             // ArrowWriter splits the sorted batch into row groups at the