19567405893

Committed 21 Nov 2025 10:21AM UTC coverage: 88.567%. First build

Build # 19567405893

Build Type

Pull #1083

github

Committed by

web-flow

Commit Message

Merge c6cd1b0c9 into ba34db99d

Pull Request Pull Request #1083: feat: new gdal source workflow optimization

Coverage Stats

6332 of 7653 new or added lines in 79 files covered. (82.74%)

116676 of 131738 relevant lines covered (88.57%)

492817.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.36

/operators/src/processing/circle_merging_quadtree/operator.rs

use std::collections::{HashMap, HashSet};

use async_trait::async_trait;
use futures::StreamExt;
use futures::stream::{BoxStream, FuturesUnordered};
use geoengine_datatypes::collections::{
    BuilderProvider, GeoFeatureCollectionRowBuilder, MultiPointCollection, VectorDataType,
};
use geoengine_datatypes::primitives::{
    BoundingBox2D, Circle, FeatureDataType, FeatureDataValue, Measurement, MultiPoint,
    MultiPointAccess, SpatialBounded, SpatialResolution, VectorQueryRectangle,
};
use geoengine_datatypes::primitives::{CacheHint, ColumnSelection};
use serde::{Deserialize, Serialize};
use snafu::ensure;

use crate::adapters::FeatureCollectionStreamExt;
use crate::engine::{
    CanonicOperatorName, ExecutionContext, InitializedVectorOperator, Operator, QueryContext,
    QueryProcessor, SingleVectorSource, TypedVectorQueryProcessor, VectorColumnInfo,
    VectorOperator, VectorQueryProcessor, VectorResultDescriptor,
};
use crate::engine::{InitializedSources, OperatorName, WorkflowOperatorPath};
use crate::error::{self, Error};
use crate::optimization::OptimizationError;
use crate::processing::circle_merging_quadtree::aggregates::MeanAggregator;
use crate::processing::circle_merging_quadtree::circle_of_points::CircleOfPoints;
use crate::processing::circle_merging_quadtree::circle_radius_model::LogScaledRadius;
use crate::util::Result;

use super::aggregates::{AttributeAggregate, AttributeAggregateType, StringSampler};
use super::circle_radius_model::CircleRadiusModel;
use super::grid::Grid;
use super::quadtree::CircleMergingQuadtree;

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct VisualPointClusteringParams {
    pub min_radius_px: f64,
    pub delta_px: f64,
    pub resolution: f64,
    radius_column: String,
    count_column: String,
    column_aggregates: HashMap<String, AttributeAggregateDef>,
}

#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub struct AttributeAggregateDef {
    pub column_name: String,
    pub aggregate_type: AttributeAggregateType,
    // if measurement is unset, it will be taken from the source result descriptor
    pub measurement: Option<Measurement>,
}

pub type VisualPointClustering = Operator<VisualPointClusteringParams, SingleVectorSource>;

impl OperatorName for VisualPointClustering {
    const TYPE_NAME: &'static str = "VisualPointClustering";
}

