chore: Clippy fixes (#182)

Signed-off-by: Anush008 <[email protected]>

Author: Anush008

src/base/json_schema.rs

@@ -47,7 +47,7 @@ impl ToJsonSchema for schema::BasicValueType {
                 }));
                 schema
                     .metadata
-                    .get_or_insert_with(|| Default::default())
+                    .get_or_insert_with(Default::default)
                     .description =
                     Some("A range, start pos (inclusive), end pos (exclusive).".to_string());
             }
@@ -80,7 +80,8 @@ impl ToJsonSchema for schema::StructSchema {
                 required: self
                     .fields
                     .iter()
-                    .filter_map(|f| (!f.value_type.nullable).then(|| f.name.to_string()))
+                    .filter(|&f| (!f.value_type.nullable))
+                    .map(|f| f.name.to_string())
                     .collect(),
                 additional_properties: Some(Schema::Bool(false).into()),
                 ..Default::default()

src/base/schema.rs

@@ -178,7 +178,7 @@ impl CollectionSchema {
         }
     }
 
-    pub fn key_field<'a>(&'a self) -> Option<&'a FieldSchema> {
+    pub fn key_field(&self) -> Option<&FieldSchema> {
         match self.kind {
             CollectionKind::Table => Some(self.row.fields.first().unwrap()),
             CollectionKind::Collection | CollectionKind::List => None,

src/base/spec.rs

@@ -132,12 +132,7 @@ impl std::fmt::Display for ValueMapping {
                     .unwrap_or_else(|_| "#(invalid json value)".to_string())
             ),
             ValueMapping::Field(v) => {
-                write!(
-                    f,
-                    "{}.{}",
-                    v.scope.as_ref().map(|s| s.as_str()).unwrap_or(""),
-                    v.field_path
-                )
+                write!(f, "{}.{}", v.scope.as_deref().unwrap_or(""), v.field_path)
             }
             ValueMapping::Struct(v) => write!(
                 f,

src/base/value.rs

@@ -405,7 +405,7 @@ impl From<KeyValue> for Value {
             KeyValue::Int64(v) => Value::Basic(BasicValue::Int64(v)),
             KeyValue::Range(v) => Value::Basic(BasicValue::Range(v)),
             KeyValue::Struct(v) => Value::Struct(FieldValues {
-                fields: v.into_iter().map(|k| Value::from(k)).collect(),
+                fields: v.into_iter().map(Value::from).collect(),
             }),
         }
     }
@@ -463,11 +463,9 @@ impl<VS> Value<VS> {
                     .map(|v| Value::<VS>::from_alternative_ref(v))
                     .collect(),
             }),
-            Value::Collection(v) => Value::Collection(v.into_iter().map(|v| v.into()).collect()),
-            Value::Table(v) => {
-                Value::Table(v.into_iter().map(|(k, v)| (k.clone(), v.into())).collect())
-            }
-            Value::List(v) => Value::List(v.into_iter().map(|v| v.into()).collect()),
+            Value::Collection(v) => Value::Collection(v.iter().map(|v| v.into()).collect()),
+            Value::Table(v) => Value::Table(v.iter().map(|(k, v)| (k.clone(), v.into())).collect()),
+            Value::List(v) => Value::List(v.iter().map(|v| v.into()).collect()),
         }
     }
 
@@ -669,7 +667,7 @@ where
                 if v.len() != fields_schema.len() {
                     api_bail!("unmatched value length");
                 }
-                Self::from_json_values(fields_schema.iter().zip(v.into_iter()))
+                Self::from_json_values(fields_schema.iter().zip(v))
             }
             serde_json::Value::Object(v) => Self::from_json_object(v, fields_schema.iter()),
             _ => api_bail!("invalid value type"),
@@ -741,7 +739,7 @@ impl BasicValue {
             ) => {
                 let vec = v
                     .into_iter()
-                    .map(|v| BasicValue::from_json(v, &element_type))
+                    .map(|v| BasicValue::from_json(v, element_type))
                     .collect::<Result<Vec<_>>>()?;
                 BasicValue::Vector(Arc::from(vec))
             }
