async fn execute(request: GetPartitionForIngester<'_>) -> Result<Vec<IngesterPartition>> {
    let GetPartitionForIngester {
        flight_client,
        catalog_cache,
        ingester_address,
        namespace_name,
        table_name,
        columns,
        predicate,
        expected_schema,
    } = request;

    let ingester_query_request = IngesterQueryRequest {
        namespace: namespace_name.to_string(),
        table: table_name.to_string(),
        columns: columns.clone(),
        predicate: Some(predicate.clone()),
    };

    let query_res = flight_client
        .query(Arc::clone(&ingester_address), ingester_query_request)
        .await;

    if let Err(FlightClientError::Flight {
        source: FlightError::GrpcError(status),
    }) = &query_res
    {
        if status.code() == tonic::Code::NotFound {
            debug!(
                ingester_address=ingester_address.as_ref(),
                %namespace_name,
                %table_name,
                "Ingester does not know namespace or table, skipping",
            );
            return Ok(vec![]);
        }
    }
    let mut perform_query = query_res
        .context(RemoteQuerySnafu {
            ingester_address: ingester_address.as_ref(),
        })
        .map_err(|e| {
            // generate a warning that is sufficient to replicate the request using CLI tooling
            warn!(
                e=%e,
                ingester_address=ingester_address.as_ref(),
                namespace=namespace_name.as_ref(),
                table=table_name.as_ref(),
                columns=columns.join(",").as_str(),
                predicate_str=%predicate,
                predicate_binary=encode_predicate_as_base64(predicate).as_str(),
                "Failed to perform ingester query",
            );

            //  need to return error until https://github.com/rust-lang/rust/issues/91345 is stable
            e
        })?;

    // read unpersisted partitions
    // map partition_id -> (PartitionMetadata, Vec<PartitionData>))
    let mut partitions: HashMap<_, _> = perform_query
        .app_metadata()
        .unpersisted_partitions
        .iter()
        .map(|(id, state)| (*id, (state.clone(), vec![])))
        .collect();

    // sort batches into partitions
    let mut num_batches = 0usize;
    let partition_ids = perform_query.app_metadata().batch_partition_ids.clone();
    while let Some(batch) = perform_query
        .next()
        .await
        .map_err(|source| FlightClientError::Flight { source })
        .context(RemoteQuerySnafu {
            ingester_address: ingester_address.as_ref(),
        })?
    {
        let partition_id = *partition_ids
            .get(num_batches)
            .context(MissingPartitionIdSnafu {
                pos: num_batches,
                ingester_address: ingester_address.as_ref(),
            })?;
        partitions
            .get_mut(&partition_id)
            .context(UnknownPartitionIdSnafu {
                partition_id,
                ingester_address: ingester_address.as_ref(),
            })?
            .1
            .push(batch);
        num_batches += 1;
    }
    debug!(%ingester_address, num_batches, "Received batches from ingester");
    trace!(?partitions, schema=?perform_query.schema(), "Detailed from ingester");
    ensure!(
        num_batches == partition_ids.len(),
        TooManyPartitionIdsSnafu {
            actual: partition_ids.len(),
            expected: num_batches,
            ingester_address: ingester_address.as_ref(),
        },
    );

    let mut ingester_partitions = vec![];
    for (partition_id, (state, batches)) in partitions {
        // do NOT filter out empty partitions, because the caller of this functions needs the attached metadata
        // to select the right parquet files and tombstones
        let partition_id = PartitionId::new(partition_id);
        let sequencer_id = catalog_cache.partition().sequencer_id(partition_id).await;
        let partition_sort_key = catalog_cache.partition().sort_key(partition_id).await;
        let ingester_partition = IngesterPartition::try_new(
            Arc::clone(&ingester_address),
            ChunkId::new(),
            Arc::clone(&namespace_name),
            Arc::clone(&table_name),
            partition_id,
            sequencer_id,
            Arc::clone(&expected_schema),
            state.parquet_max_sequence_number.map(SequenceNumber::new),
            state.tombstone_max_sequence_number.map(SequenceNumber::new),
            partition_sort_key,
            batches,
        )?;
        ingester_partitions.push(ingester_partition);
    }

    Ok(ingester_partitions)
}
/// Fetches the partitions for a single ingester
Fetches the partitions for a single ingester
Fetches the partitions for a single ingester
async fn execute(request: GetPartitionForIngester<'_>) -> Result<Vec<IngesterPartition>> {
    let GetPartitionForIngester {
        flight_client,
        catalog_cache,
        ingester_address,
        namespace_name,
        table_name,
        columns,
        predicate,
        expected_schema,
    } = request;
    let ingester_query_request = IngesterQueryRequest {
        namespace: namespace_name.to_string(),
        table: table_name.to_string(),
        columns: columns.clone(),
        predicate: Some(predicate.clone()),
    };
    let query_res = flight_client
        .query(Arc::clone(&ingester_address), ingester_query_request)
        .await;
    if let Err(FlightClientError::Flight {
        source: FlightError::GrpcError(status),
    }) = &query_res
    {
        if status.code() == tonic::Code::NotFound {
            debug!(
                ingester_address=ingester_address.as_ref(),
                %namespace_name,
                %table_name,
                "Ingester does not know namespace or table, skipping",
            );
            return Ok(vec![]);
        }
    }
    let mut perform_query = query_res
        .context(RemoteQuerySnafu {
            ingester_address: ingester_address.as_ref(),
        })
        .map_err(|e| {
            // generate a warning that is sufficient to replicate the request using CLI tooling
            warn!(
                e=%e,
                ingester_address=ingester_address.as_ref(),
                namespace=namespace_name.as_ref(),
                table=table_name.as_ref(),
                columns=columns.join(",").as_str(),
                predicate_str=%predicate,
                predicate_binary=encode_predicate_as_base64(predicate).as_str(),
                "Failed to perform ingester query",
            );
            //  need to return error until https://github.com/rust-lang/rust/issues/91345 is stable
            e
        })?;
    // read unpersisted partitions
    // map partition_id -> (PartitionMetadata, Vec<PartitionData>))
    let mut partitions: HashMap<_, _> = perform_query
        .app_metadata()
        .unpersisted_partitions
        .iter()
        .map(|(id, state)| (*id, (state.clone(), vec![])))
        .collect();
    // sort batches into partitions
    let mut num_batches = 0usize;
    let partition_ids = perform_query.app_metadata().batch_partition_ids.clone();
    while let Some(batch) = perform_query
        .next()
        .await
        .map_err(|source| FlightClientError::Flight { source })
        .context(RemoteQuerySnafu {
            ingester_address: ingester_address.as_ref(),
        })?
    {
        let partition_id = *partition_ids
            .get(num_batches)
            .context(MissingPartitionIdSnafu {
                pos: num_batches,
                ingester_address: ingester_address.as_ref(),
            })?;
        partitions
            .get_mut(&partition_id)
            .context(UnknownPartitionIdSnafu {
                partition_id,
                ingester_address: ingester_address.as_ref(),
            })?
            .1
            .push(batch);
        num_batches += 1;
    }
    debug!(%ingester_address, num_batches, "Received batches from ingester");
    trace!(?partitions, schema=?perform_query.schema(), "Detailed from ingester");
    ensure!(
        num_batches == partition_ids.len(),
        TooManyPartitionIdsSnafu {
            actual: partition_ids.len(),
            expected: num_batches,
            ingester_address: ingester_address.as_ref(),
        },
    );
    let mut ingester_partitions = vec![];
    for (partition_id, (state, batches)) in partitions {
        // do NOT filter out empty partitions, because the caller of this functions needs the attached metadata
        // to select the right parquet files and tombstones
        let partition_id = PartitionId::new(partition_id);
        let sequencer_id = catalog_cache.partition().sequencer_id(partition_id).await;
        let partition_sort_key = catalog_cache.partition().sort_key(partition_id).await;
        let ingester_partition = IngesterPartition::try_new(
            Arc::clone(&ingester_address),
            ChunkId::new(),
            Arc::clone(&namespace_name),
            Arc::clone(&table_name),
            partition_id,
            sequencer_id,
            Arc::clone(&expected_schema),
            state.parquet_max_sequence_number.map(SequenceNumber::new),
            state.tombstone_max_sequence_number.map(SequenceNumber::new),
            partition_sort_key,
            batches,
        )?;
        ingester_partitions.push(ingester_partition);
    }
    Ok(ingester_partitions)
}
async fn execute ( request : GetPartitionForIngester < ' _ > ) -> Result < Vec < IngesterPartition > > { let GetPartitionForIngester { flight_client , catalog_cache , ingester_address , namespace_name , table_name , columns , predicate , expected_schema , } = request ; let ingester_query_request = IngesterQueryRequest { namespace : namespace_name . to_string ( ) , table : table_name . to_string ( ) , columns : columns . clone ( ) , predicate : Some ( predicate . clone ( ) ) , } ; let query_res = flight_client . query ( Arc :: clone ( & ingester_address ) , ingester_query_request ) . await ; if let Err ( FlightClientError :: Flight { source : FlightError :: GrpcError ( status ) , } ) = & query_res { if status . code ( ) == tonic :: Code :: NotFound { debug ! ( ingester_address ingester_address as_ref ( ) namespace_name table_name " " ) ; return Ok ( vec ! [ ] ) ; } } let mut perform_query = query_res . context ( RemoteQuerySnafu { ingester_address : ingester_address . as_ref ( ) , } ) . map_err ( | e | { // generate a warning that is sufficient to replicate the request using CLI tooling warn ! ( e e ingester_address ingester_address as_ref ( ) namespace namespace_name as_ref ( ) table table_name as_ref ( ) columns columns join ( " " ) as_str ( ) predicate_str predicate predicate_binary encode_predicate_as_base64 ( predicate ) as_str ( ) " " ) ; //  need to return error until https://github.com/rust-lang/rust/issues/91345 is stable e } ) ? ; // read unpersisted partitions // map partition_id -> (PartitionMetadata, Vec<PartitionData>)) let mut partitions : HashMap < _ , _ > = perform_query . app_metadata ( ) . unpersisted_partitions . iter ( ) . map ( | ( id , state ) | ( * id , ( state . clone ( ) , vec ! [ ] ) ) ) . collect ( ) ; // sort batches into partitions let mut num_batches = 0usize ; let partition_ids = perform_query . app_metadata ( ) . batch_partition_ids . clone ( ) ; while let Some ( batch ) = perform_query . next ( ) . await . map_err ( | source | FlightClientError :: Flight { source } ) . context ( RemoteQuerySnafu { ingester_address : ingester_address . as_ref ( ) , } ) ? { let partition_id = * partition_ids . get ( num_batches ) . context ( MissingPartitionIdSnafu { pos : num_batches , ingester_address : ingester_address . as_ref ( ) , } ) ? ; partitions . get_mut ( & partition_id ) . context ( UnknownPartitionIdSnafu { partition_id , ingester_address : ingester_address . as_ref ( ) , } ) ? . 1 . push ( batch ) ; num_batches += 1 ; } debug ! ( ingester_address num_batches " " ) ; trace ! ( partitions schema perform_query schema ( ) " " ) ; ensure ! ( num_batches partition_ids len ( ) TooManyPartitionIdsSnafu { actual partition_ids len ( ) expected num_batches ingester_address ingester_address as_ref ( ) } ) ; let mut ingester_partitions = vec ! [ ] ; for ( partition_id , ( state , batches ) ) in partitions { // do NOT filter out empty partitions, because the caller of this functions needs the attached metadata // to select the right parquet files and tombstones let partition_id = PartitionId :: new ( partition_id ) ; let sequencer_id = catalog_cache . partition ( ) . sequencer_id ( partition_id ) . await ; let partition_sort_key = catalog_cache . partition ( ) . sort_key ( partition_id ) . await ; let ingester_partition = IngesterPartition :: try_new ( Arc :: clone ( & ingester_address ) , ChunkId :: new ( ) , Arc :: clone ( & namespace_name ) , Arc :: clone ( & table_name ) , partition_id , sequencer_id , Arc :: clone ( & expected_schema ) , state . parquet_max_sequence_number . map ( SequenceNumber :: new ) , state . tombstone_max_sequence_number . map ( SequenceNumber :: new ) , partition_sort_key , batches , ) ? ; ingester_partitions . push ( ingester_partition ) ; } Ok ( ingester_partitions ) }
Fetches the partitions for a single ingester
Fetches the partitions for a single ingester
request
GetPartitionForIngester<'_>
request
GetPartitionForIngester<'_>