Remove redundant code

Author: alamb

datafusion/datasource-parquet/src/mod.rs

@@ -25,7 +25,6 @@ pub mod file_format;
 pub mod metadata;
 mod metrics;
 mod morsel;
-mod opener;
 mod page_filter;
 mod reader;
 mod row_filter;

datafusion/datasource-parquet/src/morsel.rs

@@ -25,8 +25,6 @@ use crate::{
 };
 use arrow::array::{RecordBatch, RecordBatchOptions};
 use arrow::datatypes::{DataType, Schema};
-#[cfg(test)]
-use datafusion_datasource::file_stream::FileOpenFuture;
 use datafusion_datasource::morsel::{Morsel, MorselPlan, MorselPlanner, Morselizer};
 use datafusion_physical_expr::projection::{ProjectionExprs, Projector};
 use datafusion_physical_expr::utils::reassign_expr_columns;
@@ -58,7 +56,6 @@ use datafusion_physical_plan::metrics::{
 };
 use datafusion_pruning::{FilePruner, PruningPredicate, build_pruning_predicate};
 
-use crate::sort::reverse_row_selection;
 #[cfg(feature = "parquet_encryption")]
 use datafusion_common::config::EncryptionFactoryOptions;
 #[cfg(feature = "parquet_encryption")]
@@ -73,7 +70,7 @@ use parquet::arrow::arrow_reader::{
 use parquet::arrow::async_reader::AsyncFileReader;
 use parquet::arrow::push_decoder::{ParquetPushDecoder, ParquetPushDecoderBuilder};
 use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask};
-use parquet::file::metadata::{PageIndexPolicy, ParquetMetaDataReader, RowGroupMetaData};
+use parquet::file::metadata::{PageIndexPolicy, ParquetMetaDataReader};
 use tokio::sync::oneshot;
 
 /// Implements [`Morselizer`] for a parquet file.
@@ -152,12 +149,9 @@ impl ParquetMorselizer {
         }
     }
 
-    #[cfg(test)]
-    fn open_file(&self, partitioned_file: PartitionedFile) -> Result<FileOpenFuture> {
-        self.state.open_file(partitioned_file)
-    }
 }
 
+
 impl Deref for ParquetMorselizer {
     type Target = ParquetMorselizerState;
 
@@ -190,53 +184,6 @@ impl Debug for ParquetMorselizer {
     }
 }
 
-/// Represents a prepared access plan with optional row selection
-pub(crate) struct PreparedAccessPlan {
-    /// Row group indexes to read
-    pub(crate) row_group_indexes: Vec<usize>,
-    /// Optional row selection for filtering within row groups
-    pub(crate) row_selection: Option<parquet::arrow::arrow_reader::RowSelection>,
-}
-
-impl PreparedAccessPlan {
-    /// Create a new prepared access plan from a ParquetAccessPlan
-    pub(crate) fn from_access_plan(
-        access_plan: ParquetAccessPlan,
-        rg_metadata: &[RowGroupMetaData],
-    ) -> Result<Self> {
-        let row_group_indexes = access_plan.row_group_indexes();
-        let row_selection = access_plan.into_overall_row_selection(rg_metadata)?;
-
-        Ok(Self {
-            row_group_indexes,
-            row_selection,
-        })
-    }
-
-    /// Reverse the access plan for reverse scanning
-    pub(crate) fn reverse(
-        mut self,
-        file_metadata: &parquet::file::metadata::ParquetMetaData,
-    ) -> Result<Self> {
-        // Get the row group indexes before reversing
-        let row_groups_to_scan = self.row_group_indexes.clone();
-
-        // Reverse the row group indexes
-        self.row_group_indexes = self.row_group_indexes.into_iter().rev().collect();
-
-        // If we have a row selection, reverse it to match the new row group order
-        if let Some(row_selection) = self.row_selection {
-            self.row_selection = Some(reverse_row_selection(
-                &row_selection,
-                file_metadata,
-                &row_groups_to_scan, // Pass the original (non-reversed) row group indexes
-            )?);
-        }
-
-        Ok(self)
-    }
-}
-
 /// Result of preparing a PartitionedFile using CPU before any I/O.
 ///
 /// This captures the state computed from `PartitionedFile`, the table schema,
