clippy

crates/arroyo-connectors/src/blackhole/mod.rs

@@ -1,6 +1,5 @@
 use crate::blackhole::operator::BlackholeSinkFunc;
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::{

crates/arroyo-connectors/src/confluent/mod.rs

@@ -3,7 +3,6 @@ use crate::kafka::{
 };
 use crate::{kafka, pull_opt};
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::{

crates/arroyo-connectors/src/filesystem/delta.rs

@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::Connection;
 use arroyo_storage::BackendConfig;
 use std::collections::HashMap;

crates/arroyo-connectors/src/filesystem/mod.rs

@@ -3,7 +3,6 @@ mod sink;
 mod source;
 
 use anyhow::{anyhow, bail, Result};
-use arrow::datatypes::DataType;
 use arroyo_storage::BackendConfig;
 use regex::Regex;
 use std::collections::HashMap;

crates/arroyo-connectors/src/fluvio/mod.rs

@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;

crates/arroyo-connectors/src/impulse/mod.rs

@@ -1,7 +1,6 @@
 mod operator;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::FieldType::Primitive;

crates/arroyo-connectors/src/kafka/source/test.rs

@@ -18,7 +18,7 @@ use arroyo_operator::operator::SourceOperator;
 use arroyo_rpc::df::ArroyoSchema;
 use arroyo_rpc::formats::{Format, RawStringFormat};
 use arroyo_rpc::grpc::rpc::{CheckpointMetadata, OperatorCheckpointMetadata, OperatorMetadata};
-use arroyo_rpc::{CheckpointCompleted, ControlMessage, ControlResp};
+use arroyo_rpc::{CheckpointCompleted, ControlMessage, ControlResp, MetadataField};
 use arroyo_types::{
     single_item_hash_map, to_micros, ArrowMessage, CheckpointBarrier, SignalMessage, TaskInfo,
 };
@@ -87,7 +87,7 @@ impl KafkaTopicTester {
             schema_resolver: None,
             client_configs: HashMap::new(),
             messages_per_second: NonZeroU32::new(100).unwrap(),
-            metadata_fields: None,
+            metadata_fields: vec![],
         });
 
         let (to_control_tx, control_rx) = channel(128);
@@ -358,10 +358,10 @@ async fn test_kafka_with_metadata_fields() {
     kafka_topic_tester.create_topic().await;
 
     // Prepare metadata fields
-    let metadata_fields = Some(HashMap::from([(
-        "offset".to_string(),
-        "offset_id".to_string(),
-    )]));
+    let metadata_fields = vec![MetadataField {
+        field_name: "offset".to_string(),
+        key: "offset_id".to_string(),
+    }];
 
     // Set metadata fields in KafkaSourceFunc
     let mut kafka = KafkaSourceFunc {

crates/arroyo-connectors/src/kinesis/mod.rs

@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail, Result};
-use arrow::datatypes::DataType;
 use std::collections::HashMap;
 use typify::import_types;
 

crates/arroyo-connectors/src/mqtt/source/test.rs

@@ -125,7 +125,7 @@ impl MqttTopicTester {
             None,
             None,
             10,
-            None,
+            vec![],
         );
 
         let (to_control_tx, control_rx) = channel(128);

crates/arroyo-connectors/src/nats/mod.rs

@@ -3,7 +3,6 @@ use crate::nats::source::NatsSourceFunc;
 use crate::pull_opt;
 use anyhow::anyhow;
 use anyhow::bail;
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;

crates/arroyo-connectors/src/nexmark/mod.rs

@@ -3,7 +3,7 @@ mod operator;
 mod test;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
+use arrow::datatypes::{Field, Schema, TimeUnit};
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::{

crates/arroyo-connectors/src/polling_http/mod.rs

@@ -5,7 +5,6 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_rpc::{var_str::VarStr, OperatorConfig};
 use arroyo_types::string_to_map;
 use reqwest::{Client, Request};

crates/arroyo-connectors/src/preview/mod.rs

@@ -3,7 +3,6 @@ mod operator;
 use std::collections::HashMap;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_rpc::OperatorConfig;
 
 use arroyo_operator::connector::Connection;

crates/arroyo-connectors/src/redis/mod.rs

@@ -1,7 +1,6 @@
 mod operator;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;

crates/arroyo-connectors/src/single_file/mod.rs

@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail, Result};
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use std::collections::HashMap;
 use typify::import_types;

crates/arroyo-connectors/src/sse/mod.rs

@@ -4,7 +4,6 @@ use std::collections::HashMap;
 use std::time::Duration;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_rpc::{var_str::VarStr, OperatorConfig};
 use arroyo_types::string_to_map;
 use eventsource_client::Client;

crates/arroyo-connectors/src/stdout/mod.rs

@@ -3,7 +3,6 @@ mod operator;
 use std::collections::HashMap;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_rpc::OperatorConfig;
 use tokio::io::BufWriter;
 

crates/arroyo-connectors/src/webhook/mod.rs

@@ -5,7 +5,6 @@ use std::sync::Arc;
 use std::time::SystemTime;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_rpc::OperatorConfig;
 
 use arroyo_formats::ser::ArrowSerializer;

crates/arroyo-connectors/src/websocket/mod.rs

@@ -3,7 +3,6 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::Connection;
 use arroyo_rpc::api_types::connections::{
     ConnectionProfile, ConnectionSchema, ConnectionType, TestSourceMessage,

crates/arroyo-operator/src/connector.rs

@@ -4,9 +4,8 @@ use arrow::datatypes::{DataType, Field};
 use arroyo_rpc::api_types::connections::{
     ConnectionProfile, ConnectionSchema, ConnectionType, TestSourceMessage,
 };
-use arroyo_rpc::{primitive_to_sql, OperatorConfig};
+use arroyo_rpc::OperatorConfig;
 use arroyo_types::DisplayAsSql;
-use datafusion::sql::unparser::expr_to_sql;
 use serde::de::DeserializeOwned;
 use serde::ser::Serialize;
 use serde_json::value::Value;

crates/arroyo-planner/src/tables.rs

@@ -4,7 +4,6 @@ use std::{collections::HashMap, time::Duration};
 
 use arrow_schema::{DataType, Field, FieldRef, Schema};
 use arroyo_connectors::connector_for_type;
-use datafusion::logical_expr::expr::ScalarFunction;
 
 use crate::extension::remote_table::RemoteTableExtension;
 use crate::types::convert_data_type;
@@ -54,15 +53,14 @@ use datafusion::optimizer::unwrap_cast_in_comparison::UnwrapCastInComparison;
 use datafusion::optimizer::OptimizerRule;
 use datafusion::sql::planner::PlannerContext;
 use datafusion::sql::sqlparser;
-use datafusion::sql::sqlparser::ast::{FunctionArg, FunctionArguments, Query};
+use datafusion::sql::sqlparser::ast::Query;
 use datafusion::{
     optimizer::{optimizer::Optimizer, OptimizerContext},
     sql::{
         planner::SqlToRel,
         sqlparser::ast::{ColumnDef, ColumnOption, Statement, Value},
     },
 };
-use syn::Meta;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct ConnectorTable {
@@ -281,8 +279,6 @@ impl ConnectorTable {
                     ))
                 })?;
 
-                println!("setting metadata field for {} {:?}", struct_field.name(), f);
-
                 if let Some(key) = f.metadata_key() {
                     sf.metadata_key = Some(key.to_string());
                 }

crates/arroyo-rpc/src/api_types/connections.rs

@@ -281,7 +281,6 @@ impl ConnectionSchema {
             .iter()
             .filter(|f| f.metadata_key.is_none())
             .collect();
-        println!("non metadata fields = {:?}", non_metadata_fields);
 
         match &self.format {
             Some(Format::RawString(_)) => {

crates/arroyo-types/src/lib.rs

@@ -480,7 +480,7 @@ impl<'a> Display for DisplayAsSql<'a> {
             }
             DataType::Utf8 | DataType::LargeUtf8 => write!(f, "TEXT"),
             DataType::List(inner) => {
-                write!(f, "{}[]", DisplayAsSql(&inner.data_type()))
+                write!(f, "{}[]", DisplayAsSql(inner.data_type()))
             }
             dt => write!(f, "{}", dt),
         }