feat: add reverse arrow reader (#548)

This PR adds a new arrow reader which can read arrow file in reverse. This is to be 
used while writing to parquet file such that it yields file sorted in timestamp descending. 
Descending timestamp is the natural format for log data where users are interested in 
looking at the latest events first.

This change helps the client tools (console / pb / others) in the sense that they don't 
have to sort this on client side, neither they need to issue queries with `order by` 
keywords.

This PR also add extra time interval to object sync (5s) to ensure there is no race with 
the local_sync function.

Author: trueleo

Cargo.lock

@@ -38,7 +38,9 @@ argon2 = "0.5.0"
 async-trait = "0.1"
 base64 = "0.21"
 bytes = "1.4"
+byteorder = "1.4.3"
 bzip2 = { version = "*", features = ["static"] }
+cookie = "0.17.0"
 chrono = "0.4"
 chrono-humanize = "0.2"
 clap = { version = "4.1", default-features = false, features = [
@@ -97,8 +99,6 @@ humantime = "2.1.0"
 openid = { version = "0.12.0", default-features = false, features = ["rustls"] }
 url = "2.4.0"
 http-auth-basic = "0.3.3"
-cookies = "0.0.1"
-cookie = "0.17.0"
 
 [build-dependencies]
 cargo_toml = "0.15"
@@ -114,8 +114,8 @@ maplit = "1.0"
 rstest = "0.16"
 
 [package.metadata.parseable_ui]
-assets-url = "https://github.com/parseablehq/console/releases/download/v0.3.1/build.zip"
-assets-sha1 = "6abd7b5ca5b9c832ff58b8450cffdc83dd7172bf"
+assets-url = "https://github.com/parseablehq/console/releases/download/v0.3.3/build.zip"
+assets-sha1 = "29e1eaa2dfa081d495c1504f05b78914d074990c"
 
 [features]
 debug = []

server/Cargo.toml

@@ -16,7 +16,7 @@
  *
  */
 
-use clokwerk::{AsyncScheduler, Scheduler, TimeUnits};
+use clokwerk::{AsyncScheduler, Job, Scheduler, TimeUnits};
 use thread_priority::{ThreadBuilder, ThreadPriority};
 use tokio::sync::oneshot;
 use tokio::sync::oneshot::error::TryRecvError;
@@ -122,6 +122,8 @@ fn object_store_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Sende
                 let mut scheduler = AsyncScheduler::new();
                 scheduler
                     .every((CONFIG.parseable.upload_interval as u32).seconds())
+                    // Extra time interval is added so that this schedular does not race with local sync.
+                    .plus(5u32.seconds())
                     .run(|| async {
                         if let Err(e) = CONFIG.storage().get_object_store().sync().await {
                             log::warn!("failed to sync local data with object store. {:?}", e);

server/src/main.rs

@@ -171,7 +171,7 @@ impl Query {
             return Ok(None);
         }
         let file_format = ParquetFormat::default().with_enable_pruning(Some(true));
-        let file_sort_order = vec![vec![col(DEFAULT_TIMESTAMP_KEY).sort(true, false)]];
+        let file_sort_order = vec![vec![col(DEFAULT_TIMESTAMP_KEY).sort(false, true)]];
         let listing_options = ListingOptions::new(Arc::new(file_format))
             .with_file_extension(".parquet")
             .with_file_sort_order(file_sort_order)

server/src/query.rs

@@ -34,6 +34,8 @@ use std::any::Any;
 use std::sync::Arc;
 use std::vec;
 
+use crate::utils::arrow::reverse_reader::reverse;
+
 pub struct QueryTableProvider {
     staging: Option<MemTable>,
     // remote table
@@ -47,6 +49,17 @@ impl QueryTableProvider {
         storage: Option<Arc<ListingTable>>,
         schema: Arc<Schema>,
     ) -> Result<Self, DataFusionError> {
+        // in place reverse transform
+        let staging = if let Some(mut staged_batches) = staging {
+            staged_batches[..].reverse();
+            staged_batches
+                .iter_mut()
+                .for_each(|batch| *batch = reverse(batch));
+            Some(staged_batches)
+        } else {
+            None
+        };
+
         let memtable = staging
             .map(|records| MemTable::try_new(schema.clone(), vec![records]))
             .transpose()?;

server/src/query/table_provider.rs

@@ -41,7 +41,7 @@ use crate::{
     metrics,
     option::CONFIG,
     storage::OBJECT_STORE_DATA_GRANULARITY,
-    utils::{self, arrow::MergedRecordReader},
+    utils::{self, arrow::merged_reader::MergedReverseRecordReader},
 };
 
 const ARROW_FILE_EXTENSION: &str = "data.arrows";
@@ -198,7 +198,7 @@ pub fn convert_disk_files_to_parquet(
                 .add(file_size as i64);
         }
 
-        let record_reader = MergedRecordReader::try_new(&files).unwrap();
+        let record_reader = MergedReverseRecordReader::try_new(&files).unwrap();
 
         let parquet_file = fs::File::create(&parquet_path).map_err(|_| MoveDataError::Create)?;
 
@@ -239,8 +239,8 @@ fn parquet_writer_props() -> WriterPropertiesBuilder {
         )
         .set_sorting_columns(Some(vec![SortingColumn {
             column_idx: 0,
-            descending: false,
-            nulls_first: false,
+            descending: true,
+            nulls_first: true,
         }]))
 }
 

server/src/storage/staging.rs

@@ -25,6 +25,7 @@ use itertools::Itertools;
 
 pub mod batch_adapter;
 pub mod merged_reader;
+pub mod reverse_reader;
 
 pub use batch_adapter::adapt_batch;
 pub use merged_reader::MergedRecordReader;

server/src/utils/arrow.rs

@@ -24,8 +24,11 @@ use arrow_ipc::reader::StreamReader;
 use arrow_schema::Schema;
 use itertools::kmerge_by;
 
-use super::adapt_batch;
-use crate::event::DEFAULT_TIMESTAMP_KEY;
+use super::{
+    adapt_batch,
+    reverse_reader::{reverse, OffsetReader},
+};
+use crate::{event::DEFAULT_TIMESTAMP_KEY, utils};
 
 #[derive(Debug)]
 pub struct MergedRecordReader {
@@ -44,14 +47,42 @@ impl MergedRecordReader {
         Ok(Self { readers })
     }
 
-    pub fn merged_iter(self, schema: Arc<Schema>) -> impl Iterator<Item = RecordBatch> {
-        let adapted_readers = self.readers.into_iter().map(move |reader| reader.flatten());
+    pub fn merged_schema(&self) -> Schema {
+        Schema::try_merge(
+            self.readers
+                .iter()
+                .map(|reader| reader.schema().as_ref().clone()),
+        )
+        .unwrap()
+    }
+}
 
+#[derive(Debug)]
+pub struct MergedReverseRecordReader {
+    pub readers: Vec<StreamReader<BufReader<OffsetReader<File>>>>,
+}
+
+impl MergedReverseRecordReader {
+    pub fn try_new(files: &[PathBuf]) -> Result<Self, ()> {
+        let mut readers = Vec::with_capacity(files.len());
+        for file in files {
+            let reader =
+                utils::arrow::reverse_reader::get_reverse_reader(File::open(file).unwrap())
+                    .map_err(|_| ())?;
+            readers.push(reader);
+        }
+
+        Ok(Self { readers })
+    }
+
+    pub fn merged_iter(self, schema: Arc<Schema>) -> impl Iterator<Item = RecordBatch> {
+        let adapted_readers = self.readers.into_iter().map(|reader| reader.flatten());
         kmerge_by(adapted_readers, |a: &RecordBatch, b: &RecordBatch| {
             let a_time = get_timestamp_millis(a);
             let b_time = get_timestamp_millis(b);
-            a_time < b_time
+            a_time > b_time
         })
+        .map(|batch| reverse(&batch))
         .map(move |batch| adapt_batch(&schema, &batch))
     }