3929938005

Build Type

push

github

Committed by GitHub

Commit Message

Merge #713

Run Details

84930 of 96741 relevant lines covered (87.79%)

79640.1 hits per line

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

96.23

/operators/src/processing/time_projection/mod.rs

use std::sync::Arc;

use crate::engine::{
    CreateSpan, ExecutionContext, InitializedVectorOperator, Operator, OperatorName, QueryContext,
    SingleVectorSource, TypedVectorQueryProcessor, VectorOperator, VectorQueryProcessor,
    VectorResultDescriptor,
};
use crate::util::Result;
use async_trait::async_trait;
use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
use geoengine_datatypes::collections::{
    FeatureCollection, FeatureCollectionInfos, FeatureCollectionModifications,
};
use geoengine_datatypes::primitives::{Geometry, TimeInterval};
use geoengine_datatypes::primitives::{TimeInstance, TimeStep, VectorQueryRectangle};
use geoengine_datatypes::util::arrow::ArrowTyped;
use log::debug;
use rayon::iter::{IndexedParallelIterator, IntoParallelRefIterator, ParallelIterator};
use rayon::ThreadPool;
use serde::{Deserialize, Serialize};
use snafu::{ensure, ResultExt, Snafu};
use tracing::{span, Level};

/// Projection of time information in queries and data
///
/// This operator changes the temporal validity of the queried data.
/// In order to query all valid data, it is necessary to change the query rectangle as well.
///
pub type TimeProjection = Operator<TimeProjectionParams, SingleVectorSource>;

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

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TimeProjectionParams {
    /// Specify the time step granularity and size
    step: TimeStep,
    /// Define an anchor point for `step`
    /// If `None`, the anchor point is `1970-01-01T00:00:00Z` by default
    step_reference: Option<TimeInstance>,
}

#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)), context(suffix(false)), module(error))]
pub enum TimeProjectionError {
    #[snafu(display("Time step must be larger than zero"))]
    WindowSizeMustNotBeZero,

    #[snafu(display("Query rectangle expansion failed: {}", source))]
    CannotExpandQueryRectangle {
        source: geoengine_datatypes::error::Error,
    },

    #[snafu(display("Feature time interval expansion failed: {}", source))]
    CannotExpandFeatureTimeInterval {
        source: geoengine_datatypes::error::Error,
    },
}

#[typetag::serde]
#[async_trait]
impl VectorOperator for TimeProjection {
    async fn _initialize(
        self: Box<Self>,
        context: &dyn ExecutionContext,
    ) -> Result<Box<dyn InitializedVectorOperator>> {
        ensure!(self.params.step.step > 0, error::WindowSizeMustNotBeZero);

        let source = self.sources.vector.initialize(context).await?;

        debug!("Initializing `TimeProjection` with {:?}.", &self.params);

        let step_reference = self
            .params
            .step_reference
            // use UTC 0 as default
            .unwrap_or(TimeInstance::EPOCH_START);

        let mut result_descriptor = source.result_descriptor().clone();
        rewrite_result_descriptor(&mut result_descriptor, self.params.step, step_reference)?;

        let initialized_operator = InitializedVectorTimeProjection {
            source,
            result_descriptor,
            step: self.params.step,
            step_reference,
        };

        Ok(initialized_operator.boxed())
    }

    span_fn!(TimeProjection);
}

fn rewrite_result_descriptor(
    result_descriptor: &mut VectorResultDescriptor,
    step: TimeStep,
    step_reference: TimeInstance,
) -> Result<()> {
    if let Some(time) = result_descriptor.time {
        let start = step.snap_relative(step_reference, time.start())?;
        let end = (step.snap_relative(step_reference, time.end())? + step)?;

        result_descriptor.time = Some(TimeInterval::new(start, end)?);
    }
    Ok(())
}

pub struct InitializedVectorTimeProjection {
    source: Box<dyn InitializedVectorOperator>,
    result_descriptor: VectorResultDescriptor,
    step: TimeStep,
    step_reference: TimeInstance,
}

impl InitializedVectorOperator for InitializedVectorTimeProjection {
    fn result_descriptor(&self) -> &VectorResultDescriptor {
        &self.result_descriptor
    }

    fn query_processor(&self) -> Result<TypedVectorQueryProcessor> {
        let source_processor = self.source.query_processor()?;

        Ok(
            call_on_generic_vector_processor!(source_processor, processor => VectorTimeProjectionProcessor {
                processor,
                step: self.step,
                step_reference: self.step_reference,
            }.boxed().into()),
        )
    }
}

pub struct VectorTimeProjectionProcessor<G>
where
    G: Geometry,
{
    processor: Box<dyn VectorQueryProcessor<VectorType = FeatureCollection<G>>>,
    step: TimeStep,
    step_reference: TimeInstance,
}

#[async_trait]
impl<G> VectorQueryProcessor for VectorTimeProjectionProcessor<G>
where
    G: Geometry + ArrowTyped + 'static,
{
    type VectorType = FeatureCollection<G>;

    async fn vector_query<'a>(
        &'a self,
        query: VectorQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<Self::VectorType>>> {
        let query = self.expand_query_rectangle(query)?;
        let stream = self
            .processor
            .vector_query(query, ctx)
            .await?
            .and_then(|collection| {
                self.expand_feature_collection_result(collection, ctx.thread_pool().clone())
            })
            .boxed();
        Ok(stream)
    }
}

