29828895317

Committed 21 Jul 2026 12:08PM UTC coverage: 87.59% (-0.2%) from 87.839%

Build # 29828895317

Build Type

Pull #1192

github

Committed by

web-flow

Commit Message

Merge bd6b2979b into 1baac981e

Pull Request Pull Request #1192: feat: add Gdal process pool

Coverage Stats

3028 of 3724 new or added lines in 38 files covered. (81.31%)

16 existing lines in 5 files now uncovered.

123314 of 140785 relevant lines covered (87.59%)

481313.41 hits per line

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

97.3

/geoengine/operators/src/source/gdal_source/reader.rs

use gdal::raster::GdalType;
use geoengine_datatypes::raster::{
    EmptyGrid, GridBlit, GridBoundingBox2D, GridOrEmpty, Pixel, RasterProperties,
};
use num::FromPrimitive;

use crate::source::gdal_worker_process::{
    GdalDatasetParameters, GdalPoolDispatcher, GdalProcessPoolError, GdalProcessReadResult,
    GridAndProperties, process_common::GdalReadAdvise,
};

pub struct GdalPoolReader(GdalPoolDispatcher);

impl From<GdalPoolDispatcher> for GdalPoolReader {
    fn from(value: GdalPoolDispatcher) -> Self {
        GdalPoolReader(value)
    }
}

impl GdalPoolReader {
    #[inline]
    fn worker_instance(&self) -> &GdalPoolDispatcher {
        &self.0
    }

    /// Loads a tile using the separate process and the provided parameters and read advise.
    /// This is the method where the source operator attaches to the process pool.
    /// The method sends a message to the process pool and waits for the response. The response is then converted to a `RasterTile2D` and returned.
    /// The method also handles the case where the file is not found and the `file_not_found_handling` is set to `NoData` by returning a tile filled with nodata values.
    /// # Errors
    /// Returns a `GdalSourceError` if the process returns an error, or if the response cannot be converted to a `RasterTile2D`.
    ///
    /// # Panics
    /// Panics if the response from the process is not in the expected format, or if the grid blitting fails (which should not happen if the bounds are correct).
    pub async fn load_tile_data_process<T: Pixel + GdalType + FromPrimitive>(
        &self,
        dataset_params: GdalDatasetParameters,
        read_advise: GdalReadAdvise,
    ) -> Result<GridAndProperties<T, GridBoundingBox2D>, GdalProcessPoolError> {
        let shared_reader = crate::source::gdal_worker_process::reader::GdalPoolReader::from(
            self.worker_instance().clone(),
        );

        let GridAndProperties { grid, properties } = match shared_reader
            .read_tile_data::<T>(dataset_params, read_advise)
            .await?
        {
            GdalProcessReadResult::Grid(grid_and_properties) => *grid_and_properties,
            GdalProcessReadResult::FileNotFoundAsNoData => GridAndProperties {
                grid: GridOrEmpty::new_empty_shape(read_advise.bounds_of_target),
                properties: RasterProperties::default(),
            },
        };

        // blit the grid to the tile bounds if necessary
        let grid = if read_advise.direct_read() {
            grid
        } else {
            let mut tile_raster = GridOrEmpty::from(EmptyGrid::new(read_advise.bounds_of_target));
            tile_raster.grid_blit_from(&grid);
            tile_raster
        };

        Ok(GridAndProperties { grid, properties })
    }
}

#[cfg(test)]
mod tests {
    use geoengine_datatypes::{
        primitives::{AxisAlignedRectangle, CacheHint, SpatialPartition2D, TimeInterval},
        raster::{
            ChangeGridBounds, GeoTransform, GridShape2D, GridSize, RasterPropertiesEntry,
            RasterPropertiesEntryType, RasterPropertiesKey, RasterTile2D, TileInformation,
        },
        test_data,
    };

    use crate::source::gdal_worker_process::{
        FileNotFoundHandling, GdalDatasetGeoTransform, GdalMetadataMapping, GdalProcessPool,
        GdalProcessPoolAccess, WorkerConfig, process_common::GdalReadWindow,
    };

