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

Author: waralexrom

.github/workflows/publish.yml

@@ -618,7 +618,7 @@ jobs:
       - name: Setup Rust toolchain
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: nightly-2024-01-29
+          toolchain: nightly-2024-10-30
           target: ${{ matrix.target }}
           # override: true # this is by default on
           rustflags: ""
@@ -705,7 +705,7 @@ jobs:
       - name: Setup Rust toolchain
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: nightly-2024-01-29
+          toolchain: nightly-2024-10-30
           target: ${{ matrix.target }}
           # override: true # this is by default on
           rustflags: ""

.github/workflows/push.yml

@@ -237,7 +237,7 @@ jobs:
       - name: Install Rust
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: nightly-2024-01-29
+          toolchain: nightly-2024-10-30
           # override: true # this is by default on
           rustflags: ""
           components: rustfmt

.github/workflows/rust-cubestore-master.yml

@@ -21,7 +21,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        rust: [nightly-2024-01-29]
+        rust: [nightly-2024-10-30]
     env:
       RUST: ${{ matrix.rust }}
     steps:
@@ -304,7 +304,7 @@ jobs:
       - name: Setup Rust toolchain
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: nightly-2024-01-29
+          toolchain: nightly-2024-10-30
           target: ${{ matrix.target }}
           # override: true # this is by default on
           rustflags: ""

.github/workflows/rust-cubestore.yml

@@ -26,7 +26,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        rust: [nightly-2024-01-29]
+        rust: [nightly-2024-10-30]
     container:
       image: cubejs/rust-builder:bookworm-llvm-18
     env:
@@ -229,7 +229,7 @@ jobs:
       - name: Setup Rust toolchain
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          toolchain: nightly-2024-01-29
+          toolchain: nightly-2024-10-30
           target: ${{ matrix.target }}
           # override: true # this is by default on
           rustflags: ""

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 {