feat(query): replace Geometry deserialization with streaming bbox extraction (#19944)

Spatial index writing, statistics collection, join runtime-filter build,
and predicate pushdown now compute bounding boxes by streaming over EWKB
coordinates directly, avoiding per-row geo::Geometry heap allocations.

Author: TCeason

src/common/io/src/geometry.rs

@@ -244,14 +244,6 @@ pub fn rect_to_polygon(rect: Rect<f64>) -> Polygon<f64> {
     Polygon::new(exterior, vec![])
 }
 
-/// Process EWKB input and return SRID.
-pub fn read_srid<B: AsRef<[u8]>>(ewkb: &mut Ewkb<B>) -> Option<i32> {
-    let mut srid_processor = SridProcessor::new();
-    ewkb.process_geom(&mut srid_processor).ok()?;
-
-    srid_processor.srid
-}
-
 /// Process EWKB input and return Geometry object and SRID.
 pub fn ewkb_to_geo<B: AsRef<[u8]>>(ewkb: &mut Ewkb<B>) -> Result<(Geometry<f64>, Option<i32>)> {
     let mut ewkb_processor = EwkbProcessor::new();
@@ -265,23 +257,6 @@ pub fn ewkb_to_geo<B: AsRef<[u8]>>(ewkb: &mut Ewkb<B>) -> Result<(Geometry<f64>,
     Ok((geo, srid))
 }
 
-struct SridProcessor {
-    srid: Option<i32>,
-}
-
-impl SridProcessor {
-    fn new() -> Self {
-        Self { srid: None }
-    }
-}
-
-impl GeomProcessor for SridProcessor {
-    fn srid(&mut self, srid: Option<i32>) -> geozero::error::Result<()> {
-        self.srid = srid;
-        Ok(())
-    }
-}
-
 struct EwkbProcessor {
     geo_writer: GeoWriter,
     srid: Option<i32>,
@@ -449,3 +424,130 @@ pub fn point_to_geohash(ewkb: &[u8], precision: Option<i32>) -> Result<String> {
     encode(point.0, precision.map_or(12, |p| p as usize))
         .map_err(|e| ErrorCode::GeometryError(e.to_string()))
 }
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct Bbox {
+    min_x: f64,
+    min_y: f64,
+    max_x: f64,
+    max_y: f64,
+}
+
+impl Bbox {
+    pub fn new(x: f64, y: f64) -> Self {
+        Self {
+            min_x: x,
+            min_y: y,
+            max_x: x,
+            max_y: y,
+        }
+    }
+
+    pub fn from_corners(min_x: f64, min_y: f64, max_x: f64, max_y: f64) -> Self {
+        Self {
+            min_x,
+            min_y,
+            max_x,
+            max_y,
+        }
+    }
+
+    pub fn corners(&self) -> (f64, f64, f64, f64) {
+        (self.min_x, self.min_y, self.max_x, self.max_y)
+    }
+
+    pub fn extend(&mut self, x: f64, y: f64) {
+        if x < self.min_x {
+            self.min_x = x;
+        }
+        if x > self.max_x {
+            self.max_x = x;
+        }
+        if y < self.min_y {
+            self.min_y = y;
+        }
+        if y > self.max_y {
+            self.max_y = y;
+        }
+    }
+
+    pub fn expand(&mut self, distance: f64) {
+        self.min_x -= distance;
+        self.min_y -= distance;
+        self.max_x += distance;
+        self.max_y += distance;
+    }
+
+    pub fn intersects(&self, other: &Self) -> bool {
+        self.max_x >= other.min_x
+            && self.min_x <= other.max_x
+            && self.max_y >= other.min_y
+            && self.min_y <= other.max_y
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct EwkbBbox {
+    pub bbox: Option<Bbox>,
+    pub srid: Option<i32>,
+}
+
+pub fn ewkb_to_bbox(ewkb: &[u8]) -> Option<EwkbBbox> {
+    let mut processor = BboxProcessor::new();
+    Ewkb(ewkb).process_geom(&mut processor).ok()?;
+    Some(processor.into_ewkb_bbox())
+}
+
+struct BboxProcessor {
+    bbox: Option<Bbox>,
+    srid: Option<i32>,
+}
+
+impl BboxProcessor {
+    fn new() -> Self {
+        Self {
+            bbox: None,
+            srid: None,
+        }
+    }
+
+    fn extend(&mut self, x: f64, y: f64) {
+        if let Some(bbox) = self.bbox.as_mut() {
+            bbox.extend(x, y);
+        } else {
+            self.bbox = Some(Bbox::new(x, y));
+        }
+    }
+
+    fn into_ewkb_bbox(self) -> EwkbBbox {
+        EwkbBbox {
+            bbox: self.bbox,
+            srid: self.srid,
+        }
+    }
+}
+
+impl GeomProcessor for BboxProcessor {
+    fn srid(&mut self, srid: Option<i32>) -> geozero::error::Result<()> {
+        self.srid = srid;
+        Ok(())
+    }
+
+    fn xy(&mut self, x: f64, y: f64, _idx: usize) -> geozero::error::Result<()> {
+        self.extend(x, y);
+        Ok(())
+    }
+
+    fn coordinate(
+        &mut self,
+        x: f64,
+        y: f64,
+        _z: Option<f64>,
+        _m: Option<f64>,
+        _t: Option<f64>,
+        _tm: Option<u64>,
+        idx: usize,
+    ) -> geozero::error::Result<()> {
+        self.xy(x, y, idx)
+    }
+}

src/common/io/src/lib.rs

@@ -64,8 +64,11 @@ pub use escape::escape_string_with_quote;
 pub use geography::GEOGRAPHY_SRID;
 pub use geography::geography_format;
 pub use geometry::Axis;
+pub use geometry::Bbox;
+pub use geometry::EwkbBbox;
 pub use geometry::Extremum;
 pub use geometry::GeometryDataType;
+pub use geometry::ewkb_to_bbox;
 pub use geometry::ewkb_to_geo;
 pub use geometry::geo_to_ewkb;
 pub use geometry::geo_to_ewkt;
@@ -76,5 +79,4 @@ pub use geometry::geometry_format;
 pub use geometry::geometry_from_ewkt;
 pub use geometry::geometry_type_name;
 pub use geometry::parse_bytes_to_ewkb;
-pub use geometry::read_srid;
 pub use interval::Interval;

src/common/io/tests/it/geometry.rs

@@ -0,0 +1,70 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use databend_common_io::geometry::Bbox;
+use databend_common_io::geometry::ewkb_to_bbox;
+use databend_common_io::geometry::geometry_from_str;
+use databend_common_io::wkb::make_point;
+
+#[test]
+fn test_extract_bbox_point() {
+    let point = make_point(-74.0, 40.7);
+    let bbox = ewkb_to_bbox(&point).unwrap().bbox.unwrap();
+    assert_eq!(bbox, Bbox::from_corners(-74.0, 40.7, -74.0, 40.7));
+}
+
+#[test]
+fn test_extract_bbox_point_with_srid() {
+    let ewkb = geometry_from_str("SRID=4326;POINT(-122.35 37.55)", None).unwrap();
+    let result = ewkb_to_bbox(&ewkb).unwrap();
+    assert_eq!(
+        result.bbox.unwrap(),
+        Bbox::from_corners(-122.35, 37.55, -122.35, 37.55)
+    );
+    assert_eq!(result.srid, Some(4326));
+}
+
+#[test]
+fn test_extract_bbox_linestring() {
+    let ewkb = geometry_from_str("LINESTRING(0 0, 10 5, 20 3)", None).unwrap();
+    let bbox = ewkb_to_bbox(&ewkb).unwrap().bbox.unwrap();
+    assert_eq!(bbox, Bbox::from_corners(0.0, 0.0, 20.0, 5.0));
+}
+
+#[test]
+fn test_extract_bbox_polygon() {
+    let ewkb = geometry_from_str(
+        "POLYGON((0 0, 10 0, 10 10, 0 10, 0 0), (2 2, 8 2, 8 8, 2 8, 2 2))",
+        None,
+    )
+    .unwrap();
+    let bbox = ewkb_to_bbox(&ewkb).unwrap().bbox.unwrap();
+    assert_eq!(bbox, Bbox::from_corners(0.0, 0.0, 10.0, 10.0));
+}
+
+#[test]
+fn test_extract_bbox_point_z() {
+    // EWKB PointZ(1 2 3), little-endian, Z flag 0x80000000. Databend never stores
+    // 3D geometries today (all ingest paths funnel through to_ewkb(xy)), but the
+    // EWKB parser must still derive a correct 2D bbox if one ever appears.
+    let ewkb: &[u8] = &[
+        0x01, // little-endian
+        0x01, 0x00, 0x00, 0x80, // type = Point | Z flag
+        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xF0, 0x3F, // x = 1.0
+        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x40, // y = 2.0
+        0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x08, 0x40, // z = 3.0
+    ];
+    let bbox = ewkb_to_bbox(ewkb).unwrap().bbox.unwrap();
+    assert_eq!(bbox, Bbox::from_corners(1.0, 2.0, 1.0, 2.0));
+}

src/common/io/tests/it/main.rs

@@ -23,5 +23,6 @@ mod bincode_serialization;
 mod borsh_serialization;
 mod cursor_ext;
 mod escape;
+mod geometry;
 mod interval;
 mod serialization_format_compatibility;

src/query/expression/src/types/geometry.rs

@@ -17,7 +17,9 @@ use std::ops::Range;
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_io::Bbox;
 use databend_common_io::GEOGRAPHY_SRID;
+use databend_common_io::ewkb_to_bbox;
 use databend_common_io::ewkb_to_geo;
 use geo::Geometry;
 use geozero::ToGeo;
@@ -242,3 +244,21 @@ pub fn extract_geo_and_srid(value: ScalarRef) -> Result<Option<(Geometry<f64>, i
     };
     Ok(Some((geo, srid)))
 }
+
+/// Compute the bounding box of a geometry/geography scalar by streaming over its
+/// EWKB coordinates, avoiding the cost of materializing a `geo::Geometry`.
+///
+/// SRID follows the same rule as [`extract_geo_and_srid`]: geometries carry their
+/// own SRID (0 when absent), geographies are always [`GEOGRAPHY_SRID`].
+pub fn extract_bbox_and_srid(value: ScalarRef) -> Result<Option<(Option<Bbox>, i32)>> {
+    let (ewkb, geography_srid) = match value {
+        ScalarRef::Geometry(buf) => (buf, None),
+        ScalarRef::Geography(buf) => (buf.0, Some(GEOGRAPHY_SRID)),
+        _ => return Ok(None),
+    };
+    let Some(result) = ewkb_to_bbox(ewkb) else {
+        return Err(ErrorCode::GeometryError("invalid EWKB input"));
+    };
+    let srid = geography_srid.unwrap_or_else(|| result.srid.unwrap_or(0));
+    Ok(Some((result.bbox, srid)))
+}

src/query/functions/src/scalars/geographic/src/geometry.rs

@@ -41,6 +41,7 @@ use databend_common_expression::vectorize_with_builder_4_arg;
 use databend_common_io::Axis;
 use databend_common_io::Extremum;
 use databend_common_io::GEOGRAPHY_SRID;
+use databend_common_io::ewkb_to_bbox;
 use databend_common_io::ewkb_to_geo;
 use databend_common_io::geo_to_ewkb;
 use databend_common_io::geo_to_ewkt;
@@ -56,7 +57,6 @@ use databend_common_io::geometry::st_extreme;
 use databend_common_io::geometry_format;
 use databend_common_io::geometry_from_ewkt;
 use databend_common_io::geometry_type_name;
-use databend_common_io::read_srid;
 use geo::Area;
 use geo::BoundingRect;
 use geo::Buffer;
@@ -751,7 +751,9 @@ pub fn register(registry: &mut FunctionRegistry) {
                 }
             }
 
-            let srid = read_srid(&mut Ewkb(ewkb)).unwrap_or_default();
+            let srid = ewkb_to_bbox(ewkb)
+                .and_then(|result| result.srid)
+                .unwrap_or_default();
             output.push(srid);
         }),
     );
