style: apply rustfmt across workspace crates

Run rustfmt on CLI, coordinator, catalog_storage, streaming_planner, streaming_runtime, and wasm_runtime so cargo fmt --check passes for the main binary and CLI packages.

Author: luoluoyuyu

cli/cli/src/repl.rs

@@ -17,7 +17,7 @@ use arrow_ipc::reader::StreamReader;
 use arrow_schema::DataType;
 use comfy_table::presets::UTF8_FULL;
 use comfy_table::{Attribute, Cell, Color, ContentArrangement, Table, TableComponent};
-use protocol::cli::{function_stream_service_client::FunctionStreamServiceClient, SqlRequest};
+use protocol::cli::{SqlRequest, function_stream_service_client::FunctionStreamServiceClient};
 use rustyline::error::ReadlineError;
 use rustyline::{Config, DefaultEditor, EditMode};
 use std::fmt;
@@ -243,11 +243,7 @@ impl Repl {
             }
         }
 
-        if has_rows {
-            Ok(Some(table))
-        } else {
-            Ok(None)
-        }
+        if has_rows { Ok(Some(table)) } else { Ok(None) }
     }
 
     fn extract_value(&self, column: &dyn Array, row: usize) -> String {
@@ -317,7 +313,7 @@ impl Repl {
 
         #[cfg(unix)]
         let mut sigterm = {
-            use tokio::signal::unix::{signal, SignalKind};
+            use tokio::signal::unix::{SignalKind, signal};
             signal(SignalKind::terminate()).expect("failed to register SIGTERM handler")
         };
 
@@ -403,9 +399,7 @@ impl Repl {
             println!();
         }
 
-        if !skip_save_history
-            && let Some(ref mut ed) = repl.lock().await.editor
-        {
+        if !skip_save_history && let Some(ref mut ed) = repl.lock().await.editor {
             let _ = ed.save_history(".function-stream-cli-history");
         }
         Ok(())

src/catalog_storage/src/task/rocksdb_storage.rs

@@ -19,8 +19,8 @@ use super::proto_codec::{
     encode_task_module_bytes,
 };
 use super::storage::{StoredTaskInfo, TaskStorage};
-use crate::config::storage::RocksDBStorageConfig;
 use crate::common::ComponentState;
+use crate::config::storage::RocksDBStorageConfig;
 use anyhow::{Context, Result, anyhow};
 use rocksdb::{ColumnFamilyDescriptor, DB, IteratorMode, Options, WriteBatch};
 use std::path::Path;

src/coordinator/src/coordinator_body.rs

@@ -18,8 +18,8 @@ mod execution;
 mod execution_context;
 mod plan;
 mod runtime_context;
-mod statement;
 mod sql_classify;
+mod statement;
 mod streaming_table_options;
 mod tool;
 

src/coordinator/src/execution/executor.rs

@@ -31,13 +31,13 @@ use crate::coordinator::statement::{ConfigSource, FunctionSource};
 use crate::coordinator::streaming_table_options::{
     parse_checkpoint_interval_ms, parse_pipeline_parallelism,
 };
-use crate::streaming::job::JobManager;
-use crate::streaming::protocol::control::StopMode;
-use crate::wasm::taskexecutor::TaskManager;
 use crate::sql::schema::catalog::ExternalTable;
 use crate::sql::schema::show_create_catalog_table;
 use crate::sql::schema::table::CatalogEntity;
 use crate::stream_catalog::CatalogManager;
+use crate::streaming::job::JobManager;
+use crate::streaming::protocol::control::StopMode;
+use crate::wasm::taskexecutor::TaskManager;
 
 #[derive(Error, Debug)]
 pub enum ExecuteError {

src/coordinator/src/runtime_context.rs

@@ -16,10 +16,10 @@ use std::sync::Arc;
 
 use anyhow::Result;
 
-use crate::streaming::job::JobManager;
-use crate::wasm::taskexecutor::TaskManager;
 use crate::sql::schema::StreamSchemaProvider;
 use crate::stream_catalog::CatalogManager;
+use crate::streaming::job::JobManager;
+use crate::wasm::taskexecutor::TaskManager;
 
 /// Dependencies shared by analyze / plan / execute, analogous to installing globals in
 /// [`TaskManager`], [`CatalogManager`], and [`JobManager`].

src/coordinator/src/sql_classify.rs

@@ -15,7 +15,7 @@
 
 use std::collections::HashMap;
 
-use datafusion::common::{plan_err, Result};
+use datafusion::common::{Result, plan_err};
 use datafusion::error::DataFusionError;
 use datafusion::sql::sqlparser::ast::{
     ObjectType, ShowCreateObject, SqlOption, Statement as DFStatement,
@@ -50,9 +50,9 @@ pub fn classify_statement(stmt: DFStatement) -> Result<Box<dyn Statement>> {
         DFStatement::ShowStreamingTable => Ok(Box::new(ShowStreamingTables::new())),
         DFStatement::ShowCreate { obj_type, obj_name } => match obj_type {
             ShowCreateObject::Table => Ok(Box::new(ShowCreateTable::new(obj_name.to_string()))),
-            ShowCreateObject::StreamingTable => {
-                Ok(Box::new(ShowCreateStreamingTable::new(obj_name.to_string())))
-            }
+            ShowCreateObject::StreamingTable => Ok(Box::new(ShowCreateStreamingTable::new(
+                obj_name.to_string(),
+            ))),
             _ => plan_err!(
                 "SHOW CREATE {obj_type} is not supported; use SHOW CREATE TABLE or SHOW CREATE STREAMING TABLE <name>"
             ),
@@ -88,11 +88,12 @@ pub fn classify_statement(stmt: DFStatement) -> Result<Box<dyn Statement>> {
                     }
                     let table_name = names[0].to_string();
                     Ok(Box::new(DropStreamingTableStatement::new(
-                        table_name,
-                        *if_exists,
+                        table_name, *if_exists,
                     )))
                 }
-                _ => plan_err!("Only DROP TABLE and DROP STREAMING TABLE are supported in this SQL frontend"),
+                _ => plan_err!(
+                    "Only DROP TABLE and DROP STREAMING TABLE are supported in this SQL frontend"
+                ),
             }
         }
         DFStatement::Insert { .. } => plan_err!(

src/streaming_planner/src/common/mod.rs

@@ -33,8 +33,8 @@ pub mod topology;
 pub mod with_option_keys;
 
 // ── Re-exports from existing modules ──
-pub use function_stream_runtime_common::streaming_protocol::{CheckpointBarrier, Watermark};
 pub use arrow_ext::FsExtensionType;
+pub use function_stream_runtime_common::streaming_protocol::{CheckpointBarrier, Watermark};
 pub use time_utils::{from_nanos, to_micros, to_millis, to_nanos};
 
 // ── Re-exports from new modules ──

src/streaming_planner/src/connector/sink/runtime_config.rs

@@ -14,12 +14,12 @@ use std::collections::HashMap;
 
 use datafusion::common::{DataFusionError, Result, plan_err};
 
+use crate::common::connector_options::ConnectorOptions;
+use crate::common::with_option_keys as opt;
 use function_stream_config::global_config::{
     DEFAULT_OPERATOR_STATE_STORE_MEMORY_BYTES, DEFAULT_SINK_BUFFER_MEMORY_BYTES,
 };
 use function_stream_config::streaming_job::DEFAULT_CHECKPOINT_INTERVAL_MS;
-use crate::common::connector_options::ConnectorOptions;
-use crate::common::with_option_keys as opt;
 
 #[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub struct SinkRuntimeConfig {

src/streaming_planner/src/logical_node/aggregate.rs

@@ -30,14 +30,14 @@ use protocol::function_stream_graph::{
     SessionWindowAggregateOperator, SlidingWindowAggregateOperator, TumblingWindowAggregateOperator,
 };
 
-use crate::multifield_partial_ord;
 use crate::common::constants::{extension_node, proto_operator_name};
 use crate::common::{FsSchema, FsSchemaRef};
 use crate::logical_node::logical::{LogicalEdge, LogicalEdgeType, LogicalNode, OperatorName};
 use crate::logical_node::{
     CompiledTopologyNode, StreamingOperatorBlueprint, SystemTimestampInjectorNode,
 };
 use crate::logical_planner::planner::{NamedNode, Planner, SplitPlanOutput};
+use crate::multifield_partial_ord;
 use crate::physical::{StreamingExtensionCodec, window};
 use crate::types::{
     QualifiedField, TIMESTAMP_FIELD, WindowBehavior, WindowType, build_df_schema,

src/streaming_planner/src/logical_node/async_udf.rs

@@ -24,7 +24,6 @@ use datafusion_proto::physical_plan::to_proto::serialize_physical_expr;
 use prost::Message;
 use protocol::function_stream_graph::{AsyncUdfOperator, AsyncUdfOrdering};
 
-use crate::multifield_partial_ord;
 use crate::common::constants::extension_node;
 use crate::common::constants::sql_field;
 use crate::common::{FsSchema, FsSchemaRef};
@@ -35,6 +34,7 @@ use crate::logical_node::streaming_operator_blueprint::{
     CompiledTopologyNode, StreamingOperatorBlueprint,
 };
 use crate::logical_planner::planner::{NamedNode, Planner};
+use crate::multifield_partial_ord;
 use crate::types::{QualifiedField, build_df_schema, extract_qualified_fields};
 
 pub const NODE_TYPE_NAME: &str = extension_node::ASYNC_FUNCTION_EXECUTION;