fmt

Author: adriangb

datafusion/datasource-parquet/src/opener.rs

@@ -186,15 +186,15 @@ impl FileOpener for ParquetOpener {
                     if let Ok(rewritten) = filter_rewriter
                         .rewrite(Arc::clone(&file_schema), original_predicate)
                     {
-                        let mut filter_schema_builder = FilterSchemaBuilder::new(
-                            &file_schema,
-                            &table_schema,
-                        );
+                        let mut filter_schema_builder =
+                            FilterSchemaBuilder::new(&file_schema, &table_schema);
                         rewritten.visit(&mut filter_schema_builder)?;
                         filter_schema = filter_schema_builder.build();
                         // If we rewrote the filter we need to recompute the pruning predicates to match the new filter.
-                        page_pruning_predicate =
-                            Some(build_page_pruning_predicate(&rewritten, &filter_schema));
+                        page_pruning_predicate = Some(build_page_pruning_predicate(
+                            &rewritten,
+                            &filter_schema,
+                        ));
                         pruning_predicate = build_pruning_predicate(
                             rewritten.clone(),
                             &filter_schema,
@@ -409,24 +409,33 @@ impl<'schema> FilterSchemaBuilder<'schema> {
         }
     }
 
-    fn sort_fields(fields: &mut Vec<Arc<Field>>, table_schema: &Schema, file_schema: &Schema) {
+    fn sort_fields(
+        fields: &mut Vec<Arc<Field>>,
+        table_schema: &Schema,
+        file_schema: &Schema,
+    ) {
         fields.sort_by_key(|f| f.name().to_string());
         fields.dedup_by_key(|f| f.name().to_string());
         fields.sort_by_key(|f| {
-            let table_schema_index = table_schema.index_of(f.name()).unwrap_or(std::usize::MAX);
-            let file_schema_index = file_schema.index_of(f.name()).unwrap_or(std::usize::MAX);
+            let table_schema_index =
+                table_schema.index_of(f.name()).unwrap_or(std::usize::MAX);
+            let file_schema_index =
+                file_schema.index_of(f.name()).unwrap_or(std::usize::MAX);
             (table_schema_index, file_schema_index)
         });
     }
 
     fn build(self) -> SchemaRef {
         let mut fields = self.filter_schema_fields.into_iter().collect::<Vec<_>>();
-        FilterSchemaBuilder::sort_fields(&mut fields, self.table_schema, self.file_schema);
+        FilterSchemaBuilder::sort_fields(
+            &mut fields,
+            self.table_schema,
+            self.file_schema,
+        );
         Arc::new(Schema::new(fields))
     }
 }
 
-
 impl<'node> TreeNodeVisitor<'node> for FilterSchemaBuilder<'_> {
     type Node = Arc<dyn PhysicalExpr>;
 
@@ -441,19 +450,33 @@ impl<'node> TreeNodeVisitor<'node> for FilterSchemaBuilder<'_> {
                 self.filter_schema_fields.insert(Arc::new(field.clone()));
             } else {
                 // valid fields are the table schema's fields + the file schema's fields, preferring the table schema's fields when there is a conflict
-                let mut valid_fields = self.table_schema.fields().iter().chain(self.file_schema.fields().iter()).cloned().collect::<Vec<_>>();
-                FilterSchemaBuilder::sort_fields(&mut valid_fields, self.table_schema, self.file_schema);
-                let valid_fields = valid_fields.into_iter().map(|f| datafusion_common::Column::new_unqualified(f.name())).collect();
+                let mut valid_fields = self
+                    .table_schema
+                    .fields()
+                    .iter()
+                    .chain(self.file_schema.fields().iter())
+                    .cloned()
+                    .collect::<Vec<_>>();
+                FilterSchemaBuilder::sort_fields(
+                    &mut valid_fields,
+                    self.table_schema,
+                    self.file_schema,
+                );
+                let valid_fields = valid_fields
+                    .into_iter()
+                    .map(|f| datafusion_common::Column::new_unqualified(f.name()))
+                    .collect();
                 let field = datafusion_common::Column::new_unqualified(column.name());
-                return Err(
-                    datafusion_common::DataFusionError::SchemaError(
-                        SchemaError::FieldNotFound { field: Box::new(field), valid_fields },
-                        Box::new(None),
-                    )
-                )
+                return Err(datafusion_common::DataFusionError::SchemaError(
+                    SchemaError::FieldNotFound {
+                        field: Box::new(field),
+                        valid_fields,
+                    },
+                    Box::new(None),
+                ));
             }
         }
 
         Ok(TreeNodeRecursion::Continue)
     }
-}
+}

datafusion/datasource-parquet/src/row_filter.rs

@@ -274,7 +274,7 @@ impl<'a> FilterCandidateBuilder<'a> {
 
         let required_bytes = size_of_columns(&required_indices, metadata)?;
         let can_use_index = columns_sorted(&required_indices, metadata)?;
-    
+
         Ok(Some(FilterCandidate {
             expr: rewritten_expr,
             required_bytes,