chore(cubestore): Upgrade DF: Reimplement and use PreSerializedPlan::…

…remove_unused_tables
cube-js · Dec 17, 2024 · 783c645 · 783c645
1 parent 7956b2a
commit 783c645
Show file tree

Hide file tree

Showing 3 changed files with 425 additions and 392 deletions.
diff --git a/rust/cubestore/cubestore/src/queryplanner/query_executor.rs b/rust/cubestore/cubestore/src/queryplanner/query_executor.rs
@@ -1504,21 +1504,21 @@ impl ClusterSendExec {
         }
     }
 
-    pub fn worker_plans(&self) -> Vec<(String, PreSerializedPlan)> {
+    pub fn worker_plans(&self) -> Result<Vec<(String, PreSerializedPlan)>, CubeError> {
         let mut res = Vec::new();
         for (node_name, partitions) in self.partitions.iter() {
             res.push((
                 node_name.clone(),
-                self.serialized_plan_for_partitions(partitions),
+                self.serialized_plan_for_partitions(partitions)?,
             ));
         }
-        res
+        Ok(res)
     }
 
     fn serialized_plan_for_partitions(
         &self,
         partitions: &(Vec<(u64, RowRange)>, Vec<InlineTableId>),
-    ) -> PreSerializedPlan {
+    ) -> Result<PreSerializedPlan, CubeError> {
         let (partitions, inline_table_ids) = partitions;
         let mut ps = HashMap::<_, RowFilter>::new();
         for (id, range) in partitions {
@@ -1577,7 +1577,7 @@ impl ExecutionPlan for ClusterSendExec {
     ) -> Result<SendableRecordBatchStream, DataFusionError> {
         let (node_name, partitions) = &self.partitions[partition];
 
-        let plan = self.serialized_plan_for_partitions(partitions);
+        let plan = self.serialized_plan_for_partitions(partitions)?;
 
         let cluster = self.cluster.clone();
         let schema = self.properties.eq_properties.schema().clone();