impl<G> VectorTimeProjectionProcessor<G>
where
    G: Geometry + ArrowTyped + 'static,
{
    fn expand_query_rectangle(&self, query: VectorQueryRectangle) -> Result<VectorQueryRectangle> {
        Ok(expand_query_rectangle(
            self.step,
            self.step_reference,
            query,
        )?)
    }

    async fn expand_feature_collection_result(
        &self,
        feature_collection: FeatureCollection<G>,
        thread_pool: Arc<ThreadPool>,
    ) -> Result<FeatureCollection<G>> {
        let step = self.step;
        let step_reference = self.step_reference;

        crate::util::spawn_blocking_with_thread_pool(thread_pool, move || {
            Self::expand_feature_collection_result_inner(feature_collection, step, step_reference)
        })
        .await?
        .map_err(Into::into)
    }

    fn expand_feature_collection_result_inner(
        feature_collection: FeatureCollection<G>,
        step: TimeStep,
        step_reference: TimeInstance,
    ) -> Result<FeatureCollection<G>, TimeProjectionError> {
        let time_intervals: Option<Result<Vec<TimeInterval>, TimeProjectionError>> =
            feature_collection
                .time_intervals()
                .par_iter()
                .with_min_len(128) // TODO: find good default
                .map(|time_interval| expand_time_interval(step, step_reference, *time_interval))
                // TODO: change to [`try_collect_into_vec`](https://github.com/rayon-rs/rayon/issues/713) if available
                .try_fold_with(Vec::new(), |mut acc, time_interval| {
                    acc.push(time_interval?);
                    Ok(acc)
                })
                .try_reduce_with(|a, b| Ok([a, b].concat()));

        match time_intervals {
            Some(Ok(time_intervals)) => feature_collection
                .replace_time(&time_intervals)
                .context(error::CannotExpandFeatureTimeInterval),
            Some(Err(error)) => Err(error),
            None => Ok(feature_collection), // was empty
        }
    }
}

fn expand_query_rectangle(
    step: TimeStep,
    step_reference: TimeInstance,
    query: VectorQueryRectangle,
) -> Result<VectorQueryRectangle, TimeProjectionError> {
    Ok(VectorQueryRectangle {
        spatial_bounds: query.spatial_bounds,
        time_interval: expand_time_interval(step, step_reference, query.time_interval)?,
        spatial_resolution: query.spatial_resolution,
    })
}