#[typetag::serde]
#[async_trait]
#[allow(clippy::too_many_lines)]
impl VectorOperator for VisualPointClustering {
    async fn _initialize(
        mut self: Box<Self>,
        path: WorkflowOperatorPath,
        context: &dyn ExecutionContext,
    ) -> Result<Box<dyn InitializedVectorOperator>> {
        ensure!(
            self.params.min_radius_px > 0.0,
            error::InputMustBeGreaterThanZero {
                scope: "VisualPointClustering",
                name: "minRadius"
            }
        );
        ensure!(
            self.params.delta_px >= 0.0,
            error::InputMustBeZeroOrPositive {
                scope: "VisualPointClustering",
                name: "deltaPx"
            }
        );
        ensure!(
            self.params.resolution >= 0.0,
            error::InputMustBeZeroOrPositive {
                scope: "VisualPointClustering",
                name: "resolution"
            }
        );
        ensure!(!self.params.radius_column.is_empty(), error::EmptyInput);
        ensure!(
            self.params.count_column != self.params.radius_column,
            error::DuplicateOutputColumns
        );

        let name = CanonicOperatorName::from(&self);

        let radius_model = LogScaledRadius::new(self.params.min_radius_px, self.params.delta_px)?;

        let initialized_sources = self
            .sources
            .initialize_sources(path.clone(), context)
            .await?;
        let vector_source = initialized_sources.vector;

        ensure!(
            vector_source.result_descriptor().data_type == VectorDataType::MultiPoint,
            error::InvalidType {
                expected: VectorDataType::MultiPoint.to_string(),
                found: vector_source.result_descriptor().data_type.to_string(),
            }
        );

        // check that all input columns exist
        for column_name in self.params.column_aggregates.keys() {
            let column_name_exists = vector_source
                .result_descriptor()
                .columns
                .contains_key(column_name);

            ensure!(
                column_name_exists,
                error::MissingInputColumn {
                    name: column_name.clone()
                }
            );
        }

        // check that there are no duplicates in the output columns
        let output_names: HashSet<&String> = self
            .params
            .column_aggregates
            .values()
            .map(|def| &def.column_name)
            .collect();
        ensure!(
            output_names.len() == self.params.column_aggregates.len(),
            error::DuplicateOutputColumns
        );

        // create schema for [`ResultDescriptor`]
        let mut new_columns: HashMap<String, VectorColumnInfo> =
            HashMap::with_capacity(self.params.column_aggregates.len());
        for attribute_aggregate_def in self.params.column_aggregates.values_mut() {
            if attribute_aggregate_def.measurement.is_none() {
                // take it from source measurement

                attribute_aggregate_def.measurement = vector_source
                    .result_descriptor()
                    .column_measurement(&attribute_aggregate_def.column_name)
                    .cloned();
            }

            let data_type = match attribute_aggregate_def.aggregate_type {
                AttributeAggregateType::MeanNumber => FeatureDataType::Float,
                AttributeAggregateType::StringSample => FeatureDataType::Text,
                AttributeAggregateType::Null => {
                    return Err(Error::InvalidType {
                        expected: "not null".to_string(),
                        found: "null".to_string(),
                    });
                }
            };

            new_columns.insert(
                attribute_aggregate_def.column_name.clone(),
                VectorColumnInfo {
                    data_type,
                    measurement: attribute_aggregate_def.measurement.clone().into(),
                },
            );
        }

        // check that output schema does not interfere with count and radius columns
        ensure!(
            !new_columns.contains_key(&self.params.radius_column)
                && !new_columns.contains_key(&self.params.count_column),
            error::DuplicateOutputColumns
        );

        let in_desc = vector_source.result_descriptor();

        Ok(InitializedVisualPointClustering {
            name,
            path,
            params: self.params.clone(),
            result_descriptor: VectorResultDescriptor {
                data_type: VectorDataType::MultiPoint,
                spatial_reference: in_desc.spatial_reference,
                columns: new_columns,
                time: in_desc.time,
                bbox: in_desc.bbox,
            },
            vector_source,
            radius_model,
            radius_column: self.params.radius_column,
            count_column: self.params.count_column,
            attribute_mapping: self.params.column_aggregates,
            resolution: self.params.resolution,
        }
        .boxed())
    }

    span_fn!(VisualPointClustering);
}

pub struct InitializedVisualPointClustering {
    name: CanonicOperatorName,
    path: WorkflowOperatorPath,
    params: VisualPointClusteringParams,
    result_descriptor: VectorResultDescriptor,
    vector_source: Box<dyn InitializedVectorOperator>,
    radius_model: LogScaledRadius,
    radius_column: String,
    count_column: String,
    attribute_mapping: HashMap<String, AttributeAggregateDef>,
    resolution: f64,
}

impl InitializedVectorOperator for InitializedVisualPointClustering {
    fn query_processor(&self) -> Result<TypedVectorQueryProcessor> {
        match self.vector_source.query_processor()? {
            TypedVectorQueryProcessor::MultiPoint(source) => {
                Ok(TypedVectorQueryProcessor::MultiPoint(
                    VisualPointClusteringProcessor::new(
                        source,
                        self.radius_model,
                        self.radius_column.clone(),
                        self.count_column.clone(),
                        self.result_descriptor.clone(),
                        self.attribute_mapping.clone(),
                        self.resolution,
                    )
                    .boxed(),
                ))
            }
            TypedVectorQueryProcessor::MultiLineString(_) => Err(error::Error::InvalidVectorType {
                expected: "MultiPoint".to_owned(),
                found: "MultiLineString".to_owned(),
            }),
            TypedVectorQueryProcessor::MultiPolygon(_) => Err(error::Error::InvalidVectorType {
                expected: "MultiPoint".to_owned(),
                found: "MultiPolygon".to_owned(),
            }),
            TypedVectorQueryProcessor::Data(_) => Err(error::Error::InvalidVectorType {
                expected: "MultiPoint".to_owned(),
                found: "Data".to_owned(),
            }),
        }
    }

