From 719e4fad8aedb54299a564dbd59b58a6bd706a2c Mon Sep 17 00:00:00 2001 From: xuanyili Date: Tue, 16 Jun 2026 00:37:03 +0000 Subject: [PATCH 1/2] feat: support geometry and geography types --- Cargo.lock | 47 ++++ Cargo.toml | 1 + crates/catalog/glue/src/schema.rs | 5 +- crates/catalog/hms/src/schema.rs | 5 +- crates/iceberg/Cargo.toml | 3 +- crates/iceberg/public-api.txt | 43 +++ crates/iceberg/src/arrow/schema.rs | 197 +++++++++++++- crates/iceberg/src/arrow/value.rs | 22 +- crates/iceberg/src/avro/schema.rs | 4 +- crates/iceberg/src/spec/datatypes.rs | 249 ++++++++++++++++++ crates/iceberg/src/spec/transform.rs | 10 +- crates/iceberg/src/spec/values/datum.rs | 6 +- crates/iceberg/src/spec/values/literal.rs | 14 + crates/iceberg/src/spec/values/tests.rs | 38 +++ crates/iceberg/src/transform/bucket.rs | 10 +- crates/iceberg/src/transform/identity.rs | 8 +- crates/iceberg/src/transform/truncate.rs | 10 +- .../src/writer/file_writer/parquet_writer.rs | 135 ++++++++++ 18 files changed, 777 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 87b228a871..c4ab2e01f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3331,6 +3331,26 @@ dependencies = [ "version_check", ] +[[package]] +name = "geo-traits" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e7c353d12a704ccfab1ba8bfb1a7fe6cb18b665bf89d37f4f7890edcd260206" +dependencies = [ + "geo-types", +] + +[[package]] +name = "geo-types" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94776032c45f950d30a13af6113c2ad5625316c9abfbccee4dd5a6695f8fe0f5" +dependencies = [ + "approx", + "num-traits", + "serde", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -3788,6 +3808,7 @@ dependencies = [ "once_cell", "ordered-float 4.6.0", "parquet", + "parquet-geospatial", "pretty_assertions", "rand 0.9.5", "regex", @@ -5459,6 +5480,7 @@ dependencies = [ "num-integer", "num-traits", "object_store", + "parquet-geospatial", "paste", "ring", "seq-macro", @@ -5470,6 +5492,19 @@ dependencies = [ "zstd", ] +[[package]] +name = "parquet-geospatial" +version = "58.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32928da166f0e52c9c15d055181fb8c163e3ceb2fce12c79fdba5cfd284d98a2" +dependencies = [ + "arrow-schema", + "geo-traits", + "serde", + "serde_json", + "wkb", +] + [[package]] name = "paste" version = "1.0.15" @@ -8843,6 +8878,18 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "wkb" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a120b336c7ad17749026d50427c23d838ecb50cd64aaea6254b5030152f890a9" +dependencies = [ + "byteorder", + "geo-traits", + "num_enum", + "thiserror 1.0.69", +] + [[package]] name = "writeable" version = "0.6.3" diff --git a/Cargo.toml b/Cargo.toml index a394ec107c..663f58431b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -122,6 +122,7 @@ once_cell = "1.20" opendal = "0.57" ordered-float = "4" parquet = "58.4" +parquet-geospatial = "58.4" pilota = "0.11.10" pretty_assertions = "1.4" pyo3 = "0.28" diff --git a/crates/catalog/glue/src/schema.rs b/crates/catalog/glue/src/schema.rs index 59d51f3dc1..c49386294b 100644 --- a/crates/catalog/glue/src/schema.rs +++ b/crates/catalog/glue/src/schema.rs @@ -166,7 +166,10 @@ impl SchemaVisitor for GlueSchemaBuilder { PrimitiveType::Time | PrimitiveType::String | PrimitiveType::Uuid => { "string".to_string() } - PrimitiveType::Binary | PrimitiveType::Fixed(_) => "binary".to_string(), + PrimitiveType::Binary + | PrimitiveType::Fixed(_) + | PrimitiveType::Geometry(_) + | PrimitiveType::Geography(_) => "binary".to_string(), PrimitiveType::Decimal { precision, scale } => { format!("decimal({precision},{scale})") } diff --git a/crates/catalog/hms/src/schema.rs b/crates/catalog/hms/src/schema.rs index e21a75f976..a224680c28 100644 --- a/crates/catalog/hms/src/schema.rs +++ b/crates/catalog/hms/src/schema.rs @@ -117,7 +117,10 @@ impl SchemaVisitor for HiveSchemaBuilder { PrimitiveType::Time | PrimitiveType::String | PrimitiveType::Uuid => { "string".to_string() } - PrimitiveType::Binary | PrimitiveType::Fixed(_) => "binary".to_string(), + PrimitiveType::Binary + | PrimitiveType::Fixed(_) + | PrimitiveType::Geometry(_) + | PrimitiveType::Geography(_) => "binary".to_string(), PrimitiveType::Decimal { precision, scale } => { format!("decimal({precision},{scale})") } diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index c33445c64c..cca271de66 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -63,7 +63,8 @@ moka = { version = "0.12.10", features = ["future"] } murmur3 = { workspace = true } once_cell = { workspace = true } ordered-float = { workspace = true } -parquet = { workspace = true, features = ["async", "encryption"] } +parquet = { workspace = true, features = ["async", "encryption", "geospatial"] } +parquet-geospatial = { workspace = true } rand = { workspace = true } reqwest = { workspace = true } roaring = { workspace = true } diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 610675c83d..7116525bf5 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1597,6 +1597,8 @@ pub iceberg::spec::PrimitiveType::Decimal::scale: u32 pub iceberg::spec::PrimitiveType::Double pub iceberg::spec::PrimitiveType::Fixed(u64) pub iceberg::spec::PrimitiveType::Float +pub iceberg::spec::PrimitiveType::Geography(iceberg::spec::GeographyType) +pub iceberg::spec::PrimitiveType::Geometry(iceberg::spec::GeometryType) pub iceberg::spec::PrimitiveType::Int pub iceberg::spec::PrimitiveType::Long pub iceberg::spec::PrimitiveType::String @@ -1973,6 +1975,47 @@ impl serde_core::ser::Serialize for iceberg::spec::FieldSummary pub fn iceberg::spec::FieldSummary::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::FieldSummary pub fn iceberg::spec::FieldSummary::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::spec::GeographyType +impl iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::algorithm(&self) -> parquet_geospatial::types::Edges +pub fn iceberg::spec::GeographyType::crs(&self) -> core::option::Option<&str> +pub fn iceberg::spec::GeographyType::new(crs: core::option::Option, algorithm: parquet_geospatial::types::Edges) -> iceberg::Result +impl core::clone::Clone for iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::clone(&self) -> iceberg::spec::GeographyType +impl core::cmp::Eq for iceberg::spec::GeographyType +impl core::cmp::PartialEq for iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::eq(&self, other: &iceberg::spec::GeographyType) -> bool +impl core::default::Default for iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::default() -> Self +impl core::fmt::Debug for iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::hash::Hash for iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::hash(&self, state: &mut H) +impl core::marker::StructuralPartialEq for iceberg::spec::GeographyType +impl serde_core::ser::Serialize for iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::GeographyType +pub fn iceberg::spec::GeographyType::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub struct iceberg::spec::GeometryType +impl iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::crs(&self) -> core::option::Option<&str> +pub fn iceberg::spec::GeometryType::new(crs: core::option::Option) -> iceberg::Result +impl core::clone::Clone for iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::clone(&self) -> iceberg::spec::GeometryType +impl core::cmp::Eq for iceberg::spec::GeometryType +impl core::cmp::PartialEq for iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::eq(&self, other: &iceberg::spec::GeometryType) -> bool +impl core::default::Default for iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::default() -> iceberg::spec::GeometryType +impl core::fmt::Debug for iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::hash::Hash for iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::hash<__H: core::hash::Hasher>(&self, state: &mut __H) +impl core::marker::StructuralPartialEq for iceberg::spec::GeometryType +impl serde_core::ser::Serialize for iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::GeometryType +pub fn iceberg::spec::GeometryType::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::spec::ListType pub iceberg::spec::ListType::element_field: iceberg::spec::NestedFieldRef impl iceberg::spec::ListType diff --git a/crates/iceberg/src/arrow/schema.rs b/crates/iceberg/src/arrow/schema.rs index 923c043c74..b4dfc23708 100644 --- a/crates/iceberg/src/arrow/schema.rs +++ b/crates/iceberg/src/arrow/schema.rs @@ -32,13 +32,15 @@ use arrow_schema::{ }; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; use parquet::file::statistics::Statistics; +use parquet_geospatial::{WkbMetadata, WkbType, WkbTypeHint}; use uuid::Uuid; use crate::error::Result; use crate::spec::decimal_utils::i128_from_be_bytes; use crate::spec::{ - Datum, FIRST_FIELD_ID, ListType, MapType, NestedField, NestedFieldRef, PrimitiveLiteral, - PrimitiveType, Schema, SchemaVisitor, StructType, Type, VariantType, + Datum, FIRST_FIELD_ID, GeographyType, GeometryType, ListType, MapType, NestedField, + NestedFieldRef, PrimitiveLiteral, PrimitiveType, Schema, SchemaVisitor, StructType, Type, + VariantType, }; use crate::{Error, ErrorKind}; @@ -396,7 +398,7 @@ impl ArrowSchemaConverter { let mut results = Vec::with_capacity(fields.len()); for i in 0..fields.len() { let field = &fields[i]; - let field_type = &field_results[i]; + let field_type = self.apply_field_extension_type(field, &field_results[i])?; let id = self.get_field_id(field)?; let doc = get_field_doc(field); let nested_field = NestedField { @@ -404,7 +406,7 @@ impl ArrowSchemaConverter { doc, name: field.name().clone(), required: !field.is_nullable(), - field_type: Box::new(field_type.clone()), + field_type: Box::new(field_type), initial_default: None, write_default: None, }; @@ -412,6 +414,50 @@ impl ArrowSchemaConverter { } Ok(results) } + + fn apply_field_extension_type(&self, field: &FieldRef, field_type: &Type) -> Result { + if field.extension_type_name() != Some(WkbType::NAME) { + return Ok(field_type.clone()); + } + + let wkb_type = field.try_extension_type::().map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid geospatial Arrow extension metadata for field {}", + field.name() + ), + ) + .with_source(err) + })?; + + let crs = wkb_type.metadata().crs.as_ref().map(|crs| match crs { + serde_json::Value::String(value) => value.clone(), + other => other.to_string(), + }); + + match wkb_type.metadata().type_hint() { + WkbTypeHint::Geometry => Ok(Type::Primitive(PrimitiveType::Geometry( + GeometryType::new(crs).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid geometry CRS for field {}", field.name()), + ) + .with_source(err) + })?, + ))), + WkbTypeHint::Geography => Ok(Type::Primitive(PrimitiveType::Geography( + GeographyType::new(crs, wkb_type.metadata().algorithm.unwrap_or_default()) + .map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid geography CRS for field {}", field.name()), + ) + .with_source(err) + })?, + ))), + } + } } impl ArrowSchemaVisitor for ArrowSchemaConverter { @@ -628,15 +674,42 @@ impl SchemaVisitor for ToArrowSchemaConverter { } else { HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), field.id.to_string())]) }; - let arrow_field = + let mut arrow_field = Field::new(field.name.clone(), ty, !field.required).with_metadata(metadata); - // A variant column's storage is a struct; tag the field with the canonical - // `arrow.parquet.variant` extension type so consumers read it as a Variant, not a struct. - let arrow_field = if field.field_type.is_variant() { - arrow_field.with_extension_type(VariantExtensionType) - } else { - arrow_field - }; + + match field.field_type.as_ref() { + Type::Variant(_) => { + // A variant column's storage is a struct; tag the field with the canonical + // `arrow.parquet.variant` extension type so consumers read it as a Variant, not a struct. + arrow_field = arrow_field.with_extension_type(VariantExtensionType); + } + Type::Primitive(PrimitiveType::Geometry(geometry)) => { + let metadata = WkbMetadata::new(geometry.crs(), None); + arrow_field + .try_with_extension_type(WkbType::new(Some(metadata))) + .map_err(|err| { + Error::new( + crate::ErrorKind::DataInvalid, + "Failed to attach geometry Arrow extension metadata", + ) + .with_source(err) + })?; + } + Type::Primitive(PrimitiveType::Geography(geography)) => { + let metadata = WkbMetadata::new(geography.crs(), Some(geography.algorithm())); + arrow_field + .try_with_extension_type(WkbType::new(Some(metadata))) + .map_err(|err| { + Error::new( + crate::ErrorKind::DataInvalid, + "Failed to attach geography Arrow extension metadata", + ) + .with_source(err) + })?; + } + _ => {} + } + Ok(ArrowSchemaOrFieldOrType::Field(arrow_field)) } @@ -763,7 +836,9 @@ impl SchemaVisitor for ToArrowSchemaConverter { .map(DataType::FixedSizeBinary) .unwrap_or(DataType::LargeBinary), )), - PrimitiveType::Binary => Ok(ArrowSchemaOrFieldOrType::Type(DataType::LargeBinary)), + PrimitiveType::Binary | PrimitiveType::Geometry(_) | PrimitiveType::Geography(_) => { + Ok(ArrowSchemaOrFieldOrType::Type(DataType::LargeBinary)) + } } } @@ -1393,7 +1468,9 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; + use arrow_schema::extension::ExtensionType; use arrow_schema::{DataType, Field, Schema as ArrowSchema, TimeUnit}; + use parquet_geospatial::WkbEdges; use super::*; use crate::spec::decimal_utils::decimal_new; @@ -2211,6 +2288,100 @@ mod tests { ); } + #[test] + fn test_geospatial_arrow_schema_roundtrip() { + let schema = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required( + 1, + "geom", + Type::Primitive(PrimitiveType::Geometry( + GeometryType::new(Some("srid:4326".to_string())).unwrap(), + )), + ) + .into(), + NestedField::optional( + 2, + "geog", + Type::Primitive(PrimitiveType::Geography( + GeographyType::new(Some("srid:3857".to_string()), WkbEdges::Karney) + .unwrap(), + )), + ) + .into(), + NestedField::optional( + 3, + "geom_list", + Type::List(ListType::new( + NestedField::list_element( + 4, + Type::Primitive(PrimitiveType::Geometry(GeometryType::default())), + true, + ) + .into(), + )), + ) + .into(), + ]) + .build() + .unwrap(); + + let arrow_schema = schema_to_arrow_schema(&schema).unwrap(); + let geom = arrow_schema.field(0); + assert_eq!(geom.data_type(), &DataType::LargeBinary); + let geom_wkb = geom.try_extension_type::().unwrap(); + assert_eq!( + geom_wkb + .metadata() + .crs + .as_ref() + .and_then(|crs| crs.as_str()), + Some("srid:4326") + ); + assert!(matches!( + geom_wkb.metadata().type_hint(), + WkbTypeHint::Geometry + )); + + let geog = arrow_schema.field(1); + assert_eq!(geog.data_type(), &DataType::LargeBinary); + let geog_wkb = geog.try_extension_type::().unwrap(); + assert_eq!( + geog_wkb + .metadata() + .crs + .as_ref() + .and_then(|crs| crs.as_str()), + Some("srid:3857") + ); + assert_eq!(geog_wkb.metadata().algorithm, Some(WkbEdges::Karney)); + assert!(matches!( + geog_wkb.metadata().type_hint(), + WkbTypeHint::Geography + )); + + let list = arrow_schema.field(2); + let DataType::List(element) = list.data_type() else { + panic!("Expected list field"); + }; + assert_eq!(element.data_type(), &DataType::LargeBinary); + assert!( + matches!( + element + .try_extension_type::() + .unwrap() + .metadata() + .type_hint(), + WkbTypeHint::Geometry + ), + "Expected list element to retain WKB extension metadata" + ); + + let converted = arrow_schema_to_schema(&arrow_schema).unwrap(); + assert_eq!(converted.as_struct().fields(), schema.as_struct().fields()); + } + #[test] fn test_type_conversion() { // test primitive type diff --git a/crates/iceberg/src/arrow/value.rs b/crates/iceberg/src/arrow/value.rs index 34f6ce6e94..18ec9f735f 100644 --- a/crates/iceberg/src/arrow/value.rs +++ b/crates/iceberg/src/arrow/value.rs @@ -406,7 +406,7 @@ impl SchemaWithPartnerVisitor for ArrowArrayToIcebergStructConverter { .map(|v| v.map(|v| Literal::fixed(v.iter().cloned()))) .collect()) } - PrimitiveType::Binary => { + PrimitiveType::Binary | PrimitiveType::Geometry(_) | PrimitiveType::Geography(_) => { if let Some(array) = partner.as_any().downcast_ref::() { Ok(array .iter() @@ -702,6 +702,12 @@ pub(crate) fn create_primitive_array_single_element( (DataType::Binary, None) => Ok(Arc::new(BinaryArray::from_opt_vec(vec![ Option::<&[u8]>::None, ]))), + (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(v))) => { + Ok(Arc::new(LargeBinaryArray::from_vec(vec![v.as_slice()]))) + } + (DataType::LargeBinary, None) => Ok(Arc::new(LargeBinaryArray::from_opt_vec(vec![ + Option::<&[u8]>::None, + ]))), (DataType::Decimal128(precision, scale), Some(PrimitiveLiteral::Int128(v))) => { let array = Decimal128Array::from(vec![{ *v }]) .with_precision_and_scale(*precision, *scale) @@ -790,6 +796,13 @@ pub(crate) fn create_primitive_array_single_element( as ArrayRef, ) } + DataType::LargeBinary => { + Ok( + Arc::new(LargeBinaryArray::from_opt_vec(vec![ + Option::<&[u8]>::None, + ])) as ArrayRef, + ) + } _ => Err(Error::new( ErrorKind::Unexpected, format!("Unsupported struct field type: {:?}", f.data_type()), @@ -919,6 +932,13 @@ pub(crate) fn create_primitive_array_repeated( let vals: Vec> = vec![None; num_rows]; Arc::new(BinaryArray::from_opt_vec(vals)) } + (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(value))) => { + Arc::new(LargeBinaryArray::from_vec(vec![value; num_rows])) + } + (DataType::LargeBinary, None) => { + let vals: Vec> = vec![None; num_rows]; + Arc::new(LargeBinaryArray::from_opt_vec(vals)) + } (DataType::Decimal128(precision, scale), Some(PrimitiveLiteral::Int128(value))) => { Arc::new( Decimal128Array::from(vec![*value; num_rows]) diff --git a/crates/iceberg/src/avro/schema.rs b/crates/iceberg/src/avro/schema.rs index 528b89b7c1..e7b7f84cc9 100644 --- a/crates/iceberg/src/avro/schema.rs +++ b/crates/iceberg/src/avro/schema.rs @@ -236,7 +236,9 @@ impl SchemaVisitor for SchemaToAvroSchema { PrimitiveType::String => AvroSchema::String, PrimitiveType::Uuid => AvroSchema::Uuid, PrimitiveType::Fixed(len) => avro_fixed_schema((*len) as usize)?, - PrimitiveType::Binary => AvroSchema::Bytes, + PrimitiveType::Binary | PrimitiveType::Geometry(_) | PrimitiveType::Geography(_) => { + AvroSchema::Bytes + } PrimitiveType::Decimal { precision, scale } => { avro_decimal_schema(*precision as usize, *scale as usize)? } diff --git a/crates/iceberg/src/spec/datatypes.rs b/crates/iceberg/src/spec/datatypes.rs index 79c48c1318..43795824fc 100644 --- a/crates/iceberg/src/spec/datatypes.rs +++ b/crates/iceberg/src/spec/datatypes.rs @@ -21,10 +21,12 @@ use std::collections::HashMap; use std::convert::identity; use std::fmt; +use std::hash::{Hash, Hasher}; use std::ops::Index; use std::sync::{Arc, OnceLock}; use ::serde::de::{MapAccess, Visitor}; +use parquet_geospatial::WkbEdges; use serde::de::{Error, IntoDeserializer}; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use serde_json::Value as JsonValue; @@ -44,6 +46,7 @@ pub const MAP_VALUE_FIELD_NAME: &str = "value"; pub(crate) const MAX_DECIMAL_BYTES: u32 = 24; pub(crate) const MAX_DECIMAL_PRECISION: u32 = 38; +const DEFAULT_GEOSPATIAL_CRS: &str = "OGC:CRS84"; mod _decimal { use once_cell::sync::Lazy; @@ -231,6 +234,157 @@ impl From for Type { } } +/// Iceberg geometry type. +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash, Default)] +pub struct GeometryType { + crs: Option, +} + +impl GeometryType { + /// Creates a geometry type with an optional coordinate reference system. + pub fn new(crs: Option) -> Result { + Ok(Self { + crs: normalize_crs(crs)?, + }) + } + + /// Returns the coordinate reference system, or `None` for the Iceberg default CRS. + pub fn crs(&self) -> Option<&str> { + self.crs.as_deref() + } +} + +/// Iceberg geography type. +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)] +pub struct GeographyType { + crs: Option, + algorithm: WkbEdges, +} + +impl Default for GeographyType { + fn default() -> Self { + Self { + crs: None, + algorithm: WkbEdges::Spherical, + } + } +} + +impl Hash for GeographyType { + fn hash(&self, state: &mut H) { + self.crs.hash(state); + wkb_edges_as_str(self.algorithm).hash(state); + } +} + +impl GeographyType { + /// Creates a geography type with an optional coordinate reference system and edge interpolation algorithm. + pub fn new(crs: Option, algorithm: WkbEdges) -> Result { + Ok(Self { + crs: normalize_crs(crs)?, + algorithm, + }) + } + + /// Returns the coordinate reference system, or `None` for the Iceberg default CRS. + pub fn crs(&self) -> Option<&str> { + self.crs.as_deref() + } + + /// Returns the edge interpolation algorithm. + pub fn algorithm(&self) -> WkbEdges { + self.algorithm + } +} + +fn normalize_crs(crs: Option) -> Result> { + let Some(crs) = crs else { + return Ok(None); + }; + let crs = crs.trim().to_string(); + if crs.is_empty() { + return Err(crate::Error::new( + crate::ErrorKind::DataInvalid, + "Geospatial CRS must not be empty", + )); + } + Ok((crs != DEFAULT_GEOSPATIAL_CRS).then_some(crs)) +} + +fn wkb_edges_as_str(edges: WkbEdges) -> &'static str { + match edges { + WkbEdges::Spherical => "spherical", + WkbEdges::Vincenty => "vincenty", + WkbEdges::Thomas => "thomas", + WkbEdges::Andoyer => "andoyer", + WkbEdges::Karney => "karney", + } +} + +fn parse_wkb_edges(value: &str) -> std::result::Result { + match value.trim().to_ascii_lowercase().as_str() { + "spherical" => Ok(WkbEdges::Spherical), + "vincenty" => Ok(WkbEdges::Vincenty), + "thomas" => Ok(WkbEdges::Thomas), + "andoyer" => Ok(WkbEdges::Andoyer), + "karney" => Ok(WkbEdges::Karney), + _ => Err(format!( + "Unknown geography edge interpolation algorithm: {value}" + )), + } +} + +fn parse_geospatial_params<'a>( + value: &'a str, + type_name: &str, +) -> std::result::Result, String> { + if value == type_name { + return Ok(vec![]); + } + + let params = value + .strip_prefix(&format!("{type_name}(")) + .and_then(|s| s.strip_suffix(')')) + .ok_or_else(|| format!("Invalid {type_name} type: {value}"))?; + + if params.trim().is_empty() { + return Err(format!("{type_name} requires a non-empty CRS")); + } + + Ok(params.split(',').map(str::trim).collect()) +} + +fn parse_geometry(value: &str) -> std::result::Result { + let params = parse_geospatial_params(value.trim(), "geometry")?; + let geometry = match params.as_slice() { + [] => GeometryType::default(), + [crs] if !crs.is_empty() => { + GeometryType::new(Some((*crs).to_string())).map_err(|err| err.to_string())? + } + _ => return Err(format!("Invalid geometry type: {value}")), + }; + + Ok(PrimitiveType::Geometry(geometry)) +} + +fn parse_geography(value: &str) -> std::result::Result { + let params = parse_geospatial_params(value.trim(), "geography")?; + let geography = match params.as_slice() { + [] => GeographyType::default(), + [crs] if !crs.is_empty() => { + GeographyType::new(Some((*crs).to_string()), WkbEdges::Spherical) + .map_err(|err| err.to_string())? + } + [crs, algorithm] if !crs.is_empty() && !algorithm.is_empty() => { + GeographyType::new(Some((*crs).to_string()), parse_wkb_edges(algorithm)?) + .map_err(|err| err.to_string())? + } + _ => return Err(format!("Invalid geography type: {value}")), + }; + + Ok(PrimitiveType::Geography(geography)) +} + /// Primitive data types #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash)] #[serde(rename_all = "lowercase", remote = "Self")] @@ -274,6 +428,10 @@ pub enum PrimitiveType { Fixed(u64), /// Arbitrary-length byte array. Binary, + /// Geometry values encoded as well-known binary. + Geometry(GeometryType), + /// Geography values encoded as well-known binary. + Geography(GeographyType), } impl PrimitiveType { @@ -297,6 +455,8 @@ impl PrimitiveType { | (PrimitiveType::Uuid, PrimitiveLiteral::UInt128(_)) | (PrimitiveType::Fixed(_), PrimitiveLiteral::Binary(_)) | (PrimitiveType::Binary, PrimitiveLiteral::Binary(_)) + | (PrimitiveType::Geometry(_), PrimitiveLiteral::Binary(_)) + | (PrimitiveType::Geography(_), PrimitiveLiteral::Binary(_)) ) } } @@ -325,6 +485,10 @@ impl<'de> Deserialize<'de> for PrimitiveType { deserialize_decimal(s.into_deserializer()) } else if s.starts_with("fixed") { deserialize_fixed(s.into_deserializer()) + } else if s.starts_with("geometry") { + parse_geometry(&s).map_err(D::Error::custom) + } else if s.starts_with("geography") { + parse_geography(&s).map_err(D::Error::custom) } else { PrimitiveType::deserialize(s.into_deserializer()) } @@ -339,6 +503,9 @@ impl Serialize for PrimitiveType { serialize_decimal(precision, scale, serializer) } PrimitiveType::Fixed(l) => serialize_fixed(l, serializer), + PrimitiveType::Geometry(_) | PrimitiveType::Geography(_) => { + serializer.serialize_str(&self.to_string()) + } _ => PrimitiveType::serialize(self, serializer), } } @@ -409,6 +576,23 @@ impl fmt::Display for PrimitiveType { PrimitiveType::Uuid => write!(f, "uuid"), PrimitiveType::Fixed(size) => write!(f, "fixed({size})"), PrimitiveType::Binary => write!(f, "binary"), + PrimitiveType::Geometry(geometry) => match geometry.crs() { + Some(crs) => write!(f, "geometry({crs})"), + None => write!(f, "geometry"), + }, + PrimitiveType::Geography(geography) => { + let algorithm = geography.algorithm(); + match (geography.crs(), algorithm) { + (None, WkbEdges::Spherical) => write!(f, "geography"), + (Some(crs), WkbEdges::Spherical) => write!(f, "geography({crs})"), + (crs, algorithm) => write!( + f, + "geography({}, {})", + crs.unwrap_or(DEFAULT_GEOSPATIAL_CRS), + wkb_edges_as_str(algorithm) + ), + } + } } } } @@ -1038,6 +1222,63 @@ mod tests { ) } + #[test] + fn primitive_type_geospatial() { + let cases = vec![ + ( + r#""geometry""#, + PrimitiveType::Geometry(GeometryType::default()), + "geometry", + ), + ( + r#""geometry(srid:4326)""#, + PrimitiveType::Geometry(GeometryType::new(Some("srid:4326".to_string())).unwrap()), + "geometry(srid:4326)", + ), + ( + r#""geography""#, + PrimitiveType::Geography(GeographyType::default()), + "geography", + ), + ( + r#""geography(srid:4326)""#, + PrimitiveType::Geography( + GeographyType::new(Some("srid:4326".to_string()), WkbEdges::Spherical).unwrap(), + ), + "geography(srid:4326)", + ), + ( + r#""geography(srid:4326,karney)""#, + PrimitiveType::Geography( + GeographyType::new(Some("srid:4326".to_string()), WkbEdges::Karney).unwrap(), + ), + "geography(srid:4326, karney)", + ), + ]; + + for (json, expected, display) in cases { + let actual: PrimitiveType = serde_json::from_str(json).unwrap(); + assert_eq!(actual, expected); + assert_eq!(actual.to_string(), display); + assert_eq!( + serde_json::to_string(&actual).unwrap(), + format!(r#""{display}""#) + ); + } + + let invalid_cases = vec![ + r#""geometry()""#, + r#""geometry(a,b)""#, + r#""geography()""#, + r#""geography(srid:4326,unknown)""#, + r#""geography(a,b,c)""#, + ]; + + for json in invalid_cases { + assert!(serde_json::from_str::(json).is_err()); + } + } + #[test] fn struct_type() { let record = r#" @@ -1337,6 +1578,14 @@ mod tests { ), (PrimitiveType::Fixed(8), PrimitiveLiteral::Binary(vec![1])), (PrimitiveType::Binary, PrimitiveLiteral::Binary(vec![1])), + ( + PrimitiveType::Geometry(GeometryType::default()), + PrimitiveLiteral::Binary(vec![1]), + ), + ( + PrimitiveType::Geography(GeographyType::default()), + PrimitiveLiteral::Binary(vec![1]), + ), ]; for (ty, literal) in pairs { assert!(ty.compatible(&literal)); diff --git a/crates/iceberg/src/spec/transform.rs b/crates/iceberg/src/spec/transform.rs index 97ab638e79..56fe8e247f 100644 --- a/crates/iceberg/src/spec/transform.rs +++ b/crates/iceberg/src/spec/transform.rs @@ -160,7 +160,15 @@ impl Transform { pub fn result_type(&self, input_type: &Type) -> Result { match self { Transform::Identity => { - if matches!(input_type, Type::Primitive(_)) { + if matches!( + input_type, + Type::Primitive(PrimitiveType::Geometry(_) | PrimitiveType::Geography(_)) + ) { + Err(Error::new( + ErrorKind::DataInvalid, + format!("{input_type} is not a valid input type of identity transform",), + )) + } else if matches!(input_type, Type::Primitive(_)) { Ok(input_type.clone()) } else { Err(Error::new( diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index f170a09df5..aaaa5e74d9 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -417,8 +417,10 @@ impl Datum { PrimitiveType::Uuid => { PrimitiveLiteral::UInt128(u128::from_be_bytes(bytes.try_into()?)) } - PrimitiveType::Fixed(_) => PrimitiveLiteral::Binary(Vec::from(bytes)), - PrimitiveType::Binary => PrimitiveLiteral::Binary(Vec::from(bytes)), + PrimitiveType::Fixed(_) + | PrimitiveType::Binary + | PrimitiveType::Geometry(_) + | PrimitiveType::Geography(_) => PrimitiveLiteral::Binary(Vec::from(bytes)), PrimitiveType::Decimal { .. } => { PrimitiveLiteral::Int128(i128_from_be_bytes(bytes).ok_or_else(|| { Error::new( diff --git a/crates/iceberg/src/spec/values/literal.rs b/crates/iceberg/src/spec/values/literal.rs index 5296eff2a2..24ae89b519 100644 --- a/crates/iceberg/src/spec/values/literal.rs +++ b/crates/iceberg/src/spec/values/literal.rs @@ -534,6 +534,13 @@ impl Literal { (PrimitiveType::Binary, JsonValue::String(s)) => Ok(Some(Literal::Primitive( PrimitiveLiteral::Binary(decode_hex_bytes(&s)?), ))), + ( + PrimitiveType::Geometry(_) | PrimitiveType::Geography(_), + JsonValue::String(_), + ) => Err(Error::new( + crate::ErrorKind::DataInvalid, + "Geometry and geography defaults must be null", + )), ( PrimitiveType::Decimal { precision: _, @@ -703,6 +710,13 @@ impl Literal { (PrimitiveType::Binary, PrimitiveLiteral::Binary(val)) => { Ok(JsonValue::String(encode_hex_bytes(&val))) } + ( + PrimitiveType::Geometry(_) | PrimitiveType::Geography(_), + PrimitiveLiteral::Binary(_), + ) => Err(Error::new( + ErrorKind::DataInvalid, + "Geometry and geography defaults must be null", + )), (_, PrimitiveLiteral::Int128(val)) => match r#type { Type::Primitive(PrimitiveType::Decimal { precision: _precision, diff --git a/crates/iceberg/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index 8bf3311004..e10f023d3e 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/crates/iceberg/src/spec/values/tests.rs @@ -308,6 +308,27 @@ fn json_binary() { ); } +#[test] +fn json_geospatial_defaults_reject_non_null() { + let cases = vec![ + Type::Primitive(PrimitiveType::Geometry(Default::default())), + Type::Primitive(PrimitiveType::Geography(Default::default())), + ]; + + for ty in cases { + assert!(Literal::try_from_json(serde_json::json!("00010fff"), &ty).is_err()); + assert_eq!( + Literal::try_from_json(serde_json::Value::Null, &ty).unwrap(), + None + ); + assert!( + Literal::Primitive(PrimitiveLiteral::Binary(vec![0, 1, 15, 255])) + .try_into_json(&ty) + .is_err() + ); + } +} + #[test] fn json_fixed() { let record = r#""00010fff""#; @@ -482,6 +503,23 @@ fn avro_bytes_string() { check_avro_bytes_serde(bytes, Datum::string("iceberg"), &PrimitiveType::String); } +#[test] +fn avro_bytes_geospatial() { + let cases = vec![ + PrimitiveType::Geometry(Default::default()), + PrimitiveType::Geography(Default::default()), + ]; + + for ty in cases { + let bytes = vec![1u8, 2u8, 3u8, 4u8]; + check_avro_bytes_serde( + bytes.clone(), + Datum::new(ty.clone(), PrimitiveLiteral::Binary(bytes)), + &ty, + ); + } +} + #[test] fn avro_bytes_decimal() { // (input_bytes, decimal_num, expect_scale, expect_precision) diff --git a/crates/iceberg/src/transform/bucket.rs b/crates/iceberg/src/transform/bucket.rs index 9097f30c45..a35d077ce9 100644 --- a/crates/iceberg/src/transform/bucket.rs +++ b/crates/iceberg/src/transform/bucket.rs @@ -290,12 +290,14 @@ mod test { use crate::Result; use crate::expr::PredicateOperator; use crate::spec::PrimitiveType::{ - Binary, Date, Decimal, Fixed, Int, Long, String as StringType, Time, Timestamp, - TimestampNs, Timestamptz, TimestamptzNs, Uuid, + Binary, Date, Decimal, Fixed, Geography, Geometry, Int, Long, String as StringType, Time, + Timestamp, TimestampNs, Timestamptz, TimestamptzNs, Uuid, }; use crate::spec::Type::{Primitive, Struct}; use crate::spec::decimal_utils::decimal_new; - use crate::spec::{Datum, NestedField, PrimitiveType, StructType, Transform}; + use crate::spec::{ + Datum, GeographyType, GeometryType, NestedField, PrimitiveType, StructType, Transform, Type, + }; use crate::transform::TransformFunction; use crate::transform::test::{TestProjectionFixture, TestTransformFixture}; @@ -334,6 +336,8 @@ mod test { (Primitive(Timestamptz), Some(Primitive(Int))), (Primitive(TimestampNs), Some(Primitive(Int))), (Primitive(TimestamptzNs), Some(Primitive(Int))), + (Primitive(Geometry(GeometryType::default())), None), + (Primitive(Geography(GeographyType::default())), None), ( Struct(StructType::new(vec![ NestedField::optional(1, "a", Primitive(Timestamp)).into(), diff --git a/crates/iceberg/src/transform/identity.rs b/crates/iceberg/src/transform/identity.rs index dd096337ac..043a1ddb83 100644 --- a/crates/iceberg/src/transform/identity.rs +++ b/crates/iceberg/src/transform/identity.rs @@ -37,11 +37,11 @@ impl TransformFunction for Identity { #[cfg(test)] mod test { use crate::spec::PrimitiveType::{ - Binary, Date, Decimal, Fixed, Int, Long, String as StringType, Time, Timestamp, - TimestampNs, Timestamptz, TimestamptzNs, Uuid, + Binary, Date, Decimal, Fixed, Geography, Geometry, Int, Long, String as StringType, Time, + Timestamp, TimestampNs, Timestamptz, TimestamptzNs, Uuid, }; use crate::spec::Type::{Primitive, Struct}; - use crate::spec::{NestedField, StructType, Transform}; + use crate::spec::{GeographyType, GeometryType, NestedField, StructType, Transform}; use crate::transform::test::TestTransformFixture; #[test] @@ -83,6 +83,8 @@ mod test { (Primitive(Timestamptz), Some(Primitive(Timestamptz))), (Primitive(TimestampNs), Some(Primitive(TimestampNs))), (Primitive(TimestamptzNs), Some(Primitive(TimestamptzNs))), + (Primitive(Geometry(GeometryType::default())), None), + (Primitive(Geography(GeographyType::default())), None), ( Struct(StructType::new(vec![ NestedField::optional(1, "a", Primitive(Timestamp)).into(), diff --git a/crates/iceberg/src/transform/truncate.rs b/crates/iceberg/src/transform/truncate.rs index f60f8e4939..b83126cb35 100644 --- a/crates/iceberg/src/transform/truncate.rs +++ b/crates/iceberg/src/transform/truncate.rs @@ -195,12 +195,14 @@ mod test { use crate::Result; use crate::expr::PredicateOperator; use crate::spec::PrimitiveType::{ - Binary, Date, Decimal, Fixed, Int, Long, String as StringType, Time, Timestamp, - TimestampNs, Timestamptz, TimestamptzNs, Uuid, + Binary, Date, Decimal, Fixed, Geography, Geometry, Int, Long, String as StringType, Time, + Timestamp, TimestampNs, Timestamptz, TimestamptzNs, Uuid, }; use crate::spec::Type::{Primitive, Struct}; use crate::spec::decimal_utils::decimal_new; - use crate::spec::{Datum, NestedField, PrimitiveType, StructType, Transform}; + use crate::spec::{ + Datum, GeographyType, GeometryType, NestedField, PrimitiveType, StructType, Transform, Type, + }; use crate::transform::TransformFunction; use crate::transform::test::{TestProjectionFixture, TestTransformFixture}; @@ -243,6 +245,8 @@ mod test { (Primitive(Timestamptz), None), (Primitive(TimestampNs), None), (Primitive(TimestamptzNs), None), + (Primitive(Geometry(GeometryType::default())), None), + (Primitive(Geography(GeographyType::default())), None), ( Struct(StructType::new(vec![ NestedField::optional(1, "a", Primitive(Timestamp)).into(), diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index db9f170938..2e08d7e8f8 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -317,6 +317,10 @@ impl MinMaxColAggregator { )); }; + if matches!(ty, PrimitiveType::Geometry(_) | PrimitiveType::Geography(_)) { + return Ok(()); + } + if value.min_is_exact() { let Some(min_datum) = get_parquet_stat_min_as_datum(&ty, &value)? else { return Err(Error::new( @@ -663,7 +667,11 @@ mod tests { use arrow_schema::{DataType, Field, Fields, SchemaRef as ArrowSchemaRef}; use arrow_select::concat::concat_batches; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use parquet::basic::{EdgeInterpolationAlgorithm, LogicalType}; + use parquet::data_type::ByteArray; use parquet::file::statistics::ValueStatistics; + use parquet_geospatial::WkbEdges; + use parquet_geospatial::testing::wkb_point_xy; use tempfile::TempDir; use uuid::Uuid; @@ -2306,6 +2314,100 @@ mod tests { assert_eq!(std::fs::read_dir(temp_dir.path()).unwrap().count(), 0); } + #[tokio::test] + async fn test_parquet_writer_geospatial_logical_types() -> Result<()> { + let temp_dir = TempDir::new().unwrap(); + let file_io = FileIO::new_with_fs(); + let location_gen = DefaultLocationGenerator::with_data_location( + temp_dir.path().to_str().unwrap().to_string(), + ); + let file_name_gen = + DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet); + + let schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required( + 0, + "geom", + Type::Primitive(PrimitiveType::Geometry(GeometryType::default())), + ) + .into(), + NestedField::optional( + 1, + "geog", + Type::Primitive(PrimitiveType::Geography( + GeographyType::new(None, WkbEdges::Karney).unwrap(), + )), + ) + .into(), + ]) + .build() + .unwrap(), + ); + let arrow_schema: ArrowSchemaRef = Arc::new(schema_to_arrow_schema(&schema).unwrap()); + let geom_wkb = wkb_point_xy(1.0, 2.0); + let geog_wkb = wkb_point_xy(3.0, 4.0); + let geom = Arc::new(arrow_array::LargeBinaryArray::from_vec(vec![ + geom_wkb.as_slice(), + ])) as ArrayRef; + let geog = Arc::new(arrow_array::LargeBinaryArray::from_vec(vec![ + geog_wkb.as_slice(), + ])) as ArrayRef; + let to_write = RecordBatch::try_new(arrow_schema.clone(), vec![geom, geog]).unwrap(); + + let output_file = file_io.new_output( + location_gen.generate_location(None, &file_name_gen.generate_file_name()), + )?; + let mut pw = ParquetWriterBuilder::new(WriterProperties::builder().build(), schema) + .build(output_file) + .await?; + + pw.write(&to_write).await?; + let res = pw.close().await?; + assert_eq!(res.len(), 1); + let data_file = res + .into_iter() + .next() + .unwrap() + .content(crate::spec::DataContentType::Data) + .partition(Struct::empty()) + .partition_spec_id(0) + .build() + .unwrap(); + + assert_eq!(data_file.record_count(), 1); + assert_eq!(*data_file.value_counts(), HashMap::from([(0, 1), (1, 1)])); + assert_eq!( + *data_file.null_value_counts(), + HashMap::from([(0, 0), (1, 0)]) + ); + assert!(data_file.lower_bounds().is_empty()); + assert!(data_file.upper_bounds().is_empty()); + + let input_file = file_io.new_input(data_file.file_path())?; + let file_metadata = input_file.metadata().await?; + let reader = input_file.reader().await?; + let mut parquet_reader = ArrowFileReader::new(file_metadata, reader); + let parquet_metadata = parquet_reader.get_metadata(None).await?; + let schema_descr = parquet_metadata.file_metadata().schema_descr(); + + assert_eq!( + schema_descr.column(0).logical_type_ref(), + Some(&LogicalType::Geometry { crs: None }) + ); + assert_eq!( + schema_descr.column(1).logical_type_ref(), + Some(&LogicalType::Geography { + crs: None, + algorithm: Some(EdgeInterpolationAlgorithm::KARNEY), + }) + ); + + Ok(()) + } + #[test] fn test_min_max_aggregator() { let schema = Arc::new( @@ -2417,4 +2519,37 @@ mod tests { assert_eq!(cdc.max_chunk_size, 8192); assert_eq!(cdc.norm_level, 2); } + + #[test] + fn test_min_max_aggregator_skips_geospatial_byte_statistics() { + let schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required( + 0, + "geom", + Type::Primitive(PrimitiveType::Geometry(Default::default())), + ) + .with_id(0) + .into(), + ]) + .build() + .expect("Failed to create schema"), + ); + let mut min_max_agg = MinMaxColAggregator::new(schema); + let stats = Statistics::ByteArray(ValueStatistics::new( + Some(ByteArray::from(vec![1, 2, 3])), + Some(ByteArray::from(vec![4, 5, 6])), + None, + None, + false, + )); + + min_max_agg.update(0, stats).unwrap(); + let (lower_bounds, upper_bounds) = min_max_agg.produce(); + + assert!(lower_bounds.is_empty()); + assert!(upper_bounds.is_empty()); + } } From 2b60f9254b65adc526be4a956369c91e14acee63 Mon Sep 17 00:00:00 2001 From: xuanyili Date: Mon, 22 Jun 2026 02:15:31 +0000 Subject: [PATCH 2/2] fix: decouple geography edge algorithm API --- Cargo.lock | 4 +- crates/iceberg/public-api.txt | 29 ++++- crates/iceberg/src/arrow/schema.rs | 75 +++++++++---- crates/iceberg/src/spec/datatypes.rs | 102 +++++++++++------- crates/iceberg/src/spec/values/literal.rs | 2 +- crates/iceberg/src/spec/values/tests.rs | 4 +- crates/iceberg/src/transform/bucket.rs | 2 +- crates/iceberg/src/transform/truncate.rs | 2 +- .../src/writer/file_writer/parquet_writer.rs | 11 +- 9 files changed, 159 insertions(+), 72 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c4ab2e01f1..8253680c61 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5494,9 +5494,9 @@ dependencies = [ [[package]] name = "parquet-geospatial" -version = "58.1.0" +version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32928da166f0e52c9c15d055181fb8c163e3ceb2fce12c79fdba5cfd284d98a2" +checksum = "fce21fd6b0ed0f7a8db95bbce7c5327cdd526bcb15e2757a7b5415d9a6a2dfd9" dependencies = [ "arrow-schema", "geo-traits", diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 7116525bf5..08899bf145 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1408,6 +1408,29 @@ impl serde_core::ser::Serialize for iceberg::spec::DataFileFormat where Self: co pub fn iceberg::spec::DataFileFormat::serialize<__S>(&self, serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::DataFileFormat where Self: core::str::traits::FromStr, ::Err: core::fmt::Display pub fn iceberg::spec::DataFileFormat::deserialize<__D>(deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub enum iceberg::spec::EdgeInterpolationAlgorithm +pub iceberg::spec::EdgeInterpolationAlgorithm::Andoyer +pub iceberg::spec::EdgeInterpolationAlgorithm::Karney +pub iceberg::spec::EdgeInterpolationAlgorithm::Spherical +pub iceberg::spec::EdgeInterpolationAlgorithm::Thomas +pub iceberg::spec::EdgeInterpolationAlgorithm::Vincenty +impl core::clone::Clone for iceberg::spec::EdgeInterpolationAlgorithm +pub fn iceberg::spec::EdgeInterpolationAlgorithm::clone(&self) -> iceberg::spec::EdgeInterpolationAlgorithm +impl core::cmp::Eq for iceberg::spec::EdgeInterpolationAlgorithm +impl core::cmp::PartialEq for iceberg::spec::EdgeInterpolationAlgorithm +pub fn iceberg::spec::EdgeInterpolationAlgorithm::eq(&self, other: &iceberg::spec::EdgeInterpolationAlgorithm) -> bool +impl core::default::Default for iceberg::spec::EdgeInterpolationAlgorithm +pub fn iceberg::spec::EdgeInterpolationAlgorithm::default() -> iceberg::spec::EdgeInterpolationAlgorithm +impl core::fmt::Debug for iceberg::spec::EdgeInterpolationAlgorithm +pub fn iceberg::spec::EdgeInterpolationAlgorithm::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::hash::Hash for iceberg::spec::EdgeInterpolationAlgorithm +pub fn iceberg::spec::EdgeInterpolationAlgorithm::hash<__H: core::hash::Hasher>(&self, state: &mut __H) +impl core::marker::Copy for iceberg::spec::EdgeInterpolationAlgorithm +impl core::marker::StructuralPartialEq for iceberg::spec::EdgeInterpolationAlgorithm +impl serde_core::ser::Serialize for iceberg::spec::EdgeInterpolationAlgorithm +pub fn iceberg::spec::EdgeInterpolationAlgorithm::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::EdgeInterpolationAlgorithm +pub fn iceberg::spec::EdgeInterpolationAlgorithm::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> #[repr(u8)] pub enum iceberg::spec::FormatVersion pub iceberg::spec::FormatVersion::V1 = 1 pub iceberg::spec::FormatVersion::V2 = 2 @@ -1977,9 +2000,9 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg::spec::FieldSummary pub fn iceberg::spec::FieldSummary::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::spec::GeographyType impl iceberg::spec::GeographyType -pub fn iceberg::spec::GeographyType::algorithm(&self) -> parquet_geospatial::types::Edges +pub fn iceberg::spec::GeographyType::algorithm(&self) -> iceberg::spec::EdgeInterpolationAlgorithm pub fn iceberg::spec::GeographyType::crs(&self) -> core::option::Option<&str> -pub fn iceberg::spec::GeographyType::new(crs: core::option::Option, algorithm: parquet_geospatial::types::Edges) -> iceberg::Result +pub fn iceberg::spec::GeographyType::new(crs: core::option::Option, algorithm: iceberg::spec::EdgeInterpolationAlgorithm) -> iceberg::Result impl core::clone::Clone for iceberg::spec::GeographyType pub fn iceberg::spec::GeographyType::clone(&self) -> iceberg::spec::GeographyType impl core::cmp::Eq for iceberg::spec::GeographyType @@ -1990,7 +2013,7 @@ pub fn iceberg::spec::GeographyType::default() -> Self impl core::fmt::Debug for iceberg::spec::GeographyType pub fn iceberg::spec::GeographyType::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::hash::Hash for iceberg::spec::GeographyType -pub fn iceberg::spec::GeographyType::hash(&self, state: &mut H) +pub fn iceberg::spec::GeographyType::hash<__H: core::hash::Hasher>(&self, state: &mut __H) impl core::marker::StructuralPartialEq for iceberg::spec::GeographyType impl serde_core::ser::Serialize for iceberg::spec::GeographyType pub fn iceberg::spec::GeographyType::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer diff --git a/crates/iceberg/src/arrow/schema.rs b/crates/iceberg/src/arrow/schema.rs index b4dfc23708..f45ecbf06c 100644 --- a/crates/iceberg/src/arrow/schema.rs +++ b/crates/iceberg/src/arrow/schema.rs @@ -32,15 +32,15 @@ use arrow_schema::{ }; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; use parquet::file::statistics::Statistics; -use parquet_geospatial::{WkbMetadata, WkbType, WkbTypeHint}; +use parquet_geospatial::{WkbEdges, WkbMetadata, WkbType, WkbTypeHint}; use uuid::Uuid; use crate::error::Result; use crate::spec::decimal_utils::i128_from_be_bytes; use crate::spec::{ - Datum, FIRST_FIELD_ID, GeographyType, GeometryType, ListType, MapType, NestedField, - NestedFieldRef, PrimitiveLiteral, PrimitiveType, Schema, SchemaVisitor, StructType, Type, - VariantType, + Datum, EdgeInterpolationAlgorithm, FIRST_FIELD_ID, GeographyType, GeometryType, ListType, + MapType, NestedField, NestedFieldRef, PrimitiveLiteral, PrimitiveType, Schema, SchemaVisitor, + StructType, Type, VariantType, }; use crate::{Error, ErrorKind}; @@ -102,6 +102,26 @@ impl ExtensionType for VariantExtensionType { } } +fn edge_interpolation_algorithm_to_wkb_edges(algorithm: EdgeInterpolationAlgorithm) -> WkbEdges { + match algorithm { + EdgeInterpolationAlgorithm::Spherical => WkbEdges::Spherical, + EdgeInterpolationAlgorithm::Vincenty => WkbEdges::Vincenty, + EdgeInterpolationAlgorithm::Thomas => WkbEdges::Thomas, + EdgeInterpolationAlgorithm::Andoyer => WkbEdges::Andoyer, + EdgeInterpolationAlgorithm::Karney => WkbEdges::Karney, + } +} + +fn wkb_edges_to_edge_interpolation_algorithm(edges: WkbEdges) -> EdgeInterpolationAlgorithm { + match edges { + WkbEdges::Spherical => EdgeInterpolationAlgorithm::Spherical, + WkbEdges::Vincenty => EdgeInterpolationAlgorithm::Vincenty, + WkbEdges::Thomas => EdgeInterpolationAlgorithm::Thomas, + WkbEdges::Andoyer => EdgeInterpolationAlgorithm::Andoyer, + WkbEdges::Karney => EdgeInterpolationAlgorithm::Karney, + } +} + /// A post order arrow schema visitor. /// /// For order of methods called, please refer to [`visit_schema`]. @@ -447,14 +467,19 @@ impl ArrowSchemaConverter { })?, ))), WkbTypeHint::Geography => Ok(Type::Primitive(PrimitiveType::Geography( - GeographyType::new(crs, wkb_type.metadata().algorithm.unwrap_or_default()) - .map_err(|err| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid geography CRS for field {}", field.name()), - ) - .with_source(err) - })?, + GeographyType::new( + crs, + wkb_edges_to_edge_interpolation_algorithm( + wkb_type.metadata().algorithm.unwrap_or_default(), + ), + ) + .map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid geography CRS for field {}", field.name()), + ) + .with_source(err) + })?, ))), } } @@ -491,6 +516,7 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter { } }; + let value = self.apply_field_extension_type(element_field, &value)?; let id = self.get_field_id(element_field)?; let doc = get_field_doc(element_field); let mut element_field = @@ -515,6 +541,8 @@ impl ArrowSchemaVisitor for ArrowSchemaConverter { let key_field = &fields[0]; let value_field = &fields[1]; + let key_value = self.apply_field_extension_type(key_field, &key_value)?; + let value = self.apply_field_extension_type(value_field, &value)?; let key_id = self.get_field_id(key_field)?; let key_doc = get_field_doc(key_field); @@ -689,19 +717,24 @@ impl SchemaVisitor for ToArrowSchemaConverter { .try_with_extension_type(WkbType::new(Some(metadata))) .map_err(|err| { Error::new( - crate::ErrorKind::DataInvalid, + ErrorKind::DataInvalid, "Failed to attach geometry Arrow extension metadata", ) .with_source(err) })?; } Type::Primitive(PrimitiveType::Geography(geography)) => { - let metadata = WkbMetadata::new(geography.crs(), Some(geography.algorithm())); + let metadata = WkbMetadata::new( + geography.crs(), + Some(edge_interpolation_algorithm_to_wkb_edges( + geography.algorithm(), + )), + ); arrow_field .try_with_extension_type(WkbType::new(Some(metadata))) .map_err(|err| { Error::new( - crate::ErrorKind::DataInvalid, + ErrorKind::DataInvalid, "Failed to attach geography Arrow extension metadata", ) .with_source(err) @@ -1293,6 +1326,7 @@ pub fn datum_to_arrow_type_with_ree(datum: &Datum) -> DataType { PrimitiveType::Uuid => make_ree(DataType::Binary), PrimitiveType::Fixed(_) => make_ree(DataType::Binary), PrimitiveType::Binary => make_ree(DataType::Binary), + PrimitiveType::Geometry(_) | PrimitiveType::Geography(_) => make_ree(DataType::LargeBinary), PrimitiveType::Decimal { precision, scale } => { make_ree(DataType::Decimal128(*precision as u8, *scale as i8)) } @@ -1474,7 +1508,9 @@ mod tests { use super::*; use crate::spec::decimal_utils::decimal_new; - use crate::spec::{Literal, Schema}; + use crate::spec::{ + EdgeInterpolationAlgorithm as IcebergEdgeInterpolationAlgorithm, Literal, Schema, + }; /// Create a simple field with metadata. fn simple_field(name: &str, ty: DataType, nullable: bool, value: &str) -> Field { @@ -2305,8 +2341,11 @@ mod tests { 2, "geog", Type::Primitive(PrimitiveType::Geography( - GeographyType::new(Some("srid:3857".to_string()), WkbEdges::Karney) - .unwrap(), + GeographyType::new( + Some("srid:3857".to_string()), + IcebergEdgeInterpolationAlgorithm::Karney, + ) + .unwrap(), )), ) .into(), diff --git a/crates/iceberg/src/spec/datatypes.rs b/crates/iceberg/src/spec/datatypes.rs index 43795824fc..acd6e96f83 100644 --- a/crates/iceberg/src/spec/datatypes.rs +++ b/crates/iceberg/src/spec/datatypes.rs @@ -21,12 +21,10 @@ use std::collections::HashMap; use std::convert::identity; use std::fmt; -use std::hash::{Hash, Hasher}; use std::ops::Index; use std::sync::{Arc, OnceLock}; use ::serde::de::{MapAccess, Visitor}; -use parquet_geospatial::WkbEdges; use serde::de::{Error, IntoDeserializer}; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use serde_json::Value as JsonValue; @@ -254,32 +252,42 @@ impl GeometryType { } } +/// Iceberg geography edge interpolation algorithm. +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Copy, Hash, Default)] +#[serde(rename_all = "lowercase")] +pub enum EdgeInterpolationAlgorithm { + /// Spherical edge interpolation. + #[default] + Spherical, + /// Vincenty edge interpolation. + Vincenty, + /// Thomas edge interpolation. + Thomas, + /// Andoyer edge interpolation. + Andoyer, + /// Karney edge interpolation. + Karney, +} + /// Iceberg geography type. -#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)] +#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash)] pub struct GeographyType { crs: Option, - algorithm: WkbEdges, + algorithm: EdgeInterpolationAlgorithm, } impl Default for GeographyType { fn default() -> Self { Self { crs: None, - algorithm: WkbEdges::Spherical, + algorithm: EdgeInterpolationAlgorithm::Spherical, } } } -impl Hash for GeographyType { - fn hash(&self, state: &mut H) { - self.crs.hash(state); - wkb_edges_as_str(self.algorithm).hash(state); - } -} - impl GeographyType { /// Creates a geography type with an optional coordinate reference system and edge interpolation algorithm. - pub fn new(crs: Option, algorithm: WkbEdges) -> Result { + pub fn new(crs: Option, algorithm: EdgeInterpolationAlgorithm) -> Result { Ok(Self { crs: normalize_crs(crs)?, algorithm, @@ -292,7 +300,7 @@ impl GeographyType { } /// Returns the edge interpolation algorithm. - pub fn algorithm(&self) -> WkbEdges { + pub fn algorithm(&self) -> EdgeInterpolationAlgorithm { self.algorithm } } @@ -311,23 +319,25 @@ fn normalize_crs(crs: Option) -> Result> { Ok((crs != DEFAULT_GEOSPATIAL_CRS).then_some(crs)) } -fn wkb_edges_as_str(edges: WkbEdges) -> &'static str { - match edges { - WkbEdges::Spherical => "spherical", - WkbEdges::Vincenty => "vincenty", - WkbEdges::Thomas => "thomas", - WkbEdges::Andoyer => "andoyer", - WkbEdges::Karney => "karney", +fn edge_interpolation_algorithm_as_str(algorithm: EdgeInterpolationAlgorithm) -> &'static str { + match algorithm { + EdgeInterpolationAlgorithm::Spherical => "spherical", + EdgeInterpolationAlgorithm::Vincenty => "vincenty", + EdgeInterpolationAlgorithm::Thomas => "thomas", + EdgeInterpolationAlgorithm::Andoyer => "andoyer", + EdgeInterpolationAlgorithm::Karney => "karney", } } -fn parse_wkb_edges(value: &str) -> std::result::Result { +fn parse_edge_interpolation_algorithm( + value: &str, +) -> std::result::Result { match value.trim().to_ascii_lowercase().as_str() { - "spherical" => Ok(WkbEdges::Spherical), - "vincenty" => Ok(WkbEdges::Vincenty), - "thomas" => Ok(WkbEdges::Thomas), - "andoyer" => Ok(WkbEdges::Andoyer), - "karney" => Ok(WkbEdges::Karney), + "spherical" => Ok(EdgeInterpolationAlgorithm::Spherical), + "vincenty" => Ok(EdgeInterpolationAlgorithm::Vincenty), + "thomas" => Ok(EdgeInterpolationAlgorithm::Thomas), + "andoyer" => Ok(EdgeInterpolationAlgorithm::Andoyer), + "karney" => Ok(EdgeInterpolationAlgorithm::Karney), _ => Err(format!( "Unknown geography edge interpolation algorithm: {value}" )), @@ -371,14 +381,16 @@ fn parse_geography(value: &str) -> std::result::Result { let params = parse_geospatial_params(value.trim(), "geography")?; let geography = match params.as_slice() { [] => GeographyType::default(), - [crs] if !crs.is_empty() => { - GeographyType::new(Some((*crs).to_string()), WkbEdges::Spherical) - .map_err(|err| err.to_string())? - } - [crs, algorithm] if !crs.is_empty() && !algorithm.is_empty() => { - GeographyType::new(Some((*crs).to_string()), parse_wkb_edges(algorithm)?) - .map_err(|err| err.to_string())? - } + [crs] if !crs.is_empty() => GeographyType::new( + Some((*crs).to_string()), + EdgeInterpolationAlgorithm::Spherical, + ) + .map_err(|err| err.to_string())?, + [crs, algorithm] if !crs.is_empty() && !algorithm.is_empty() => GeographyType::new( + Some((*crs).to_string()), + parse_edge_interpolation_algorithm(algorithm)?, + ) + .map_err(|err| err.to_string())?, _ => return Err(format!("Invalid geography type: {value}")), }; @@ -583,13 +595,15 @@ impl fmt::Display for PrimitiveType { PrimitiveType::Geography(geography) => { let algorithm = geography.algorithm(); match (geography.crs(), algorithm) { - (None, WkbEdges::Spherical) => write!(f, "geography"), - (Some(crs), WkbEdges::Spherical) => write!(f, "geography({crs})"), + (None, EdgeInterpolationAlgorithm::Spherical) => write!(f, "geography"), + (Some(crs), EdgeInterpolationAlgorithm::Spherical) => { + write!(f, "geography({crs})") + } (crs, algorithm) => write!( f, "geography({}, {})", crs.unwrap_or(DEFAULT_GEOSPATIAL_CRS), - wkb_edges_as_str(algorithm) + edge_interpolation_algorithm_as_str(algorithm) ), } } @@ -1243,14 +1257,22 @@ mod tests { ( r#""geography(srid:4326)""#, PrimitiveType::Geography( - GeographyType::new(Some("srid:4326".to_string()), WkbEdges::Spherical).unwrap(), + GeographyType::new( + Some("srid:4326".to_string()), + EdgeInterpolationAlgorithm::Spherical, + ) + .unwrap(), ), "geography(srid:4326)", ), ( r#""geography(srid:4326,karney)""#, PrimitiveType::Geography( - GeographyType::new(Some("srid:4326".to_string()), WkbEdges::Karney).unwrap(), + GeographyType::new( + Some("srid:4326".to_string()), + EdgeInterpolationAlgorithm::Karney, + ) + .unwrap(), ), "geography(srid:4326, karney)", ), diff --git a/crates/iceberg/src/spec/values/literal.rs b/crates/iceberg/src/spec/values/literal.rs index 24ae89b519..be98f03f8d 100644 --- a/crates/iceberg/src/spec/values/literal.rs +++ b/crates/iceberg/src/spec/values/literal.rs @@ -538,7 +538,7 @@ impl Literal { PrimitiveType::Geometry(_) | PrimitiveType::Geography(_), JsonValue::String(_), ) => Err(Error::new( - crate::ErrorKind::DataInvalid, + ErrorKind::DataInvalid, "Geometry and geography defaults must be null", )), ( diff --git a/crates/iceberg/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index e10f023d3e..f860f76251 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/crates/iceberg/src/spec/values/tests.rs @@ -311,8 +311,8 @@ fn json_binary() { #[test] fn json_geospatial_defaults_reject_non_null() { let cases = vec![ - Type::Primitive(PrimitiveType::Geometry(Default::default())), - Type::Primitive(PrimitiveType::Geography(Default::default())), + Primitive(PrimitiveType::Geometry(Default::default())), + Primitive(PrimitiveType::Geography(Default::default())), ]; for ty in cases { diff --git a/crates/iceberg/src/transform/bucket.rs b/crates/iceberg/src/transform/bucket.rs index a35d077ce9..bcf3ea2040 100644 --- a/crates/iceberg/src/transform/bucket.rs +++ b/crates/iceberg/src/transform/bucket.rs @@ -296,7 +296,7 @@ mod test { use crate::spec::Type::{Primitive, Struct}; use crate::spec::decimal_utils::decimal_new; use crate::spec::{ - Datum, GeographyType, GeometryType, NestedField, PrimitiveType, StructType, Transform, Type, + Datum, GeographyType, GeometryType, NestedField, PrimitiveType, StructType, Transform, }; use crate::transform::TransformFunction; use crate::transform::test::{TestProjectionFixture, TestTransformFixture}; diff --git a/crates/iceberg/src/transform/truncate.rs b/crates/iceberg/src/transform/truncate.rs index b83126cb35..52b9480e16 100644 --- a/crates/iceberg/src/transform/truncate.rs +++ b/crates/iceberg/src/transform/truncate.rs @@ -201,7 +201,7 @@ mod test { use crate::spec::Type::{Primitive, Struct}; use crate::spec::decimal_utils::decimal_new; use crate::spec::{ - Datum, GeographyType, GeometryType, NestedField, PrimitiveType, StructType, Transform, Type, + Datum, GeographyType, GeometryType, NestedField, PrimitiveType, StructType, Transform, }; use crate::transform::TransformFunction; use crate::transform::test::{TestProjectionFixture, TestTransformFixture}; diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index 2e08d7e8f8..cdc935158a 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -670,7 +670,6 @@ mod tests { use parquet::basic::{EdgeInterpolationAlgorithm, LogicalType}; use parquet::data_type::ByteArray; use parquet::file::statistics::ValueStatistics; - use parquet_geospatial::WkbEdges; use parquet_geospatial::testing::wkb_point_xy; use tempfile::TempDir; use uuid::Uuid; @@ -679,7 +678,10 @@ mod tests { use crate::arrow::schema_to_arrow_schema; use crate::io::FileIO; use crate::spec::decimal_utils::{decimal_mantissa, decimal_new, decimal_scale}; - use crate::spec::{PrimitiveLiteral, Struct, *}; + use crate::spec::{ + EdgeInterpolationAlgorithm as IcebergEdgeInterpolationAlgorithm, PrimitiveLiteral, Struct, + *, + }; use crate::writer::file_writer::location_generator::{ DefaultFileNameGenerator, DefaultLocationGenerator, FileNameGenerator, LocationGenerator, }; @@ -2338,7 +2340,8 @@ mod tests { 1, "geog", Type::Primitive(PrimitiveType::Geography( - GeographyType::new(None, WkbEdges::Karney).unwrap(), + GeographyType::new(None, IcebergEdgeInterpolationAlgorithm::Karney) + .unwrap(), )), ) .into(), @@ -2371,7 +2374,7 @@ mod tests { .into_iter() .next() .unwrap() - .content(crate::spec::DataContentType::Data) + .content(DataContentType::Data) .partition(Struct::empty()) .partition_spec_id(0) .build()