Executor

Settings-default.toml

@@ -24,6 +24,9 @@ origin_coordinate_y = 0.0
 tile_shape_pixels_x = 512
 tile_shape_pixels_y = 512
 
+[executor]
+queue_size = 10
+
 [query_context]
 chunk_byte_size = 1048576 # TODO: find reasonable default
 

datatypes/src/collections/feature_collection.rs

@@ -23,7 +23,7 @@ use std::ops::{Bound, RangeBounds};
 use std::rc::Rc;
 use std::sync::Arc;
 
-use crate::primitives::Coordinate2D;
+use crate::primitives::{BoundingBox2D, Coordinate2D};
 use crate::primitives::{
     CategoryDataRef, FeatureData, FeatureDataRef, FeatureDataType, FeatureDataValue, FloatDataRef,
     Geometry, IntDataRef, TextDataRef, TimeInterval,
@@ -39,6 +39,7 @@ use crate::{
     collections::{FeatureCollectionError, IntoGeometryOptionsIterator},
     operations::reproject::CoordinateProjection,
 };
+use geo::intersects::Intersects;
 use std::iter::FromIterator;
 
 use super::{geo_feature_collection::ReplaceRawArrayCoords, GeometryCollection};
@@ -811,6 +812,21 @@ impl<'a, GeometryRef> FeatureCollectionRow<'a, GeometryRef> {
     }
 }
 
+impl<'a, GR> Intersects<BoundingBox2D> for FeatureCollectionRow<'a, GR>
+where
+    GR: Intersects<BoundingBox2D>,
+{
+    fn intersects(&self, rhs: &BoundingBox2D) -> bool {
+        self.geometry.intersects(rhs)
+    }
+}
+
+impl<'a, GR> Intersects<TimeInterval> for FeatureCollectionRow<'a, GR> {
+    fn intersects(&self, rhs: &TimeInterval) -> bool {
+        self.time_interval.intersects(rhs)
+    }
+}
+
 pub struct FeatureCollectionIterator<'a, GeometryIter> {
     geometries: GeometryIter,
     time_intervals: std::slice::Iter<'a, TimeInterval>,

datatypes/src/primitives/bounding_box.rs

@@ -688,6 +688,17 @@ mod tests {
         assert!(bbox.contains_bbox(&bbox_in));
     }
 
+    #[test]
+    fn bounding_box_contains_bbox_equal() {
+        let ll = Coordinate2D::new(1.0, 1.0);
+        let ur = Coordinate2D::new(4.0, 4.0);
+        let bbox = BoundingBox2D::new(ll, ur).unwrap();
+
+        let bbox_in = BoundingBox2D::new(ll, ur).unwrap();
+
+        assert!(bbox.contains_bbox(&bbox_in));
+    }
+
     #[test]
     fn bounding_box_contains_bbox_overlap() {
         let ll = Coordinate2D::new(1.0, 1.0);

datatypes/src/primitives/multi_line_string.rs

@@ -3,14 +3,14 @@ use std::convert::TryFrom;
 use arrow::array::{ArrayBuilder, BooleanArray};
 use arrow::error::ArrowError;
 use float_cmp::{ApproxEq, F64Margin};
-use geo::algorithm::intersects::Intersects;
+use geo::intersects::Intersects;
 use serde::{Deserialize, Serialize};
 use snafu::ensure;
 
 use crate::collections::VectorDataType;
 use crate::error::Error;
 use crate::primitives::{
-    error, BoundingBox2D, GeometryRef, MultiPoint, PrimitivesError, TypedGeometry,
+    error, BoundingBox2D, GeometryRef, MultiPoint, PrimitivesError, SpatialBounded, TypedGeometry,
 };
 use crate::primitives::{Coordinate2D, Geometry};
 use crate::util::arrow::{downcast_array, ArrowTyped};
@@ -280,6 +280,19 @@ impl<'g> MultiLineStringAccess for MultiLineStringRef<'g> {
         &self.point_coordinates
     }
 }
