/// Return data to send as a response back to the Querier per its request
Return data to send as a response back to the Querier per its request
Return data to send as a response back to the Querier per its request
pub async fn prepare_data_to_querier(
    ingest_data: &Arc<IngesterData>,
    request: &IngesterQueryRequest,
) -> Result<IngesterQueryResponse> {
    debug!(?request, "prepare_data_to_querier");
    let mut unpersisted_partitions = BTreeMap::new();
    let mut found_namespace = false;
    let mut batches = vec![];
    let mut batch_partition_ids = vec![];
    for (sequencer_id, sequencer_data) in ingest_data.sequencers() {
        debug!(sequencer_id=%sequencer_id.get());
        let namespace_data = match sequencer_data.namespace(&request.namespace) {
            Some(namespace_data) => {
                debug!(namespace=%request.namespace, "found namespace");
                found_namespace = true;
                namespace_data
            }
            None => {
                continue;
            }
        };
        let table_data = match namespace_data.table_data(&request.table) {
            Some(table_data) => {
                debug!(table_name=%request.table, "found table");
                table_data
            }
            None => {
                continue;
            }
        };
        let unpersisted_partition_data = {
            let table_data = table_data.read().await;
            table_data.unpersisted_partition_data()
        };
        debug!(?unpersisted_partition_data);
        for partition in unpersisted_partition_data {
            // include partition in `unpersisted_partitions` even when there we might filter out all the data, because
            // the metadata (e.g. max persisted parquet file) is important for the querier.
            unpersisted_partitions
                .insert(partition.partition_id, partition.partition_status.clone());
            // extract payload
            let partition_id = partition.partition_id;
            let maybe_batch =
                prepare_data_to_querier_for_partition(ingest_data.exec(), partition, request)
                    .await?;
            if let Some(batch) = maybe_batch {
                batches.push(Arc::new(batch));
                batch_partition_ids.push(partition_id);
            }
        }
    }
    ensure!(
        found_namespace,
        NamespaceNotFoundSnafu {
            namespace_name: &request.namespace,
        },
    );
    debug!(?unpersisted_partitions);
    ensure!(
        !unpersisted_partitions.is_empty(),
        TableNotFoundSnafu {
            namespace_name: &request.namespace,
            table_name: &request.table
        },
    );
    debug!(
        num_batches=%batches.len(),
        table_name=%request.table,
        "prepare_data_to_querier found batches"
    );
    // ------------------------------------------------
    // ensure consistent record batch schemas
    //
    // This is required to be able to transmit the batches via Arrow Flight.
    //
    // Note that we need one batch per partition -- not a single batch for all of them -- because the querier also mixes
    // parquet data in and needs to have per-partition batches for de-duplication.
    let schema = merge_record_batch_schemas(&batches);
    let batches = batches
        .into_iter()
        .map(|batch| merge_record_batches(schema.as_arrow(), vec![batch]).transpose().expect("Batch should not be empty/non-existing at this point because we have ruled that out earlier.").map(Arc::new))
        .collect::<Result<Vec<Arc<RecordBatch>>, ArrowError>>().context(InterPartitionSchemaApplicationSnafu)?;
    // ------------------------------------------------
    // Make a stream for this batch
    let dummy_metrics = ExecutionPlanMetricsSet::new();
    let mem_metrics = MemTrackingMetrics::new(&dummy_metrics, 0);
    let stream = SizedRecordBatchStream::new(schema.as_arrow(), batches, mem_metrics);
    Ok(IngesterQueryResponse::new(
        Box::pin(stream),
        schema,
        unpersisted_partitions,
        batch_partition_ids,
    ))
}
pub async fn prepare_data_to_querier ( ingest_data : & Arc < IngesterData > , request : & IngesterQueryRequest , ) -> Result < IngesterQueryResponse > { debug ! ( request " " ) ; let mut unpersisted_partitions = BTreeMap :: new ( ) ; let mut found_namespace = false ; let mut batches = vec ! [ ] ; let mut batch_partition_ids = vec ! [ ] ; for ( sequencer_id , sequencer_data ) in ingest_data . sequencers ( ) { debug ! ( sequencer_id sequencer_id get ( ) ) ; let namespace_data = match sequencer_data . namespace ( & request . namespace ) { Some ( namespace_data ) => { debug ! ( namespace request namespace " " ) ; found_namespace = true ; namespace_data } None => { continue ; } } ; let table_data = match namespace_data . table_data ( & request . table ) { Some ( table_data ) => { debug ! ( table_name request table " " ) ; table_data } None => { continue ; } } ; let unpersisted_partition_data = { let table_data = table_data . read ( ) . await ; table_data . unpersisted_partition_data ( ) } ; debug ! ( unpersisted_partition_data ) ; for partition in unpersisted_partition_data { // include partition in `unpersisted_partitions` even when there we might filter out all the data, because // the metadata (e.g. max persisted parquet file) is important for the querier. unpersisted_partitions . insert ( partition . partition_id , partition . partition_status . clone ( ) ) ; // extract payload let partition_id = partition . partition_id ; let maybe_batch = prepare_data_to_querier_for_partition ( ingest_data . exec ( ) , partition , request ) . await ? ; if let Some ( batch ) = maybe_batch { batches . push ( Arc :: new ( batch ) ) ; batch_partition_ids . push ( partition_id ) ; } } } ensure ! ( found_namespace NamespaceNotFoundSnafu { namespace_name request namespace } ) ; debug ! ( unpersisted_partitions ) ; ensure ! ( unpersisted_partitions is_empty ( ) TableNotFoundSnafu { namespace_name request namespace table_name request table } ) ; debug ! ( num_batches batches len ( ) table_name request table " " ) ; // ------------------------------------------------ // ensure consistent record batch schemas // // This is required to be able to transmit the batches via Arrow Flight. // // Note that we need one batch per partition -- not a single batch for all of them -- because the querier also mixes // parquet data in and needs to have per-partition batches for de-duplication. let schema = merge_record_batch_schemas ( & batches ) ; let batches = batches . into_iter ( ) . map ( | batch | merge_record_batches ( schema . as_arrow ( ) , vec ! [ batch ] ) . transpose ( ) . expect ( " " ) . map ( Arc :: new ) ) . collect :: < Result < Vec < Arc < RecordBatch > > , ArrowError > > ( ) . context ( InterPartitionSchemaApplicationSnafu ) ? ; // ------------------------------------------------ // Make a stream for this batch let dummy_metrics = ExecutionPlanMetricsSet :: new ( ) ; let mem_metrics = MemTrackingMetrics :: new ( & dummy_metrics , 0 ) ; let stream = SizedRecordBatchStream :: new ( schema . as_arrow ( ) , batches , mem_metrics ) ; Ok ( IngesterQueryResponse :: new ( Box :: pin ( stream ) , schema , unpersisted_partitions , batch_partition_ids , ) ) }
Return data to send as a response back to the Querier per its request
Return data to send as a response back to the Querier per its request
ingest_data
&Arc<IngesterData>
request
&IngesterQueryRequest