fn read_filter(
        &self,
        mut ctx: IOxSessionContext,
        predicate: &Predicate,
        selection: Selection<'_>,
    ) -> Result<SendableRecordBatchStream, QueryChunkError> {
        ctx.set_metadata("storage", "ingester");
        ctx.set_metadata("projection", format!("{}", selection));
        trace!(?selection, "selection");

        // Get all record batches from their snapshots
        let mut batches = vec![];
        for snapshot in &self.data {
            // Only return columns in the selection
            let batch = snapshot.scan(selection).context(FilterColumnsSnafu {})?;
            if let Some(batch) = batch {
                batches.push(batch);
            }
        }

        // Combine record batches into one batch and padding null values as needed
        // Schema of all record batches after mergeing
        let schema = merge_record_batch_schemas(&batches);
        let batch =
            merge_record_batches(schema.as_arrow(), batches).context(ConcatBatchesSnafu {})?;

        let mut stream_batches = vec![];
        if let Some(mut batch) = batch {
            // Apply predicate to filter data
            if let Some(filter_expr) = predicate.filter_expr() {
                // Since the predicate may include columns that the batches do not have,
                // we need to rewrite the predicate to replace those column names with NULL
                let mut rewriter = MissingColumnsToNull::new(&schema);
                let filter_expr = filter_expr
                    .rewrite(&mut rewriter)
                    .context(RewritingFilterPredicateSnafu {})?;
                let df_phy_expr =
                    df_physical_expr_from_schema_and_expr(schema.as_arrow(), filter_expr)
                        .context(ToPhysicalExprSnafu)?;
                let num_rows_before = batch.num_rows();
                batch = batch_filter(&batch, &df_phy_expr).context(FilterBatchSnafu)?;
                let num_rows_after = batch.num_rows();
                debug!(
                    ?num_rows_before,
                    ?num_rows_after,
                    "predicate pushdown for QueryableBatch"
                );
            }

            if batch.num_rows() > 0 {
                stream_batches.push(Arc::new(batch));
            }
        }

        // Return stream of data
        let dummy_metrics = ExecutionPlanMetricsSet::new();
        let mem_metrics = MemTrackingMetrics::new(&dummy_metrics, 0);
        let stream = SizedRecordBatchStream::new(schema.as_arrow(), stream_batches, mem_metrics);
        Ok(Box::pin(stream))
    }
/// Provides access to raw `QueryChunk` data as an
/// asynchronous stream of `RecordBatch`es filtered by a *required*
/// predicate. Note that not all chunks can evaluate all types of
/// predicates and this function will return an error
/// if requested to evaluate with a predicate that is not supported
///
/// This is the analog of the `TableProvider` in DataFusion
///
/// The reason we can't simply use the `TableProvider` trait
/// directly is that the data for a particular Table lives in
/// several chunks within a partition, so there needs to be an
/// implementation of `TableProvider` that stitches together the
/// streams from several different `QueryChunk`s.
Provides access to raw `QueryChunk` data as an
asynchronous stream of `RecordBatch`es filtered by a *required
predicate. Note that not all chunks can evaluate all types of
predicates and this function will return an error
if requested to evaluate with a predicate that is not supported

This is the analog of the `TableProvider` in DataFusion