    fn result_descriptor(&self) -> &VectorResultDescriptor {
        &self.result_descriptor
    }

    fn canonic_name(&self) -> CanonicOperatorName {
        self.name.clone()
    }

    fn name(&self) -> &'static str {
        VisualPointClustering::TYPE_NAME
    }

    fn path(&self) -> WorkflowOperatorPath {
        self.path.clone()
    }

    fn optimize(
        &self,
        target_resolution: SpatialResolution,
    ) -> Result<Box<dyn VectorOperator>, OptimizationError> {
        Ok(VisualPointClustering {
            params: self.params.clone(),
            sources: SingleVectorSource {
                vector: self.vector_source.optimize(target_resolution)?,
            },
        }
        .boxed())
    }
}

pub struct VisualPointClusteringProcessor {
    source: Box<dyn VectorQueryProcessor<VectorType = MultiPointCollection>>,
    radius_model: LogScaledRadius,
    radius_column: String,
    count_column: String,
    result_descriptor: VectorResultDescriptor,
    attribute_mapping: HashMap<String, AttributeAggregateDef>,
    resolution: f64,
}

impl VisualPointClusteringProcessor {
    fn new(
        source: Box<dyn VectorQueryProcessor<VectorType = MultiPointCollection>>,
        radius_model: LogScaledRadius,
        radius_column: String,
        count_column: String,
        result_descriptor: VectorResultDescriptor,
        attribute_mapping: HashMap<String, AttributeAggregateDef>,
        resolution: f64,
    ) -> Self {
        Self {
            source,
            radius_model,
            radius_column,
            count_column,
            result_descriptor,
            attribute_mapping,
            resolution,
        }
    }

    fn create_point_collection(
        circles_of_points: impl Iterator<Item = CircleOfPoints>,
        radius_column: &str,
        count_column: &str,
        columns: &HashMap<String, FeatureDataType>,
        cache_hint: CacheHint,
        resolution: f64,
    ) -> Result<MultiPointCollection> {
        let mut builder = MultiPointCollection::builder();

        for (column, &data_type) in columns {
            builder.add_column(column.clone(), data_type)?;
        }

        builder.add_column(radius_column.to_string(), FeatureDataType::Float)?;
        builder.add_column(count_column.to_string(), FeatureDataType::Int)?;

        let mut builder = builder.finish_header();

        for circle_of_points in circles_of_points {
            builder.push_geometry(MultiPoint::new(vec![circle_of_points.circle.center()])?);
            builder.push_time_interval(circle_of_points.time_aggregate);

            builder.push_data(
                radius_column,
                FeatureDataValue::Float(circle_of_points.circle.radius() / resolution),
            )?;
            builder.push_data(
                count_column,
                FeatureDataValue::Int(circle_of_points.number_of_points() as i64),
            )?;

            for (column, data_type) in circle_of_points.attribute_aggregates {
                match data_type {
                    AttributeAggregate::MeanNumber(mean_aggregator) => {
                        builder
                            .push_data(&column, FeatureDataValue::Float(mean_aggregator.mean))?;
                    }
                    AttributeAggregate::StringSample(string_sampler) => {
                        builder.push_data(
                            &column,
                            FeatureDataValue::Text(string_sampler.strings.join(", ")),
                        )?;
                    }
                    AttributeAggregate::Null => {
                        builder.push_null(&column)?;
                    }
                }
            }

            builder.finish_row();
        }

        builder.cache_hint(cache_hint);

        builder.build().map_err(Into::into)
    }

