pub fn read_filter(
        &self,
        _predicate: &Predicate,
        selection: Selection<'_>,
        schema: SchemaRef,
        meta: &IoxMetadata,
    ) -> Result<SendableRecordBatchStream, ReadError> {
        let path = ParquetFilePath::from(meta).object_store_path();
        trace!(path=?path, "fetching parquet data for filtered read");

        // Indices of columns in the schema needed to read
        let projection: Vec<usize> = column_indices(selection, Arc::clone(&schema));

        // Compute final (output) schema after selection
        let schema = Arc::new(Schema::new(
            projection
                .iter()
                .map(|i| schema.field(*i).clone())
                .collect(),
        ));

        let (tx, rx) = tokio::sync::mpsc::channel(2);

        // Run async dance here to make sure any error returned
        // `download_and_scan_parquet` is sent back to the reader and
        // not silently ignored
        let object_store = Arc::clone(&self.object_store);
        let handle = tokio::task::spawn(async move {
            let download_result =
                download_and_scan_parquet(projection, path, object_store, tx.clone()).await;

            // If there was an error returned from download_and_scan_parquet send it back to the receiver.
            if let Err(e) = download_result {
                warn!(error=%e, "Parquet download & scan failed");
                let e = ArrowError::ExternalError(Box::new(e));
                if let Err(e) = tx.send(ArrowResult::Err(e)).await {
                    // if no one is listening, there is no one else to hear our screams
                    debug!(%e, "Error sending result of download function. Receiver is closed.");
                }
            }
        });

        // returned stream simply reads off the rx channel
        Ok(AdapterStream::adapt(
            schema,
            rx,
            Some(Arc::new(AutoAbortJoinHandle::new(handle))),
        ))
    }
/// Pull the Parquet-encoded [`RecordBatch`] at the file path derived from
/// the provided [`ParquetFilePath`].
///
/// The `selection` projection is pushed down to the Parquet deserializer.
///
/// This impl fetches the associated Parquet file bytes from object storage,
/// temporarily persisting them to a local temp file to feed to the arrow
/// reader.
///
/// No caching is performed by `read_filter()`, and each call to
/// `read_filter()` will re-download the parquet file unless the underlying
/// object store impl caches the fetched bytes.
Pull the Parquet-encoded [`RecordBatch`] at the file path derived from
the provided [`ParquetFilePath`].

The `selection` projection is pushed down to the Parquet deserializer.

This impl fetches the associated Parquet file bytes from object storage,
temporarily persisting them to a local temp file to feed to the arrow
reader.

No caching is performed by `read_filter()`, and each call to
`read_filter()` will re-download the parquet file unless the underlying
object store impl caches the fetched bytes.
Pull the Parquet - encoded [ ` RecordBatch ` ] at the file path derived from the provided [ ` ParquetFilePath ` ] . The ` selection ` projection is pushed down to the Parquet deserializer . This impl fetches the associated Parquet file bytes from object storage temporarily persisting them to a local temp file to feed to the arrow reader . No caching is performed by ` read_filter () ` and each call to ` read_filter () ` will re - download the parquet file unless the underlying object store impl caches the fetched bytes .
pub fn read_filter(
        &self,
        _predicate: &Predicate,
        selection: Selection<'_>,
        schema: SchemaRef,
        meta: &IoxMetadata,
    ) -> Result<SendableRecordBatchStream, ReadError> {
        let path = ParquetFilePath::from(meta).object_store_path();
        trace!(path=?path, "fetching parquet data for filtered read");
        // Indices of columns in the schema needed to read
        let projection: Vec<usize> = column_indices(selection, Arc::clone(&schema));
        // Compute final (output) schema after selection
        let schema = Arc::new(Schema::new(
            projection
                .iter()
                .map(|i| schema.field(*i).clone())
                .collect(),
        ));
        let (tx, rx) = tokio::sync::mpsc::channel(2);
        // Run async dance here to make sure any error returned
        // `download_and_scan_parquet` is sent back to the reader and
        // not silently ignored
        let object_store = Arc::clone(&self.object_store);
        let handle = tokio::task::spawn(async move {
            let download_result =
                download_and_scan_parquet(projection, path, object_store, tx.clone()).await;
            // If there was an error returned from download_and_scan_parquet send it back to the receiver.
            if let Err(e) = download_result {
                warn!(error=%e, "Parquet download & scan failed");
                let e = ArrowError::ExternalError(Box::new(e));
                if let Err(e) = tx.send(ArrowResult::Err(e)).await {
                    // if no one is listening, there is no one else to hear our screams
                    debug!(%e, "Error sending result of download function. Receiver is closed.");
                }
            }
        });
        // returned stream simply reads off the rx channel
        Ok(AdapterStream::adapt(
            schema,
            rx,
            Some(Arc::new(AutoAbortJoinHandle::new(handle))),
        ))
    }
pub fn read_filter ( & self , _predicate : & Predicate , selection : Selection < ' _ > , schema : SchemaRef , meta : & IoxMetadata , ) -> Result < SendableRecordBatchStream , ReadError > { let path = ParquetFilePath :: from ( meta ) . object_store_path ( ) ; trace ! ( path path " " ) ; // Indices of columns in the schema needed to read let projection : Vec < usize > = column_indices ( selection , Arc :: clone ( & schema ) ) ; // Compute final (output) schema after selection let schema = Arc :: new ( Schema :: new ( projection . iter ( ) . map ( | i | schema . field ( * i ) . clone ( ) ) . collect ( ) , ) ) ; let ( tx , rx ) = tokio :: sync :: mpsc :: channel ( 2 ) ; // Run async dance here to make sure any error returned // `download_and_scan_parquet` is sent back to the reader and // not silently ignored let object_store = Arc :: clone ( & self . object_store ) ; let handle = tokio :: task :: spawn ( async move { let download_result = download_and_scan_parquet ( projection , path , object_store , tx . clone ( ) ) . await ; // If there was an error returned from download_and_scan_parquet send it back to the receiver. if let Err ( e ) = download_result { warn ! ( error e " " ) ; let e = ArrowError :: ExternalError ( Box :: new ( e ) ) ; if let Err ( e ) = tx . send ( ArrowResult :: Err ( e ) ) . await { // if no one is listening, there is no one else to hear our screams debug ! ( e " " ) ; } } } ) ; // returned stream simply reads off the rx channel Ok ( AdapterStream :: adapt ( schema , rx , Some ( Arc :: new ( AutoAbortJoinHandle :: new ( handle ) ) ) , ) ) }
Pull the Parquet-encoded [`RecordBatch`] at the file path derived from
the provided [`ParquetFilePath`].
Pull the Parquet - encoded [ ` RecordBatch ` ] at the file path derived from the provided [ ` ParquetFilePath ` ] .
&self
_predicate
&Predicate
selection
Selection<'_>
schema
SchemaRef
meta
&IoxMetadata
&self
_predicate
&Predicate
selection
Selection<'_>
schema
SchemaRef
meta
&IoxMetadata