29957817213

Committed 22 Jul 2026 08:32PM UTC coverage: 87.585% (-0.2%) from 87.738%

Build # 29957817213

Build Type

push

github

Committed by

web-flow

Commit Message

feat: add Gdal process pool (#1192)

* ipc-channel based gdal process

Co-authored-by: Mika <78550479+0xmycf@users.noreply.github.com>

* use thread and transfer only tile grid data

* LazyLock

* gdal pool

* working pool, tests, lints

* revert arrow serialization for gdal ipc data

* fix end of file

* clarify field name

* update gdalsource-process binary handling

* fix lazy gdalsource-process path detection

* justfile: add run --release

* remove or downgrade debug logs in hot paths

* gdal pool: add more complex affinity calculation

* gdal process pool with more caching workers

* simplified pool again

* batching pool

* gdal pool next iteration

* more structure

* more constants, less unwrap. Add decay.

* make load_tile_data_process more reusable

* adapt multi tile source part 1

* adapt  gdal multi source. sort errors.

* fix typo

* adapt ci bench

* remove LLVM_PROFILE from worker processes.

* update backend justfile run

* rename pool, pool worker and reorder imports

* use process-wide gdal options in gdal worker process

* put "Grid" into type name + remove comments

* explain llcov related env modification

* rename gdal dataset holding struct

* calculate composite thereshold from others

* remove comment

* rename GdaProcess config fields and add docs

* restructure gdal handling and sources

* fmt

* change env_remove logic for coverage

* refactor common worker spwaning

* stac back to 0.17

* add  leader & follower feature to gdal pool

* add default viscurl params to gdal dataset params

* lints

* sequential build bins

* complete hashing, avod stalling in group read feature, serial tests

* deaktivate porp no loger valid test

* fix: remove trailing space in VSI_CACHE_SIZE default

* chore: remove commented-out dead gdal geotransform tests

* docs: explain leader/follower cancellation semantics in read_data

* fix: set VSICURL globals once at worker startup instead of per-request

* refactor: make gdal request deduplicati... (continued)

Coverage Stats

3022 of 3724 new or added lines in 38 files covered. (81.15%)

21 existing lines in 7 files now uncovered.

123306 of 140785 relevant lines covered (87.58%)

481313.37 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 / 29957817213

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