16354926043

Committed 17 Jul 2025 08:09PM UTC coverage: 88.872%. First build

Build # 16354926043

Build Type

Pull #1061

github

Committed by

web-flow

Commit Message

Merge 6307e082f into b8910c811

Pull Request Pull Request #1061: feat(operators): skip empty tiles and merge masks in onnx; remove trace/debug in release mode

Run Details

414 of 569 new or added lines in 44 files covered. (72.76%)

111581 of 125552 relevant lines covered (88.87%)

80316.26 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::StreamExt;
use futures::stream::BoxStream;
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 {
            tracing::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 {
            tracing::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::StreamExt;
11	use futures::stream::BoxStream;
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 {
NEW 52	tracing::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,	×
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 {
NEW 91	tracing::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 / 16354926043

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