fn expand_time_interval(
    step: TimeStep,
    step_reference: TimeInstance,
    time_interval: TimeInterval,
) -> Result<TimeInterval, TimeProjectionError> {
    let start = step.snap_relative_preserve_bounds(step_reference, time_interval.start());
    let mut end = step.snap_relative_preserve_bounds(step_reference, time_interval.end());

    // Since snap_relative snaps to the "left side", we have to add one step to the end
    // This only applies if `time_interval.end()` is not the snap point itself.
    if end < time_interval.end() {
        end = match end + step {
            Ok(end) => end,
            Err(_) => TimeInstance::MAX, // `TimeInterval::MAX` can overflow
        };
    }

    TimeInterval::new(start, end).context(error::CannotExpandQueryRectangle)
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::{
        engine::{MockExecutionContext, MockQueryContext},
        mock::MockFeatureCollectionSource,
    };
    use geoengine_datatypes::{
        collections::{MultiPointCollection, VectorDataType},
        primitives::{
            BoundingBox2D, DateTime, MultiPoint, SpatialResolution, TimeGranularity, TimeInterval,
        },
        spatial_reference::SpatialReference,
        util::test::TestDefault,
    };

    #[test]
    #[allow(clippy::too_many_lines)]
    fn test_expand_query_time_interval() {
        fn assert_time_interval_transform<T1: TryInto<TimeInstance>, T2: TryInto<TimeInstance>>(
            t1: T1,
            t2: T1,
            step: TimeStep,
            step_reference: TimeInstance,
            t1_expanded: T2,
            t2_expanded: T2,
        ) where
            T1::Error: std::fmt::Debug,
            T2::Error: std::fmt::Debug,
        {
            let result = expand_time_interval(
                step,
                step_reference,
                TimeInterval::new(t1.try_into().unwrap(), t2.try_into().unwrap()).unwrap(),
            )
            .unwrap();
            let expected = TimeInterval::new(
                t1_expanded.try_into().unwrap(),
                t2_expanded.try_into().unwrap(),
            )
            .unwrap();

            assert_eq!(
                result,
                expected,
                "[{}, {}) != [{}, {})",
                result.start().as_datetime_string(),
                result.end().as_datetime_string(),
                expected.start().as_datetime_string(),
                expected.end().as_datetime_string(),
            );
        }

        assert_time_interval_transform(
            DateTime::new_utc(2010, 1, 1, 0, 0, 0),
            DateTime::new_utc(2011, 1, 1, 0, 0, 0),
            TimeStep {
                granularity: TimeGranularity::Years,
                step: 1,
            },
            TimeInstance::from_millis(0).unwrap(),
            DateTime::new_utc(2010, 1, 1, 0, 0, 0),
            DateTime::new_utc(2011, 1, 1, 0, 0, 0),
        );

        assert_time_interval_transform(
            DateTime::new_utc(2010, 4, 3, 0, 0, 0),
            DateTime::new_utc(2010, 5, 14, 0, 0, 0),
            TimeStep {
                granularity: TimeGranularity::Years,
                step: 1,
            },
            TimeInstance::from_millis(0).unwrap(),
            DateTime::new_utc(2010, 1, 1, 0, 0, 0),
            DateTime::new_utc(2011, 1, 1, 0, 0, 0),
        );

        assert_time_interval_transform(
            DateTime::new_utc(2009, 4, 3, 0, 0, 0),
            DateTime::new_utc(2010, 5, 14, 0, 0, 0),
            TimeStep {
                granularity: TimeGranularity::Years,
                step: 1,
            },
            TimeInstance::from_millis(0).unwrap(),
            DateTime::new_utc(2009, 1, 1, 0, 0, 0),
            DateTime::new_utc(2011, 1, 1, 0, 0, 0),
        );

        assert_time_interval_transform(
            DateTime::new_utc(2009, 4, 3, 0, 0, 0),
            DateTime::new_utc(2010, 5, 14, 0, 0, 0),
            TimeStep {
                granularity: TimeGranularity::Months,
                step: 6,
            },
            TimeInstance::from(DateTime::new_utc(2010, 3, 1, 0, 0, 0)),
            DateTime::new_utc(2009, 3, 1, 0, 0, 0),
            DateTime::new_utc(2010, 9, 1, 0, 0, 0),
        );

        assert_time_interval_transform(
            DateTime::new_utc(2009, 4, 3, 0, 0, 0),
            DateTime::new_utc(2010, 5, 14, 0, 0, 0),
            TimeStep {
                granularity: TimeGranularity::Months,
                step: 6,
            },
            TimeInstance::from(DateTime::new_utc(2020, 1, 1, 0, 0, 0)),
            DateTime::new_utc(2009, 1, 1, 0, 0, 0),
            DateTime::new_utc(2010, 7, 1, 0, 0, 0),
        );

        assert_time_interval_transform(
            TimeInstance::MIN,
            TimeInstance::MAX,
            TimeStep {
                granularity: TimeGranularity::Months,
                step: 6,
            },
            TimeInstance::from(DateTime::new_utc(2010, 3, 1, 0, 0, 0)),
            TimeInstance::MIN,
            TimeInstance::MAX,
        );

        assert_time_interval_transform(
            TimeInstance::MIN + 1,
            TimeInstance::MAX - 1,
            TimeStep {
                granularity: TimeGranularity::Months,
                step: 6,
            },
            TimeInstance::from(DateTime::new_utc(2010, 3, 1, 0, 0, 0)),
            TimeInstance::MIN,
            TimeInstance::MAX,
        );
    }

    #[tokio::test]
    async fn single_year() {
        let execution_context = MockExecutionContext::test_default();
        let query_context = MockQueryContext::test_default();

        let source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),
                vec![
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2010, 12, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 6, 3, 0, 0, 0),
                        DateTime::new_utc(2010, 7, 14, 0, 0, 0),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2010, 3, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                ],
                Default::default(),
            )
            .unwrap(),
        );

        let time_projection = TimeProjection {
            sources: SingleVectorSource {
                vector: source.boxed(),
            },
            params: TimeProjectionParams {
                step: TimeStep {
                    granularity: TimeGranularity::Years,
                    step: 1,
                },
                step_reference: None,
            },
        };

        let query_processor = time_projection
            .boxed()
            .initialize(&execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let mut stream = query_processor
            .vector_query(
                VectorQueryRectangle {
                    spatial_bounds: BoundingBox2D::new((0., 0.).into(), (2., 2.).into()).unwrap(),
                    time_interval: TimeInterval::new(
                        DateTime::new_utc(2010, 4, 3, 0, 0, 0),
                        DateTime::new_utc(2010, 5, 14, 0, 0, 0),
                    )
                    .unwrap(),
                    spatial_resolution: SpatialResolution::one(),
                },
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(collection) = stream.next().await {
            result.push(collection.unwrap());
        }

        assert_eq!(result.len(), 1);

        let expected = MultiPointCollection::from_data(
            MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),
            vec![
                TimeInterval::new(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                )
                .unwrap(),
                TimeInterval::new(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                )
                .unwrap(),
                TimeInterval::new(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                )
                .unwrap(),
            ],
            Default::default(),
        )
        .unwrap();

        assert_eq!(result[0], expected);
    }

    #[tokio::test]
    async fn over_a_year() {
        let execution_context = MockExecutionContext::test_default();
        let query_context = MockQueryContext::test_default();

        let source = MockFeatureCollectionSource::single(
            MultiPointCollection::from_data(
                MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),
                vec![
                    TimeInterval::new(
                        DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2010, 12, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2009, 6, 3, 0, 0, 0),
                        DateTime::new_utc(2010, 7, 14, 0, 0, 0),
                    )
                    .unwrap(),
                    TimeInterval::new(
                        DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                        DateTime::new_utc_with_millis(2011, 3, 31, 23, 59, 59, 999),
                    )
                    .unwrap(),
                ],
                Default::default(),
            )
            .unwrap(),
        );

        let time_projection = TimeProjection {
            sources: SingleVectorSource {
                vector: source.boxed(),
            },
            params: TimeProjectionParams {
                step: TimeStep {
                    granularity: TimeGranularity::Years,
                    step: 1,
                },
                step_reference: None,
            },
        };

        let query_processor = time_projection
            .boxed()
            .initialize(&execution_context)
            .await
            .unwrap()
            .query_processor()
            .unwrap()
            .multi_point()
            .unwrap();

        let mut stream = query_processor
            .vector_query(
                VectorQueryRectangle {
                    spatial_bounds: BoundingBox2D::new((0., 0.).into(), (2., 2.).into()).unwrap(),
                    time_interval: TimeInterval::new(
                        DateTime::new_utc(2010, 4, 3, 0, 0, 0),
                        DateTime::new_utc(2010, 5, 14, 0, 0, 0),
                    )
                    .unwrap(),
                    spatial_resolution: SpatialResolution::one(),
                },
                &query_context,
            )
            .await
            .unwrap();

        let mut result = Vec::new();
        while let Some(collection) = stream.next().await {
            result.push(collection.unwrap());
        }

        assert_eq!(result.len(), 1);

        let expected = MultiPointCollection::from_data(
            MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),
            vec![
                TimeInterval::new(
                    DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                )
                .unwrap(),
                TimeInterval::new(
                    DateTime::new_utc(2009, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2011, 1, 1, 0, 0, 0),
                )
                .unwrap(),
                TimeInterval::new(
                    DateTime::new_utc(2010, 1, 1, 0, 0, 0),
                    DateTime::new_utc(2012, 1, 1, 0, 0, 0),
                )
                .unwrap(),
            ],
            Default::default(),
        )
        .unwrap();

        assert_eq!(result[0], expected);
    }

    #[test]
    fn it_rewrites_result_descriptor() {
        let mut result_descriptor = VectorResultDescriptor {
            data_type: VectorDataType::MultiPoint,
            spatial_reference: SpatialReference::epsg_4326().into(),
            columns: Default::default(),
            time: Some(TimeInterval::new_unchecked(30_000, 90_000)),
            bbox: None,
        };

        rewrite_result_descriptor(
            &mut result_descriptor,
            TimeStep {
                granularity: TimeGranularity::Minutes,
                step: 1,
            },
            TimeInstance::from_millis_unchecked(0),
        )
        .unwrap();

        assert_eq!(
            result_descriptor,
            VectorResultDescriptor {
                data_type: VectorDataType::MultiPoint,
                spatial_reference: SpatialReference::epsg_4326().into(),
                columns: Default::default(),
                time: Some(TimeInterval::new_unchecked(0, 120_000)),
                bbox: None,
            }
        );
    }
}