    fn aggregate_from_feature_data(
        feature_data: FeatureDataValue,
        aggregate_type: AttributeAggregateType,
    ) -> AttributeAggregate {
        match (feature_data, aggregate_type) {
            (
                FeatureDataValue::Category(value) | FeatureDataValue::NullableCategory(Some(value)),
                AttributeAggregateType::MeanNumber,
            ) => AttributeAggregate::MeanNumber(MeanAggregator::from_value(f64::from(value))),
            (
                FeatureDataValue::Float(value) | FeatureDataValue::NullableFloat(Some(value)),
                AttributeAggregateType::MeanNumber,
            ) => AttributeAggregate::MeanNumber(MeanAggregator::from_value(value)),
            (
                FeatureDataValue::Int(value) | FeatureDataValue::NullableInt(Some(value)),
                AttributeAggregateType::MeanNumber,
            ) => AttributeAggregate::MeanNumber(MeanAggregator::from_value(value as f64)),
            (
                FeatureDataValue::Text(value) | FeatureDataValue::NullableText(Some(value)),
                AttributeAggregateType::StringSample,
            ) => AttributeAggregate::StringSample(StringSampler::from_value(value)),
            (
                FeatureDataValue::DateTime(value) | FeatureDataValue::NullableDateTime(Some(value)),
                AttributeAggregateType::StringSample,
            ) => AttributeAggregate::StringSample(StringSampler::from_value(value.to_string())),
            _ => AttributeAggregate::Null,
        }
    }
}

struct GridFoldState {
    grid: Grid<LogScaledRadius>,
    column_mapping: HashMap<String, AttributeAggregateDef>,
    cache_hint: CacheHint,
}

#[async_trait]
impl QueryProcessor for VisualPointClusteringProcessor {
    type Output = MultiPointCollection;
    type SpatialBounds = BoundingBox2D;
    type Selection = ColumnSelection;
    type ResultDescription = VectorResultDescriptor;

    async fn _query<'a>(
        &'a self,
        query: VectorQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<Self::Output>>> {
        // we aggregate all points into one collection

        let column_schema = self
            .result_descriptor
            .columns
            .iter()
            .map(|(name, column_info)| (name.clone(), column_info.data_type))
            .collect();
        let scaled_radius_model = self.radius_model.with_scaled_radii(self.resolution)?;

        let initial_grid_fold_state = Result::<GridFoldState>::Ok(GridFoldState {
            grid: Grid::new(query.spatial_bounds().spatial_bounds(), scaled_radius_model),
            column_mapping: self.attribute_mapping.clone(),
            cache_hint: CacheHint::max_duration(),
        });

        let grid_future = self.source.query(query.clone(), ctx).await?.fold(
            initial_grid_fold_state,
            |state, feature_collection| async move {
                // TODO: worker thread

                let GridFoldState {
                    mut grid,
                    column_mapping,
                    mut cache_hint,
                } = state?;

                let feature_collection = feature_collection?;

                for feature in &feature_collection {
                    // TODO: pre-aggregate multi-points differently?
                    for coordinate in feature.geometry.points() {
                        let circle =
                            Circle::from_coordinate(coordinate, grid.radius_model().min_radius());

                        let mut attribute_aggregates = HashMap::with_capacity(column_mapping.len());

                        for (
                            src_column,
                            AttributeAggregateDef {
                                column_name: tgt_column,
                                aggregate_type,
                                measurement: _,
                            },
                        ) in &column_mapping
                        {
                            let attribute_aggregate =
                                if let Some(feature_data) = feature.get(src_column) {
                                    Self::aggregate_from_feature_data(feature_data, *aggregate_type)
                                } else {
                                    AttributeAggregate::Null
                                };

                            attribute_aggregates.insert(tgt_column.clone(), attribute_aggregate);
                        }

                        grid.insert(CircleOfPoints::new_with_one_point(
                            circle,
                            feature.time_interval,
                            attribute_aggregates,
                        ));
                    }
                }

                cache_hint.merge_with(&feature_collection.cache_hint);

                Ok(GridFoldState {
                    grid,
                    column_mapping,
                    cache_hint,
                })
            },
        );

        let stream = FuturesUnordered::new();
        stream.push(grid_future);

        let stream = stream.map(move |grid| {
            let GridFoldState {
                grid,
                column_mapping: _,
                cache_hint,
            } = grid?;

            let mut cmq = CircleMergingQuadtree::new(
                query.spatial_bounds().spatial_bounds(),
                *grid.radius_model(),
                1,
            );

            // TODO: worker thread
            for circle_of_points in grid.drain() {
                cmq.insert_circle(circle_of_points);
            }

            Self::create_point_collection(
                cmq.into_iter(),
                &self.radius_column,
                &self.count_column,
                &column_schema,
                cache_hint,
                self.resolution,
            )
        });

        Ok(stream.merge_chunks(ctx.chunk_byte_size().into()).boxed())
    }