    use super::*;

    // TODO (low): name / test
    async fn load_ndvi_jan_2014_by_process(
        gdal_read_advice: GdalReadAdvise,
        tile_information: TileInformation,
        gdal_worker: GdalPoolDispatcher,
    ) -> Result<RasterTile2D<u8>, GdalProcessPoolError> {
        let dataset_params = GdalDatasetParameters {
            file_path: test_data!("raster/modis_ndvi/MOD13A2_M_NDVI_2014-01-01.TIFF").into(),
            rasterband_channel: 1,
            geo_transform: GdalDatasetGeoTransform {
                origin_coordinate: (-180., 90.).into(),
                x_pixel_size: 0.1,
                y_pixel_size: -0.1,
            },
            width: 3600,
            height: 1800,
            file_not_found_handling: FileNotFoundHandling::NoData,
            no_data_value: Some(0.),
            properties_mapping: Some(vec![
                GdalMetadataMapping {
                    source_key: RasterPropertiesKey {
                        domain: None,
                        key: "AREA_OR_POINT".to_string(),
                    },
                    target_type: RasterPropertiesEntryType::String,
                    target_key: RasterPropertiesKey {
                        domain: None,
                        key: "AREA_OR_POINT".to_string(),
                    },
                },
                GdalMetadataMapping {
                    source_key: RasterPropertiesKey {
                        domain: Some("IMAGE_STRUCTURE".to_string()),
                        key: "COMPRESSION".to_string(),
                    },
                    target_type: RasterPropertiesEntryType::String,
                    target_key: RasterPropertiesKey {
                        domain: Some("IMAGE_STRUCTURE_INFO".to_string()),
                        key: "COMPRESSION".to_string(),
                    },
                },
            ]),
            gdal_open_options: None,
            gdal_config_options: None,
            allow_alphaband_as_mask: true,
            retry: None,
        };

        GdalPoolReader::from(gdal_worker)
            .load_tile_data_process::<u8>(dataset_params, gdal_read_advice)
            .await
            .map(|r| {
                RasterTile2D::new_with_tile_info_and_properties(
                    TimeInterval::default(),
                    tile_information,
                    0,
                    r.grid.unbounded(),
                    r.properties,
                    CacheHint::default(),
                )
            })
            .map_err(Into::into)
    }

    fn tile_information_with_partition_and_shape(
        partition: SpatialPartition2D,
        shape: GridShape2D,
    ) -> TileInformation {
        let real_geotransform = GeoTransform::new(
            partition.upper_left(),
            partition.size_x() / shape.axis_size_x() as f64,
            -partition.size_y() / shape.axis_size_y() as f64,
        );

        TileInformation {
            tile_size_in_pixels: shape,
            global_tile_position: [0, 0].into(),
            global_geo_transform: real_geotransform,
        }
    }