@@ -774,7 +772,7 @@ impl serde::Serialize for Value<ScopeValue> {
     }
 }
 
-impl<'a> serde::Serialize for TableEntry<'a> {
+impl serde::Serialize for TableEntry<'_> {
     fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
         let &TableEntry(key, value) = self;
         let mut seq = serializer.serialize_seq(Some(value.0.fields.len() + 1))?;
@@ -873,7 +871,7 @@ pub struct TypedValue<'a> {
     pub v: &'a Value,
 }
 
-impl<'a> Serialize for TypedValue<'a> {
+impl Serialize for TypedValue<'_> {
     fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
         match (self.t, self.v) {
             (ValueType::Basic(_), v) => v.serialize(serializer),

src/builder/analyzed_flow.rs

@@ -83,9 +83,9 @@ impl AnalyzedTransientFlow {
             analyzer::analyze_transient_flow(&transient_flow, &ctx, registry)?;
         Ok(Self {
             transient_flow_instance: transient_flow,
-            data_schema: data_schema,
+            data_schema,
             execution_plan: execution_plan_fut.await?,
-            output_type: output_type,
+            output_type,
         })
     }
 }

src/builder/analyzer.rs

@@ -107,7 +107,7 @@ impl TryInto<StructSchema> for &StructSchemaBuilder {
             fields: Arc::new(
                 self.fields
                     .iter()
-                    .map(|f| FieldSchema::<ValueType>::from_alternative(f))
+                    .map(FieldSchema::<ValueType>::from_alternative)
                     .collect::<Result<Vec<_>>>()?,
             ),
         })
@@ -306,7 +306,7 @@ impl CollectorBuilder {
             api_bail!("Collector is already used");
         }
         let common_schema =
