fix(cubestore): Allowing window functions in cluster send planning

Author: waralexrom

rust/cubestore/cubestore/src/queryplanner/planning.rs

@@ -1597,6 +1597,7 @@ fn pull_up_cluster_send(mut p: LogicalPlan) -> Result<LogicalPlan, DataFusionErr
         LogicalPlan::Extension { .. } => return Ok(p),
         // These nodes collect results from multiple partitions, return unchanged.
         LogicalPlan::Aggregate { .. }
+        | LogicalPlan::Window { .. }
         | LogicalPlan::Repartition { .. }
         | LogicalPlan::Limit { .. } => return Ok(p),
         // Collects results but let's push sort,fetch underneath the input.

rust/cubestore/cubestore/src/queryplanner/query_executor.rs

@@ -710,11 +710,9 @@ impl CubeTable {
                 let mut options = TableParquetOptions::new();
                 options.global = state.config_options().execution.parquet.clone();
 
-                let parquet_source = ParquetSource::new(
-                    options,
-                    get_reader_options_customizer(state.config()),
-                )
-                .with_parquet_file_reader_factory(self.parquet_metadata_cache.clone());
+                let parquet_source =
+                    ParquetSource::new(options, get_reader_options_customizer(state.config()))
+                        .with_parquet_file_reader_factory(self.parquet_metadata_cache.clone());
                 let parquet_source = if let Some(phys_pred) = &physical_predicate {
                     parquet_source.with_predicate(index_schema.clone(), phys_pred.clone())
                 } else {
@@ -792,11 +790,9 @@ impl CubeTable {
 
                     let mut options = TableParquetOptions::new();
                     options.global = state.config_options().execution.parquet.clone();
-                    let parquet_source = ParquetSource::new(
-                        options,
-                        get_reader_options_customizer(state.config()),
-                    )
-                    .with_parquet_file_reader_factory(self.parquet_metadata_cache.clone());
+                    let parquet_source =
+                        ParquetSource::new(options, get_reader_options_customizer(state.config()))
+                            .with_parquet_file_reader_factory(self.parquet_metadata_cache.clone());
                     let parquet_source = if let Some(phys_pred) = &physical_predicate {
                         parquet_source.with_predicate(index_schema.clone(), phys_pred.clone())
                     } else {