    fn result_descriptor(&self) -> &VectorResultDescriptor {
        &self.result_descriptor
    }
}

#[cfg(test)]
mod tests {
    use geoengine_datatypes::collections::ChunksEqualIgnoringCacheHint;
    use geoengine_datatypes::primitives::BoundingBox2D;
    use geoengine_datatypes::primitives::CacheHint;
    use geoengine_datatypes::primitives::FeatureData;
    use geoengine_datatypes::primitives::TimeInterval;
    use geoengine_datatypes::util::test::TestDefault;

    use crate::{engine::MockExecutionContext, mock::MockFeatureCollectionSource};

    use super::*;

    #[tokio::test]
    async fn simple_test() {
        let mut coordinates = vec![(0.0, 0.1); 9];
        coordinates.extend_from_slice(&[(50.0, 50.1)]);

        let input = MultiPointCollection::from_data(
            MultiPoint::many(coordinates).unwrap(),
            vec![TimeInterval::default(); 10],
            HashMap::default(),
            CacheHint::default(),
        )
        .unwrap();

        let operator = VisualPointClustering {
            params: VisualPointClusteringParams {
                min_radius_px: 8.,
                delta_px: 1.,
                resolution: 0.1,
                radius_column: "radius".to_string(),
                count_column: "count".to_string(),
                column_aggregates: Default::default(),
            },
            sources: SingleVectorSource {
                vector: MockFeatureCollectionSource::single(input).boxed(),
            },
        };

        let execution_context = MockExecutionContext::test_default();

        let initialized_operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap();

        let query_processor = initialized_operator
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let query_context = execution_context.mock_query_context_test_default();

        let qrect = VectorQueryRectangle::new(
            BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
            TimeInterval::default(),
            ColumnSelection::all(),
        );

        let query = query_processor.query(qrect, &query_context).await.unwrap();

        let result: Vec<MultiPointCollection> = query.map(Result::unwrap).collect().await;

        assert_eq!(result.len(), 1);
        assert!(
            result[0].chunks_equal_ignoring_cache_hint(
                &MultiPointCollection::from_slices(
                    &[(0.0, 0.099_999_999_999_999_99), (50.0, 50.1)],
                    &[TimeInterval::default(); 2],
                    &[
                        ("count", FeatureData::Int(vec![9, 1])),
                        (
                            "radius",
                            FeatureData::Float(vec![10.197_224_577_336_218, 8.])
                        )
                    ],
                )
                .unwrap()
            )
        );
    }