-            try_make_common_struct_schemas(&self.schema, &schema).with_context(|| {
+            try_make_common_struct_schemas(&self.schema, schema).with_context(|| {
                 format!(
                     "Collectors are sent with entries in incompatible schemas:\n  {}\n  {}\n",
                     self.schema, schema
@@ -377,7 +377,7 @@ impl DataScopeBuilder {
             i += 1;
 
             struct_schema = match &field.value_type.typ {
-                ValueTypeBuilder::Struct(struct_type) => &struct_type,
+                ValueTypeBuilder::Struct(struct_type) => struct_type,
                 _ => {
                     api_bail!("Field {} is not a struct", field_path[0..(i + 1)].join("."));
                 }
@@ -508,7 +508,7 @@ fn analyze_value_mapping(
             (
                 AnalyzedValueMapping::Field(AnalyzedFieldReference {
                     local: local_field_ref,
-                    scope_up_level: scope_up_level as u32,
+                    scope_up_level,
                 }),
                 EnrichedValueType::from_alternative(value_type)?,
             )
@@ -559,7 +559,7 @@ fn add_collector(
     })
 }
 
-impl<'a> AnalyzerContext<'a> {
+impl AnalyzerContext<'_> {
     pub(super) fn analyze_source_op(
         &self,
         scope: &mut DataScopeBuilder,
@@ -592,8 +592,7 @@ impl<'a> AnalyzerContext<'a> {
         let source_id = metadata.map(|metadata| {
             let existing_source_ids = existing_source_states
                 .iter()
-                .map(|v| v.iter())
-                .flatten()
+                .flat_map(|v| v.iter())
                 .filter_map(|state| {
                     if state.key_schema == key_schema_no_attrs {
                         Some(state.source_id)
@@ -741,7 +740,7 @@ impl<'a> AnalyzerContext<'a> {
                     self.analyze_op_scope(
                         &mut exec_scope,
                         &op.op_scope.ops,
-                        parent_scopes.prepend(&scope),
+                        parent_scopes.prepend(scope),
                     )?
                 };
                 let op_name = reactive_op.name.clone();
@@ -873,8 +872,7 @@ impl<'a> AnalyzerContext<'a> {
             .map(|setup_state| -> Result<i32> {
                 let existing_target_ids = existing_target_states
                     .iter()
-                    .map(|v| v.iter())
-                    .flatten()
+                    .flat_map(|v| v.iter())
                     .map(|state| state.common.target_id)
                     .collect::<HashSet<_>>();
                 let target_id = if existing_target_ids.len() == 1 {
@@ -888,15 +886,13 @@ impl<'a> AnalyzerContext<'a> {
                 };
                 let max_schema_version_id = existing_target_states
                     .iter()
-                    .map(|v| v.iter())
-                    .flatten()
+                    .flat_map(|v| v.iter())
                     .map(|s| s.common.max_schema_version_id)
                     .max()
                     .unwrap_or(0);
                 let reusable_schema_version_ids = existing_target_states
                     .iter()
-                    .map(|v| v.iter())
-                    .flatten()
+                    .flat_map(|v| v.iter())
                     .map(|s| {
                         Ok({
                             if export_factory.will_keep_all_existing_data(
@@ -1010,8 +1006,7 @@ pub fn analyze_flow(
     let existing_metadata_versions = || {
         existing_flow_ss
             .iter()
-            .map(|flow_ss| flow_ss.metadata.possible_versions())
-            .flatten()
+            .flat_map(|flow_ss| flow_ss.metadata.possible_versions())
     };
 
     let mut source_states_by_name = HashMap::<&str, Vec<&SourceSetupState>>::new();
@@ -1037,8 +1032,7 @@ pub fn analyze_flow(
 
     let mut setup_state = setup::FlowSetupState::<setup::DesiredMode> {
         seen_flow_metadata_version: existing_flow_ss
-            .map(|flow_ss| flow_ss.seen_flow_metadata_version)
-            .flatten(),
+            .and_then(|flow_ss| flow_ss.seen_flow_metadata_version),
         metadata: FlowSetupMetadata {
             last_source_id: existing_metadata_versions()
                 .map(|metadata| metadata.last_source_id)
@@ -1052,14 +1046,13 @@ pub fn analyze_flow(
         },
         tracking_table: db_tracking_setup::TrackingTableSetupState {
             table_name: existing_flow_ss
-                .map(|flow_ss| {
+                .and_then(|flow_ss| {
                     flow_ss
                         .tracking_table
                         .current
                         .as_ref()
                         .map(|v| v.table_name.clone())
                 })
-                .flatten()
                 .unwrap_or_else(|| db_tracking_setup::default_tracking_table_name(&flow_inst.name)),
             version_id: db_tracking_setup::CURRENT_TRACKING_TABLE_VERSION,
         },
@@ -1078,7 +1071,7 @@ pub fn analyze_flow(
             .map(|source_op| {
                 let existing_source_states = source_states_by_name.get(source_op.name.as_str());
                 analyzer_ctx.analyze_source_op(
-                    &mut root_exec_scope.data,
+                    root_exec_scope.data,
                     source_op.clone(),
                     Some(&mut setup_state.metadata),
                     existing_source_states,
@@ -1095,7 +1088,7 @@ pub fn analyze_flow(
             .iter()
             .map(|export_op| {
                 analyzer_ctx.analyze_export_op(
-                    &mut root_exec_scope.data,
+                    root_exec_scope.data,
                     export_op.clone(),
                     Some(&mut setup_state),
                     &target_states_by_name_type,

src/builder/flow_builder.rs

@@ -85,13 +85,13 @@ impl DataScopeRef {
                 if let Some(child) = child {
                     DataScopeRef(child)
                 } else {
-                    let new_scope = self.make_child_scope(&entry.key())?;
+                    let new_scope = self.make_child_scope(entry.key())?;
                     entry.insert(Arc::downgrade(&new_scope.0));
                     new_scope
                 }
             }
             Entry::Vacant(entry) => {
-                let new_scope = self.make_child_scope(&entry.key())?;
+                let new_scope = self.make_child_scope(entry.key())?;
                 entry.insert(Arc::downgrade(&new_scope.0));
                 new_scope
             }
@@ -118,7 +118,7 @@ impl DataScopeRef {
             .1
             .value_type
             .typ;
-        for field in (&field_path[1..]).iter() {
+        for field in field_path[1..].iter() {
             let struct_builder = match field_typ {
                 ValueTypeBuilder::Struct(struct_type) => struct_type,
                 _ => bail!("expect struct type"),
@@ -250,10 +250,9 @@ impl DataSlice {
             spec::ValueMapping::Field(v) => &v.field_path,
             _ => return Err(PyException::new_err("expect field path")),
         };
-        Ok(self
-            .scope
+        self.scope
             .get_child_scope(field_path.clone())
-            .into_py_result()?)
+            .into_py_result()
     }
 }
 
@@ -420,10 +419,10 @@ impl FlowBuilder {
         Ok(result)
     }
 
-    pub fn constant<'py>(
+    pub fn constant(
         &self,
         value_type: py::Pythonized<schema::EnrichedValueType>,
-        value: Bound<'py, PyAny>,
+        value: Bound<'_, PyAny>,
     ) -> PyResult<DataSlice> {
         let schema = value_type.into_inner();
         let value = py::value_from_py_object(&schema.typ, &value)?;
@@ -502,7 +501,7 @@ impl FlowBuilder {
                 let _ = analyzer_ctx.analyze_reactive_op(scope, &reactive_op, parent_scopes)?;
 
                 reactive_ops.push(reactive_op);
-                let result = Self::last_field_to_data_slice(&scope.data, common_scope.clone())
+                let result = Self::last_field_to_data_slice(scope.data, common_scope.clone())
                     .into_py_result()?;
                 Ok(result)
             },
@@ -706,10 +705,10 @@ impl std::fmt::Display for FlowBuilder {
             )?;
         }
         for field in self.direct_input_fields.iter() {
-            write!(f, "Direct input {}: {}\n", field.name, field.value_type)?;
+            writeln!(f, "Direct input {}: {}", field.name, field.value_type)?;
         }
         if !self.direct_input_fields.is_empty() {
-            write!(f, "\n")?;
+            writeln!(f)?;
         }
         for op in self.reactive_ops.iter() {
             write!(

src/execution/db_tracking.rs

@@ -91,7 +91,7 @@ pub async fn precommit_source_tracking_info(
         .bind(source_key_json) // $2
         .bind(max_process_ordinal) // $3
         .bind(sqlx::types::Json(staging_target_keys)) // $4
-        .bind(memoization_info.map(|m| sqlx::types::Json(m))) // $5
+        .bind(memoization_info.map(sqlx::types::Json)) // $5
         .execute(db_executor)
         .await?;
     Ok(())

src/execution/db_tracking_setup.rs

@@ -101,15 +101,15 @@ impl ResourceSetupStatusCheck for TrackingTableSetupStatusCheck {
 
     fn describe_changes(&self) -> Vec<String> {
         let mut changes: Vec<String> = vec![];
-        if self.desired_state.is_some() && self.legacy_table_names.len() > 0 {
+        if self.desired_state.is_some() && !self.legacy_table_names.is_empty() {
             changes.push(format!(
                 "Rename legacy tracking tables: {}. ",
                 self.legacy_table_names.join(", ")
             ));
         }
         match (self.min_existing_version_ids, &self.desired_state) {
             (None, Some(state)) => {
-                changes.push(format!("Create the tracking table: {}. ", state.table_name).into())
+                changes.push(format!("Create the tracking table: {}. ", state.table_name))
             }
             (Some(min_version_id), Some(desired)) => {
                 if min_version_id < desired.version_id {
@@ -139,7 +139,7 @@ impl ResourceSetupStatusCheck for TrackingTableSetupStatusCheck {
         match (self.min_existing_version_ids, &self.desired_state) {
             (None, Some(_)) => SetupChangeType::Create,
             (Some(min_version_id), Some(desired)) => {
-                if min_version_id == desired.version_id && self.legacy_table_names.len() == 0 {
+                if min_version_id == desired.version_id && self.legacy_table_names.is_empty() {
                     SetupChangeType::NoChange
                 } else if min_version_id < desired.version_id {
                     SetupChangeType::Update