    #[tokio::test]
    async fn test_load_tile_data_process() {
        let output_shape: GridShape2D = [8, 8].into();
        let output_bounds =
            SpatialPartition2D::new_unchecked((-180., 90.).into(), (-179.2, 89.2).into());

        let gdal_read_advice = GdalReadAdvise {
            gdal_read_widow: GdalReadWindow::new([0, 0].into(), output_shape),
            read_window_bounds: GridBoundingBox2D::new([0, 0], [7, 7]).unwrap(),
            bounds_of_target: GridBoundingBox2D::new([0, 0], [7, 7]).unwrap(),
            flip_y: false,
        };

        let tile_information =
            tile_information_with_partition_and_shape(output_bounds, output_shape);

        let gpp = GdalProcessPool::new(2, 2, 2, WorkerConfig::default());
        let gw = gpp.get_gdal_worker();

        let RasterTile2D {
            global_geo_transform: _,
            grid_array: grid,
            tile_position: _,
            band: _,
            time: _,
            properties,
            cache_hint: _,
        } = load_ndvi_jan_2014_by_process(gdal_read_advice, tile_information, gw)
            .await
            .unwrap();

        assert!(!grid.is_empty());

        let grid = grid.into_materialized_masked_grid();

        assert_eq!(grid.inner_grid.data.len(), 64);
        assert_eq!(
            grid.inner_grid.data,
            &[
                255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
                255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
                255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
                255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255
            ]
        );

        assert_eq!(grid.validity_mask.data.len(), 64);
        assert_eq!(grid.validity_mask.data, &[true; 64]);

        assert!((properties.scale_option()).is_none());
        assert!(properties.offset_option().is_none());
        assert_eq!(
            properties.get_property(&RasterPropertiesKey {
                domain: Some("IMAGE_STRUCTURE_INFO".to_string()),
                key: "COMPRESSION".to_string(),
            }),
            Some(&RasterPropertiesEntry::String("LZW".to_string()))
        );
        assert_eq!(
            properties.get_property(&RasterPropertiesKey {
                domain: None,
                key: "AREA_OR_POINT".to_string(),
            }),
            Some(&RasterPropertiesEntry::String("Area".to_string()))
        );
    }
}

