chore(cubestore): Upgrade DF: Return correct EmissionType values in Cube ExecutionPlans

Author: srh

rust/cubestore/cubestore/src/queryplanner/panic.rs

@@ -122,7 +122,7 @@ impl PanicWorkerExec {
             properties: PlanProperties::new(
                 EquivalenceProperties::new(Arc::new(Schema::empty())),
                 Partitioning::UnknownPartitioning(1),
-                EmissionType::Both, // Well, neither.
+                EmissionType::Incremental,  // Not really applicable.
                 Boundedness::Bounded,
             ),
         }

rust/cubestore/cubestore/src/queryplanner/providers/query_cache.rs

@@ -70,7 +70,7 @@ impl TableProvider for InfoSchemaQueryCacheTableProvider {
             properties: PlanProperties::new(
                 EquivalenceProperties::new(schema),
                 Partitioning::UnknownPartitioning(1),
-                EmissionType::Both, // TODO upgrade DF: which?
+                EmissionType::Final,
                 Boundedness::Bounded,
             ),
         };

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

@@ -1052,11 +1052,6 @@ impl ExecutionPlan for CubeTableExec {
         self.schema.clone()
     }
 
-    // TODO upgrade DF
-    // fn output_partitioning(&self) -> Partitioning {
-    //     Partitioning::UnknownPartitioning(self.partition_execs.len())
-    // }
-
     fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
         self.partition_execs.iter().collect()
     }
@@ -1347,7 +1342,7 @@ impl ClusterSendExec {
         PlanProperties::new(
             eq_properties,
             Partitioning::UnknownPartitioning(partitions_num),
-            EmissionType::Both, // Or Final, but we should implement streaming.
+            EmissionType::Final, // We should implement streaming.
             input_properties.boundedness.clone(),
         )
     }

rust/cubestore/cubestore/src/queryplanner/topk/execute.rs

@@ -84,7 +84,7 @@ impl AggregateTopKExec {
         let cache = PlanProperties::new(
             EquivalenceProperties::new(schema.clone()),
             Partitioning::UnknownPartitioning(1),
-            EmissionType::Both, // TODO upgrade DF
+            EmissionType::Final,
             Boundedness::Bounded,
         );