async fn init_file(
        &self,
        partition: usize,
    ) -> DataFusionResult<Arc<SerializedFileReader<CachedFile>>> {
        let end_dl_chunk_start = Self::download_footer(self.files[partition].clone());
        let file_schema = self.file_schema.clone();
        let file = self.files[partition].clone();
        let projection = self.projection.clone();

        // Reading the footer is blocking so it should be started on a specific thread
        tokio::task::spawn_blocking(move || {
            let file_reader = Arc::new(
                SerializedFileReader::new(file.clone())
                    .map_err(|e| DataFusionError::ParquetError(e))?,
            );
            let mut arrow_reader = ParquetFileArrowReader::new(file_reader.clone());

            let schema_eq = file_schema
                .fields()
                .iter()
                .zip(arrow_reader.get_schema()?.fields().iter())
                .all(|x| {
                    x.0.name() == x.1.name()
                        && x.0.data_type() == x.1.data_type()
                        && x.0.is_nullable() == x.1.is_nullable()
                });

            if !schema_eq {
                return Err(DataFusionError::Plan(format!(
                    "Expected and parsed schema fields are not equal: \n{:?} \n!= \n{:?}",
                    file_schema.fields(),
                    arrow_reader.get_schema()?.fields()
                )));
            }
            // prefetch usefull byte ranges
            let metadata = file_reader.metadata();
            for i in 0..metadata.num_row_groups() {
                for proj in &projection {
                    let rg_metadata = metadata.row_group(i);
                    let col_metadata = rg_metadata.column(*proj);
                    let (start, length) = col_metadata.byte_range();
                    if start < end_dl_chunk_start {
                        file.prefetch(start, length as usize);
                    }
                }
            }
            Ok(file_reader)
        })
        .await
        .unwrap()
    }
/// Read the footer and schedule the downloads of all the required chunks
Read the footer and schedule the downloads of all the required chunks
Read the footer and schedule the downloads of all the required chunks
async fn init_file(
        &self,
        partition: usize,
    ) -> DataFusionResult<Arc<SerializedFileReader<CachedFile>>> {
        let end_dl_chunk_start = Self::download_footer(self.files[partition].clone());
        let file_schema = self.file_schema.clone();
        let file = self.files[partition].clone();
        let projection = self.projection.clone();
        // Reading the footer is blocking so it should be started on a specific thread
        tokio::task::spawn_blocking(move || {
            let file_reader = Arc::new(
                SerializedFileReader::new(file.clone())
                    .map_err(|e| DataFusionError::ParquetError(e))?,
            );
            let mut arrow_reader = ParquetFileArrowReader::new(file_reader.clone());
            let schema_eq = file_schema
                .fields()
                .iter()
                .zip(arrow_reader.get_schema()?.fields().iter())
                .all(|x| {
                    x.0.name() == x.1.name()
                        && x.0.data_type() == x.1.data_type()
                        && x.0.is_nullable() == x.1.is_nullable()
                });
            if !schema_eq {
                return Err(DataFusionError::Plan(format!(
                    "Expected and parsed schema fields are not equal: \n{:?} \n!= \n{:?}",
                    file_schema.fields(),
                    arrow_reader.get_schema()?.fields()
                )));
            }
            // prefetch usefull byte ranges
            let metadata = file_reader.metadata();
            for i in 0..metadata.num_row_groups() {
                for proj in &projection {
                    let rg_metadata = metadata.row_group(i);
                    let col_metadata = rg_metadata.column(*proj);
                    let (start, length) = col_metadata.byte_range();
                    if start < end_dl_chunk_start {
                        file.prefetch(start, length as usize);
                    }
                }
            }
            Ok(file_reader)
        })
        .await
        .unwrap()
    }
async fn init_file ( & self , partition : usize , ) -> DataFusionResult < Arc < SerializedFileReader < CachedFile > > > { let end_dl_chunk_start = Self :: download_footer ( self . files [ partition ] . clone ( ) ) ; let file_schema = self . file_schema . clone ( ) ; let file = self . files [ partition ] . clone ( ) ; let projection = self . projection . clone ( ) ; // Reading the footer is blocking so it should be started on a specific thread tokio :: task :: spawn_blocking ( move | | { let file_reader = Arc :: new ( SerializedFileReader :: new ( file . clone ( ) ) . map_err ( | e | DataFusionError :: ParquetError ( e ) ) ? , ) ; let mut arrow_reader = ParquetFileArrowReader :: new ( file_reader . clone ( ) ) ; let schema_eq = file_schema . fields ( ) . iter ( ) . zip ( arrow_reader . get_schema ( ) ? . fields ( ) . iter ( ) ) . all ( | x | { x . 0 . name ( ) == x . 1 . name ( ) && x . 0 . data_type ( ) == x . 1 . data_type ( ) && x . 0 . is_nullable ( ) == x . 1 . is_nullable ( ) } ) ; if ! schema_eq { return Err ( DataFusionError :: Plan ( format ! ( " \n \n \n " file_schema fields ( ) arrow_reader get_schema ( ) fields ( ) ) ) ) ; } // prefetch usefull byte ranges let metadata = file_reader . metadata ( ) ; for i in 0 .. metadata . num_row_groups ( ) { for proj in & projection { let rg_metadata = metadata . row_group ( i ) ; let col_metadata = rg_metadata . column ( * proj ) ; let ( start , length ) = col_metadata . byte_range ( ) ; if start < end_dl_chunk_start { file . prefetch ( start , length as usize ) ; } } } Ok ( file_reader ) } ) . await . unwrap ( ) }
Read the footer and schedule the downloads of all the required chunks
Read the footer and schedule the downloads of all the required chunks
&self
partition
usize
&self
partition
usize