Provides access to raw ` QueryChunk ` data as an asynchronous stream of ` RecordBatch ` es filtered by a * required predicate . Note that not all chunks can evaluate all types of predicates and this function will return an error if requested to evaluate with a predicate that is not supported This is the analog of the ` TableProvider ` in DataFusion
fn read_filter(
        &self,
        mut ctx: IOxSessionContext,
        predicate: &Predicate,
        selection: Selection<'_>,
    ) -> Result<SendableRecordBatchStream, QueryChunkError> {
        ctx.set_metadata("storage", "ingester");
        ctx.set_metadata("projection", format!("{}", selection));
        trace!(?selection, "selection");
        // Get all record batches from their snapshots
        let mut batches = vec![];
        for snapshot in &self.data {
            // Only return columns in the selection
            let batch = snapshot.scan(selection).context(FilterColumnsSnafu {})?;
            if let Some(batch) = batch {
                batches.push(batch);
            }
        }
        // Combine record batches into one batch and padding null values as needed
        // Schema of all record batches after mergeing
        let schema = merge_record_batch_schemas(&batches);
        let batch =
            merge_record_batches(schema.as_arrow(), batches).context(ConcatBatchesSnafu {})?;
        let mut stream_batches = vec![];
        if let Some(mut batch) = batch {
            // Apply predicate to filter data
            if let Some(filter_expr) = predicate.filter_expr() {
                // Since the predicate may include columns that the batches do not have,
                // we need to rewrite the predicate to replace those column names with NULL
                let mut rewriter = MissingColumnsToNull::new(&schema);
                let filter_expr = filter_expr
                    .rewrite(&mut rewriter)
                    .context(RewritingFilterPredicateSnafu {})?;
                let df_phy_expr =
                    df_physical_expr_from_schema_and_expr(schema.as_arrow(), filter_expr)
                        .context(ToPhysicalExprSnafu)?;
                let num_rows_before = batch.num_rows();
                batch = batch_filter(&batch, &df_phy_expr).context(FilterBatchSnafu)?;
                let num_rows_after = batch.num_rows();
                debug!(
                    ?num_rows_before,
                    ?num_rows_after,
                    "predicate pushdown for QueryableBatch"
                );
            }
            if batch.num_rows() > 0 {
                stream_batches.push(Arc::new(batch));
            }
        }
        // Return stream of data
        let dummy_metrics = ExecutionPlanMetricsSet::new();
        let mem_metrics = MemTrackingMetrics::new(&dummy_metrics, 0);
        let stream = SizedRecordBatchStream::new(schema.as_arrow(), stream_batches, mem_metrics);
        Ok(Box::pin(stream))
    }
fn read_filter ( & self , mut ctx : IOxSessionContext , predicate : & Predicate , selection : Selection < ' _ > , ) -> Result < SendableRecordBatchStream , QueryChunkError > { ctx . set_metadata ( " " , " " ) ; ctx . set_metadata ( " " , format ! ( " " selection ) ) ; trace ! ( selection " " ) ; // Get all record batches from their snapshots let mut batches = vec ! [ ] ; for snapshot in & self . data { // Only return columns in the selection let batch = snapshot . scan ( selection ) . context ( FilterColumnsSnafu { } ) ? ; if let Some ( batch ) = batch { batches . push ( batch ) ; } } // Combine record batches into one batch and padding null values as needed // Schema of all record batches after mergeing let schema = merge_record_batch_schemas ( & batches ) ; let batch = merge_record_batches ( schema . as_arrow ( ) , batches ) . context ( ConcatBatchesSnafu { } ) ? ; let mut stream_batches = vec ! [ ] ; if let Some ( mut batch ) = batch { // Apply predicate to filter data if let Some ( filter_expr ) = predicate . filter_expr ( ) { // Since the predicate may include columns that the batches do not have, // we need to rewrite the predicate to replace those column names with NULL let mut rewriter = MissingColumnsToNull :: new ( & schema ) ; let filter_expr = filter_expr . rewrite ( & mut rewriter ) . context ( RewritingFilterPredicateSnafu { } ) ? ; let df_phy_expr = df_physical_expr_from_schema_and_expr ( schema . as_arrow ( ) , filter_expr ) . context ( ToPhysicalExprSnafu ) ? ; let num_rows_before = batch . num_rows ( ) ; batch = batch_filter ( & batch , & df_phy_expr ) . context ( FilterBatchSnafu ) ? ; let num_rows_after = batch . num_rows ( ) ; debug ! ( num_rows_before num_rows_after " " ) ; } if batch . num_rows ( ) > 0 { stream_batches . push ( Arc :: new ( batch ) ) ; } } // Return stream of data let dummy_metrics = ExecutionPlanMetricsSet :: new ( ) ; let mem_metrics = MemTrackingMetrics :: new ( & dummy_metrics , 0 ) ; let stream = SizedRecordBatchStream :: new ( schema . as_arrow ( ) , stream_batches , mem_metrics ) ; Ok ( Box :: pin ( stream ) ) }
Provides access to raw `QueryChunk` data as an
asynchronous stream of `RecordBatch`es filtered by a *required
predicate.
Provides access to raw ` QueryChunk ` data as an asynchronous stream of ` RecordBatch ` es filtered by a * required predicate .
&self
ctx
IOxSessionContext
predicate
&Predicate
selection
Selection<'_>
&self
ctx
IOxSessionContext
predicate
&Predicate
selection
Selection<'_>