explanatory comments

Author: parmesant

server/Cargo.toml

@@ -15,13 +15,12 @@ arrow-array = { version = "53.0.0" }
 arrow-json = "53.0.0"
 arrow-ipc = { version = "53.0.0", features = ["zstd"] }
 arrow-select = "53.0.0"
-# datafusion = { git = "https://github.com/apache/datafusion.git", rev = "a64df83502821f18067fb4ff65dd217815b305c9" }
 datafusion = "42.0.0"
-object_store = { version = "0.11.0", features = ["cloud", "aws"] }  # cannot update object_store as datafusion has not caught up
+object_store = { version = "0.11.0", features = ["cloud", "aws"] }
 parquet = "53.0.0"
 arrow-flight = { version = "53.0.0", features = [ "tls" ] }
-tonic = {version = "0.12.1", features = ["tls", "transport", "gzip", "zstd"] }
-tonic-web = "0.12.1"
+tonic = {version = "0.12.3", features = ["tls", "transport", "gzip", "zstd"] }
+tonic-web = "0.12.3"
 tower-http = { version = "0.6.1", features = ["cors"] }
 
 ### actix dependencies

server/src/cli.rs

@@ -421,6 +421,9 @@ impl Cli {
                      .help("Set a fixed memory limit for query"),
              )
              .arg(
+                 // RowGroupSize controls the number of rows present in one row group
+                 // More rows = better compression but HIGHER Memory consumption during read/write
+                 // 1048576 is the default value for DataFusion 
                  Arg::new(Self::ROW_GROUP_SIZE)
                      .long(Self::ROW_GROUP_SIZE)
                      .env("P_PARQUET_ROW_GROUP_SIZE")

server/src/query.rs

@@ -86,9 +86,20 @@ impl Query {
             .with_prefer_existing_sort(true)
             .with_round_robin_repartition(true);
 
+        // For more details refer https://datafusion.apache.org/user-guide/configs.html
+
+        // Reduce the number of rows read (if possible)
         config.options_mut().execution.parquet.enable_page_index = true;
+
+        // Pushdown filters allows DF to push the filters as far down in the plan as possible
+        // and thus, reducing the number of rows decoded
         config.options_mut().execution.parquet.pushdown_filters = true;
+
+        // Reorder filters allows DF to decide the order of filters minimizing the cost of filter evaluation
         config.options_mut().execution.parquet.reorder_filters = true;
+
+        // Enable StringViewArray
+        // https://www.influxdata.com/blog/faster-queries-with-stringview-part-one-influxdb/
         config
             .options_mut()
             .execution

server/src/query/stream_schema_provider.rs

@@ -216,9 +216,9 @@ async fn collect_from_snapshot(
 
 fn partitioned_files(
     manifest_files: Vec<catalog::manifest::File>,
-    table_schema: &Schema,
-    target_partition: usize,
+    table_schema: &Schema
 ) -> (Vec<Vec<PartitionedFile>>, datafusion::common::Statistics) {
+    let target_partition = num_cpus::get();
     let mut partitioned_files = Vec::from_iter((0..target_partition).map(|_| Vec::new()));
     let mut column_statistics = HashMap::<String, Option<catalog::column::TypedStatistics>>::new();
     let mut count = 0;
@@ -436,7 +436,7 @@ impl TableProvider for StandardTableProvider {
             );
         }
 
-        let (partitioned_files, statistics) = partitioned_files(manifest_files, &self.schema, 1);
+        let (partitioned_files, statistics) = partitioned_files(manifest_files, &self.schema);
         let remote_exec = create_parquet_physical_plan(
             ObjectStoreUrl::parse(glob_storage.store_url()).unwrap(),
             partitioned_files,
@@ -520,7 +520,7 @@ async fn get_cache_exectuion_plan(
         })
         .collect();
 
-    let (partitioned_files, statistics) = partitioned_files(cached, &schema, 1);
+    let (partitioned_files, statistics) = partitioned_files(cached, &schema);
     let plan = create_parquet_physical_plan(
         ObjectStoreUrl::parse("file:///").unwrap(),
         partitioned_files,
@@ -571,7 +571,7 @@ async fn get_hottier_exectuion_plan(
         })
         .collect();
 
-    let (partitioned_files, statistics) = partitioned_files(hot_tier_files, &schema, 1);
+    let (partitioned_files, statistics) = partitioned_files(hot_tier_files, &schema);
     let plan = create_parquet_physical_plan(
         ObjectStoreUrl::parse("file:///").unwrap(),
         partitioned_files,