pub async fn to_series_and_groups(
        &self,
        series_set_plans: SeriesSetPlans,
    ) -> Result<Vec<Either>> {
        let SeriesSetPlans {
            mut plans,
            group_columns,
        } = series_set_plans;

        if plans.is_empty() {
            return Ok(vec![]);
        }

        // sort plans by table (measurement) name
        plans.sort_by(|a, b| a.table_name.cmp(&b.table_name));

        // Run the plans in parallel
        let handles = plans
            .into_iter()
            .map(|plan| {
                let ctx = self.child_ctx("to_series_set");
                self.run(async move {
                    let SeriesSetPlan {
                        table_name,
                        plan,
                        tag_columns,
                        field_columns,
                    } = plan;

                    let tag_columns = Arc::new(tag_columns);

                    let physical_plan = ctx.prepare_plan(&plan).await?;

                    let it = ctx.execute_stream(physical_plan).await?;

                    SeriesSetConverter::default()
                        .convert(table_name, tag_columns, field_columns, it)
                        .await
                        .map_err(|e| {
                            Error::Execution(format!(
                                "Error executing series set conversion: {}",
                                e
                            ))
                        })
                })
            })
            .collect::<Vec<_>>();

        // join_all ensures that the results are consumed in the same order they
        // were spawned maintaining the guarantee to return results ordered
        // by table name and plan sort order.
        let all_series_sets = futures::future::try_join_all(handles).await?;

        // convert to series sets
        let mut data: Vec<Series> = vec![];
        for series_sets in all_series_sets {
            for series_set in series_sets {
                // If all timestamps of returned columns are nulls,
                // there must be no data. We need to check this because
                // aggregate (e.g. count, min, max) returns one row that are
                // all null (even the values of aggregate) for min, max and 0 for count.
                // For influx read_group's series and group, we do not want to return 0
                // for count either.
                if series_set.is_timestamp_all_null() {
                    continue;
                }

                let series: Vec<Series> = series_set
                    .try_into()
                    .map_err(|e| Error::Execution(format!("Error converting to series: {}", e)))?;
                data.extend(series);
            }
        }

        // If we have group columns, sort the results, and create the
        // appropriate groups
        if let Some(group_columns) = group_columns {
            let grouper = GroupGenerator::new(group_columns);
            grouper
                .group(data)
                .map_err(|e| Error::Execution(format!("Error forming groups: {}", e)))
        } else {
            let data = data.into_iter().map(|series| series.into()).collect();
            Ok(data)
        }
    }
/// Executes the SeriesSetPlans on the query executor, in
/// parallel, producing series or groups
///
/// TODO make this streaming rather than buffering the results
Executes the SeriesSetPlans on the query executor, in
parallel, producing series or groups


