Add alternate arrow reader

server/Cargo.toml

@@ -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/src/main.rs

@@ -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/query.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/table_provider.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/storage/staging.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/utils/arrow.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/merged_reader.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))
     }