@@ -771,7 +773,7 @@ pub fn register(registry: &mut FunctionRegistry) {
 
                 // All representations of the geo types supported by crates under the GeoRust organization, have not implemented srid().
                 // Currently, the srid() of all types returns the default value `None`, so we need to parse it manually here.
-                let Some(from_srid) = read_srid(&mut Ewkb(original)) else {
+                let Some(from_srid) = ewkb_to_bbox(original).and_then(|result| result.srid) else {
                     ctx.set_error(row, "input geometry must has the correct SRID".to_string());
                     builder.commit_row();
                     return;

src/query/service/src/pipelines/processors/transforms/hash_join/runtime_filter/local_builder.rs

@@ -24,9 +24,8 @@ use databend_common_expression::HashMethodKind;
 use databend_common_expression::RawExpr;
 use databend_common_expression::Scalar;
 use databend_common_expression::types::DataType;
-use databend_common_expression::types::geometry::extract_geo_and_srid;
+use databend_common_expression::types::geometry::extract_bbox_and_srid;
 use databend_common_functions::BUILTIN_FUNCTIONS;
-use geo::algorithm::bounding_rect::BoundingRect;
 
 use crate::physical_plans::SpatialRuntimeFilterMode;
 use crate::pipelines::processors::transforms::hash_join::desc::RuntimeFilterDesc;
@@ -168,7 +167,7 @@ impl SingleFilterBuilder {
         }
 
         for value in column.iter() {
-            let Some((geo, srid)) = extract_geo_and_srid(value)? else {
+            let Some((bbox, srid)) = extract_bbox_and_srid(value)? else {
                 continue;
             };
 
@@ -182,10 +181,9 @@ impl SingleFilterBuilder {
                 self.spatial_srid = Some(srid);
             }
 
-            if let Some(rect) = geo.bounding_rect() {
-                let min = rect.min();
-                let max = rect.max();
-                self.spatial_rects.push((min.x, min.y, max.x, max.y));
+            if let Some(bbox) = bbox {
+                let (min_x, min_y, max_x, max_y) = bbox.corners();
+                self.spatial_rects.push((min_x, min_y, max_x, max_y));
             }
         }
 

src/query/storages/common/index/src/spatial_predicate.rs

@@ -26,10 +26,8 @@ use databend_common_expression::TableSchemaRef;
 use databend_common_expression::types::DataType;
 use databend_common_expression::visit_expr;
 use databend_common_functions::SPATIAL_INDEX_FUNCTIONS;
-use databend_common_io::ewkb_to_geo;
-use geo::BoundingRect;
+use databend_common_io::ewkb_to_bbox;
 use geo::Rect;
-use geozero::wkb::Ewkb;
 use unicase::Ascii;
 
 use crate::scalar_to_distance_threshold;
@@ -139,10 +137,12 @@ impl<'a> SpatialPredicateVisitor<'a> {
     fn scalar_to_query(scalar: &Scalar) -> Option<(Option<Rect<f64>>, i32)> {
         match scalar {
             Scalar::Geometry(buffer) => {
-                let mut ewkb = Ewkb(buffer.as_slice());
-                let (geom, srid) = ewkb_to_geo(&mut ewkb).ok()?;
-                let rect = geom.bounding_rect();
-                Some((rect, srid.unwrap_or(0)))
+                let result = ewkb_to_bbox(buffer.as_slice())?;
+                let rect = result.bbox.map(|bbox| {
+                    let (min_x, min_y, max_x, max_y) = bbox.corners();
+                    Rect::new((min_x, min_y), (max_x, max_y))
+                });
+                Some((rect, result.srid.unwrap_or(0)))
             }
             _ => None,
         }