Executes the SeriesSetPlans on the query executor in parallel producing series or groups
pub async fn to_series_and_groups(
        &self,
        series_set_plans: SeriesSetPlans,
    ) -> Result<Vec<Either>> {
        let SeriesSetPlans {
            mut plans,
            group_columns,
        } = series_set_plans;
        if plans.is_empty() {
            return Ok(vec![]);
        }
        // sort plans by table (measurement) name
        plans.sort_by(|a, b| a.table_name.cmp(&b.table_name));
        // Run the plans in parallel
        let handles = plans
            .into_iter()
            .map(|plan| {
                let ctx = self.child_ctx("to_series_set");
                self.run(async move {
                    let SeriesSetPlan {
                        table_name,
                        plan,
                        tag_columns,
                        field_columns,
                    } = plan;
                    let tag_columns = Arc::new(tag_columns);
                    let physical_plan = ctx.prepare_plan(&plan).await?;
                    let it = ctx.execute_stream(physical_plan).await?;
                    SeriesSetConverter::default()
                        .convert(table_name, tag_columns, field_columns, it)
                        .await
                        .map_err(|e| {
                            Error::Execution(format!(
                                "Error executing series set conversion: {}",
                                e
                            ))
                        })
                })
            })
            .collect::<Vec<_>>();
        // join_all ensures that the results are consumed in the same order they
        // were spawned maintaining the guarantee to return results ordered
        // by table name and plan sort order.
        let all_series_sets = futures::future::try_join_all(handles).await?;
        // convert to series sets
        let mut data: Vec<Series> = vec![];
        for series_sets in all_series_sets {
            for series_set in series_sets {
                // If all timestamps of returned columns are nulls,
                // there must be no data. We need to check this because
                // aggregate (e.g. count, min, max) returns one row that are
                // all null (even the values of aggregate) for min, max and 0 for count.
                // For influx read_group's series and group, we do not want to return 0
                // for count either.
                if series_set.is_timestamp_all_null() {
                    continue;
                }
                let series: Vec<Series> = series_set
                    .try_into()
                    .map_err(|e| Error::Execution(format!("Error converting to series: {}", e)))?;
                data.extend(series);
            }
        }
        // If we have group columns, sort the results, and create the
        // appropriate groups
        if let Some(group_columns) = group_columns {
            let grouper = GroupGenerator::new(group_columns);
            grouper
                .group(data)
                .map_err(|e| Error::Execution(format!("Error forming groups: {}", e)))
        } else {
            let data = data.into_iter().map(|series| series.into()).collect();
            Ok(data)
        }
    }
pub async fn to_series_and_groups ( & self , series_set_plans : SeriesSetPlans , ) -> Result < Vec < Either > > { let SeriesSetPlans { mut plans , group_columns , } = series_set_plans ; if plans . is_empty ( ) { return Ok ( vec ! [ ] ) ; } // sort plans by table (measurement) name plans . sort_by ( | a , b | a . table_name . cmp ( & b . table_name ) ) ; // Run the plans in parallel let handles = plans . into_iter ( ) . map ( | plan | { let ctx = self . child_ctx ( " " ) ; self . run ( async move { let SeriesSetPlan { table_name , plan , tag_columns , field_columns , } = plan ; let tag_columns = Arc :: new ( tag_columns ) ; let physical_plan = ctx . prepare_plan ( & plan ) . await ? ; let it = ctx . execute_stream ( physical_plan ) . await ? ; SeriesSetConverter :: default ( ) . convert ( table_name , tag_columns , field_columns , it ) . await . map_err ( | e | { Error :: Execution ( format ! ( " " e ) ) } ) } ) } ) . collect :: < Vec < _ > > ( ) ; // join_all ensures that the results are consumed in the same order they // were spawned maintaining the guarantee to return results ordered // by table name and plan sort order. let all_series_sets = futures :: future :: try_join_all ( handles ) . await ? ; // convert to series sets let mut data : Vec < Series > = vec ! [ ] ; for series_sets in all_series_sets { for series_set in series_sets { // If all timestamps of returned columns are nulls, // there must be no data. We need to check this because // aggregate (e.g. count, min, max) returns one row that are // all null (even the values of aggregate) for min, max and 0 for count. // For influx read_group's series and group, we do not want to return 0 // for count either. if series_set . is_timestamp_all_null ( ) { continue ; } let series : Vec < Series > = series_set . try_into ( ) . map_err ( | e | Error :: Execution ( format ! ( " " e ) ) ) ? ; data . extend ( series ) ; } } // If we have group columns, sort the results, and create the // appropriate groups if let Some ( group_columns ) = group_columns { let grouper = GroupGenerator :: new ( group_columns ) ; grouper . group ( data ) . map_err ( | e | Error :: Execution ( format ! ( " " e ) ) ) } else { let data = data . into_iter ( ) . map ( | series | series . into ( ) ) . collect ( ) ; Ok ( data ) } }
Executes the SeriesSetPlans on the query executor, in
parallel, producing series or groups
Executes the SeriesSetPlans on the query executor in parallel producing series or groups
&self
series_set_plans
SeriesSetPlans
&self
series_set_plans
SeriesSetPlans