fmt

ArroyoSystems · Nov 14, 2024 · e0e0a4b · e0e0a4b
1 parent eec4f91
commit e0e0a4b
Show file tree

Hide file tree

Showing 28 changed files with 221 additions and 153 deletions.
diff --git a/crates/arroyo-api/src/cloud.rs b/crates/arroyo-api/src/cloud.rs
@@ -1,7 +1,7 @@
-use axum_extra::headers::Authorization;
+use crate::{rest_utils::ErrorResp, AuthData, OrgMetadata};
 use axum_extra::headers::authorization::Bearer;
+use axum_extra::headers::Authorization;
 use axum_extra::TypedHeader;
-use crate::{rest_utils::ErrorResp, AuthData, OrgMetadata};
 use cornucopia_async::Database;
 
 pub(crate) async fn authenticate(

diff --git a/crates/arroyo-api/src/lib.rs b/crates/arroyo-api/src/lib.rs
@@ -3,7 +3,7 @@ use axum::Json;
 use cornucopia_async::DatabaseSource;
 use http::StatusCode;
 use serde::{Deserialize, Serialize};
-use std::net::{SocketAddr};
+use std::net::SocketAddr;
 use time::OffsetDateTime;
 use tokio::net::TcpListener;
 use tonic::transport::Channel;
@@ -138,11 +138,9 @@ pub async fn start_server(database: DatabaseSource, guard: ShutdownGuard) -> any
     );
 
     info!("Starting API server on {:?}", local_addr);
-    guard.into_spawn_task(wrap_start(
-        "api",
-        local_addr,
-        async { axum::serve(listener, app.into_make_service()).await },
-    ));
+    guard.into_spawn_task(wrap_start("api", local_addr, async {
+        axum::serve(listener, app.into_make_service()).await
+    }));
 
     Ok(local_addr.port())
 }

diff --git a/crates/arroyo-api/src/rest_utils.rs b/crates/arroyo-api/src/rest_utils.rs
@@ -3,9 +3,9 @@ use arroyo_server_common::log_event;
 use axum::extract::rejection::JsonRejection;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
-use axum::{Json};
-use axum_extra::headers::Authorization;
+use axum::Json;
 use axum_extra::headers::authorization::Bearer;
+use axum_extra::headers::Authorization;
 use axum_extra::TypedHeader;
 use serde_json::json;
 use tracing::{error, warn};

diff --git a/crates/arroyo-connectors/src/filesystem/sink/mod.rs b/crates/arroyo-connectors/src/filesystem/sink/mod.rs
@@ -21,6 +21,7 @@ use arroyo_storage::StorageProvider;
 use async_trait::async_trait;
 use bincode::{Decode, Encode};
 use chrono::{DateTime, Utc};
+use datafusion::execution::SessionStateBuilder;
 use datafusion::prelude::concat;
 use datafusion::{
     common::{Column, Result as DFResult},
@@ -35,7 +36,6 @@ use datafusion::{
     physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner},
     scalar::ScalarValue,
 };
-use datafusion::execution::SessionStateBuilder;
 use futures::{stream::FuturesUnordered, Future};
 use futures::{stream::StreamExt, TryStreamExt};
 use object_store::{multipart::PartId, path::Path, MultipartId};