    #[tokio::test]
    async fn simple_test_with_aggregate() {
        let mut coordinates = vec![(0.0, 0.1); 9];
        coordinates.extend_from_slice(&[(50.0, 50.1)]);

        let input = MultiPointCollection::from_slices(
            &MultiPoint::many(coordinates).unwrap(),
            &[TimeInterval::default(); 10],
            &[(
                "foo",
                FeatureData::Float(vec![1., 2., 3., 4., 5., 6., 7., 8., 9., 10.]),
            )],
        )
        .unwrap();

        let operator = VisualPointClustering {
            params: VisualPointClusteringParams {
                min_radius_px: 8.,
                delta_px: 1.,
                resolution: 0.1,
                radius_column: "radius".to_string(),
                count_column: "count".to_string(),
                column_aggregates: [(
                    "foo".to_string(),
                    AttributeAggregateDef {
                        column_name: "bar".to_string(),
                        aggregate_type: AttributeAggregateType::MeanNumber,
                        measurement: None,
                    },
                )]
                .iter()
                .cloned()
                .collect(),
            },
            sources: SingleVectorSource {
                vector: MockFeatureCollectionSource::single(input).boxed(),
            },
        };

        let execution_context = MockExecutionContext::test_default();

        let initialized_operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap();

        let query_processor = initialized_operator
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let query_context = execution_context.mock_query_context_test_default();

        let qrect = VectorQueryRectangle::new(
            BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
            TimeInterval::default(),
            ColumnSelection::all(),
        );

        let query = query_processor.query(qrect, &query_context).await.unwrap();

        let result: Vec<MultiPointCollection> = query.map(Result::unwrap).collect().await;

        assert_eq!(result.len(), 1);
        assert!(
            result[0].chunks_equal_ignoring_cache_hint(
                &MultiPointCollection::from_slices(
                    &[(0.0, 0.099_999_999_999_999_99), (50.0, 50.1)],
                    &[TimeInterval::default(); 2],
                    &[
                        ("count", FeatureData::Int(vec![9, 1])),
                        (
                            "radius",
                            FeatureData::Float(vec![10.197_224_577_336_218, 8.])
                        ),
                        ("bar", FeatureData::Float(vec![5., 10.]))
                    ],
                )
                .unwrap()
            )
        );
    }

    #[tokio::test]
    async fn aggregate_of_null() {
        let mut coordinates = vec![(0.0, 0.1); 2];
        coordinates.extend_from_slice(&[(50.0, 50.1); 2]);

        let input = MultiPointCollection::from_slices(
            &MultiPoint::many(coordinates).unwrap(),
            &[TimeInterval::default(); 4],
            &[(
                "foo",
                FeatureData::NullableInt(vec![Some(1), None, None, None]),
            )],
        )
        .unwrap();

        let operator = VisualPointClustering {
            params: VisualPointClusteringParams {
                min_radius_px: 8.,
                delta_px: 1.,
                resolution: 0.1,
                radius_column: "radius".to_string(),
                count_column: "count".to_string(),
                column_aggregates: [(
                    "foo".to_string(),
                    AttributeAggregateDef {
                        column_name: "foo".to_string(),
                        aggregate_type: AttributeAggregateType::MeanNumber,
                        measurement: None,
                    },
                )]
                .iter()
                .cloned()
                .collect(),
            },
            sources: SingleVectorSource {
                vector: MockFeatureCollectionSource::single(input).boxed(),
            },
        };

        let execution_context = MockExecutionContext::test_default();

        let initialized_operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap();

        let query_processor = initialized_operator
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let query_context = execution_context.mock_query_context_test_default();

        let qrect = VectorQueryRectangle::new(
            BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
            TimeInterval::default(),
            ColumnSelection::all(),
        );

        let query = query_processor.query(qrect, &query_context).await.unwrap();

        let result: Vec<MultiPointCollection> = query.map(Result::unwrap).collect().await;

        assert_eq!(result.len(), 1);
        assert!(
            result[0].chunks_equal_ignoring_cache_hint(
                &MultiPointCollection::from_slices(
                    &[(0.0, 0.1), (50.0, 50.1)],
                    &[TimeInterval::default(); 2],
                    &[
                        ("count", FeatureData::Int(vec![2, 2])),
                        (
                            "radius",
                            FeatureData::Float(vec![8.693_147_180_559_945, 8.693_147_180_559_945])
                        ),
                        ("foo", FeatureData::NullableFloat(vec![Some(1.), None]))
                    ],
                )
                .unwrap()
            )
        );
    }