+impl<'g> SpatialBounded for MultiLineStringRef<'g> {
+    fn spatial_bounds(&self) -> BoundingBox2D {
+        let coords = self.point_coordinates.iter().flat_map(|&x| x.iter());
+        BoundingBox2D::from_coord_ref_iter(coords)
+            .expect("there must be at least one coordinate in a multilinestring")
+    }
+}
+
+impl<'g> Intersects<BoundingBox2D> for MultiLineStringRef<'g> {
+    fn intersects(&self, rhs: &BoundingBox2D) -> bool {
+        self.spatial_bounds().intersects_bbox(rhs)
+    }
+}
 
 impl<'g> From<MultiLineStringRef<'g>> for geojson::Geometry {
     fn from(geometry: MultiLineStringRef<'g>) -> geojson::Geometry {
@@ -354,6 +367,32 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_ref_intersects() {
+        let coordinates = vec![vec![(0.0, 0.0).into(), (10.0, 10.0).into()]];
+        let multi_line_string_ref =
+            MultiLineStringRef::new(coordinates.iter().map(AsRef::as_ref).collect()).unwrap();
+
+        assert!(
+            multi_line_string_ref.intersects(&BoundingBox2D::new_unchecked(
+                (-1., -1.,).into(),
+                (11., 11.).into()
+            ))
+        );
+        assert!(
+            multi_line_string_ref.intersects(&BoundingBox2D::new_unchecked(
+                (2., 2.,).into(),
+                (9., 9.).into()
+            ))
+        );
+        assert!(
+            !multi_line_string_ref.intersects(&BoundingBox2D::new_unchecked(
+                (-2., -2.,).into(),
+                (-2., 12.).into()
+            ))
+        );
+    }
+
     #[test]
     fn approx_equal() {
         let a = MultiLineString::new(vec![

datatypes/src/primitives/multi_point.rs

@@ -3,6 +3,7 @@ use std::convert::{TryFrom, TryInto};
 use arrow::array::{ArrayBuilder, BooleanArray};
 use arrow::error::ArrowError;
 use float_cmp::{ApproxEq, F64Margin};
+use geo::intersects::Intersects;
 use serde::{Deserialize, Serialize};
 use snafu::ensure;
 
@@ -286,6 +287,14 @@ where
     }
 }
 
+impl<'f> Intersects<BoundingBox2D> for MultiPointRef<'f> {
+    fn intersects(&self, rhs: &BoundingBox2D) -> bool {
+        self.point_coordinates
+            .iter()
+            .any(|c| rhs.contains_coordinate(c))
+    }
+}
+
 impl ApproxEq for &MultiPoint {
     type Margin = F64Margin;
 
@@ -392,4 +401,23 @@ mod tests {
 
         assert!(!approx_eq!(&MultiPoint, &a, &b, F64Margin::default()));
     }
+
+    #[test]
+    fn ref_intersects_bbox() -> Result<()> {
+        let bbox = BoundingBox2D::new((0.0, 0.0).into(), (1.0, 1.0).into())?;
+
+        let v1: Vec<Coordinate2D> = vec![(0.5, 0.5).into()];
+        let v2: Vec<Coordinate2D> = vec![(1.0, 1.0).into()];
+        let v3: Vec<Coordinate2D> = vec![(0.5, 0.5).into(), (1.5, 1.5).into()];
+        let v4: Vec<Coordinate2D> = vec![(1.1, 1.1).into()];
+        let v5: Vec<Coordinate2D> = vec![(-0.1, -0.1).into(), (1.1, 1.1).into()];
+
+        assert!(MultiPointRef::new(&v1)?.intersects(&bbox));
+        assert!(MultiPointRef::new(&v2)?.intersects(&bbox));
+        assert!(MultiPointRef::new(&v3)?.intersects(&bbox));
+        assert!(!MultiPointRef::new(&v4)?.intersects(&bbox));
+        assert!(!MultiPointRef::new(&v5)?.intersects(&bbox));
+
+        Ok(())
+    }
 }

datatypes/src/primitives/multi_polygon.rs

@@ -10,7 +10,8 @@ use snafu::ensure;
 use crate::collections::VectorDataType;
 use crate::error::Error;
 use crate::primitives::{
-    error, BoundingBox2D, GeometryRef, MultiLineString, PrimitivesError, TypedGeometry,
+    error, BoundingBox2D, GeometryRef, MultiLineString, PrimitivesError, SpatialBounded,
+    TypedGeometry,
 };
 use crate::primitives::{Coordinate2D, Geometry};
 use crate::util::arrow::{downcast_array, ArrowTyped};
@@ -356,6 +357,25 @@ impl<'g> MultiPolygonAccess for MultiPolygonRef<'g> {
     }
 }
 
+impl<'g> SpatialBounded for MultiPolygonRef<'g> {
+    fn spatial_bounds(&self) -> BoundingBox2D {
+        let outer_ring_coords = self
+            .polygons
+            .iter()
+            // Use exterior ring (first ring of a polygon)
+            .filter_map(|p| p.iter().next())
+            .flat_map(|&exterior| exterior.iter());
+        BoundingBox2D::from_coord_ref_iter(outer_ring_coords)
+            .expect("there must be at least one coordinate in a multipolygon")
+    }
+}
+
+impl<'g> Intersects<BoundingBox2D> for MultiPolygonRef<'g> {
+    fn intersects(&self, rhs: &BoundingBox2D) -> bool {
+        self.spatial_bounds().intersects_bbox(rhs)
+    }
+}
+
 impl<'g> From<MultiPolygonRef<'g>> for geojson::Geometry {
     fn from(geometry: MultiPolygonRef<'g>) -> geojson::Geometry {
         geojson::Geometry::new(match geometry.polygons.len() {
@@ -481,6 +501,48 @@ mod tests {
         assert_eq!(aggregate(&multi_polygon), aggregate(&multi_polygon_ref));
     }
 
+    #[test]
+    fn test_ref_intersects() {
+        let coordinates = vec![vec![
+            vec![
+                (0.0, 0.0).into(),
+                (10.0, 0.0).into(),
+                (10.0, 10.0).into(),
+                (0.0, 10.0).into(),
+                (0.0, 0.0).into(),
+            ],
+            vec![
+                (4.0, 4.0).into(),
+                (6.0, 4.0).into(),
+                (6.0, 6.0).into(),
+                (4.0, 6.0).into(),
+                (4.0, 4.0).into(),
+            ],
+        ]];
+        let multi_polygon_ref = MultiPolygonRef::new(
+            coordinates
+                .iter()
+                .map(|r| r.iter().map(AsRef::as_ref).collect())
+                .collect(),
+        )
+        .unwrap();
+
+        assert!(multi_polygon_ref.intersects(&BoundingBox2D::new_unchecked(
+            (-1., -1.,).into(),
+            (11., 11.).into()
+        )));
+
+        assert!(multi_polygon_ref.intersects(&BoundingBox2D::new_unchecked(
+            (4.5, 4.5,).into(),
+            (5.5, 5.5).into()
+        )));
+
+        assert!(!multi_polygon_ref.intersects(&BoundingBox2D::new_unchecked(
+            (-11., -1.,).into(),
+            (-1., 11.).into()
+        )));
+    }
+
     #[test]
     fn approx_equal() {
         let a = MultiPolygon::new(vec![

datatypes/src/primitives/no_geometry.rs

@@ -4,6 +4,7 @@ use std::convert::TryFrom;
 use arrow::array::{Array, ArrayBuilder, ArrayData, ArrayRef, BooleanArray, JsonEqual};
 use arrow::datatypes::DataType;
 use arrow::error::ArrowError;
+use geo::prelude::Intersects;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
@@ -26,6 +27,12 @@ impl Geometry for NoGeometry {
     }
 }
 
+impl Intersects<BoundingBox2D> for NoGeometry {
+    fn intersects(&self, _rhs: &BoundingBox2D) -> bool {
+        true
+    }
+}
+
 impl GeometryRef for NoGeometry {}
 
 impl TryFrom<TypedGeometry> for NoGeometry {

datatypes/src/primitives/spatial_partition.rs

@@ -167,23 +167,23 @@ impl SpatialPartition2D {
     }
 
     fn contains_x(&self, other: &Self) -> bool {
-        crate::util::ranges::value_in_range(
+        crate::util::ranges::value_in_range_inclusive(
             other.upper_left_coordinate.x,
             self.upper_left_coordinate.x,
             self.lower_right_coordinate.x,
-        ) && crate::util::ranges::value_in_range(
+        ) && crate::util::ranges::value_in_range_inclusive(
             other.lower_right_coordinate.x,
             self.upper_left_coordinate.x,
             self.lower_right_coordinate.x,
         )
     }
 
     fn contains_y(&self, other: &Self) -> bool {
-        crate::util::ranges::value_in_range_inv(
+        crate::util::ranges::value_in_range_inclusive(
             other.lower_right_coordinate.y,
             self.lower_right_coordinate.y,
             self.upper_left_coordinate.y,
-        ) && crate::util::ranges::value_in_range_inv(
+        ) && crate::util::ranges::value_in_range_inclusive(
             other.upper_left_coordinate.y,
             self.lower_right_coordinate.y,
             self.upper_left_coordinate.y,
@@ -305,6 +305,14 @@ mod tests {
         assert!(!p2.contains(&p1));
     }
 
+    #[test]
+    fn it_contains_itself() {
+        let p1 = SpatialPartition2D::new_unchecked((0., 1.).into(), (1., 0.).into());
+        let p2 = SpatialPartition2D::new_unchecked((0., 1.).into(), (1., 0.).into());
+        assert!(p1.contains(&p2));
+        assert!(p2.contains(&p1));
+    }
+
     #[test]
     fn it_contains_coord() {
         let p1 = SpatialPartition2D::new_unchecked((0., 1.).into(), (1., 0.).into());

operators/Cargo.toml

@@ -52,6 +52,7 @@ uuid = { version = "0.8", features = ["serde", "v4", "v5"] }
 [dev-dependencies]
 async-stream = "0.3"
 geo-rand = { git = "https://github.com/lelongg/geo-rand", tag = "v0.3.0" }
+tokio-util = "0.6"
 rand = "0.8"
 
 

operators/src/error.rs

@@ -270,10 +270,18 @@ pub enum Error {
         source: crate::util::statistics::StatisticsError,
     },
 
+    #[cfg(feature = "pro")]
+    #[snafu(display("Executor error: {}", source))]
+    Executor {
+        #[snafu(implicit)]
+        source: crate::pro::executor::error::ExecutorError,
+    },
+
     #[snafu(display("SparseTilesFillAdapter error: {}", source))]
     SparseTilesFillAdapter {
         source: crate::adapters::SparseTilesFillAdapterError,
     },
+
     #[snafu(context(false))]
     ExpressionOperator {
         source: crate::processing::ExpressionError,

operators/src/pro/executor/error.rs

@@ -0,0 +1,39 @@
+use snafu::Snafu;
+use tokio::sync::mpsc::error::SendError;
+use tokio::sync::oneshot::error::RecvError;
+use tokio::task::JoinError;
+
+pub type Result<T> = std::result::Result<T, ExecutorError>;
+
+#[derive(Debug, Clone, Snafu)]
+pub enum ExecutorError {
+    Submission { message: String },
+    Panic,
+    Cancelled,
+}
+
+impl From<JoinError> for ExecutorError {
+    fn from(src: JoinError) -> Self {
+        if src.is_cancelled() {
+            ExecutorError::Cancelled
+        } else {
+            ExecutorError::Panic
+        }
+    }
+}
+
+impl<T> From<SendError<T>> for ExecutorError {
+    fn from(e: SendError<T>) -> Self {
+        Self::Submission {
+            message: e.to_string(),
+        }
+    }
+}
+
+impl From<RecvError> for ExecutorError {
+    fn from(e: RecvError) -> Self {
+        Self::Submission {
+            message: e.to_string(),
+        }
+    }
+}