@@ -212,7 +212,7 @@ fn partition_string_for_fields_and_time(
 fn compile_expression(expr: &Expr, schema: ArroyoSchemaRef) -> Result<Arc<dyn PhysicalExpr>> {
     let physical_planner = DefaultPhysicalPlanner::default();
     let session_state = SessionStateBuilder::new().build();
-    
+
     let plan = physical_planner.create_physical_expr(
         expr,
         &(schema.schema.as_ref().clone()).try_into()?,

diff --git a/crates/arroyo-connectors/src/kinesis/source.rs b/crates/arroyo-connectors/src/kinesis/source.rs
@@ -337,7 +337,7 @@ impl KinesisSourceFunc {
     }
 
     async fn init_client(&mut self) {
-        let mut loader =  aws_config::defaults(BehaviorVersion::v2024_03_28());
+        let mut loader = aws_config::defaults(BehaviorVersion::v2024_03_28());
         if let Some(region) = &self.aws_region {
             loader = loader.region(Region::new(region.clone()));
         }

diff --git a/crates/arroyo-controller/src/schedulers/kubernetes/mod.rs b/crates/arroyo-controller/src/schedulers/kubernetes/mod.rs
@@ -8,11 +8,11 @@ use arroyo_rpc::grpc::rpc::{HeartbeatNodeReq, RegisterNodeReq, WorkerFinishedReq
 use arroyo_types::{WorkerId, JOB_ID_ENV, RUN_ID_ENV};
 use async_trait::async_trait;
 use k8s_openapi::api::core::v1::Pod;
+use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
 use kube::api::{DeleteParams, ListParams};
 use kube::{Api, Client};
 use serde_json::json;
 use std::time::Duration;
-use k8s_openapi::apimachinery::pkg::api::resource::Quantity;
 use tonic::Status;
 use tracing::{info, warn};
 

diff --git a/crates/arroyo-planner/src/extension/aggregate.rs b/crates/arroyo-planner/src/extension/aggregate.rs
@@ -28,9 +28,14 @@ use datafusion_proto::physical_plan::DefaultPhysicalExtensionCodec;
 use datafusion_proto::{physical_plan::AsExecutionPlan, protobuf::PhysicalPlanNode};
 use prost::Message;
 
-use crate::{builder::{NamedNode, Planner, SplitPlanOutput}, fields_with_qualifiers, multifield_partial_ord, physical::ArroyoPhysicalExtensionCodec, schema_from_df_fields, schema_from_df_fields_with_metadata, DFField, WindowBehavior};
-use crate::physical::window;
 use super::{ArroyoExtension, NodeWithIncomingEdges, TimestampAppendExtension};
+use crate::physical::window;
+use crate::{
+    builder::{NamedNode, Planner, SplitPlanOutput},
+    fields_with_qualifiers, multifield_partial_ord,
+    physical::ArroyoPhysicalExtensionCodec,
+    schema_from_df_fields, schema_from_df_fields_with_metadata, DFField, WindowBehavior,
+};
 
 pub(crate) const AGGREGATE_EXTENSION_NAME: &str = "AggregateExtension";
 
@@ -331,7 +336,7 @@ impl AggregateExtension {
         let timestamp_column =
             Column::new(timestamp_field.qualifier().cloned(), timestamp_field.name());
         aggregate_fields.insert(window_index, window_field.clone());
-        
+
         let window_expression = Expr::ScalarFunction(ScalarFunction {
             func: window(),
             args: vec![

diff --git a/crates/arroyo-planner/src/extension/debezium.rs b/crates/arroyo-planner/src/extension/debezium.rs
@@ -25,7 +25,12 @@ pub struct DebeziumUnrollingExtension {
     primary_key_names: Arc<Vec<String>>,
 }
 
-multifield_partial_ord!(DebeziumUnrollingExtension, input, primary_keys, primary_key_names);
+multifield_partial_ord!(
+    DebeziumUnrollingExtension,
+    input,
+    primary_keys,
+    primary_key_names
+);
 
 impl DebeziumUnrollingExtension {
     pub(crate) fn as_debezium_schema(

diff --git a/crates/arroyo-planner/src/extension/key_calculation.rs b/crates/arroyo-planner/src/extension/key_calculation.rs
@@ -11,7 +11,12 @@ use datafusion::logical_expr::{Expr, LogicalPlan, UserDefinedLogicalNodeCore};
 use datafusion_proto::{physical_plan::AsExecutionPlan, protobuf::PhysicalPlanNode};
 use prost::Message;
 
-use crate::{builder::{NamedNode, Planner}, fields_with_qualifiers, multifield_partial_ord, physical::ArroyoPhysicalExtensionCodec, schema_from_df_fields_with_metadata};
+use crate::{
+    builder::{NamedNode, Planner},
+    fields_with_qualifiers, multifield_partial_ord,
+    physical::ArroyoPhysicalExtensionCodec,
+    schema_from_df_fields_with_metadata,
+};
 
 use super::{ArroyoExtension, NodeWithIncomingEdges};
 

diff --git a/crates/arroyo-planner/src/extension/mod.rs b/crates/arroyo-planner/src/extension/mod.rs
@@ -114,7 +114,6 @@ macro_rules! multifield_partial_ord {
 }
     }
 
-
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub(crate) struct TimestampAppendExtension {
     pub(crate) input: LogicalPlan,
@@ -187,7 +186,17 @@ pub(crate) struct AsyncUDFExtension {
     pub(crate) final_schema: DFSchemaRef,
 }
 
-multifield_partial_ord!(AsyncUDFExtension, input, name, udf, arg_exprs, final_exprs, ordered, max_concurrency, timeout);
+multifield_partial_ord!(
+    AsyncUDFExtension,
+    input,
+    name,
+    udf,
+    arg_exprs,
+    final_exprs,
+    ordered,
+    max_concurrency,
+    timeout
+);
 
 impl ArroyoExtension for AsyncUDFExtension {
     fn node_name(&self) -> Option<NamedNode> {

diff --git a/crates/arroyo-planner/src/extension/remote_table.rs b/crates/arroyo-planner/src/extension/remote_table.rs
@@ -11,7 +11,11 @@ use datafusion::logical_expr::{Expr, LogicalPlan, UserDefinedLogicalNodeCore};
 use datafusion_proto::{physical_plan::AsExecutionPlan, protobuf::PhysicalPlanNode};
 use prost::Message;
 
-use crate::{builder::{NamedNode, Planner}, multifield_partial_ord, physical::ArroyoPhysicalExtensionCodec};
+use crate::{
+    builder::{NamedNode, Planner},
+    multifield_partial_ord,
+    physical::ArroyoPhysicalExtensionCodec,
+};
 
 use super::{ArroyoExtension, NodeWithIncomingEdges};
 

diff --git a/crates/arroyo-planner/src/extension/sink.rs b/crates/arroyo-planner/src/extension/sink.rs
@@ -11,7 +11,11 @@ use datafusion::logical_expr::{Expr, Extension, LogicalPlan, UserDefinedLogicalN
 
 use prost::Message;
 
-use crate::{builder::{NamedNode, Planner}, multifield_partial_ord, tables::Table};
+use crate::{
+    builder::{NamedNode, Planner},
+    multifield_partial_ord,
+    tables::Table,
+};
 
 use super::{
     debezium::ToDebeziumExtension, remote_table::RemoteTableExtension, ArroyoExtension,

diff --git a/crates/arroyo-planner/src/extension/table_source.rs b/crates/arroyo-planner/src/extension/table_source.rs
@@ -10,7 +10,12 @@ use prost::Message;
 
 use super::{ArroyoExtension, DebeziumUnrollingExtension, NodeWithIncomingEdges};
 use crate::tables::FieldSpec;
-use crate::{builder::{NamedNode, Planner}, multifield_partial_ord, schema_from_df_fields, schemas::add_timestamp_field, tables::ConnectorTable};
+use crate::{
+    builder::{NamedNode, Planner},
+    multifield_partial_ord, schema_from_df_fields,
+    schemas::add_timestamp_field,
+    tables::ConnectorTable,
+};
 pub(crate) const TABLE_SOURCE_NAME: &str = "TableSourceExtension";
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]

diff --git a/crates/arroyo-planner/src/extension/watermark_node.rs b/crates/arroyo-planner/src/extension/watermark_node.rs
@@ -1,5 +1,6 @@
 use crate::builder::{NamedNode, Planner};
 use crate::extension::{ArroyoExtension, NodeWithIncomingEdges};
+use crate::multifield_partial_ord;
 use crate::schemas::add_timestamp_field;
 use arroyo_datastream::logical::{LogicalEdge, LogicalEdgeType, LogicalNode, OperatorName};
 use arroyo_rpc::df::{ArroyoSchema, ArroyoSchemaRef};
@@ -12,7 +13,6 @@ use datafusion_proto::physical_plan::DefaultPhysicalExtensionCodec;
 use prost::Message;
 use std::fmt::Formatter;
 use std::sync::Arc;
-use crate::multifield_partial_ord;
 
 pub(crate) const WATERMARK_NODE_NAME: &str = "WatermarkNode";
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -24,7 +24,13 @@ pub struct WatermarkNode {
     timestamp_index: usize,
 }
 
-multifield_partial_ord!(WatermarkNode, input, qualifier, watermark_expression, timestamp_index);
+multifield_partial_ord!(
+    WatermarkNode,
+    input,
+    qualifier,
+    watermark_expression,
+    timestamp_index
+);
 
 impl UserDefinedLogicalNodeCore for WatermarkNode {
     fn name(&self) -> &str {

diff --git a/crates/arroyo-planner/src/functions.rs b/crates/arroyo-planner/src/functions.rs
@@ -66,11 +66,7 @@ pub fn register_all(registry: &mut dyn FunctionRegistry) {
         .register_udf(Arc::new(create_udf(
             "extract_json",
             vec![DataType::Utf8, DataType::Utf8],
-            DataType::List(Arc::new(Field::new(
-                "item",
-                DataType::Utf8,
-                true,
-            ))),
+            DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
             Volatility::Immutable,
             Arc::new(extract_json),
         )))