@@ -267,40 +214,6 @@ struct MetadataLoadedParquetOpen {
 }
 
 impl ParquetMorselizerState {
-    /// Build a stream for a single parquet file using the copied opener logic.
-    ///
-    /// This remains private so the public integration point for parquet scans is
-    /// the morsel API rather than the legacy `FileOpener` trait.
-    #[cfg(test)]
-    fn open_file(
-        self: &Arc<Self>,
-        partitioned_file: PartitionedFile,
-    ) -> Result<FileOpenFuture> {
-        let prepared = self.prepare_open_file(partitioned_file)?;
-        Ok(Box::pin(async move {
-            #[cfg(feature = "parquet_encryption")]
-            let prepared = {
-                let mut prepared = prepared;
-                if prepared
-                    .state
-                    .get_encryption_context()
-                    .needs_file_decryption_properties()
-                {
-                    prepared.file_decryption_properties = prepared
-                        .state
-                        .load_file_decryption_properties(
-                            prepared.partitioned_file.object_meta.location.clone(),
-                        )
-                        .await?;
-                }
-                prepared
-            };
-            #[cfg(not(feature = "parquet_encryption"))]
-            let prepared = prepared;
-            ParquetMorselizerState::execute_prepared_open(prepared).await
-        }))
-    }
-
     /// Perform the CPU-only setup for opening a parquet file.
     fn prepare_open_file(
         self: &Arc<Self>,
@@ -436,18 +349,6 @@ impl ParquetMorselizerState {
         Ok(None)
     }
 
-    /// Continue opening a parquet file once the CPU-only setup has completed.
-    #[cfg(test)]
-    async fn execute_prepared_open(
-        prepared: PreparedParquetOpen,
-    ) -> Result<BoxStream<'static, Result<RecordBatch>>> {
-        let Some(prepared) = Self::prune_prepared_file(prepared)? else {
-            return Ok(futures::stream::empty().boxed());
-        };
-        let loaded = Self::load_prepared_metadata(prepared).await?;
-        Self::execute_metadata_loaded_open(loaded).await
-    }
-
     /// CPU-only file pruning performed before metadata I/O begins.
     ///
     /// Returns `None` if the file was completely pruned
@@ -727,8 +628,7 @@ impl ParquetMorselizerState {
         }
 
         // Prepare the access plan (extract row groups and row selection)
-        let mut prepared_plan =
-            PreparedAccessPlan::from_access_plan(access_plan, rg_metadata)?;
+        let mut prepared_plan = access_plan.prepare(rg_metadata)?;
 
         // ----------------------------------------------------------
         // Step: potentially reverse the access plan for performance.
@@ -1953,7 +1853,7 @@ mod test {
         let expr = col("a").eq(lit(1));
         let predicate = logical2physical(&expr, &schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 1);
         assert_eq!(num_rows, 3);
@@ -1962,7 +1862,7 @@ mod test {
         let expr = col("b").eq(lit(ScalarValue::Float32(Some(5.0))));
         let predicate = logical2physical(&expr, &schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2008,7 +1908,7 @@ mod test {
         // Otherwise we assume it already happened at the planning stage and won't re-do the work here
         let predicate = make_dynamic_expr(logical2physical(&expr, &table_schema));
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 1);
         assert_eq!(num_rows, 3);
@@ -2019,7 +1919,7 @@ mod test {
         // Otherwise we assume it already happened at the planning stage and won't re-do the work here
         let predicate = make_dynamic_expr(logical2physical(&expr, &table_schema));
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2075,7 +1975,7 @@ mod test {
         let expr = col("part").eq(lit(1)).and(col("b").eq(lit(1.0)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 1);
         assert_eq!(num_rows, 3);
@@ -2084,7 +1984,7 @@ mod test {
         let expr = col("part").eq(lit(2)).and(col("b").eq(lit(1.0)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2093,7 +1993,7 @@ mod test {
         let expr = col("part").eq(lit(1)).and(col("b").eq(lit(7.0)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2102,7 +2002,7 @@ mod test {
         let expr = col("part").eq(lit(2)).and(col("b").eq(lit(7.0)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2148,7 +2048,7 @@ mod test {
         let expr = col("part").eq(lit(1)).or(col("a").eq(lit(1)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 1);
         assert_eq!(num_rows, 3);
@@ -2157,7 +2057,7 @@ mod test {
         let expr = col("part").eq(lit(1)).or(col("a").eq(lit(3)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 1);
         assert_eq!(num_rows, 3);
@@ -2166,7 +2066,7 @@ mod test {
         let expr = col("part").eq(lit(2)).or(col("a").eq(lit(1)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 1);
         assert_eq!(num_rows, 1);
@@ -2175,7 +2075,7 @@ mod test {
         let expr = col("part").eq(lit(2)).or(col("a").eq(lit(3)));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2228,7 +2128,7 @@ mod test {
         let expr = col("a").eq(lit(42));
         let predicate = logical2physical(&expr, &table_schema);
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2237,7 +2137,7 @@ mod test {
         // This allows dynamic filters to prune partitions/files even if they are populated late into execution.
         let predicate = make_dynamic_expr(logical2physical(&expr, &table_schema));
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2247,7 +2147,7 @@ mod test {
         let expr = col("part").eq(lit(2));
         let predicate = make_dynamic_expr(logical2physical(&expr, &table_schema));
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2256,7 +2156,7 @@ mod test {
         let expr = col("part").eq(lit(2)).and(col("a").eq(lit(42)));
         let predicate = make_dynamic_expr(logical2physical(&expr, &table_schema));
         let opener = make_opener(predicate);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let (num_batches, num_rows) = count_batches_and_rows(stream).await;
         assert_eq!(num_batches, 0);
         assert_eq!(num_rows, 0);
@@ -2307,12 +2207,12 @@ mod test {
 
         // Test normal scan (forward)
         let opener = make_opener(false);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let forward_values = collect_int32_values(stream).await;
 
         // Test reverse scan
         let opener = make_opener(true);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let reverse_values = collect_int32_values(stream).await;
 
         // The forward scan should return data in the order written
@@ -2352,7 +2252,7 @@ mod test {
         );
 
         let opener_values =
-            collect_int32_values(opener.open_file(file.clone()).unwrap().await.unwrap())
+            collect_int32_values(open_via_morselizer(&opener, file.clone()).await)
                 .await;
         let morsel_values =
             collect_int32_values(open_via_morselizer(&opener, file).await).await;
@@ -2387,15 +2287,11 @@ mod test {
         // With a single row group, forward and reverse should be the same
         // (only the row group order is reversed, not the rows within)
         let opener_forward = make_opener(false);
-        let stream_forward = opener_forward
-            .open_file(file.clone())
-            .unwrap()
-            .await
-            .unwrap();
+        let stream_forward = open_via_morselizer(&opener_forward, file.clone()).await;
         let (batches_forward, _) = count_batches_and_rows(stream_forward).await;
 
         let opener_reverse = make_opener(true);
-        let stream_reverse = opener_reverse.open_file(file).unwrap().await.unwrap();
+        let stream_reverse = open_via_morselizer(&opener_reverse, file).await;
         let (batches_reverse, _) = count_batches_and_rows(stream_reverse).await;
 
         // Both should have the same number of batches since there's only one row group
@@ -2466,7 +2362,7 @@ mod test {
 
         // Forward scan: RG0(3,4), RG1(5,6,7,8), RG2(9,10)
         let opener = make_opener(false);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let forward_values = collect_int32_values(stream).await;
 
         // Forward scan should produce: RG0(3,4), RG1(5,6,7,8), RG2(9,10)
@@ -2482,7 +2378,7 @@ mod test {
         // - RG1 is read second, WITH RG1's selection (select all) -> 5, 6, 7, 8
         // - RG0 is read third, WITH RG0's selection (skip 2, select 2) -> 3, 4
         let opener = make_opener(true);
-        let stream = opener.open_file(file).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file).await;
         let reverse_values = collect_int32_values(stream).await;
 
         // Correct expected result: row groups reversed but each keeps its own selection
@@ -2568,7 +2464,7 @@ mod test {
         // Forward scan: RG0(1), RG2(5), RG3(7)
         // Note: RG1 is completely skipped
         let opener = make_opener(false);
-        let stream = opener.open_file(file.clone()).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file.clone()).await;
         let forward_values = collect_int32_values(stream).await;
 
         assert_eq!(
@@ -2581,7 +2477,7 @@ mod test {
         // WITHOUT the bug fix, this would return WRONG values
         // because the RowSelection would be incorrectly mapped
         let opener = make_opener(true);
-        let stream = opener.open_file(file).unwrap().await.unwrap();
+        let stream = open_via_morselizer(&opener, file).await;
         let reverse_values = collect_int32_values(stream).await;
 
         assert_eq!(