dannycjones commented on code in PR #2933:
URL: https://github.com/apache/iceberg-rust/pull/2933#discussion_r3934269127
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -2506,6 +2516,96 @@ 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,
IcebergEdgeInterpolationAlgorithm::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(DataContentType::Data)
+ .partition(Struct::empty())
+ .partition_spec_id(0)
+ .build()
+ .unwrap();
+
+ assert_eq!(data_file.record_count(), 1);
+ assert!(data_file.lower_bounds().is_empty());
+ assert!(data_file.upper_bounds().is_empty());
Review Comment:
Can you include an assertion message about why the bounds are empty? Is it
simply because the field is optional, and this is left as a follow-up?
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -100,6 +102,57 @@ 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,
+ }
+}
Review Comment:
Why not `impl From<EdgeInterpolationAlgorithm> for WkbEdges`? That way we
get idiomatic/easy translation from the iceberg-rust type into the Parquet type
(and the same for any future file format equivalent).
Same in reverse: `impl From<WkbEdges> for EdgeInterpolationAlgorithm`.
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -2506,6 +2516,96 @@ 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,
IcebergEdgeInterpolationAlgorithm::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(DataContentType::Data)
+ .partition(Struct::empty())
+ .partition_spec_id(0)
+ .build()
+ .unwrap();
+
+ assert_eq!(data_file.record_count(), 1);
+ 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(Some("srid:0".to_string())))
+ );
+ assert_eq!(
+ schema_descr.column(1).logical_type_ref(),
+ Some(&LogicalType::geography(
+ Some("srid:0".to_string()),
+ Some(EdgeInterpolationAlgorithm::KARNEY),
+ ))
+ );
Review Comment:
Why are we expecting `srid:0` here? Isn't the default to be `None`, implying
`OGC:CRS84`? This looks like a bug.
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -230,6 +232,173 @@ 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() || crs.contains([',', ')']) {
+ return Err(crate::Error::new(
+ crate::ErrorKind::DataInvalid,
+ "Geospatial CRS must be non-empty and must not contain ',' or ')'",
+ ));
+ }
+ Ok((crs != DEFAULT_GEOSPATIAL_CRS && crs !=
EQUIVALENT_DEFAULT_GEOSPATIAL_CRS).then_some(crs))
+}
+
+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_edge_interpolation_algorithm(
+ value: &str,
+) -> std::result::Result<EdgeInterpolationAlgorithm, String> {
+ match value.trim().to_ascii_lowercase().as_str() {
+ "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}"
+ )),
+ }
+}
Review Comment:
Implement this on `EdgeInterpolationAlgorithm` itself.
```suggestion
impl EdgeInterpolationAlgorithm {
fn parse(
value: &str,
) -> std::result::Result<Self, iceberg::Error> {
let parsed = match value.trim().to_ascii_lowercase().as_str() {
"spherical" => Self::Spherical,
"vincenty" => Self::Vincenty,
"thomas" => Self::Thomas,
"andoyer" => Self::Andoyer,
"karney" => Self::Karney,
_ => return Error::new(
ErrorKind::DataInvalid,
format!("Unknown geography edge interpolation algorithm:
{value}"),
)),
}
Ok(parsed)
}
}
```
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -324,6 +499,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)
Review Comment:
This diverges from fixed binary and decimal above - is there a good reason,
or should we instead implement the equivalent for geometry and geography?
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -43,6 +43,8 @@ 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";
+const EQUIVALENT_DEFAULT_GEOSPATIAL_CRS: &str = "EPSG:4326";
Review Comment:
What's the reason for normalizing this back to `OGC:CRS84`? Is there
anything we can point to, such as the Iceberg or Parquet spec, to explain this?
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -1042,6 +1243,56 @@ mod tests {
)
}
+ #[test]
+ fn primitive_type_geospatial() {
+ let cases = vec![
+ (
+ r#""geometry""#,
+ PrimitiveType::Geometry(GeometryType::default()),
+ "geometry",
+ ),
+ (
+ r#""geometry ( EPSG:3857 )""#,
+
PrimitiveType::Geometry(GeometryType::new(Some("EPSG:3857".to_string())).unwrap()),
+ "geometry(EPSG:3857)",
+ ),
+ (
+ r#""geography""#,
+ PrimitiveType::Geography(GeographyType::default()),
+ "geography",
+ ),
+ (
+ r#""geography ( OGC:CRS27 , karney )""#,
+ PrimitiveType::Geography(
+ GeographyType::new(
+ Some("OGC:CRS27".to_string()),
+ EdgeInterpolationAlgorithm::Karney,
+ )
+ .unwrap(),
+ ),
+ "geography(OGC:CRS27, 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}""#)
+ );
Review Comment:
nit: please can you add some assertion messages so it's clearer?
```suggestion
assert_eq!(actual, expected, "parsed primitive type did not
match expectation");
assert_eq!(actual.to_string(), display, "display impl did not
match");
assert_eq!(
serde_json::to_string(&actual).unwrap(),
format!(r#""{display}""#),
"JSON serialization did not match expectation",
);
```
##########
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(
+ ErrorKind::DataInvalid,
+ "Geometry and geography defaults must be null",
+ )),
Review Comment:
This seems wrong. While we may not allow defaults, there is a valid
deserialization from JSON in the spec.
We need to add the check elsewhere in the code to disallow defaults.
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -396,22 +449,55 @@ 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 =
iceberg_crs_from_wkb_metadata(wkb_type.metadata().crs.as_ref())?;
+
+ match wkb_type.metadata().type_hint() {
+ WkbTypeHint::Geometry =>
Ok(Type::Primitive(PrimitiveType::Geometry(
+ GeometryType::new(crs)?,
+ ))),
+ WkbTypeHint::Geography =>
Ok(Type::Primitive(PrimitiveType::Geography(
+ GeographyType::new(
+ crs,
+ wkb_edges_to_edge_interpolation_algorithm(
+ wkb_type.metadata().algorithm.unwrap_or_default(),
+ ),
Review Comment:
Given this is the Parquet default, I'd rather we explicitly default for
Iceberg.
```suggestion
wkb_edges_to_edge_interpolation_algorithm(
wkb_type.metadata().algorithm.unwrap_or(WkbEdges::Spherical),
),
```
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -100,6 +102,73 @@ 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,
+ }
+}
+
+fn iceberg_crs_from_wkb_metadata(crs: Option<&serde_json::Value>) ->
Result<Option<String>> {
+ let Some(crs) = crs else {
+ return Ok(None);
+ };
+
+ match crs {
+ serde_json::Value::String(crs) => Ok(Some(crs.clone())),
Review Comment:
Did we remove this limit again? I don't see it in the PR right now.
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -230,6 +232,173 @@ 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,
+}
Review Comment:
Java equivalent with case insensitivity:
https://github.com/apache/iceberg/blob/27bfd00d67b2d00e3d4ea3e96832248bdc5107f6/api/src/main/java/org/apache/iceberg/types/Types.java#L608-L628
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -230,6 +232,173 @@ 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() || crs.contains([',', ')']) {
+ return Err(crate::Error::new(
+ crate::ErrorKind::DataInvalid,
+ "Geospatial CRS must be non-empty and must not contain ',' or ')'",
+ ));
+ }
+ Ok((crs != DEFAULT_GEOSPATIAL_CRS && crs !=
EQUIVALENT_DEFAULT_GEOSPATIAL_CRS).then_some(crs))
+}
+
+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_edge_interpolation_algorithm(
+ value: &str,
+) -> std::result::Result<EdgeInterpolationAlgorithm, String> {
+ match value.trim().to_ascii_lowercase().as_str() {
+ "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}"
+ )),
+ }
+}
Review Comment:
There's a few bits like this in this file, where the functions would be
better suited as methods implemented on the type itself.
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -100,6 +102,57 @@ 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,
+ }
+}
+
+fn iceberg_crs_from_wkb_metadata(crs: Option<&serde_json::Value>) ->
Result<Option<String>> {
+ match crs {
+ None => Ok(None),
+ Some(serde_json::Value::String(crs)) => Ok(Some(crs.clone())),
Review Comment:
So all but `PROJJSON` are just pass through and the Parquet reader will
handle it?
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -230,6 +232,173 @@ 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,
+}
Review Comment:
Deriving equals here will result in case sensitive comparison of the CRS. I
imagine this is not desired, so we should implement equality manually.
Similar issue with hashing (although unclear to me yet how we depend on it).
I think this is an opportunity to use a stronger type for CRS. This would
allow us to implement `Eq` and `Hash` for that type, and we can leave the
derive here. Alternatively, we can ensure that we always transform it into a
consistent representation by tranforming it all lowercase or similar.
Something roughly like this... and we can evolve it later.
```rust
/// Geospatial Coordinate Reference System (CRS).
struct Crs {
raw_string: String,
}
// Case insensitive
impl Eq for Crs { ... }
// Case insensitive
impl Hash for Crs { ... }
/// Iceberg geography type.
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash)]
pub struct GeographyType {
crs: Option<Crs>,
algorithm: EdgeInterpolationAlgorithm,
}
/// Iceberg geometry type.
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash, Default)]
pub struct GeometryType {
crs: Option<Crs>,
}
```
With a dedicated type, we can also enforce normalization as part of
construction.
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -408,6 +590,25 @@ 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, EdgeInterpolationAlgorithm::Spherical) => write!(f,
"geography"),
+ (Some(crs), EdgeInterpolationAlgorithm::Spherical) => {
+ write!(f, "geography({crs})")
+ }
+ (crs, algorithm) => write!(
+ f,
+ "geography({}, {})",
+ crs.unwrap_or(DEFAULT_GEOSPATIAL_CRS),
+ edge_interpolation_algorithm_as_str(algorithm)
+ ),
+ }
+ }
Review Comment:
Geography in iceberg-java is including spherical in its toString method, why
not be consistent?
https://github.com/apache/iceberg/blob/27bfd00d67b2d00e3d4ea3e96832248bdc5107f6/api/src/test/java/org/apache/iceberg/types/TestTypes.java#L181-L182
##########
crates/iceberg/public-api.txt:
##########
@@ -1463,6 +1463,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, <Self as
core::str::traits::FromStr>::Err: core::fmt::Display
pub fn iceberg::spec::DataFileFormat::deserialize<__D>(deserializer: __D) ->
core::result::Result<Self, <__D as serde_core::de::Deserializer>::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<Self, <__D as
serde_core::de::Deserializer>::Error> where __D:
serde_core::de::Deserializer<'de>
Review Comment:
I'm wondering if we really want to export all of this into the `spec`
module, or if we might want to namespace this under `geospatial` and only
re-export the main Geometry types.
My thinking is it may be better to put all the geospatial spec stuff in its
own `spec::geospatial` module, we can re-export more if we feel its justified
later without breaking changes.
Do you have any thoughts on this, @blackmwk @CTTY?
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -408,6 +590,25 @@ 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"),
+ },
Review Comment:
I don't see a variant in the spec without CRS provided. Should we write the
default?
https://iceberg.apache.org/spec/#schemas
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -396,22 +449,55 @@ 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])?;
Review Comment:
(sorry, this comment isn't super actionable)
I need to better understand Variant's approach here. It appears to attach
the field extension type earlier (I think because of field ID shenanigans), and
I'm wondering if we should align or not. If we choose not to align, I think a
comment is necessary on why they are separate.
##########
crates/iceberg/src/spec/datatypes.rs:
##########
@@ -230,6 +232,173 @@ 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,
+}
Review Comment:
We should include tests for equality and hash, to show if case sensitivity
matters or not.
##########
crates/iceberg/src/arrow/schema.rs:
##########
@@ -628,15 +717,31 @@ 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);
+ }
Review Comment:
I think we should update this to be `try_with_extension_type` for
consistency (or update geospatial types to use `with_extension_type` if
justified).
--
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]
