paleolimbot commented on code in PR #2933:
URL: https://github.com/apache/iceberg-rust/pull/2933#discussion_r3697336926
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -2211,6 +2324,103 @@ 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(),
+ )),
Review Comment:
We should use an `AUTHORITY:CODE` value, here (e.g., `EPSG:3857`)...this was
recently clarified in the Iceberg spec. We shouldn't use `EPSG:4326`, because
that is the default CRS (typically represented by `None` here).
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -396,22 +418,71 @@ 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 {
id,
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,
};
results.push(Arc::new(nested_field));
}
Ok(results)
}
+
+ fn apply_field_extension_type(&self, field: &FieldRef, field_type: &Type)
-> Result<Type> {
+ if field.extension_type_name() != Some(WkbType::NAME) {
+ return Ok(field_type.clone());
+ }
+
+ let wkb_type = field.try_extension_type::<WkbType>().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(),
+ });
Review Comment:
Iceberg has a slightly different CRS restriction than Parquet, where it
greatly prefers Authority:Code over PROJJSON (because schemas are written so
frequently to iceberg files, and PROJJSON is sufficiently verbose that it
causes performance/space issues.
Because of this (and the buggy `WkbType` until arrow-rs 59.1.0), you may
want to implement `WkbType` yourself specifically for Iceberg (e.g., with a
utility to extract the iceberg-preferred CRS, which can be derived from
PROJJSON using the `id` member).
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -396,22 +418,71 @@ 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 {
id,
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,
};
results.push(Arc::new(nested_field));
}
Ok(results)
}
+
+ fn apply_field_extension_type(&self, field: &FieldRef, field_type: &Type)
-> Result<Type> {
+ if field.extension_type_name() != Some(WkbType::NAME) {
+ return Ok(field_type.clone());
+ }
+
+ let wkb_type = field.try_extension_type::<WkbType>().map_err(|err| {
Review Comment:
Unfortunately, until arrow-rs 59.1.0 there are some corner cases that slip
through here (fixed in https://github.com/apache/arrow-rs/pull/10065 ). This is
still OK, it just won't work for geography and does a few funny things with
corner case CRSes until the arrow-rs dependency updates.
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -2211,6 +2324,103 @@ 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()),
Review Comment:
This should also be an authority:code combination, and one that is valid for
Geography (EPSG:3857 is not). OGC:CRS27 would work here (although this is
typically `None`, for the default of EPSG:4326).
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -231,6 +232,171 @@ impl From<MapType> for Type {
}
}
+/// Iceberg geometry type.
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash, Default)]
+pub struct GeometryType {
+ crs: Option<String>,
+}
+
+impl GeometryType {
+ /// Creates a geometry type with an optional coordinate reference system.
+ pub fn new(crs: Option<String>) -> Result<Self> {
+ 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 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, Hash)]
+pub struct GeographyType {
+ crs: Option<String>,
+ algorithm: EdgeInterpolationAlgorithm,
+}
+
+impl Default for GeographyType {
+ fn default() -> Self {
+ Self {
+ crs: None,
+ algorithm: EdgeInterpolationAlgorithm::Spherical,
+ }
+ }
+}
+
+impl GeographyType {
+ /// Creates a geography type with an optional coordinate reference system
and edge interpolation algorithm.
+ pub fn new(crs: Option<String>, algorithm: EdgeInterpolationAlgorithm) ->
Result<Self> {
+ 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) -> EdgeInterpolationAlgorithm {
+ self.algorithm
+ }
+}
+
+fn normalize_crs(crs: Option<String>) -> Result<Option<String>> {
+ 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))
Review Comment:
This should also check if the CRS is `EPSG:4326` in addition to `OGC:CRS84`.
Because of the Iceberg spec's specification that longitude is always first and
latitude is always second, the two are equivalent and are best stored in an
iceberg table as the default value.
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -2211,6 +2324,103 @@ 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()),
+ IcebergEdgeInterpolationAlgorithm::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::<WkbType>().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::<WkbType>().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::<WkbType>()
+ .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());
+ }
+
Review Comment:
It would be good to check the roundtripping of the `Geometry` and
`Geography` with the defaults (`None`) for the parameters. This may run into
one of the bugs I fixed in the arrow-rs WkbType but it's a very common case and
should work here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
