wirybeaver commented on code in PR #2933:
URL: https://github.com/apache/iceberg-rust/pull/2933#discussion_r3997432200
##########
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:
Done in 255e0211. I added a comment and assertion messages explaining that
geospatial bounds are intentionally omitted until zonemap statistics are
implemented.
##########
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:
Partially implemented in 255e0211: `From<WkbEdges> for
EdgeInterpolationAlgorithm` now handles the inbound conversion. Rust orphan
rules prevent the reverse `From<EdgeInterpolationAlgorithm> for WkbEdges`
because both `From` and the target `WkbEdges` type are external, so the
outbound direction remains a small named adapter.
##########
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:
Done in 255e0211. CRS is now a dedicated internal type with case-insensitive
equality and hashing while preserving the original spelling for display. The
focused test verifies both equality and matching hashes.
##########
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:
Done in 255e0211. Parsing and string conversion now live on
`EdgeInterpolationAlgorithm`.
##########
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:
Done in 255e0211. Geography display now always includes both the canonical
CRS and edge algorithm, matching iceberg-java, including `spherical`.
##########
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:
Done in 255e0211. Added the requested assertion messages.
##########
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:
Done in 255e0211. Parsing remains tolerant of omitted defaults, while
display/JSON serialization now emits the canonical forms `geometry(OGC:CRS84)`
and `geography(OGC:CRS84, spherical)`.
##########
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:
Done in 255e0211. The missing Parquet edge metadata now explicitly maps to
`WkbEdges::Spherical` before conversion.
--
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]
