7276350647

Committed 20 Dec 2023 01:21PM UTC coverage: 89.798% (-0.03%) from 89.823%

Build # 7276350647

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #906 from geo-engine/raster_result_describer

result descriptors for query processors

Run Details

1080 of 1240 new or added lines in 43 files covered. (87.1%)

11 existing lines in 3 files now uncovered.

115920 of 129090 relevant lines covered (89.8%)

58689.64 hits per line

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

0.0

/operators/src/processing/map_query.rs

use crate::adapters::{FillerTileCacheExpirationStrategy, SparseTilesFillAdapter};
use crate::engine::{
    QueryContext, RasterQueryProcessor, RasterResultDescriptor, VectorQueryProcessor,
    VectorResultDescriptor,
};
use crate::util::Result;
use async_trait::async_trait;
use futures::stream::BoxStream;
use futures::StreamExt;
use geoengine_datatypes::primitives::{RasterQueryRectangle, VectorQueryRectangle};
use geoengine_datatypes::raster::{RasterTile2D, TilingSpecification};

/// This `QueryProcessor` allows to rewrite a query. It does not change the data. Results of the children are forwarded.
pub(crate) struct MapQueryProcessor<S, Q, A, R> {
    source: S,
    result_descriptor: R,
    query_fn: Q,
    additional_data: A,
}

impl<S, Q, A, R> MapQueryProcessor<S, Q, A, R> {
    pub fn new(source: S, result_descriptor: R, query_fn: Q, additional_data: A) -> Self {
        Self {
            source,
            result_descriptor,
            query_fn,
            additional_data,
        }
    }
}

#[async_trait]
impl<S, Q> RasterQueryProcessor
    for MapQueryProcessor<S, Q, TilingSpecification, RasterResultDescriptor>
where
    S: RasterQueryProcessor,
    Q: Fn(RasterQueryRectangle) -> Result<Option<RasterQueryRectangle>> + Sync + Send,
{
    type RasterType = S::RasterType;
    async fn raster_query<'a>(
        &'a self,
        query: RasterQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<RasterTile2D<S::RasterType>>>> {
        let rewritten_query = (self.query_fn)(query.clone())?;

        if let Some(rewritten_query) = rewritten_query {
            self.source.raster_query(rewritten_query, ctx).await
        } else {
            log::debug!("Query was rewritten to empty query. Returning empty / filled stream.");
            let s = futures::stream::empty();

            // TODO: The input of the `SparseTilesFillAdapter` is empty here, so we can't derive the expiration, as there are no tiles to derive them from.
            //       As this is the result of the query not being rewritten, we should check if the expiration could also be `max`, because this error
            //       will be persistent and we might as well cache the empty stream.
            Ok(SparseTilesFillAdapter::new_like_subquery(
                s,
                &query,
                self.additional_data,
                FillerTileCacheExpirationStrategy::NoCache,
            )
            .boxed())
        }
    }

    fn raster_result_descriptor(&self) -> &RasterResultDescriptor {
        &self.result_descriptor
    }
}

#[async_trait]
impl<S, Q> VectorQueryProcessor for MapQueryProcessor<S, Q, (), VectorResultDescriptor>
where
    S: VectorQueryProcessor,
    Q: Fn(VectorQueryRectangle) -> Result<Option<VectorQueryRectangle>> + Sync + Send,
    S::VectorType: Send,
{
    type VectorType = S::VectorType;
    async fn vector_query<'a>(
        &'a self,
        query: VectorQueryRectangle,
        ctx: &'a dyn QueryContext,
    ) -> Result<BoxStream<'a, Result<Self::VectorType>>> {
        let rewritten_query = (self.query_fn)(query)?;
        if let Some(rewritten_query) = rewritten_query {
            self.source.vector_query(rewritten_query, ctx).await
        } else {
            log::debug!("Query was rewritten to empty query. Returning empty stream.");
            Ok(Box::pin(futures::stream::empty())) // TODO: should be empty collection?
        }
    }

    fn vector_result_descriptor(&self) -> &VectorResultDescriptor {
        self.source.vector_result_descriptor()
    }
}

