vortex_geo/extension/
linestring.rs1use std::sync::Arc;
9
10use arrow_array::ArrayRef as ArrowArrayRef;
11use arrow_schema::DataType;
12use arrow_schema::Field;
13use arrow_schema::extension::ExtensionType;
14use geo_traits::to_geo::ToGeoGeometry;
15use geo_types::Geometry;
16use geoarrow::array::GeoArrowArrayAccessor;
17use geoarrow::array::IntoArrow;
18use geoarrow::array::LineStringArray;
19use geoarrow::datatypes::CoordType;
20use geoarrow::datatypes::LineStringType;
21use prost::Message;
22use vortex_array::ArrayRef;
23use vortex_array::ExecutionCtx;
24use vortex_array::IntoArray;
25use vortex_array::arrays::ExtensionArray;
26use vortex_array::arrays::extension::ExtensionArrayExt;
27use vortex_array::arrow::ArrowExport;
28use vortex_array::arrow::ArrowExportVTable;
29use vortex_array::arrow::ArrowImport;
30use vortex_array::arrow::ArrowImportVTable;
31use vortex_array::arrow::ArrowSession;
32use vortex_array::arrow::ArrowSessionExt;
33use vortex_array::arrow::FromArrowArray;
34use vortex_array::dtype::DType;
35use vortex_array::dtype::Nullability;
36use vortex_array::dtype::arrow::FromArrowType;
37use vortex_array::dtype::extension::ExtDType;
38use vortex_array::dtype::extension::ExtId;
39use vortex_array::dtype::extension::ExtVTable;
40use vortex_array::scalar::ScalarValue;
41use vortex_error::VortexError;
42use vortex_error::VortexResult;
43use vortex_error::vortex_bail;
44use vortex_error::vortex_ensure;
45use vortex_error::vortex_err;
46use vortex_session::registry::CachedId;
47use vortex_session::registry::Id;
48
49use super::GeoMetadata;
50use super::coordinate::Dimension;
51use super::coordinate::coordinate_dimension;
52use super::coordinate::coordinate_storage_dtype;
53use super::geo_metadata_from_arrow;
54use super::geoarrow_metadata;
55use super::geoarrow_to_wkb;
56
57#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
60pub struct LineString;
61
62impl ExtVTable for LineString {
63 type Metadata = GeoMetadata;
64 type NativeValue<'a> = &'a ScalarValue;
66
67 fn id(&self) -> ExtId {
68 static ID: CachedId = CachedId::new("vortex.geo.linestring");
69 *ID
70 }
71
72 fn serialize_metadata(&self, metadata: &Self::Metadata) -> VortexResult<Vec<u8>> {
73 Ok(metadata.encode_to_vec())
74 }
75
76 fn deserialize_metadata(&self, metadata: &[u8]) -> VortexResult<Self::Metadata> {
77 Ok(GeoMetadata::decode(metadata)?)
78 }
79
80 fn validate_dtype(ext_dtype: &ExtDType<Self>) -> VortexResult<()> {
81 linestring_dimension(ext_dtype.storage_dtype()).map(|_| ())
82 }
83
84 fn unpack_native<'a>(
85 _ext_dtype: &'a ExtDType<Self>,
86 storage_value: &'a ScalarValue,
87 ) -> VortexResult<&'a ScalarValue> {
88 Ok(storage_value)
89 }
90}
91
92pub(crate) fn linestring_storage_dtype(dim: Dimension, nullability: Nullability) -> DType {
94 let coords = coordinate_storage_dtype(dim, Nullability::NonNullable);
95 DType::List(Arc::new(coords), nullability)
96}
97
98pub(crate) fn linestring_dimension(dtype: &DType) -> VortexResult<Dimension> {
100 let DType::List(coords, _) = dtype else {
101 vortex_bail!("linestring storage must be a List of coordinates, was {dtype}");
102 };
103 coordinate_dimension(coords)
104}
105
106static ARROW_LINESTRING: CachedId = CachedId::new(LineStringType::NAME);
107
108fn linestring_type(geo_metadata: &GeoMetadata, dimension: Dimension) -> LineStringType {
111 LineStringType::new(dimension.into(), geoarrow_metadata(geo_metadata))
112}
113
114pub(crate) fn linestring_geometries(
117 storage: &ArrayRef,
118 ctx: &mut ExecutionCtx,
119) -> VortexResult<Vec<Geometry<f64>>> {
120 linestring_array(storage, ctx)?
121 .iter()
122 .map(|geometry| -> VortexResult<Geometry<f64>> {
123 Ok(geometry
124 .ok_or_else(|| vortex_err!("geo: null geometry is not supported"))?
125 .map_err(|e| vortex_err!("geo: geometry access failed: {e}"))?
126 .to_geometry())
127 })
128 .collect()
129}
130
131fn linestring_array(storage: &ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<LineStringArray> {
133 let linestring_type = linestring_type(
134 &GeoMetadata::default(),
135 linestring_dimension(storage.dtype())?,
136 );
137 let session = ctx.session().clone();
138 let arrow = session.arrow().execute_arrow(storage.clone(), None, ctx)?;
139 LineStringArray::try_from((arrow.as_ref(), linestring_type))
140 .map_err(|e| vortex_err!("failed to construct LineStringArray: {e}"))
141}
142
143pub struct LineStringData(ExtensionArray);
145
146impl TryFrom<ExtensionArray> for LineStringData {
147 type Error = VortexError;
148
149 fn try_from(ext: ExtensionArray) -> Result<Self, Self::Error> {
150 vortex_ensure!(
151 ext.ext_dtype().is::<LineString>(),
152 "expected a LineString extension array"
153 );
154 Ok(LineStringData(ext))
155 }
156}
157
158impl LineStringData {
159 pub fn to_wkb(&self, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
161 geoarrow_to_wkb(&linestring_array(self.0.storage_array(), ctx)?)
162 }
163}
164
165impl ArrowExportVTable for LineString {
166 fn arrow_ext_id(&self) -> Id {
167 *ARROW_LINESTRING
168 }
169
170 fn vortex_id(&self) -> Id {
171 self.id()
172 }
173
174 fn to_arrow_field(
175 &self,
176 name: &str,
177 dtype: &DType,
178 session: &ArrowSession,
179 ) -> VortexResult<Option<Field>> {
180 let ext_type = dtype.as_extension();
181 let geo_metadata = ext_type.metadata::<LineString>();
182 let dimension = linestring_dimension(ext_type.storage_dtype())?;
183
184 let mut field = session.to_arrow_field(name, ext_type.storage_dtype())?;
185 field.try_with_extension_type(linestring_type(geo_metadata, dimension))?;
186
187 Ok(Some(field))
188 }
189
190 fn execute_arrow(
191 &self,
192 array: ArrayRef,
193 target: &Field,
194 ctx: &mut ExecutionCtx,
195 ) -> VortexResult<ArrowExport> {
196 let is_linestring = array
197 .dtype()
198 .as_extension_opt()
199 .map(|ext| ext.is::<LineString>())
200 .unwrap_or(false);
201 if !is_linestring {
202 return Ok(ArrowExport::Unsupported(array));
203 }
204
205 let Ok(linestring_meta) = target.try_extension_type::<LineStringType>() else {
206 return Ok(ArrowExport::Unsupported(array));
207 };
208 if linestring_meta.coord_type() != CoordType::Separated {
209 return Ok(ArrowExport::Unsupported(array));
210 }
211
212 let executed = array.execute::<ExtensionArray>(ctx)?;
213 let storage = executed.storage_array().clone();
214
215 let storage_field = Field::new(
216 String::new(),
217 target.data_type().clone(),
218 target.is_nullable(),
219 );
220 let session = ctx.session().clone();
221 let arrow_storage = session
222 .arrow()
223 .execute_arrow(storage, Some(&storage_field), ctx)?;
224
225 let linestrings = LineStringArray::try_from((arrow_storage.as_ref(), linestring_meta))
227 .map_err(|e| vortex_err!("failed to construct LineStringArray: {e}"))?;
228
229 Ok(ArrowExport::Exported(Arc::new(linestrings.into_arrow())))
230 }
231}
232
233impl ArrowImportVTable for LineString {
234 fn arrow_ext_id(&self) -> Id {
235 *ARROW_LINESTRING
236 }
237
238 fn from_arrow_field(&self, field: &Field) -> VortexResult<Option<DType>> {
243 let (dimension, metadata) =
244 if let Ok(linestring_meta) = field.try_extension_type::<LineStringType>() {
245 vortex_ensure!(
246 linestring_meta.coord_type() == CoordType::Separated,
247 "geoarrow.linestring with interleaved coordinates is not supported; \
248 re-encode with separated (struct) coordinates"
249 );
250 (
251 linestring_meta.dimension().into(),
252 geo_metadata_from_arrow(linestring_meta.metadata()),
253 )
254 } else {
255 if field.extension_type_name() != Some(LineStringType::NAME) {
258 return Ok(None);
259 }
260 let DType::List(coords, _) = DType::from_arrow(field) else {
261 return Ok(None);
262 };
263 let DType::Struct(fields, _) = coords.as_ref() else {
264 return Ok(None);
265 };
266 let Ok(dimension) = Dimension::from_field_names(fields.names()) else {
267 return Ok(None);
268 };
269 (dimension, GeoMetadata::default())
270 };
271
272 let storage_dtype = linestring_storage_dtype(dimension, field.is_nullable().into());
273 Ok(Some(DType::Extension(
274 ExtDType::try_with_vtable(LineString, metadata, storage_dtype)?.erased(),
275 )))
276 }
277
278 fn from_arrow_array(
279 &self,
280 array: ArrowArrayRef,
281 field: &Field,
282 dtype: &DType,
283 ) -> VortexResult<ArrowImport> {
284 let Some(ext_dtype) = dtype.as_extension_opt() else {
285 return Ok(ArrowImport::Unsupported(array));
286 };
287 if !ext_dtype.is::<LineString>()
288 || field.try_extension_type::<LineStringType>().is_err()
289 || !matches!(array.data_type(), DataType::List(_))
290 {
291 return Ok(ArrowImport::Unsupported(array));
292 }
293
294 let storage = ArrayRef::from_arrow(array.as_ref(), field.is_nullable())?;
295 Ok(ArrowImport::Imported(
296 ExtensionArray::try_new(ext_dtype.clone(), storage)?.into_array(),
297 ))
298 }
299}
300
301#[cfg(test)]
302mod tests {
303 use rstest::rstest;
304 use vortex_array::dtype::DType;
305 use vortex_array::dtype::Nullability;
306 use vortex_array::dtype::PType;
307 use vortex_array::dtype::extension::ExtDType;
308 use vortex_error::VortexResult;
309
310 use super::LineString;
311 use super::linestring_storage_dtype;
312 use crate::extension::GeoMetadata;
313 use crate::extension::coordinate::Dimension;
314
315 fn geo_meta() -> GeoMetadata {
316 GeoMetadata {
317 crs: Some("EPSG:4326".to_string()),
318 }
319 }
320
321 #[rstest]
323 #[case::xy(Dimension::Xy)]
324 #[case::xyz(Dimension::Xyz)]
325 #[case::xym(Dimension::Xym)]
326 #[case::xyzm(Dimension::Xyzm)]
327 fn linestring_validates_every_dimension(#[case] dim: Dimension) -> VortexResult<()> {
328 let storage = linestring_storage_dtype(dim, Nullability::NonNullable);
329 ExtDType::<LineString>::try_new(geo_meta(), storage)?;
330 Ok(())
331 }
332
333 #[test]
336 fn linestring_rejects_invalid_storage() -> VortexResult<()> {
337 let primitive = DType::Primitive(PType::F64, Nullability::NonNullable);
338 assert!(ExtDType::<LineString>::try_new(geo_meta(), primitive).is_err());
339 Ok(())
340 }
341}