9989247261

Committed 18 Jul 2024 09:43AM UTC coverage: 90.682% (+0.03%) from 90.651%

Build # 9989247261

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #968 from geo-engine/first_last_nodata_fill

First_last_nodata_fill

Run Details

700 of 751 new or added lines in 18 files covered. (93.21%)

13 existing lines in 11 files now uncovered.

133925 of 147687 relevant lines covered (90.68%)

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

geo-engine / geoengine / 9989247261

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