1	use crate::adapters::{FillerTileCacheExpirationStrategy, SparseTilesFillAdapter};
2	use crate::engine::{
3	QueryContext, RasterQueryProcessor, RasterResultDescriptor, VectorQueryProcessor,
4	VectorResultDescriptor,
5	};
6	use crate::util::Result;
7	use async_trait::async_trait;
8	use futures::stream::BoxStream;
9	use futures::StreamExt;
10	use geoengine_datatypes::primitives::{RasterQueryRectangle, VectorQueryRectangle};
11	use geoengine_datatypes::raster::{RasterTile2D, TilingSpecification};
12
13	/// This `QueryProcessor` allows to rewrite a query. It does not change the data. Results of the children are forwarded.
14	pub(crate) struct MapQueryProcessor<S, Q, A, R> {
15	source: S,
16	result_descriptor: R,
17	query_fn: Q,
18	additional_data: A,
19	}
20
21	impl<S, Q, A, R> MapQueryProcessor<S, Q, A, R> {
NEW 22	pub fn new(source: S, result_descriptor: R, query_fn: Q, additional_data: A) -> Self {	×
23	Self {	×
24	source,	×
NEW 25	result_descriptor,	×
26	query_fn,	×
27	additional_data,	×
28	}	×
29	}	×
30	}
31
32	#[async_trait]
33	impl<S, Q> RasterQueryProcessor
34	for MapQueryProcessor<S, Q, TilingSpecification, RasterResultDescriptor>
35	where
36	S: RasterQueryProcessor,
37	Q: Fn(RasterQueryRectangle) -> Result<Option<RasterQueryRectangle>> + Sync + Send,
38	{
39	type RasterType = S::RasterType;
40	async fn raster_query<'a>(	×
41	&'a self,	×
42	query: RasterQueryRectangle,	×
43	ctx: &'a dyn QueryContext,	×
44	) -> Result<BoxStream<'a, Result<RasterTile2D<S::RasterType>>>> {	×
45	let rewritten_query = (self.query_fn)(query.clone())?;	×
46
47	if let Some(rewritten_query) = rewritten_query {	×
48	self.source.raster_query(rewritten_query, ctx).await	×
49	} else {
50	log::debug!("Query was rewritten to empty query. Returning empty / filled stream.");	×
51	let s = futures::stream::empty();	×
52		×
53	// TODO: The input of the `SparseTilesFillAdapter` is empty here, so we can't derive the expiration, as there are no tiles to derive them from.	×
54	// As this is the result of the query not being rewritten, we should check if the expiration could also be `max`, because this error	×
55	// will be persistent and we might as well cache the empty stream.	×
56	Ok(SparseTilesFillAdapter::new_like_subquery(	×
57	s,	×
58	&query,	×
59	self.additional_data,	×
60	FillerTileCacheExpirationStrategy::NoCache,	×
61	)	×
62	.boxed())	×
63	}
64	}	×
65
NEW 66	fn raster_result_descriptor(&self) -> &RasterResultDescriptor {	×
NEW 67	&self.result_descriptor	×
NEW 68	}	×
69	}
70
71	#[async_trait]
72	impl<S, Q> VectorQueryProcessor for MapQueryProcessor<S, Q, (), VectorResultDescriptor>
73	where
74	S: VectorQueryProcessor,
75	Q: Fn(VectorQueryRectangle) -> Result<Option<VectorQueryRectangle>> + Sync + Send,
76	S::VectorType: Send,
77	{
78	type VectorType = S::VectorType;
79	async fn vector_query<'a>(	×
80	&'a self,	×
81	query: VectorQueryRectangle,	×
82	ctx: &'a dyn QueryContext,	×
83	) -> Result<BoxStream<'a, Result<Self::VectorType>>> {	×
84	let rewritten_query = (self.query_fn)(query)?;	×
85	if let Some(rewritten_query) = rewritten_query {	×
86	self.source.vector_query(rewritten_query, ctx).await	×
87	} else {
88	log::debug!("Query was rewritten to empty query. Returning empty stream.");	×
89	Ok(Box::pin(futures::stream::empty())) // TODO: should be empty collection?	×
90	}
91	}	×
92
NEW 93	fn vector_result_descriptor(&self) -> &VectorResultDescriptor {	×
NEW 94	self.source.vector_result_descriptor()	×
NEW 95	}	×
96	}

geo-engine / geoengine / 7276350647

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