avoid some clones

Author: alamb

datafusion/core/src/datasource/physical_plan/parquet/access_plan.rs

@@ -191,8 +191,8 @@ impl ParquetAccessPlan {
     /// Note there is no entry for the (entirely) skipped row group 1.
     ///
     /// [`ArrowReaderBuilder::with_row_selection`]: parquet::arrow::arrow_reader::ArrowReaderBuilder::with_row_selection
-    pub fn overall_row_selection(
-        &self,
+    pub fn into_overall_row_selection(
+        self,
         row_group_meta_data: &[RowGroupMetaData],
     ) -> Option<RowSelection> {
         assert_eq!(row_group_meta_data.len(), self.row_groups.len());
@@ -206,7 +206,7 @@ impl ParquetAccessPlan {
 
         let total_selection: RowSelection = self
             .row_groups
-            .iter()
+            .into_iter()
             .zip(row_group_meta_data.iter())
             .flat_map(|(rg, rg_meta)| {
                 match rg {
@@ -216,8 +216,7 @@ impl ParquetAccessPlan {
                         vec![RowSelector::select(rg_meta.num_rows() as usize)]
                     }
                     RowGroupAccess::Selection(selection) => {
-                        // todo avoid these clones
-                        let selection: Vec<RowSelector> = selection.clone().into();
+                        let selection: Vec<RowSelector> = selection.into();
                         selection
                     }
                 }
@@ -353,7 +352,7 @@ mod test {
         row_group_access: Vec<RowGroupAccess>,
     ) -> Option<RowSelection> {
         let access_plan = ParquetAccessPlan::new(row_group_access);
-        access_plan.overall_row_selection(row_group_metadata())
+        access_plan.into_overall_row_selection(row_group_metadata())
     }
 
     static ROW_GROUP_METADATA: OnceLock<Vec<RowGroupMetaData>> = OnceLock::new();

datafusion/core/src/datasource/physical_plan/parquet/opener.rs

@@ -184,7 +184,10 @@ impl FileOpener for ParquetOpener {
                 }
             }
 
-            if let Some(row_selection) = access_plan.overall_row_selection(rg_metadata) {
+            let row_group_indexes = access_plan.row_group_indexes();
+            if let Some(row_selection) =
+                access_plan.into_overall_row_selection(rg_metadata)
+            {
                 builder = builder.with_row_selection(row_selection);
             }
 
@@ -195,7 +198,7 @@ impl FileOpener for ParquetOpener {
             let stream = builder
                 .with_projection(mask)
                 .with_batch_size(batch_size)
-                .with_row_groups(access_plan.row_group_indexes())
+                .with_row_groups(row_group_indexes)
                 .build()?;
 
             let adapted = stream