1	use std::sync::Arc;
2
3	use crate::engine::{
4	CreateSpan, ExecutionContext, InitializedVectorOperator, Operator, OperatorName, QueryContext,
5	SingleVectorSource, TypedVectorQueryProcessor, VectorOperator, VectorQueryProcessor,
6	VectorResultDescriptor,
7	};
8	use crate::util::Result;
9	use async_trait::async_trait;
10	use futures::stream::BoxStream;
11	use futures::{StreamExt, TryStreamExt};
12	use geoengine_datatypes::collections::{
13	FeatureCollection, FeatureCollectionInfos, FeatureCollectionModifications,
14	};
15	use geoengine_datatypes::primitives::{Geometry, TimeInterval};
16	use geoengine_datatypes::primitives::{TimeInstance, TimeStep, VectorQueryRectangle};
17	use geoengine_datatypes::util::arrow::ArrowTyped;
18	use log::debug;
19	use rayon::iter::{IndexedParallelIterator, IntoParallelRefIterator, ParallelIterator};
20	use rayon::ThreadPool;
21	use serde::{Deserialize, Serialize};
22	use snafu::{ensure, ResultExt, Snafu};
23	use tracing::{span, Level};
24
25	/// Projection of time information in queries and data
26	///
27	/// This operator changes the temporal validity of the queried data.
28	/// In order to query all valid data, it is necessary to change the query rectangle as well.
29	///
30	pub type TimeProjection = Operator<TimeProjectionParams, SingleVectorSource>;
31
32	impl OperatorName for TimeProjection {
33	const TYPE_NAME: &'static str = "TimeProjection";
34	}
35
36	#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]	×
37	pub struct TimeProjectionParams {
38	/// Specify the time step granularity and size
39	step: TimeStep,
40	/// Define an anchor point for `step`
41	/// If `None`, the anchor point is `1970-01-01T00:00:00Z` by default
42	step_reference: Option<TimeInstance>,
43	}
44
45	#[derive(Debug, Snafu)]	×
46	#[snafu(visibility(pub(crate)), context(suffix(false)), module(error))]
47	pub enum TimeProjectionError {
48	#[snafu(display("Time step must be larger than zero"))]
49	WindowSizeMustNotBeZero,
50
51	#[snafu(display("Query rectangle expansion failed: {}", source))]
52	CannotExpandQueryRectangle {
53	source: geoengine_datatypes::error::Error,
54	},
55
56	#[snafu(display("Feature time interval expansion failed: {}", source))]
57	CannotExpandFeatureTimeInterval {
58	source: geoengine_datatypes::error::Error,
59	},
60	}
61
62	#[typetag::serde]	×
63	#[async_trait]
64	impl VectorOperator for TimeProjection {
65	async fn _initialize(	2✔
66	self: Box<Self>,	2✔
67	context: &dyn ExecutionContext,	2✔
68	) -> Result<Box<dyn InitializedVectorOperator>> {	2✔
69	ensure!(self.params.step.step > 0, error::WindowSizeMustNotBeZero);	2✔
70
71	let source = self.sources.vector.initialize(context).await?;	2✔
72
73	debug!("Initializing `TimeProjection` with {:?}.", &self.params);	2✔
74
75	let step_reference = self	2✔
76	.params	2✔
77	.step_reference	2✔
78	// use UTC 0 as default	2✔
79	.unwrap_or(TimeInstance::EPOCH_START);	2✔
80		2✔
81	let mut result_descriptor = source.result_descriptor().clone();	2✔
82	rewrite_result_descriptor(&mut result_descriptor, self.params.step, step_reference)?;	2✔
83
84	let initialized_operator = InitializedVectorTimeProjection {	2✔
85	source,	2✔
86	result_descriptor,	2✔
87	step: self.params.step,	2✔
88	step_reference,	2✔
89	};	2✔
90		2✔
91	Ok(initialized_operator.boxed())	2✔
92	}	4✔
93
94	span_fn!(TimeProjection);	×
95	}
96
97	fn rewrite_result_descriptor(
98	result_descriptor: &mut VectorResultDescriptor,
99	step: TimeStep,
100	step_reference: TimeInstance,
101	) -> Result<()> {
102	if let Some(time) = result_descriptor.time {	3✔
103	let start = step.snap_relative(step_reference, time.start())?;	1✔
104	let end = (step.snap_relative(step_reference, time.end())? + step)?;	1✔
105
106	result_descriptor.time = Some(TimeInterval::new(start, end)?);	1✔
107	}	2✔
108	Ok(())	3✔
109	}	3✔
110
111	pub struct InitializedVectorTimeProjection {
112	source: Box<dyn InitializedVectorOperator>,
113	result_descriptor: VectorResultDescriptor,
114	step: TimeStep,
115	step_reference: TimeInstance,
116	}
117
118	impl InitializedVectorOperator for InitializedVectorTimeProjection {
119	fn result_descriptor(&self) -> &VectorResultDescriptor {	×
120	&self.result_descriptor	×
121	}	×
122
123	fn query_processor(&self) -> Result<TypedVectorQueryProcessor> {	2✔
124	let source_processor = self.source.query_processor()?;	2✔
125
126	Ok(
127	call_on_generic_vector_processor!(source_processor, processor => VectorTimeProjectionProcessor {	2✔
128	processor,	2✔
129	step: self.step,	2✔
130	step_reference: self.step_reference,	2✔
131	}.boxed().into()),	2✔
132	)
133	}	2✔
134	}
135
136	pub struct VectorTimeProjectionProcessor<G>
137	where
138	G: Geometry,
139	{
140	processor: Box<dyn VectorQueryProcessor<VectorType = FeatureCollection<G>>>,
141	step: TimeStep,
142	step_reference: TimeInstance,
143	}
144
145	#[async_trait]
146	impl<G> VectorQueryProcessor for VectorTimeProjectionProcessor<G>
147	where
148	G: Geometry + ArrowTyped + 'static,
149	{
150	type VectorType = FeatureCollection<G>;
151
152	async fn vector_query<'a>(	2✔
153	&'a self,	2✔
154	query: VectorQueryRectangle,	2✔
155	ctx: &'a dyn QueryContext,	2✔
156	) -> Result<BoxStream<'a, Result<Self::VectorType>>> {	2✔
157	let query = self.expand_query_rectangle(query)?;	2✔
158	let stream = self	2✔
159	.processor	2✔
160	.vector_query(query, ctx)	2✔
161	.await?	×
162	.and_then(\|collection\| {	2✔
163	self.expand_feature_collection_result(collection, ctx.thread_pool().clone())	2✔
164	})	2✔
165	.boxed();	2✔
166	Ok(stream)	2✔
167	}	4✔
168	}
169
170	impl<G> VectorTimeProjectionProcessor<G>
171	where
172	G: Geometry + ArrowTyped + 'static,
173	{
174	fn expand_query_rectangle(&self, query: VectorQueryRectangle) -> Result<VectorQueryRectangle> {	2✔
175	Ok(expand_query_rectangle(	2✔
176	self.step,	2✔
177	self.step_reference,	2✔
178	query,	2✔
179	)?)	2✔
180	}	2✔
181
182	async fn expand_feature_collection_result(	2✔
183	&self,	2✔
184	feature_collection: FeatureCollection<G>,	2✔
185	thread_pool: Arc<ThreadPool>,	2✔
186	) -> Result<FeatureCollection<G>> {	2✔
187	let step = self.step;	2✔
188	let step_reference = self.step_reference;	2✔
189		2✔
190	crate::util::spawn_blocking_with_thread_pool(thread_pool, move \|\| {	2✔
191	Self::expand_feature_collection_result_inner(feature_collection, step, step_reference)	2✔
192	})	2✔
193	.await?	2✔
194	.map_err(Into::into)	2✔
195	}	2✔
196
197	fn expand_feature_collection_result_inner(	2✔
198	feature_collection: FeatureCollection<G>,	2✔
199	step: TimeStep,	2✔
200	step_reference: TimeInstance,	2✔
201	) -> Result<FeatureCollection<G>, TimeProjectionError> {	2✔
202	let time_intervals: Option<Result<Vec<TimeInterval>, TimeProjectionError>> =	2✔
203	feature_collection	2✔
204	.time_intervals()	2✔
205	.par_iter()	2✔
206	.with_min_len(128) // TODO: find good default	2✔
207	.map(\|time_interval\| expand_time_interval(step, step_reference, *time_interval))	6✔
208	// TODO: change to [`try_collect_into_vec`](https://github.com/rayon-rs/rayon/issues/713) if available	2✔
209	.try_fold_with(Vec::new(), \|mut acc, time_interval\| {	2✔
210	acc.push(time_interval?);	6✔
211	Ok(acc)	6✔
212	})	6✔
213	.try_reduce_with(\|a, b\| Ok([a, b].concat()));	2✔
214
215	match time_intervals {	2✔
216	Some(Ok(time_intervals)) => feature_collection	2✔
217	.replace_time(&time_intervals)	2✔
218	.context(error::CannotExpandFeatureTimeInterval),	2✔
219	Some(Err(error)) => Err(error),	×
220	None => Ok(feature_collection), // was empty	×
221	}
222	}	2✔
223	}
224
225	fn expand_query_rectangle(	2✔
226	step: TimeStep,	2✔
227	step_reference: TimeInstance,	2✔
228	query: VectorQueryRectangle,	2✔
229	) -> Result<VectorQueryRectangle, TimeProjectionError> {	2✔
230	Ok(VectorQueryRectangle {	2✔
231	spatial_bounds: query.spatial_bounds,	2✔
232	time_interval: expand_time_interval(step, step_reference, query.time_interval)?,	2✔
233	spatial_resolution: query.spatial_resolution,	2✔
234	})
235	}	2✔
236
237	fn expand_time_interval(	15✔
238	step: TimeStep,	15✔
239	step_reference: TimeInstance,	15✔
240	time_interval: TimeInterval,	15✔
241	) -> Result<TimeInterval, TimeProjectionError> {	15✔
242	let start = step.snap_relative_preserve_bounds(step_reference, time_interval.start());	15✔
243	let mut end = step.snap_relative_preserve_bounds(step_reference, time_interval.end());	15✔
244		15✔
245	// Since snap_relative snaps to the "left side", we have to add one step to the end	15✔
246	// This only applies if `time_interval.end()` is not the snap point itself.	15✔
247	if end < time_interval.end() {	15✔
248	end = match end + step {	14✔
249	Ok(end) => end,	12✔
250	Err(_) => TimeInstance::MAX, // `TimeInterval::MAX` can overflow	2✔
251	};
252	}	1✔
253
254	TimeInterval::new(start, end).context(error::CannotExpandQueryRectangle)	15✔
255	}	15✔
256
257	#[cfg(test)]
258	mod tests {
259	use super::*;
260
261	use crate::{
262	engine::{MockExecutionContext, MockQueryContext},
263	mock::MockFeatureCollectionSource,
264	};
265	use geoengine_datatypes::{
266	collections::{MultiPointCollection, VectorDataType},
267	primitives::{
268	BoundingBox2D, DateTime, MultiPoint, SpatialResolution, TimeGranularity, TimeInterval,
269	},
270	spatial_reference::SpatialReference,
271	util::test::TestDefault,
272	};
273
274	#[test]	1✔
275	#[allow(clippy::too_many_lines)]
276	fn test_expand_query_time_interval() {	1✔
277	fn assert_time_interval_transform<T1: TryInto<TimeInstance>, T2: TryInto<TimeInstance>>(	7✔
278	t1: T1,	7✔
279	t2: T1,	7✔
280	step: TimeStep,	7✔
281	step_reference: TimeInstance,	7✔
282	t1_expanded: T2,	7✔
283	t2_expanded: T2,	7✔
284	) where	7✔
285	T1::Error: std::fmt::Debug,	7✔
286	T2::Error: std::fmt::Debug,	7✔
287	{	7✔
288	let result = expand_time_interval(	7✔
289	step,	7✔
290	step_reference,	7✔
291	TimeInterval::new(t1.try_into().unwrap(), t2.try_into().unwrap()).unwrap(),	7✔
292	)	7✔
293	.unwrap();	7✔
294	let expected = TimeInterval::new(	7✔
295	t1_expanded.try_into().unwrap(),	7✔
296	t2_expanded.try_into().unwrap(),	7✔
297	)	7✔
298	.unwrap();	7✔
299		7✔
300	assert_eq!(	7✔
301	result,	1✔
302	expected,	1✔
303	"[{}, {}) != [{}, {})",	1✔
304	result.start().as_datetime_string(),	×
305	result.end().as_datetime_string(),	×
306	expected.start().as_datetime_string(),	×
307	expected.end().as_datetime_string(),	×
308	);	1✔
309	}	7✔
310		1✔
311	assert_time_interval_transform(	1✔
312	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
313	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
314	TimeStep {	1✔
315	granularity: TimeGranularity::Years,	1✔
316	step: 1,	1✔
317	},	1✔
318	TimeInstance::from_millis(0).unwrap(),	1✔
319	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
320	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
321	);	1✔
322		1✔
323	assert_time_interval_transform(	1✔
324	DateTime::new_utc(2010, 4, 3, 0, 0, 0),	1✔
325	DateTime::new_utc(2010, 5, 14, 0, 0, 0),	1✔
326	TimeStep {	1✔
327	granularity: TimeGranularity::Years,	1✔
328	step: 1,	1✔
329	},	1✔
330	TimeInstance::from_millis(0).unwrap(),	1✔
331	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
332	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
333	);	1✔
334		1✔
335	assert_time_interval_transform(	1✔
336	DateTime::new_utc(2009, 4, 3, 0, 0, 0),	1✔
337	DateTime::new_utc(2010, 5, 14, 0, 0, 0),	1✔
338	TimeStep {	1✔
339	granularity: TimeGranularity::Years,	1✔
340	step: 1,	1✔
341	},	1✔
342	TimeInstance::from_millis(0).unwrap(),	1✔
343	DateTime::new_utc(2009, 1, 1, 0, 0, 0),	1✔
344	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
345	);	1✔
346		1✔
347	assert_time_interval_transform(	1✔
348	DateTime::new_utc(2009, 4, 3, 0, 0, 0),	1✔
349	DateTime::new_utc(2010, 5, 14, 0, 0, 0),	1✔
350	TimeStep {	1✔
351	granularity: TimeGranularity::Months,	1✔
352	step: 6,	1✔
353	},	1✔
354	TimeInstance::from(DateTime::new_utc(2010, 3, 1, 0, 0, 0)),	1✔
355	DateTime::new_utc(2009, 3, 1, 0, 0, 0),	1✔
356	DateTime::new_utc(2010, 9, 1, 0, 0, 0),	1✔
357	);	1✔
358		1✔
359	assert_time_interval_transform(	1✔
360	DateTime::new_utc(2009, 4, 3, 0, 0, 0),	1✔
361	DateTime::new_utc(2010, 5, 14, 0, 0, 0),	1✔
362	TimeStep {	1✔
363	granularity: TimeGranularity::Months,	1✔
364	step: 6,	1✔
365	},	1✔
366	TimeInstance::from(DateTime::new_utc(2020, 1, 1, 0, 0, 0)),	1✔
367	DateTime::new_utc(2009, 1, 1, 0, 0, 0),	1✔
368	DateTime::new_utc(2010, 7, 1, 0, 0, 0),	1✔
369	);	1✔
370		1✔
371	assert_time_interval_transform(	1✔
372	TimeInstance::MIN,	1✔
373	TimeInstance::MAX,	1✔
374	TimeStep {	1✔
375	granularity: TimeGranularity::Months,	1✔
376	step: 6,	1✔
377	},	1✔
378	TimeInstance::from(DateTime::new_utc(2010, 3, 1, 0, 0, 0)),	1✔
379	TimeInstance::MIN,	1✔
380	TimeInstance::MAX,	1✔
381	);	1✔
382		1✔
383	assert_time_interval_transform(	1✔
384	TimeInstance::MIN + 1,	1✔
385	TimeInstance::MAX - 1,	1✔
386	TimeStep {	1✔
387	granularity: TimeGranularity::Months,	1✔
388	step: 6,	1✔
389	},	1✔
390	TimeInstance::from(DateTime::new_utc(2010, 3, 1, 0, 0, 0)),	1✔
391	TimeInstance::MIN,	1✔
392	TimeInstance::MAX,	1✔
393	);	1✔
394	}	1✔
395
396	#[tokio::test]	1✔
397	async fn single_year() {	1✔
398	let execution_context = MockExecutionContext::test_default();	1✔
399	let query_context = MockQueryContext::test_default();	1✔
400		1✔
401	let source = MockFeatureCollectionSource::single(	1✔
402	MultiPointCollection::from_data(	1✔
403	MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),	1✔
404	vec![	1✔
405	TimeInterval::new(	1✔
406	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
407	DateTime::new_utc_with_millis(2010, 12, 31, 23, 59, 59, 999),	1✔
408	)	1✔
409	.unwrap(),	1✔
410	TimeInterval::new(	1✔
411	DateTime::new_utc(2010, 6, 3, 0, 0, 0),	1✔
412	DateTime::new_utc(2010, 7, 14, 0, 0, 0),	1✔
413	)	1✔
414	.unwrap(),	1✔
415	TimeInterval::new(	1✔
416	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
417	DateTime::new_utc_with_millis(2010, 3, 31, 23, 59, 59, 999),	1✔
418	)	1✔
419	.unwrap(),	1✔
420	],	1✔
421	Default::default(),	1✔
422	)	1✔
423	.unwrap(),	1✔
424	);	1✔
425		1✔
426	let time_projection = TimeProjection {	1✔
427	sources: SingleVectorSource {	1✔
428	vector: source.boxed(),	1✔
429	},	1✔
430	params: TimeProjectionParams {	1✔
431	step: TimeStep {	1✔
432	granularity: TimeGranularity::Years,	1✔
433	step: 1,	1✔
434	},	1✔
435	step_reference: None,	1✔
436	},	1✔
437	};	1✔
438
439	let query_processor = time_projection	1✔
440	.boxed()	1✔
441	.initialize(&execution_context)	1✔
442	.await	×
443	.unwrap()	1✔
444	.query_processor()	1✔
445	.unwrap()	1✔
446	.multi_point()	1✔
447	.unwrap();	1✔
448
449	let mut stream = query_processor	1✔
450	.vector_query(	1✔
451	VectorQueryRectangle {	1✔
452	spatial_bounds: BoundingBox2D::new((0., 0.).into(), (2., 2.).into()).unwrap(),	1✔
453	time_interval: TimeInterval::new(	1✔
454	DateTime::new_utc(2010, 4, 3, 0, 0, 0),	1✔
455	DateTime::new_utc(2010, 5, 14, 0, 0, 0),	1✔
456	)	1✔
457	.unwrap(),	1✔
458	spatial_resolution: SpatialResolution::one(),	1✔
459	},	1✔
460	&query_context,	1✔
461	)	1✔
462	.await	×
463	.unwrap();	1✔
464		1✔
465	let mut result = Vec::new();	1✔
466	while let Some(collection) = stream.next().await {	2✔
467	result.push(collection.unwrap());	1✔
468	}	1✔
469
470	assert_eq!(result.len(), 1);	1✔
471
472	let expected = MultiPointCollection::from_data(	1✔
473	MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),	1✔
474	vec![	1✔
475	TimeInterval::new(	1✔
476	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
477	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
478	)	1✔
479	.unwrap(),	1✔
480	TimeInterval::new(	1✔
481	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
482	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
483	)	1✔
484	.unwrap(),	1✔
485	TimeInterval::new(	1✔
486	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
487	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
488	)	1✔
489	.unwrap(),	1✔
490	],	1✔
491	Default::default(),	1✔
492	)	1✔
493	.unwrap();	1✔
494		1✔
495	assert_eq!(result[0], expected);	1✔
496	}
497
498	#[tokio::test]	1✔
499	async fn over_a_year() {	1✔
500	let execution_context = MockExecutionContext::test_default();	1✔
501	let query_context = MockQueryContext::test_default();	1✔
502		1✔
503	let source = MockFeatureCollectionSource::single(	1✔
504	MultiPointCollection::from_data(	1✔
505	MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),	1✔
506	vec![	1✔
507	TimeInterval::new(	1✔
508	DateTime::new_utc(2009, 1, 1, 0, 0, 0),	1✔
509	DateTime::new_utc_with_millis(2010, 12, 31, 23, 59, 59, 999),	1✔
510	)	1✔
511	.unwrap(),	1✔
512	TimeInterval::new(	1✔
513	DateTime::new_utc(2009, 6, 3, 0, 0, 0),	1✔
514	DateTime::new_utc(2010, 7, 14, 0, 0, 0),	1✔
515	)	1✔
516	.unwrap(),	1✔
517	TimeInterval::new(	1✔
518	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
519	DateTime::new_utc_with_millis(2011, 3, 31, 23, 59, 59, 999),	1✔
520	)	1✔
521	.unwrap(),	1✔
522	],	1✔
523	Default::default(),	1✔
524	)	1✔
525	.unwrap(),	1✔
526	);	1✔
527		1✔
528	let time_projection = TimeProjection {	1✔
529	sources: SingleVectorSource {	1✔
530	vector: source.boxed(),	1✔
531	},	1✔
532	params: TimeProjectionParams {	1✔
533	step: TimeStep {	1✔
534	granularity: TimeGranularity::Years,	1✔
535	step: 1,	1✔
536	},	1✔
537	step_reference: None,	1✔
538	},	1✔
539	};	1✔
540
541	let query_processor = time_projection	1✔
542	.boxed()	1✔
543	.initialize(&execution_context)	1✔
544	.await	×
545	.unwrap()	1✔
546	.query_processor()	1✔
547	.unwrap()	1✔
548	.multi_point()	1✔
549	.unwrap();	1✔
550
551	let mut stream = query_processor	1✔
552	.vector_query(	1✔
553	VectorQueryRectangle {	1✔
554	spatial_bounds: BoundingBox2D::new((0., 0.).into(), (2., 2.).into()).unwrap(),	1✔
555	time_interval: TimeInterval::new(	1✔
556	DateTime::new_utc(2010, 4, 3, 0, 0, 0),	1✔
557	DateTime::new_utc(2010, 5, 14, 0, 0, 0),	1✔
558	)	1✔
559	.unwrap(),	1✔
560	spatial_resolution: SpatialResolution::one(),	1✔
561	},	1✔
562	&query_context,	1✔
563	)	1✔
564	.await	×
565	.unwrap();	1✔
566		1✔
567	let mut result = Vec::new();	1✔
568	while let Some(collection) = stream.next().await {	2✔
569	result.push(collection.unwrap());	1✔
570	}	1✔
571
572	assert_eq!(result.len(), 1);	1✔
573
574	let expected = MultiPointCollection::from_data(	1✔
575	MultiPoint::many(vec![(0., 0.), (1., 1.), (2., 2.)]).unwrap(),	1✔
576	vec![	1✔
577	TimeInterval::new(	1✔
578	DateTime::new_utc(2009, 1, 1, 0, 0, 0),	1✔
579	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
580	)	1✔
581	.unwrap(),	1✔
582	TimeInterval::new(	1✔
583	DateTime::new_utc(2009, 1, 1, 0, 0, 0),	1✔
584	DateTime::new_utc(2011, 1, 1, 0, 0, 0),	1✔
585	)	1✔
586	.unwrap(),	1✔
587	TimeInterval::new(	1✔
588	DateTime::new_utc(2010, 1, 1, 0, 0, 0),	1✔
589	DateTime::new_utc(2012, 1, 1, 0, 0, 0),	1✔
590	)	1✔
591	.unwrap(),	1✔
592	],	1✔
593	Default::default(),	1✔
594	)	1✔
595	.unwrap();	1✔
596		1✔
597	assert_eq!(result[0], expected);	1✔
598	}
599
600	#[test]	1✔
601	fn it_rewrites_result_descriptor() {	1✔
602	let mut result_descriptor = VectorResultDescriptor {	1✔
603	data_type: VectorDataType::MultiPoint,	1✔
604	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
605	columns: Default::default(),	1✔
606	time: Some(TimeInterval::new_unchecked(30_000, 90_000)),	1✔
607	bbox: None,	1✔
608	};	1✔
609		1✔
610	rewrite_result_descriptor(	1✔
611	&mut result_descriptor,	1✔
612	TimeStep {	1✔
613	granularity: TimeGranularity::Minutes,	1✔
614	step: 1,	1✔
615	},	1✔
616	TimeInstance::from_millis_unchecked(0),	1✔
617	)	1✔
618	.unwrap();	1✔
619		1✔
620	assert_eq!(	1✔
621	result_descriptor,	1✔
622	VectorResultDescriptor {	1✔
623	data_type: VectorDataType::MultiPoint,	1✔
624	spatial_reference: SpatialReference::epsg_4326().into(),	1✔
625	columns: Default::default(),	1✔
626	time: Some(TimeInterval::new_unchecked(0, 120_000)),	1✔
627	bbox: None,	1✔
628	}	1✔
629	);	1✔
630	}	1✔
631	}

geo-engine / geoengine / 3929938005

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