pub async fn to_series_set(
        &self,
        series_set_plans: SeriesSetPlans,
    ) -> Result<Vec<SeriesSetItem>, Error> {
        let SeriesSetPlans { mut plans } = series_set_plans;

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

        // sort plans by table 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 executor_type = ExecutorType::Query;
                let ctx = self.new_context(executor_type);

                self.run(
                    async move {
                        let SeriesSetPlan {
                            table_name,
                            plan,
                            tag_columns,
                            field_columns,
                            num_prefix_tag_group_columns,
                        } = plan;

                        let tag_columns = Arc::new(tag_columns);

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

                        let it = ctx
                            .execute(physical_plan)
                            .await
                            .context(SeriesSetExecution)?;

                        SeriesSetConverter::default()
                            .convert(
                                table_name,
                                tag_columns,
                                field_columns,
                                num_prefix_tag_group_columns,
                                it,
                            )
                            .await
                            .context(SeriesSetConversion)
                    },
                    executor_type,
                )
            })
            .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 the plan sort order.
        let handles = future::try_join_all(handles).await?;
        let mut results = vec![];
        for handle in handles {
            results.extend(handle?.into_iter());
        }

        Ok(results)
    }
/// Executes the SeriesSetPlans on the query exectutor, in
/// parallel, combining the results into the returned collection
/// of items.
///
/// The SeriesSets are guaranteed to come back ordered by table_name.
Executes the SeriesSetPlans on the query exectutor, in
parallel, combining the results into the returned collection
of items.

The SeriesSets are guaranteed to come back ordered by table_name.
Executes the SeriesSetPlans on the query exectutor in parallel combining the results into the returned collection of items . The SeriesSets are guaranteed to come back ordered by table_name .
pub async fn to_series_set(
        &self,
        series_set_plans: SeriesSetPlans,
    ) -> Result<Vec<SeriesSetItem>, Error> {
        let SeriesSetPlans { mut plans } = series_set_plans;
        if plans.is_empty() {
            return Ok(vec![]);
        }
        // sort plans by table 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 executor_type = ExecutorType::Query;
                let ctx = self.new_context(executor_type);
                self.run(
                    async move {
                        let SeriesSetPlan {
                            table_name,
                            plan,
                            tag_columns,
                            field_columns,
                            num_prefix_tag_group_columns,
                        } = plan;
                        let tag_columns = Arc::new(tag_columns);
                        let physical_plan = ctx
                            .prepare_plan(&plan)
                            .context(DataFusionPhysicalPlanning)?;
                        let it = ctx
                            .execute(physical_plan)
                            .await
                            .context(SeriesSetExecution)?;
                        SeriesSetConverter::default()
                            .convert(
                                table_name,
                                tag_columns,
                                field_columns,
                                num_prefix_tag_group_columns,
                                it,
                            )
                            .await
                            .context(SeriesSetConversion)
                    },
                    executor_type,
                )
            })
            .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 the plan sort order.
        let handles = future::try_join_all(handles).await?;
        let mut results = vec![];
        for handle in handles {
            results.extend(handle?.into_iter());
        }
        Ok(results)
    }
pub async fn to_series_set ( & self , series_set_plans : SeriesSetPlans , ) -> Result < Vec < SeriesSetItem > , Error > { let SeriesSetPlans { mut plans } = series_set_plans ; if plans . is_empty ( ) { return Ok ( vec ! [ ] ) ; } // sort plans by table 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 executor_type = ExecutorType :: Query ; let ctx = self . new_context ( executor_type ) ; self . run ( async move { let SeriesSetPlan { table_name , plan , tag_columns , field_columns , num_prefix_tag_group_columns , } = plan ; let tag_columns = Arc :: new ( tag_columns ) ; let physical_plan = ctx . prepare_plan ( & plan ) . context ( DataFusionPhysicalPlanning ) ? ; let it = ctx . execute ( physical_plan ) . await . context ( SeriesSetExecution ) ? ; SeriesSetConverter :: default ( ) . convert ( table_name , tag_columns , field_columns , num_prefix_tag_group_columns , it , ) . await . context ( SeriesSetConversion ) } , executor_type , ) } ) . 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 the plan sort order. let handles = future :: try_join_all ( handles ) . await ? ; let mut results = vec ! [ ] ; for handle in handles { results . extend ( handle ? . into_iter ( ) ) ; } Ok ( results ) }
Executes the SeriesSetPlans on the query exectutor, in
parallel, combining the results into the returned collection
of items.
Executes the SeriesSetPlans on the query exectutor in parallel combining the results into the returned collection of items .
&self
series_set_plans
SeriesSetPlans
&self
series_set_plans
SeriesSetPlans