etseidl commented on code in PR #8524:
URL: https://github.com/apache/arrow-rs/pull/8524#discussion_r2412190677
##########
parquet/src/column/writer/encoder.rs:
##########
@@ -145,10 +152,12 @@ impl<T: DataType> ColumnValueEncoderImpl<T> {
fn write_slice(&mut self, slice: &[T::T]) -> Result<()> {
if self.statistics_enabled != EnabledStatistics::None
- // INTERVAL has undefined sort order, so don't write min/max stats
for it
+ // INTERVAL, Geometry, and Geography have undefined sort order,so
don't write min/max stats for them
&& self.descr.converted_type() != ConvertedType::INTERVAL
{
- if let Some((min, max)) = self.min_max(slice, None) {
+ if let Some(accumulator) = self.geo_stats_accumulator.as_mut() {
+ update_geo_stats_accumulator(accumulator.as_mut(),
slice.iter());
Review Comment:
```suggestion
if let Some(accumulator) =
self.geo_stats_accumulator.as_deref_mut() {
update_geo_stats_accumulator(accumulator, slice.iter());
```
##########
parquet/src/geospatial/accumulator.rs:
##########
@@ -0,0 +1,359 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! This module provides implementations and traits for building
[GeospatialStatistics]
+
+use std::sync::{Arc, OnceLock};
+
+use crate::{
+ errors::ParquetError, geospatial::statistics::GeospatialStatistics,
+ schema::types::ColumnDescPtr,
+};
+
+/// Create a new [GeoStatsAccumulator] instance
+pub fn new_geo_stats_accumulator(descr: &ColumnDescPtr) -> Box<dyn
GeoStatsAccumulator> {
+ ACCUMULATOR_FACTORY
+ .get_or_init(|| Arc::new(DefaultGeoStatsAccumulatorFactory::default()))
+ .new_accumulator(descr)
+}
+
+/// Initialize the global [GeoStatsAccumulatorFactory]
Review Comment:
```suggestion
/// Initialize the global [`GeoStatsAccumulatorFactory`]
```
##########
parquet/src/arrow/arrow_writer/byte_array.rs:
##########
@@ -536,6 +549,14 @@ impl ColumnValueEncoder for ByteArrayEncoder {
_ => self.fallback.flush_data_page(min_value, max_value),
}
}
+
+ fn flush_geospatial_statistics(&mut self) ->
Option<Box<GeospatialStatistics>> {
+ if let Some(accumulator) = self.geo_stats_accumulator.as_mut() {
+ accumulator.finish()
+ } else {
+ None
+ }
Review Comment:
```suggestion
self.geo_stats_accumulator.as_mut().map(|a| a.finish())?
```
##########
parquet/src/column/writer/encoder.rs:
##########
@@ -307,6 +326,14 @@ impl<T: DataType> ColumnValueEncoder for
ColumnValueEncoderImpl<T> {
variable_length_bytes: self.variable_length_bytes.take(),
})
}
+
+ fn flush_geospatial_statistics(&mut self) ->
Option<Box<GeospatialStatistics>> {
+ if let Some(accumulator) = self.geo_stats_accumulator.as_mut() {
+ accumulator.finish()
+ } else {
+ None
+ }
Review Comment:
```suggestion
self.geo_stats_accumulator.as_mut().map(|a| a.finish())?
```
##########
parquet/src/arrow/arrow_writer/mod.rs:
##########
@@ -491,6 +499,18 @@ impl ArrowWriterOptions {
..self
}
}
+
+ /// Explicitly specify the Parquet schema to be used
+ ///
+ /// If omitted (the default), the [ArrowSchemaConverter] is used to
compute the
+ /// Parquet [SchemaDescriptor]. This may be used When the
[SchemaDescriptor] is
+ /// already known or must be calculated using custom logic.
Review Comment:
```suggestion
/// If omitted (the default), the [`ArrowSchemaConverter`] is used to
compute the
/// Parquet [`SchemaDescriptor`]. This may be used When the
[`SchemaDescriptor`] is
/// already known or must be calculated using custom logic.
```
##########
parquet/src/column/writer/encoder.rs:
##########
@@ -145,10 +152,12 @@ impl<T: DataType> ColumnValueEncoderImpl<T> {
fn write_slice(&mut self, slice: &[T::T]) -> Result<()> {
if self.statistics_enabled != EnabledStatistics::None
- // INTERVAL has undefined sort order, so don't write min/max stats
for it
+ // INTERVAL, Geometry, and Geography have undefined sort order,so
don't write min/max stats for them
Review Comment:
```suggestion
// INTERVAL, Geometry, and Geography have undefined sort order,
so don't write min/max stats for them
```
##########
parquet/src/column/writer/encoder.rs:
##########
@@ -367,3 +394,17 @@ fn replace_zero<T: ParquetValueType>(val: &T, descr:
&ColumnDescriptor, replace:
_ => val.clone(),
}
}
+
+fn update_geo_stats_accumulator<'a, T, I>(bounder: &mut dyn
GeoStatsAccumulator, iter: I)
+where
+ T: ParquetValueType + 'a,
+ I: Iterator<Item = &'a T>,
+{
+ if !bounder.is_valid() {
+ return;
+ }
+
+ for val in iter {
+ bounder.update_wkb(val.as_bytes());
+ }
Review Comment:
This could go inside `if bounder.is_valid()` instead. 🤷
##########
parquet/src/geospatial/accumulator.rs:
##########
@@ -0,0 +1,359 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! This module provides implementations and traits for building
[GeospatialStatistics]
Review Comment:
```suggestion
//! This module provides implementations and traits for building
[`GeospatialStatistics`]
```
##########
parquet/src/arrow/arrow_writer/byte_array.rs:
##########
@@ -447,13 +450,23 @@ impl ColumnValueEncoder for ByteArrayEncoder {
let statistics_enabled = props.statistics_enabled(descr.path());
+ let geo_stats_accumulator = if matches!(
+ descr.logical_type(),
+ Some(LogicalType::Geometry) | Some(LogicalType::Geography)
+ ) {
+ Some(new_geo_stats_accumulator(descr))
+ } else {
+ None
+ };
Review Comment:
I see this pattern twice. I'm wondering if `new_geo_stats_accumulator` could
be `try_new_geo_stats_accumulator` and perform the check on logical type
internally, returning `None` if not `Geometry` or `Geography`.
##########
parquet/src/geospatial/accumulator.rs:
##########
@@ -0,0 +1,359 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! This module provides implementations and traits for building
[GeospatialStatistics]
+
+use std::sync::{Arc, OnceLock};
+
+use crate::{
+ errors::ParquetError, geospatial::statistics::GeospatialStatistics,
+ schema::types::ColumnDescPtr,
+};
+
+/// Create a new [GeoStatsAccumulator] instance
+pub fn new_geo_stats_accumulator(descr: &ColumnDescPtr) -> Box<dyn
GeoStatsAccumulator> {
+ ACCUMULATOR_FACTORY
+ .get_or_init(|| Arc::new(DefaultGeoStatsAccumulatorFactory::default()))
+ .new_accumulator(descr)
+}
+
+/// Initialize the global [GeoStatsAccumulatorFactory]
+///
+/// This may only be done once before any calls to [new_geo_stats_accumulator].
Review Comment:
```suggestion
/// This may only be done once before any calls to
[`new_geo_stats_accumulator`].
```
##########
parquet/src/geospatial/statistics.rs:
##########
@@ -58,6 +56,16 @@ impl GeospatialStatistics {
geospatial_types,
}
}
+
+ /// Optional list of geometry type identifiers, where None represents lack
of information
Review Comment:
```suggestion
/// Optional list of geometry type identifiers, where `None` represents
lack of information
```
##########
parquet/src/geospatial/accumulator.rs:
##########
@@ -0,0 +1,359 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! This module provides implementations and traits for building
[GeospatialStatistics]
+
+use std::sync::{Arc, OnceLock};
+
+use crate::{
+ errors::ParquetError, geospatial::statistics::GeospatialStatistics,
+ schema::types::ColumnDescPtr,
+};
+
+/// Create a new [GeoStatsAccumulator] instance
+pub fn new_geo_stats_accumulator(descr: &ColumnDescPtr) -> Box<dyn
GeoStatsAccumulator> {
+ ACCUMULATOR_FACTORY
+ .get_or_init(|| Arc::new(DefaultGeoStatsAccumulatorFactory::default()))
+ .new_accumulator(descr)
+}
+
+/// Initialize the global [GeoStatsAccumulatorFactory]
+///
+/// This may only be done once before any calls to [new_geo_stats_accumulator].
+/// Clients may use this to implement support for builds of the Parquet crate
without
+/// geospatial support or to implement support for Geography bounding using
external
+/// dependencies.
+pub fn init_geo_stats_accumulator_factory(
+ factory: Arc<dyn GeoStatsAccumulatorFactory>,
+) -> Result<(), ParquetError> {
+ if ACCUMULATOR_FACTORY.set(factory).is_err() {
+ Err(ParquetError::General(
+ "Global GeoStatsAccumulatorFactory already set".to_string(),
+ ))
+ } else {
+ Ok(())
+ }
+}
+
+/// Global accumulator factory instance
+static ACCUMULATOR_FACTORY: OnceLock<Arc<dyn GeoStatsAccumulatorFactory>> =
OnceLock::new();
+
+/// Factory for [GeospatialStatistics] accumulators
Review Comment:
```suggestion
/// Factory for [`GeospatialStatistics`] accumulators
```
I'm getting lazy...there are more below. We usually use fixed font for types
et al.
##########
parquet/src/geospatial/statistics.rs:
##########
@@ -58,6 +56,16 @@ impl GeospatialStatistics {
geospatial_types,
}
}
+
+ /// Optional list of geometry type identifiers, where None represents lack
of information
+ pub fn geospatial_types(&self) -> Option<&Vec<i32>> {
+ self.geospatial_types.as_ref()
+ }
+
+ /// Optional bounding defining the spatial extent, where None represents a
lack of information.
Review Comment:
```suggestion
/// Optional bounding defining the spatial extent, where `None`
represents a lack of information.
```
--
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]
