upgrade datafusion, arrow, object_store (#1772)

* upgrade datafusion, arrow, object_store

* upgrade df to 55.0.0

* preserve order to true

* track parquet null counts in manifest

Author: nikhilsinhaparseable

Cargo.lock

@@ -3,31 +3,31 @@ name = "parseable"
 version = "3.1.1"
 authors = ["Parseable Team <hi@parseable.com>"]
 edition = "2024"
-rust-version = "1.91.0"
+rust-version = "1.94.0"
 categories = ["logs", "observability", "metrics", "traces"]
 build = "build.rs"
 
 [dependencies]
 # Arrow and DataFusion ecosystem
-arrow = "58.0.0"
-arrow-array = "58.0.0"
-arrow-flight = { version = "58.0.0", features = [
+arrow = "59.2.0"
+arrow-array = "59.2.0"
+arrow-flight = { version = "59.2.0", features = [
     "tls-aws-lc",
     "tls-native-roots",
 ] }
-arrow-ipc = { version = "58.0.0", features = ["zstd"] }
-arrow-json = "58.0.0"
-arrow-schema = { version = "58.0.0", features = ["serde"] }
-arrow-select = "58.0.0"
-datafusion = "53.0.0"
-datafusion-proto = "53.0.0"
-object_store = { version = "0.13.1", features = [
+arrow-ipc = { version = "59.2.0", features = ["zstd"] }
+arrow-json = "59.2.0"
+arrow-schema = { version = "59.2.0", features = ["serde"] }
+arrow-select = "59.2.0"
+datafusion = "55.0.0"
+datafusion-proto = "55.0.0"
+object_store = { version = "0.13.2", features = [
     "cloud",
     "aws",
     "azure",
     "gcp",
 ] }
-parquet = "58.0.0"
+parquet = "59.2.0"
 
 # Web server and HTTP-related
 actix-cors = "0.7.0"
@@ -222,7 +222,7 @@ anyhow = "1.0"
 
 [dev-dependencies]
 rstest = "0.26.1"
-arrow = "58.0.0"
+arrow = "59.2.0"
 temp-dir = "0.1.14"
 
 [package.metadata.parseable_ui]

Cargo.toml

@@ -201,6 +201,8 @@ fn is_valid_float_range(min: f64, max: f64) -> bool {
 pub struct Column {
     pub name: String,
     pub stats: Option<TypedStatistics>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub null_count: Option<u64>,
     pub uncompressed_size: u64,
     pub compressed_size: u64,
 }

src/catalog/column.rs

@@ -22,6 +22,7 @@ use itertools::Itertools;
 use parquet::file::{
     metadata::{RowGroupMetaData, SortingColumn},
     reader::FileReader,
+    statistics::Statistics,
 };
 
 use crate::metastore::metastore_traits::MetastoreObject;
@@ -267,6 +268,10 @@ fn column_statistics(row_groups: &[RowGroupMetaData]) -> HashMap<String, Column>
             if let Some(entry) = columns.get_mut(&col_name) {
                 entry.compressed_size += col.compressed_size() as u64;
                 entry.uncompressed_size += col.uncompressed_size() as u64;
+                entry.null_count = entry
+                    .null_count
+                    .zip(col.statistics().and_then(Statistics::null_count_opt))
+                    .and_then(|(current, other)| current.checked_add(other));
                 if let Some(other) = col.statistics().and_then(|stats| stats.try_into().ok()) {
                     entry.stats = entry.stats.clone().and_then(|this| this.update(other));
                 }
@@ -276,6 +281,7 @@ fn column_statistics(row_groups: &[RowGroupMetaData]) -> HashMap<String, Column>
                     Column {
                         name: col_name,
                         stats: col.statistics().and_then(|stats| stats.try_into().ok()),
+                        null_count: col.statistics().and_then(Statistics::null_count_opt),
                         uncompressed_size: col.uncompressed_size() as u64,
                         compressed_size: col.compressed_size() as u64,
                     },
@@ -301,6 +307,7 @@ mod codec_tests {
                     .map(|c| Column {
                         name: format!("some_reasonably_long_column_name_{c}"),
                         stats: None,
+                        null_count: None,
                         uncompressed_size: 1024,
                         compressed_size: 512,
                     })

src/catalog/manifest.rs

@@ -297,7 +297,7 @@ mod tests {
         types::Int64Type,
     };
     use arrow_ipc::writer::{
-        CompressionContext, DictionaryTracker, IpcDataGenerator, IpcWriteOptions, StreamWriter,
+        DictionaryTracker, IpcDataGenerator, IpcWriteContext, IpcWriteOptions, StreamWriter,
         write_message,
     };
     use arrow_schema::{DataType, Field, Schema};
@@ -389,7 +389,7 @@ mod tests {
         let options = IpcWriteOptions::default();
         let mut dictionary_tracker = DictionaryTracker::new(error_on_replacement);
         let data_gen = IpcDataGenerator {};
-        let mut compression_context = CompressionContext::default();
+        let mut compression_context = IpcWriteContext::default();
 
         let mut buf = Vec::new();
         let rb1 = rb(1);

src/parseable/staging/reader.rs

@@ -750,7 +750,7 @@ impl Stream {
             };
             props = props
                 .set_column_bloom_filter_enabled(column_path.clone(), true)
-                .set_column_bloom_filter_ndv(column_path, METRIC_NAME_BLOOM_FILTER_NDV)
+                .set_column_bloom_filter_max_ndv(column_path, METRIC_NAME_BLOOM_FILTER_NDV)
                 .set_bloom_filter_position(bloom_filter_position);
         }
         sorting_column_vec.push(SortingColumn {

src/parseable/streams.rs

@@ -25,14 +25,10 @@ use datafusion::{
         listing::{ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl},
     },
     error::DataFusionError,
-    logical_expr::col,
 };
 use itertools::Itertools;
 
-use crate::{
-    OBJECT_STORE_DATA_GRANULARITY, event::DEFAULT_TIMESTAMP_KEY, storage::ObjectStorage,
-    utils::time::TimeRange,
-};
+use crate::{OBJECT_STORE_DATA_GRANULARITY, storage::ObjectStorage, utils::time::TimeRange};
 
 use super::PartialTimeFilter;
 
@@ -121,23 +117,14 @@ impl ListingTableBuilder {
         self,
         schema: Arc<Schema>,
         map: impl Fn(Vec<String>) -> Vec<ListingTableUrl>,
-        time_partition: Option<String>,
     ) -> Result<Option<Arc<ListingTable>>, DataFusionError> {
         if self.listing.is_empty() {
             return Ok(None);
         }
 
-        let file_sort_order = vec![vec![
-            time_partition
-                .map_or_else(|| col(DEFAULT_TIMESTAMP_KEY), col)
-                .sort(true, false),
-        ]];
         let file_format = ParquetFormat::default().with_enable_pruning(true);
-        let listing_options = ListingOptions::new(Arc::new(file_format))
-            .with_file_extension(".parquet")
-            .with_file_sort_order(file_sort_order)
-            .with_collect_stat(true)
-            .with_target_partitions(1);
+        let listing_options =
+            ListingOptions::new(Arc::new(file_format)).with_file_extension(".parquet");
         let config = ListingTableConfig::new_with_multi_paths(map(self.listing))
             .with_listing_options(listing_options)
             .with_schema(schema);

src/query/listing_table_builder.rs

@@ -284,6 +284,7 @@ impl Query {
         let mut config = SessionConfig::default()
             .with_parquet_pruning(true)
             .with_prefer_existing_sort(true)
+            .with_collect_statistics(true)
             //batch size has been made configurable via environment variable
             //default value is 20000
             .with_batch_size(PARSEABLE.options.execution_batch_size)
@@ -296,6 +297,24 @@ impl Query {
         // Reorder filters allows DF to decide the order of filters minimizing the cost of filter evaluation
         config.options_mut().execution.parquet.reorder_filters = true;
         config.options_mut().execution.parquet.binary_as_string = true;
+        // Allow unordered Parquet scans to split files into row-group morsels and let idle scan
+        // partitions steal work. Ordered scans set FileScanConfig::preserve_order, which prevents
+        // file reassignment while retaining their advertised output ordering.
+        config
+            .options_mut()
+            .execution
+            .enable_file_stream_work_stealing = true;
+        config.options_mut().optimizer.repartition_file_scans = true;
+
+        // Feed changing TopK / aggregate bounds into Parquet pruning. This is especially useful
+        // for observability queries such as `ORDER BY p_timestamp DESC LIMIT N`.
+        config
+            .options_mut()
+            .optimizer
+            .enable_dynamic_filter_pushdown = true;
+        config.options_mut().optimizer.enable_topk_aggregation = true;
+        config.options_mut().optimizer.enable_topk_repartition = true;
+        config.options_mut().optimizer.enable_sort_pushdown = true;
         // Bump footer-read hint from the 512 KiB default. Streams with
         // many label columns + page-indexed value columns can have
         // parquet footers in the 1-2 MiB range; sizing the hint above
@@ -458,6 +477,7 @@ impl Query {
                 LogicalPlan::Explain(Explain {
                     explain_format: plan.explain_format,
                     verbose: plan.verbose,
+                    show_statistics: plan.show_statistics,
                     stringified_plans: vec![
                         transformed
                             .data