1	use gdal::raster::GdalType;
2	use geoengine_datatypes::raster::{
3	EmptyGrid, GridBlit, GridBoundingBox2D, GridOrEmpty, Pixel, RasterProperties,
4	};
5	use num::FromPrimitive;
6
7	use crate::source::gdal_worker_process::{
8	GdalDatasetParameters, GdalPoolDispatcher, GdalProcessPoolError, GdalProcessReadResult,
9	GridAndProperties, process_common::GdalReadAdvise,
10	};
11
12	pub struct GdalPoolReader(GdalPoolDispatcher);
13
14	impl From<GdalPoolDispatcher> for GdalPoolReader {
15	fn from(value: GdalPoolDispatcher) -> Self {	1,767✔
16	GdalPoolReader(value)	1,767✔
17	}	1,767✔
18	}
19
20	impl GdalPoolReader {
21	#[inline]
22	fn worker_instance(&self) -> &GdalPoolDispatcher {	1,767✔
23	&self.0	1,767✔
24	}	1,767✔
25
26	/// Loads a tile using the separate process and the provided parameters and read advise.
27	/// This is the method where the source operator attaches to the process pool.
28	/// The method sends a message to the process pool and waits for the response. The response is then converted to a `RasterTile2D` and returned.
29	/// The method also handles the case where the file is not found and the `file_not_found_handling` is set to `NoData` by returning a tile filled with nodata values.
30	/// # Errors
31	/// Returns a `GdalSourceError` if the process returns an error, or if the response cannot be converted to a `RasterTile2D`.
32	///
33	/// # Panics
34	/// Panics if the response from the process is not in the expected format, or if the grid blitting fails (which should not happen if the bounds are correct).
35	pub async fn load_tile_data_process<T: Pixel + GdalType + FromPrimitive>(	1,767✔
36	&self,	1,767✔
37	dataset_params: GdalDatasetParameters,	1,767✔
38	read_advise: GdalReadAdvise,	1,767✔
39	) -> Result<GridAndProperties<T, GridBoundingBox2D>, GdalProcessPoolError> {	1,767✔
40	let shared_reader = crate::source::gdal_worker_process::reader::GdalPoolReader::from(	1,767✔
41	self.worker_instance().clone(),	1,767✔
42	);
43
44	let GridAndProperties { grid, properties } = match shared_reader	1,767✔
45	.read_tile_data::<T>(dataset_params, read_advise)	1,767✔
46	.await?	1,767✔
47	{
48	GdalProcessReadResult::Grid(grid_and_properties) => *grid_and_properties,	1,755✔
NEW 49	GdalProcessReadResult::FileNotFoundAsNoData => GridAndProperties {	×
NEW 50	grid: GridOrEmpty::new_empty_shape(read_advise.bounds_of_target),	×
NEW 51	properties: RasterProperties::default(),	×
NEW 52	},	×
53	};
54
55	// blit the grid to the tile bounds if necessary
56	let grid = if read_advise.direct_read() {	1,755✔
57	grid	1,137✔
58	} else {
59	let mut tile_raster = GridOrEmpty::from(EmptyGrid::new(read_advise.bounds_of_target));	618✔
60	tile_raster.grid_blit_from(&grid);	618✔
61	tile_raster	618✔
62	};
63
64	Ok(GridAndProperties { grid, properties })	1,755✔
65	}	1,755✔
66	}
67
68	#[cfg(test)]
69	mod tests {
70	use geoengine_datatypes::{
71	primitives::{AxisAlignedRectangle, CacheHint, SpatialPartition2D, TimeInterval},
72	raster::{
73	ChangeGridBounds, GeoTransform, GridShape2D, GridSize, RasterPropertiesEntry,
74	RasterPropertiesEntryType, RasterPropertiesKey, RasterTile2D, TileInformation,
75	},
76	test_data,
77	};
78
79	use crate::source::gdal_worker_process::{
80	FileNotFoundHandling, GdalDatasetGeoTransform, GdalMetadataMapping, GdalProcessPool,
81	GdalProcessPoolAccess, WorkerConfig, process_common::GdalReadWindow,
82	};
83
84	use super::*;
85
86	// TODO (low): name / test
87	async fn load_ndvi_jan_2014_by_process(	1✔
88	gdal_read_advice: GdalReadAdvise,	1✔
89	tile_information: TileInformation,	1✔
90	gdal_worker: GdalPoolDispatcher,	1✔
91	) -> Result<RasterTile2D<u8>, GdalProcessPoolError> {	1✔
92	let dataset_params = GdalDatasetParameters {	1✔
93	file_path: test_data!("raster/modis_ndvi/MOD13A2_M_NDVI_2014-01-01.TIFF").into(),	1✔
94	rasterband_channel: 1,	1✔
95	geo_transform: GdalDatasetGeoTransform {	1✔
96	origin_coordinate: (-180., 90.).into(),	1✔
97	x_pixel_size: 0.1,	1✔
98	y_pixel_size: -0.1,	1✔
99	},	1✔
100	width: 3600,	1✔
101	height: 1800,	1✔
102	file_not_found_handling: FileNotFoundHandling::NoData,	1✔
103	no_data_value: Some(0.),	1✔
104	properties_mapping: Some(vec![	1✔
105	GdalMetadataMapping {	1✔
106	source_key: RasterPropertiesKey {	1✔
107	domain: None,	1✔
108	key: "AREA_OR_POINT".to_string(),	1✔
109	},	1✔
110	target_type: RasterPropertiesEntryType::String,	1✔
111	target_key: RasterPropertiesKey {	1✔
112	domain: None,	1✔
113	key: "AREA_OR_POINT".to_string(),	1✔
114	},	1✔
115	},	1✔
116	GdalMetadataMapping {	1✔
117	source_key: RasterPropertiesKey {	1✔
118	domain: Some("IMAGE_STRUCTURE".to_string()),	1✔
119	key: "COMPRESSION".to_string(),	1✔
120	},	1✔
121	target_type: RasterPropertiesEntryType::String,	1✔
122	target_key: RasterPropertiesKey {	1✔
123	domain: Some("IMAGE_STRUCTURE_INFO".to_string()),	1✔
124	key: "COMPRESSION".to_string(),	1✔
125	},	1✔
126	},	1✔
127	]),	1✔
128	gdal_open_options: None,	1✔
129	gdal_config_options: None,	1✔
130	allow_alphaband_as_mask: true,	1✔
131	retry: None,	1✔
132	};	1✔
133
134	GdalPoolReader::from(gdal_worker)	1✔
135	.load_tile_data_process::<u8>(dataset_params, gdal_read_advice)	1✔
136	.await	1✔
137	.map(\|r\| {	1✔
138	RasterTile2D::new_with_tile_info_and_properties(	1✔
139	TimeInterval::default(),	1✔
140	tile_information,	1✔
141	0,
142	r.grid.unbounded(),	1✔
143	r.properties,	1✔
144	CacheHint::default(),	1✔
145	)
146	})	1✔
147	.map_err(Into::into)	1✔
148	}	1✔
149
150	fn tile_information_with_partition_and_shape(	1✔
151	partition: SpatialPartition2D,	1✔
152	shape: GridShape2D,	1✔
153	) -> TileInformation {	1✔
154	let real_geotransform = GeoTransform::new(	1✔
155	partition.upper_left(),	1✔
156	partition.size_x() / shape.axis_size_x() as f64,	1✔
157	-partition.size_y() / shape.axis_size_y() as f64,	1✔
158	);
159
160	TileInformation {	1✔
161	tile_size_in_pixels: shape,	1✔
162	global_tile_position: [0, 0].into(),	1✔
163	global_geo_transform: real_geotransform,	1✔
164	}	1✔
165	}	1✔
166
167	#[tokio::test]
168	async fn test_load_tile_data_process() {	1✔
169	let output_shape: GridShape2D = [8, 8].into();	1✔
170	let output_bounds =	1✔
171	SpatialPartition2D::new_unchecked((-180., 90.).into(), (-179.2, 89.2).into());	1✔
172
173	let gdal_read_advice = GdalReadAdvise {	1✔
174	gdal_read_widow: GdalReadWindow::new([0, 0].into(), output_shape),	1✔
175	read_window_bounds: GridBoundingBox2D::new([0, 0], [7, 7]).unwrap(),	1✔
176	bounds_of_target: GridBoundingBox2D::new([0, 0], [7, 7]).unwrap(),	1✔
177	flip_y: false,	1✔
178	};	1✔
179
180	let tile_information =	1✔
181	tile_information_with_partition_and_shape(output_bounds, output_shape);	1✔
182
183	let gpp = GdalProcessPool::new(2, 2, 2, WorkerConfig::default());	1✔
184	let gw = gpp.get_gdal_worker();	1✔
185
186	let RasterTile2D {	1✔
187	global_geo_transform: _,	1✔
188	grid_array: grid,	1✔
189	tile_position: _,	1✔
190	band: _,	1✔
191	time: _,	1✔
192	properties,	1✔
193	cache_hint: _,	1✔
194	} = load_ndvi_jan_2014_by_process(gdal_read_advice, tile_information, gw)	1✔
195	.await	1✔
196	.unwrap();	1✔
197
198	assert!(!grid.is_empty());	1✔
199
200	let grid = grid.into_materialized_masked_grid();	1✔
201
202	assert_eq!(grid.inner_grid.data.len(), 64);	1✔
203	assert_eq!(	1✔
204	grid.inner_grid.data,
205	&[
206	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
207	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
208	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
209	255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255
210	]
211	);
212
213	assert_eq!(grid.validity_mask.data.len(), 64);	1✔
214	assert_eq!(grid.validity_mask.data, &[true; 64]);	1✔
215
216	assert!((properties.scale_option()).is_none());	1✔
217	assert!(properties.offset_option().is_none());	1✔
218	assert_eq!(	1✔
219	properties.get_property(&RasterPropertiesKey {	1✔
220	domain: Some("IMAGE_STRUCTURE_INFO".to_string()),	1✔
221	key: "COMPRESSION".to_string(),	1✔
222	}),	1✔
223	Some(&RasterPropertiesEntry::String("LZW".to_string()))	1✔
224	);
225	assert_eq!(	1✔
226	properties.get_property(&RasterPropertiesKey {	1✔
227	domain: None,	1✔
228	key: "AREA_OR_POINT".to_string(),	1✔
229	}),	1✔
230	Some(&RasterPropertiesEntry::String("Area".to_string()))	1✔
231	);	1✔
232	}	1✔
233	}

geo-engine / geoengine / 29828895317

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