feat: add more geo functions (#4888)

* chore: add type conversion for array types

* feat: add h3_cells_contains

* refactor: resolve lint issues

* feat: add sphere distance function

* feat: euclidean distance between h3 centroids

* test: round float number

* feat: add more geospatial functions

* test: add tests for geometry functions

* refactor: move wkt function to dedicated module

* feat: add st_area

* refactor: only allow sphere distance between points

src/common/function/Cargo.toml

@@ -9,7 +9,7 @@ workspace = true
 
 [features]
 default = ["geo"]
-geo = ["geohash", "h3o", "s2"]
+geo = ["geohash", "h3o", "s2", "wkt", "geo-types", "dep:geo"]
 
 [dependencies]
 api.workspace = true
@@ -28,6 +28,8 @@ common-version.workspace = true
 datafusion.workspace = true
 datatypes.workspace = true
 derive_more = { version = "1", default-features = false, features = ["display"] }
+geo = { version = "0.29", optional = true }
+geo-types = { version = "0.7", optional = true }
 geohash = { version = "0.13", optional = true }
 h3o = { version = "0.6", optional = true }
 jsonb.workspace = true
@@ -44,6 +46,7 @@ sql.workspace = true
 statrs = "0.16"
 store-api.workspace = true
 table.workspace = true
+wkt = { version = "0.11", optional = true }
 
 [dev-dependencies]
 ron = "0.7"

src/common/function/src/scalars/geo.rs

@@ -17,7 +17,10 @@ pub(crate) mod encoding;
 mod geohash;
 mod h3;
 mod helpers;
+mod measure;
+mod relation;
 mod s2;
+mod wkt;
 
 use crate::function_registry::FunctionRegistry;
 
@@ -48,6 +51,7 @@ impl GeoFunctions {
         registry.register(Arc::new(h3::H3CellToChildrenSize));
         registry.register(Arc::new(h3::H3CellToChildPos));
         registry.register(Arc::new(h3::H3ChildPosToCell));
+        registry.register(Arc::new(h3::H3CellContains));
 
         // h3 grid traversal
         registry.register(Arc::new(h3::H3GridDisk));
@@ -55,10 +59,27 @@ impl GeoFunctions {
         registry.register(Arc::new(h3::H3GridDistance));
         registry.register(Arc::new(h3::H3GridPathCells));
 
+        // h3 measurement
+        registry.register(Arc::new(h3::H3CellDistanceSphereKm));
+        registry.register(Arc::new(h3::H3CellDistanceEuclideanDegree));
+
         // s2
         registry.register(Arc::new(s2::S2LatLngToCell));
         registry.register(Arc::new(s2::S2CellLevel));
         registry.register(Arc::new(s2::S2CellToToken));
         registry.register(Arc::new(s2::S2CellParent));
+
+        // spatial data type
+        registry.register(Arc::new(wkt::LatLngToPointWkt));
+
+        // spatial relation
+        registry.register(Arc::new(relation::STContains));
+        registry.register(Arc::new(relation::STWithin));
+        registry.register(Arc::new(relation::STIntersects));
+
+        // spatial measure
+        registry.register(Arc::new(measure::STDistance));
+        registry.register(Arc::new(measure::STDistanceSphere));
+        registry.register(Arc::new(measure::STArea));
     }
 }

src/common/function/src/scalars/geo/h3.rs

@@ -23,8 +23,8 @@ use datatypes::prelude::ConcreteDataType;
 use datatypes::scalars::{Scalar, ScalarVectorBuilder};
 use datatypes::value::{ListValue, Value};
 use datatypes::vectors::{
-    BooleanVectorBuilder, Int32VectorBuilder, ListVectorBuilder, MutableVector,
-    StringVectorBuilder, UInt64VectorBuilder, UInt8VectorBuilder, VectorRef,
+    BooleanVectorBuilder, Float64VectorBuilder, Int32VectorBuilder, ListVectorBuilder,
+    MutableVector, StringVectorBuilder, UInt64VectorBuilder, UInt8VectorBuilder, VectorRef,
 };
 use derive_more::Display;
 use h3o::{CellIndex, LatLng, Resolution};
@@ -38,6 +38,7 @@ static CELL_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
     vec![
         ConcreteDataType::int64_datatype(),
         ConcreteDataType::uint64_datatype(),
+        ConcreteDataType::string_datatype(),
     ]
 });
 
@@ -952,6 +953,181 @@ impl Function for H3GridPathCells {
     }
 }
 