    #[tokio::test]
    async fn text_aggregate() {
        let mut coordinates = vec![(0.0, 0.1); 2];
        coordinates.extend_from_slice(&[(50.0, 50.1); 2]);
        coordinates.extend_from_slice(&[(25.0, 25.1); 2]);

        let input = MultiPointCollection::from_slices(
            &MultiPoint::many(coordinates).unwrap(),
            &[TimeInterval::default(); 6],
            &[(
                "text",
                FeatureData::NullableText(vec![
                    Some("foo".to_string()),
                    Some("bar".to_string()),
                    Some("foo".to_string()),
                    None,
                    None,
                    None,
                ]),
            )],
        )
        .unwrap();

        let operator = VisualPointClustering {
            params: VisualPointClusteringParams {
                min_radius_px: 8.,
                delta_px: 1.,
                resolution: 0.1,
                radius_column: "radius".to_string(),
                count_column: "count".to_string(),
                column_aggregates: [(
                    "text".to_string(),
                    AttributeAggregateDef {
                        column_name: "text".to_string(),
                        aggregate_type: AttributeAggregateType::StringSample,
                        measurement: None,
                    },
                )]
                .iter()
                .cloned()
                .collect(),
            },
            sources: SingleVectorSource {
                vector: MockFeatureCollectionSource::single(input).boxed(),
            },
        };

        let execution_context = MockExecutionContext::test_default();

        let initialized_operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap();

        let query_processor = initialized_operator
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let query_context = execution_context.mock_query_context_test_default();

        let qrect = VectorQueryRectangle::new(
            BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
            TimeInterval::default(),
            ColumnSelection::all(),
        );

        let query = query_processor.query(qrect, &query_context).await.unwrap();

        let result: Vec<MultiPointCollection> = query.map(Result::unwrap).collect().await;

        assert_eq!(result.len(), 1);
        assert!(
            result[0].chunks_equal_ignoring_cache_hint(
                &MultiPointCollection::from_slices(
                    &[(0.0, 0.1), (50.0, 50.1), (25.0, 25.1)],
                    &[TimeInterval::default(); 3],
                    &[
                        ("count", FeatureData::Int(vec![2, 2, 2])),
                        (
                            "radius",
                            FeatureData::Float(vec![
                                8.693_147_180_559_945,
                                8.693_147_180_559_945,
                                8.693_147_180_559_945
                            ])
                        ),
                        (
                            "text",
                            FeatureData::NullableText(vec![
                                Some("foo, bar".to_string()),
                                Some("foo".to_string()),
                                None
                            ])
                        )
                    ],
                )
                .unwrap()
            )
        );
    }

    #[tokio::test]
    async fn it_attaches_cache_hint() {
        let mut coordinates = vec![(0.0, 0.1); 2];
        coordinates.extend_from_slice(&[(50.0, 50.1); 2]);
        coordinates.extend_from_slice(&[(25.0, 25.1); 2]);

        let mut input = MultiPointCollection::from_slices(
            &MultiPoint::many(coordinates).unwrap(),
            &[TimeInterval::default(); 6],
            &[(
                "text",
                FeatureData::NullableText(vec![
                    Some("foo".to_string()),
                    Some("bar".to_string()),
                    Some("foo".to_string()),
                    None,
                    None,
                    None,
                ]),
            )],
        )
        .unwrap();

        let cache_hint = CacheHint::seconds(1234);

        input.cache_hint = cache_hint;

        let operator = VisualPointClustering {
            params: VisualPointClusteringParams {
                min_radius_px: 8.,
                delta_px: 1.,
                resolution: 0.1,
                radius_column: "radius".to_string(),
                count_column: "count".to_string(),
                column_aggregates: [(
                    "text".to_string(),
                    AttributeAggregateDef {
                        column_name: "text".to_string(),
                        aggregate_type: AttributeAggregateType::StringSample,
                        measurement: None,
                    },
                )]
                .iter()
                .cloned()
                .collect(),
            },
            sources: SingleVectorSource {
                vector: MockFeatureCollectionSource::single(input).boxed(),
            },
        };

        let execution_context = MockExecutionContext::test_default();

        let initialized_operator = operator
            .boxed()
            .initialize(WorkflowOperatorPath::initialize_root(), &execution_context)
            .await
            .unwrap();

        let query_processor = initialized_operator
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let query_context = execution_context.mock_query_context_test_default();

        let qrect = VectorQueryRectangle::new(
            BoundingBox2D::new((-180., -90.).into(), (180., 90.).into()).unwrap(),
            TimeInterval::default(),
            ColumnSelection::all(),
        );

        let query = query_processor.query(qrect, &query_context).await.unwrap();

        let result: Vec<MultiPointCollection> = query.map(Result::unwrap).collect().await;

        assert_eq!(result.len(), 1);
        assert_eq!(result[0].cache_hint.expires(), cache_hint.expires());
    }
}

geo-engine / geoengine / 19567405893

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous