update

Author: luoluoyuyu

src/runtime/streaming/operators/grouping/incremental_aggregate.rs

@@ -465,7 +465,7 @@ impl IncrementalAggregatingFunc {
                         state
                     });
 
-                    for (idx, v) in agg.state_cols.iter().zip(state.into_iter()) {
+                    for (idx, v) in agg.state_cols.iter().zip(state) {
                         states[*idx].push(v);
                     }
                 }
@@ -543,7 +543,7 @@ impl IncrementalAggregatingFunc {
             }
         }
 
-        let mut cols = self.key_converter.convert_rows(rows.into_iter())?;
+        let mut cols = self.key_converter.convert_rows(rows)?;
         cols.push(Arc::new(accumulator_builder.finish()));
         cols.push(Arc::new(args_row_builder.finish()));
         cols.push(Arc::new(count_builder.finish()));
@@ -612,7 +612,7 @@ impl IncrementalAggregatingFunc {
             mem::take(&mut self.updated_keys).into_iter().unzip();
         let mut deleted_keys = vec![];
 
-        for (k, retract) in updated_keys.iter().zip(updated_values.into_iter()) {
+        for (k, retract) in updated_keys.iter().zip(updated_values) {
             let append = self.evaluate(&k.0)?;
 
             if let Some(v) = retract {

src/sql/logical_planner/streaming_planner.rs

@@ -341,7 +341,7 @@ impl PlanToGraphVisitor<'_> {
         let node_index = self.graph.add_node(execution_unit);
         self.add_index_to_traversal(node_index);
 
-        for (source, edge) in input_nodes.into_iter().zip(routing_edges.into_iter()) {
+        for (source, edge) in input_nodes.into_iter().zip(routing_edges) {
             self.graph.add_edge(source, node_index, edge);
         }
 

src/sql/types/data_type.rs

@@ -98,14 +98,11 @@ fn convert_simple_data_type(
             }
         },
         SQLDataType::Date => Ok(DataType::Date32),
-        SQLDataType::Time(None, tz_info) => {
+        SQLDataType::Time(None, tz_info)
             if matches!(tz_info, TimezoneInfo::None)
-                || matches!(tz_info, TimezoneInfo::WithoutTimeZone)
-            {
-                Ok(DataType::Time64(TimeUnit::Nanosecond))
-            } else {
-                return plan_err!("Unsupported SQL type {sql_type:?}");
-            }
+                || matches!(tz_info, TimezoneInfo::WithoutTimeZone) =>
+        {
+            Ok(DataType::Time64(TimeUnit::Nanosecond))
         }
         SQLDataType::Numeric(exact_number_info) | SQLDataType::Decimal(exact_number_info) => {
             let (precision, scale) = match *exact_number_info {

tests/integration/framework/instance.py

@@ -20,8 +20,7 @@
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Optional, Type
 
-from .paths import PathManager
-from .config import InstanceConfig
+from .config import InstanceConfig, PathManager
 from .process import FunctionStreamProcess
 from .utils import find_free_port, wait_for_port
 from .workspace import InstanceWorkspace