+/// Tests if cells contains given cells
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct H3CellContains;
+
+impl Function for H3CellContains {
+    fn name(&self) -> &str {
+        "h3_cells_contains"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        let multi_cell_types = vec![
+            ConcreteDataType::list_datatype(ConcreteDataType::int64_datatype()),
+            ConcreteDataType::list_datatype(ConcreteDataType::uint64_datatype()),
+            ConcreteDataType::list_datatype(ConcreteDataType::string_datatype()),
+            ConcreteDataType::string_datatype(),
+        ];
+
+        let mut signatures = Vec::with_capacity(multi_cell_types.len() * CELL_TYPES.len());
+        for multi_cell_type in &multi_cell_types {
+            for cell_type in CELL_TYPES.as_slice() {
+                signatures.push(TypeSignature::Exact(vec![
+                    multi_cell_type.clone(),
+                    cell_type.clone(),
+                ]));
+            }
+        }
+
+        Signature::one_of(signatures, Volatility::Stable)
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let cells_vec = &columns[0];
+        let cell_this_vec = &columns[1];
+
+        let size = cell_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let mut result = None;
+            if let (cells, Some(cell_this)) = (
+                cells_from_value(cells_vec.get(i))?,
+                cell_from_value(cell_this_vec.get(i))?,
+            ) {
+                result = Some(false);
+
+                for cell_that in cells.iter() {
+                    // get cell resolution, and find cell_this's parent at
+                    //  this solution, test if cell_that equals the parent
+                    let resolution = cell_that.resolution();
+                    if let Some(cell_this_parent) = cell_this.parent(resolution) {
+                        if cell_this_parent == *cell_that {
+                            result = Some(true);
+                            break;
+                        }
+                    }
+                }
+            }
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Get WGS84 great circle distance of two cell centroid
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct H3CellDistanceSphereKm;
+
+impl Function for H3CellDistanceSphereKm {
+    fn name(&self) -> &str {
+        "h3_distance_sphere_km"
+    }
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        signature_of_double_cells()
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let cell_this_vec = &columns[0];
+        let cell_that_vec = &columns[1];
+        let size = cell_this_vec.len();
+
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let result = match (
+                cell_from_value(cell_this_vec.get(i))?,
+                cell_from_value(cell_that_vec.get(i))?,
+            ) {
+                (Some(cell_this), Some(cell_that)) => {
+                    let centroid_this = LatLng::from(cell_this);
+                    let centroid_that = LatLng::from(cell_that);
+
+                    Some(centroid_this.distance_km(centroid_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Get Euclidean distance of two cell centroid
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct H3CellDistanceEuclideanDegree;
+
+impl H3CellDistanceEuclideanDegree {
+    fn distance(centroid_this: LatLng, centroid_that: LatLng) -> f64 {
+        ((centroid_this.lat() - centroid_that.lat()).powi(2)
+            + (centroid_this.lng() - centroid_that.lng()).powi(2))
+        .sqrt()
+    }
+}
+
+impl Function for H3CellDistanceEuclideanDegree {
+    fn name(&self) -> &str {
+        "h3_distance_degree"
+    }
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        signature_of_double_cells()
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let cell_this_vec = &columns[0];
+        let cell_that_vec = &columns[1];
+        let size = cell_this_vec.len();
+
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let result = match (
+                cell_from_value(cell_this_vec.get(i))?,
+                cell_from_value(cell_that_vec.get(i))?,
+            ) {
+                (Some(cell_this), Some(cell_that)) => {
+                    let centroid_this = LatLng::from(cell_this);
+                    let centroid_that = LatLng::from(cell_that);
+
+                    let dist = Self::distance(centroid_this, centroid_that);
+                    Some(dist)
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
 fn value_to_resolution(v: Value) -> Result<Resolution> {
     let r = match v {
         Value::Int8(v) => v as u8,
@@ -1073,7 +1249,126 @@ fn cell_from_value(v: Value) -> Result<Option<CellIndex>> {
                 })
                 .context(error::ExecuteSnafu)?,
         ),
+        Value::String(s) => Some(
+            CellIndex::from_str(s.as_utf8())
+                .map_err(|e| {
+                    BoxedError::new(PlainError::new(
+                        format!("H3 error: {}", e),
+                        StatusCode::EngineExecuteQuery,
+                    ))
+                })
+                .context(error::ExecuteSnafu)?,
+        ),
         _ => None,
     };
     Ok(cell)
 }
+
+/// extract cell array from all possible types including:
+/// - int64 list
+/// - uint64 list
+/// - string list
+/// - comma-separated string
+fn cells_from_value(v: Value) -> Result<Vec<CellIndex>> {
+    match v {
+        Value::List(list) => match list.datatype() {
+            ConcreteDataType::Int64(_) => list
+                .items()
+                .iter()
+                .map(|v| {
+                    if let Value::Int64(v) = v {
+                        CellIndex::try_from(*v as u64)
+                            .map_err(|e| {
+                                BoxedError::new(PlainError::new(
+                                    format!("H3 error: {}", e),
+                                    StatusCode::EngineExecuteQuery,
+                                ))
+                            })
+                            .context(error::ExecuteSnafu)
+                    } else {
+                        Err(BoxedError::new(PlainError::new(
+                            "Invalid data type in array".to_string(),
+                            StatusCode::EngineExecuteQuery,
+                        )))
+                        .context(error::ExecuteSnafu)
+                    }
+                })
+                .collect::<Result<Vec<CellIndex>>>(),
+            ConcreteDataType::UInt64(_) => list
+                .items()
+                .iter()
+                .map(|v| {
+                    if let Value::UInt64(v) = v {
+                        CellIndex::try_from(*v)
+                            .map_err(|e| {
+                                BoxedError::new(PlainError::new(
+                                    format!("H3 error: {}", e),
+                                    StatusCode::EngineExecuteQuery,
+                                ))
+                            })
+                            .context(error::ExecuteSnafu)
+                    } else {
+                        Err(BoxedError::new(PlainError::new(
+                            "Invalid data type in array".to_string(),
+                            StatusCode::EngineExecuteQuery,
+                        )))
+                        .context(error::ExecuteSnafu)
+                    }
+                })
+                .collect::<Result<Vec<CellIndex>>>(),
+            ConcreteDataType::String(_) => list
+                .items()
+                .iter()
+                .map(|v| {
+                    if let Value::String(v) = v {
+                        CellIndex::from_str(v.as_utf8().trim())
+                            .map_err(|e| {
+                                BoxedError::new(PlainError::new(
+                                    format!("H3 error: {}", e),
+                                    StatusCode::EngineExecuteQuery,
+                                ))
+                            })
+                            .context(error::ExecuteSnafu)
+                    } else {
+                        Err(BoxedError::new(PlainError::new(
+                            "Invalid data type in array".to_string(),
+                            StatusCode::EngineExecuteQuery,
+                        )))
+                        .context(error::ExecuteSnafu)
+                    }
+                })
+                .collect::<Result<Vec<CellIndex>>>(),
+            _ => Ok(vec![]),
+        },
+        Value::String(csv) => {
+            let str_seq = csv.as_utf8().split(',');
+            str_seq
+                .map(|v| {
+                    CellIndex::from_str(v.trim())
+                        .map_err(|e| {
+                            BoxedError::new(PlainError::new(
+                                format!("H3 error: {}", e),
+                                StatusCode::EngineExecuteQuery,
+                            ))
+                        })
+                        .context(error::ExecuteSnafu)
+                })
+                .collect::<Result<Vec<CellIndex>>>()
+        }
+        _ => Ok(vec![]),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_h3_euclidean_distance() {
+        let point_this = LatLng::new(42.3521, -72.1235).expect("incorrect lat lng");
+        let point_that = LatLng::new(42.45, -72.1260).expect("incorrect lat lng");
+
+        let dist = H3CellDistanceEuclideanDegree::distance(point_this, point_that);
+        assert_eq!(dist, 0.09793191512474639);
+    }
+}

src/common/function/src/scalars/geo/measure.rs

@@ -0,0 +1,195 @@
+// Copyright 2023 Greptime Team
+//
+// 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 common_error::ext::{BoxedError, PlainError};
+use common_error::status_code::StatusCode;
+use common_query::error::{self, Result};
+use common_query::prelude::{Signature, TypeSignature};
+use datafusion::logical_expr::Volatility;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{Float64VectorBuilder, MutableVector, VectorRef};
+use derive_more::Display;
+use geo::algorithm::line_measures::metric_spaces::Euclidean;
+use geo::{Area, Distance, Haversine};
+use geo_types::Geometry;
+use snafu::ResultExt;
+
+use super::helpers::{ensure_columns_len, ensure_columns_n};
+use super::wkt::parse_wkt;
+use crate::function::{Function, FunctionContext};
+
+/// Return WGS84(SRID: 4326) euclidean distance between two geometry object, in degree
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STDistance;
+
+impl Function for STDistance {
+    fn name(&self) -> &str {
+        "st_distance"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(Euclidean::distance(&geom_this, &geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Return great circle distance between two geometry object, in meters
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STDistanceSphere;
+
+impl Function for STDistanceSphere {
+    fn name(&self) -> &str {
+        "st_distance_sphere_m"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    match (geom_this, geom_that) {
+                        (Geometry::Point(this), Geometry::Point(that)) => {
+                            Some(Haversine::distance(this, that))
+                        }
+                        _ => {
+                            Err(BoxedError::new(PlainError::new(
+                                "Great circle distance between non-point objects are not supported for now.".to_string(),
+                                StatusCode::Unsupported,
+                            ))).context(error::ExecuteSnafu)?
+                        }
+                    }
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Return area of given geometry object
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STArea;
+
+impl Function for STArea {
+    fn name(&self) -> &str {
+        "st_area"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 1);
+
+        let wkt_vec = &columns[0];
+
+        let size = wkt_vec.len();
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt = wkt_vec.get(i).as_string();
+
+            let result = if let Some(wkt) = wkt {
+                let geom = parse_wkt(&wkt)?;
+                Some(geom.unsigned_area())
+            } else {
+                None
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}

src/common/function/src/scalars/geo/relation.rs

@@ -0,0 +1,190 @@
+// Copyright 2023 Greptime Team
+//
+// 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 common_query::error::Result;
+use common_query::prelude::{Signature, TypeSignature};
+use datafusion::logical_expr::Volatility;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{BooleanVectorBuilder, MutableVector, VectorRef};
+use derive_more::Display;
+use geo::algorithm::contains::Contains;
+use geo::algorithm::intersects::Intersects;
+use geo::algorithm::within::Within;
+
+use super::helpers::{ensure_columns_len, ensure_columns_n};
+use super::wkt::parse_wkt;
+use crate::function::{Function, FunctionContext};
+
+/// Test if spatial relationship: contains
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STContains;
+
+impl Function for STContains {
+    fn name(&self) -> &str {
+        "st_contains"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(geom_this.contains(&geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Test if spatial relationship: within
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STWithin;
+
+impl Function for STWithin {
+    fn name(&self) -> &str {
+        "st_within"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(geom_this.is_within(&geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Test if spatial relationship: within
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STIntersects;
+
+impl Function for STIntersects {
+    fn name(&self) -> &str {
+        "st_intersects"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(geom_this.intersects(&geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}

src/common/function/src/scalars/geo/wkt.rs

@@ -0,0 +1,100 @@
+// Copyright 2023 Greptime Team
+//
+// 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 common_error::ext::{BoxedError, PlainError};
+use common_error::status_code::StatusCode;
+use common_query::error::{self, Result};
+use common_query::prelude::{Signature, TypeSignature};
+use datafusion::logical_expr::Volatility;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{MutableVector, StringVectorBuilder, VectorRef};
+use derive_more::Display;
+use geo_types::{Geometry, Point};
+use once_cell::sync::Lazy;
+use snafu::ResultExt;
+use wkt::{ToWkt, TryFromWkt};
+
+use super::helpers::{ensure_columns_len, ensure_columns_n};
+use crate::function::{Function, FunctionContext};
+
+static COORDINATE_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
+    vec![
+        ConcreteDataType::float32_datatype(),
+        ConcreteDataType::float64_datatype(),
+    ]
+});
+
+/// Return WGS84(SRID: 4326) euclidean distance between two geometry object, in degree
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct LatLngToPointWkt;
+
+impl Function for LatLngToPointWkt {
+    fn name(&self) -> &str {
+        "wkt_point_from_latlng"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::string_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        let mut signatures = Vec::new();
+        for coord_type in COORDINATE_TYPES.as_slice() {
+            signatures.push(TypeSignature::Exact(vec![
+                // latitude
+                coord_type.clone(),
+                // longitude
+                coord_type.clone(),
+            ]));
+        }
+        Signature::one_of(signatures, Volatility::Stable)
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let lat_vec = &columns[0];
+        let lng_vec = &columns[1];
+
+        let size = lat_vec.len();
+        let mut results = StringVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let lat = lat_vec.get(i).as_f64_lossy();
+            let lng = lng_vec.get(i).as_f64_lossy();
+
+            let result = match (lat, lng) {
+                (Some(lat), Some(lng)) => Some(Point::new(lng, lat).wkt_string()),
+                _ => None,
+            };
+
+            results.push(result.as_deref());
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+pub(super) fn parse_wkt(s: &str) -> Result<Geometry> {
+    Geometry::try_from_wkt_str(s)
+        .map_err(|e| {
+            BoxedError::new(PlainError::new(
+                format!("Fail to parse WKT: {}", e),
+                StatusCode::EngineExecuteQuery,
+            ))
+        })
+        .context(error::ExecuteSnafu)
+}