refactor: per review

GreptimeTeam · Dec 19, 2024 · 201b683 · 201b683
1 parent 1bc84a9
commit 201b683
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 7 deletions.
diff --git a/src/flow/src/adapter.rs b/src/flow/src/adapter.rs
@@ -565,7 +565,7 @@ impl FlowWorkerManager {
         let default_interval = Duration::from_secs(1);
         let mut avg_spd = 0; // rows/sec
         let mut since_last_run = tokio::time::Instant::now();
-        let run_per_trace = 5;
+        let run_per_trace = 10;
         let mut run_cnt = 0;
         loop {
             // TODO(discord9): only run when new inputs arrive or scheduled to

diff --git a/src/flow/src/compute/render/reduce.rs b/src/flow/src/compute/render/reduce.rs
@@ -47,7 +47,6 @@ impl Context<'_, '_> {
         reduce_plan: &ReducePlan,
         output_type: &RelationType,
     ) -> Result<CollectionBundle<Batch>, Error> {
-        common_telemetry::debug!("render reduce batch");
         let accum_plan = if let ReducePlan::Accumulable(accum_plan) = reduce_plan {
             if !accum_plan.distinct_aggrs.is_empty() {
                 NotImplementedSnafu {

diff --git a/src/flow/src/expr.rs b/src/flow/src/expr.rs
@@ -240,11 +240,8 @@ impl Batch {
                     dts.push(datatypes::prelude::ConcreteDataType::null_datatype())
                 }
             }
-            if self.batch.is_empty() {
-                other.batch.iter().map(|v| v.data_type()).collect_vec()
-            } else {
-                self.batch.iter().map(|v| v.data_type()).collect_vec()
-            }
+
+            dts
         };
 
         let batch_builders = dts