fix: more permissive deserialization, arrow v58 (#985)

### Description

We have to patch to a lot of branches, so we might not even want to
merge it. But I _think_ we'll need v58 to get native geometry reading
(see stac-utils/rustac-py#251), which GDAL is
writing.

Includes several relaxes of our **stac-geoparquet** parsing to handle
common mistakes. We should probably make a validator.

Author: gadomski

Cargo.toml

@@ -34,10 +34,10 @@ rust-version = "1.88"
 
 [workspace.dependencies]
 anyhow = "1.0"
-arrow-array = "57.0.0"
-arrow-cast = "57.0.0"
-arrow-json = "57.0.0"
-arrow-schema = "57.0.0"
+arrow-array = "58.0.0"
+arrow-cast = "58.0.0"
+arrow-json = "58.0.0"
+arrow-schema = "58.0.0"
 assert-json-diff = "2.0"
 assert_cmd = "2.1"
 async-recursion = "1.1.1"
@@ -51,7 +51,7 @@ chrono = "0.4.39"
 clap = "4.5"
 clap_complete = "4.5"
 cql2 = "0.5.0"
-duckdb = { git = "https://github.com/duckdb/duckdb-rs", rev = "a2639608d946690e04b1d5f6448302d57bb99d7e" }
+duckdb = { git = "https://github.com/gadomski/duckdb-rs", branch = "arrow-v58" }
 fluent-uri = "0.4.1"
 futures = "0.3.31"
 futures-core = "0.3.31"
@@ -73,8 +73,8 @@ libduckdb-sys = "1.3.0"
 log = "0.4.25"
 mime = "0.3.17"
 mockito = "1.5"
-object_store = "0.12.0"
-parquet = { version = "57.0.0" }
+object_store = "0.13.0"
+parquet = { version = "58.0.0" }
 quote = "1.0"
 reqwest = { version = "0.13.1", features = ["query"] }
 referencing = { version = "0.45.0", features = ["retrieve-async"] }
@@ -102,3 +102,8 @@ tracing-indicatif = "0.3.9"
 url = "2.3"
 webpki-roots = "1.0.0"
 wkb = "0.9.0"
+
+[patch.crates-io]
+geoarrow-array = { git = "https://github.com/geoarrow/geoarrow-rs", branch = "arrow-58" }
+geoarrow-schema = { git = "https://github.com/geoarrow/geoarrow-rs", branch = "arrow-58" }
+geoparquet = { git = "https://github.com/geoarrow/geoarrow-rs", branch = "arrow-58" }

crates/cli/src/lib.rs

@@ -106,11 +106,11 @@ pub struct Rustac {
     /// When records are ordered spatially or temporally, lower values result in smaller row groups (better for selective queries),
     /// while higher values result in larger row groups (better compression).
     #[arg(
-        long = "parquet-max-row-group-size",
+        long = "parquet-max-row-group-row-count",
         global = true,
         verbatim_doc_comment
     )]
-    parquet_max_row_group_size: Option<usize>,
+    parquet_max_row_group_row_count: Option<usize>,
 
     #[arg(
         long,
@@ -398,6 +398,7 @@ impl Rustac {
                         );
                     }
                     let input_format = self.input_format(infile.as_deref());
+                    tracing::debug!("Reading as {input_format}");
                     let can_stream = matches!(input_format, Format::NdJson | Format::Geoparquet(_));
                     if can_stream {
                         let items = self.get_item_stream(infile.as_deref()).await?;
@@ -826,8 +827,9 @@ impl Rustac {
             let mut writer_options = WriterOptions::new()
                 .with_compression(self.parquet_compression.or(Some(default_compression())));
 
-            if let Some(max_row_group_size) = self.parquet_max_row_group_size {
-                writer_options = writer_options.with_max_row_group_size(max_row_group_size);
+            if let Some(max_row_group_row_count) = self.parquet_max_row_group_row_count {
+                writer_options =
+                    writer_options.with_max_row_group_row_count(max_row_group_row_count);
             }
 
             Format::Geoparquet(writer_options)

crates/cli/tests/test_cli.rs

@@ -137,13 +137,13 @@ fn output_format() {
         "rustac",
         "--output-format",
         "parquet",
-        "--parquet-max-row-group-size",
+        "--parquet-max-row-group-row-count",
         "50000",
         "translate",
     ]);
     assert_eq!(
         rustac.output_format(None),
-        Format::Geoparquet(WriterOptions::new().with_max_row_group_size(50000))
+        Format::Geoparquet(WriterOptions::new().with_max_row_group_row_count(50000))
     );
 
     let rustac = Rustac::parse_from([
@@ -152,7 +152,7 @@ fn output_format() {
         "parquet",
         "--parquet-compression",
         "snappy",
-        "--parquet-max-row-group-size",
+        "--parquet-max-row-group-row-count",
         "100000",
         "translate",
     ]);
@@ -161,7 +161,7 @@ fn output_format() {
         Format::Geoparquet(
             WriterOptions::new()
                 .with_compression(Some(Compression::SNAPPY))
-                .with_max_row_group_size(100000)
+                .with_max_row_group_row_count(100000)
         )
     );
 }

crates/core/src/datetime.rs

@@ -1,12 +1,12 @@
 //! Datetime utilities.
 
 use crate::{Error, Result};
-use chrono::{DateTime, FixedOffset};
+use chrono::{DateTime, NaiveDateTime, Utc};
 
 /// A start and end datetime.
-pub type Interval = (Option<DateTime<FixedOffset>>, Option<DateTime<FixedOffset>>);
+pub type Interval = (Option<DateTime<Utc>>, Option<DateTime<Utc>>);
 
-/// Parse a datetime or datetime interval into a start and end datetime.
+/// Parses a datetime or datetime interval into a start and end datetime.
 ///
 /// Returns `None` to indicate an open interval.
 ///
@@ -35,21 +35,38 @@ pub fn parse(datetime: &str) -> Result<Interval> {
     } else if datetime == ".." {
         Err(Error::InvalidDatetime(datetime.to_string()))
     } else {
-        let datetime = DateTime::parse_from_rfc3339(datetime).map(Some)?;
+        let datetime = parse_datetime_permissively(datetime).map(Some)?;
         Ok((datetime, datetime))
     }
 }
 
-fn parse_one(s: &str) -> Result<Option<DateTime<FixedOffset>>> {
+/// Parses a single datetime permissively.
+pub fn parse_datetime_permissively(s: &str) -> Result<DateTime<Utc>> {
+    match DateTime::parse_from_rfc3339(&s) {
+        Ok(datetime) => Ok(datetime.to_utc()),
+        Err(err) => {
+            log::warn!(
+                "error when parsing item datetime as rfc3339 ({err}), trying to parse as naive datetime"
+            );
+            let (mut datetime, remainder) =
+                NaiveDateTime::parse_and_remainder(&s, "%Y-%m-%dT%H:%M:%S")?;
+            // This isn't super efficient but we're in a read-invalid-data path, so I think it's fine.
+            if !remainder.is_empty() && remainder.starts_with(".") {
+                datetime = NaiveDateTime::parse_from_str(&s, "%Y-%m-%dT%H:%M:%S%.f")?;
+            }
+            Ok(datetime.and_utc())
+        }
+    }
+}
+
+fn parse_one(s: &str) -> Result<Option<DateTime<Utc>>> {
     if s == ".." {
         Ok(None)
     } else if s.is_empty() {
         log::warn!("an empty string in a datetime interval are invalid, converting to \"..\"");
         Ok(None)
     } else {
-        DateTime::parse_from_rfc3339(s)
-            .map(Some)
-            .map_err(Error::from)
+        parse_datetime_permissively(s).map(Some)
     }
 }
 

crates/core/src/geoarrow/json.rs

@@ -38,12 +38,11 @@ const TOP_LEVEL_KEYS: [&str; 10] = [
     "collection",
 ];
 
-use crate::Error;
+use crate::{Error, datetime::parse_datetime_permissively};
 use arrow_array::{RecordBatchReader, cast::*, types::*, *};
 use arrow_cast::display::{ArrayFormatter, FormatOptions};
 use arrow_json::JsonSerializable;
 use arrow_schema::*;
-use chrono::DateTime;
 use geo_traits::to_geo::{
     ToGeoGeometry, ToGeoGeometryCollection, ToGeoLineString, ToGeoMultiLineString, ToGeoMultiPoint,
     ToGeoMultiPolygon, ToGeoPoint, ToGeoPolygon, ToGeoRect,
@@ -535,7 +534,33 @@ pub(crate) fn record_batch_to_json_rows(
     }
     rows.into_iter()
         .map(|row| {
-            let row = row.unwrap();
+            let mut row = row.unwrap();
+            // Force it, in case
+            if row.contains_key("type") {
+                row.insert("type".to_string(), "Feature".to_string().into());
+            }
+            if let Some(id) = row.remove("id") {
+                if id.is_string() {
+                    row.insert("id".to_string(), id);
+                } else {
+                    tracing::warn!("id field is not a string, coercing");
+                    row.insert("id".to_string(), id.to_string().into());
+                }
+            }
+            if let Some(stac_extensions) = row.remove("stac_extensions") {
+                if stac_extensions.is_string() {
+                    tracing::warn!(
+                        "stac_extensions field is a string, coercing to list of strings"
+                    );
+                    let _ = row.insert(
+                        "stac_extensions".to_string(),
+                        serde_json::from_str::<Vec<String>>(stac_extensions.as_str().unwrap())?
+                            .into(),
+                    );
+                } else {
+                    row.insert("stac_extensions".to_string(), stac_extensions);
+                }
+            }
             unflatten(row)
         })
         .collect::<Result<_, _>>()
@@ -562,13 +587,8 @@ fn unflatten(
         if let Some(value) = item.remove(&key) {
             if DATETIME_COLUMNS.contains(&key.as_str()) {
                 if let Some(value) = value.as_str() {
-                    let _ = properties.insert(
-                        key,
-                        DateTime::parse_from_rfc3339(value)?
-                            .to_utc()
-                            .to_rfc3339()
-                            .into(),
-                    );
+                    let _ = properties
+                        .insert(key, parse_datetime_permissively(value)?.to_rfc3339().into());
                 }
             } else {
                 let _ = properties.insert(key, value);

crates/core/src/geoparquet.rs

@@ -23,8 +23,8 @@ pub fn default_compression() -> Compression {
     Compression::ZSTD(ZstdLevel::try_new(15).unwrap())
 }
 
-/// Default stac-geoparquet max row group size
-pub const DEFAULT_STAC_MAX_ROW_GROUP_SIZE: usize = 150_000;
+/// Default stac-geoparquet max row group row count
+pub const DEFAULT_STAC_MAX_ROW_GROUP_ROW_COUNT: usize = 150_000;
 
 /// The stac-geoparquet metadata key.
 pub const METADATA_KEY: &str = "stac-geoparquet";
@@ -39,7 +39,7 @@ pub struct WriterOptions {
     pub compression: Option<Compression>,
 
     /// Maximum number of rows in a row group
-    pub max_row_group_size: usize,
+    pub max_row_group_row_count: usize,
 }
 
 /// An encoder for writing stac-geoparquet
@@ -84,10 +84,10 @@ impl WriterOptions {
     /// ```
     /// use stac::geoparquet::WriterOptions;
     ///
-    /// let options = WriterOptions::new().with_max_row_group_size(50000);
+    /// let options = WriterOptions::new().with_max_row_group_row_count(50000);
     /// ```
-    pub fn with_max_row_group_size(mut self, size: usize) -> Self {
-        self.max_row_group_size = size;
+    pub fn with_max_row_group_row_count(mut self, size: usize) -> Self {
+        self.max_row_group_row_count = size;
         self
     }
 }
@@ -96,7 +96,7 @@ impl Default for WriterOptions {
     fn default() -> Self {
         Self {
             compression: Some(default_compression()),
-            max_row_group_size: DEFAULT_STAC_MAX_ROW_GROUP_SIZE,
+            max_row_group_row_count: DEFAULT_STAC_MAX_ROW_GROUP_ROW_COUNT,
         }
     }
 }
@@ -277,7 +277,7 @@ impl<W: Write + Send> WriterBuilder<W> {
     /// let cursor = Cursor::new(Vec::new());
     /// let options = WriterOptions::new()
     ///     .with_compression(Compression::SNAPPY)
-    ///     .with_max_row_group_size(50000);
+    ///     .with_max_row_group_row_count(50000);
     /// let writer = WriterBuilder::new(cursor)
     ///     .writer_options(options)
     ///     .build(vec![item])
@@ -733,7 +733,7 @@ impl From<WriterOptions> for WriterProperties {
         if let Some(compression) = value.compression {
             builder = builder.set_compression(compression);
         }
-        builder = builder.set_max_row_group_size(value.max_row_group_size);
+        builder = builder.set_max_row_group_row_count(Some(value.max_row_group_row_count));
         builder.build()
     }
 }
@@ -877,7 +877,7 @@ mod tests {
         let items: Vec<Item> = (0..100).map(|_| item.clone()).collect();
 
         let mut cursor = Cursor::new(Vec::new());
-        let options = super::WriterOptions::new().with_max_row_group_size(25);
+        let options = super::WriterOptions::new().with_max_row_group_row_count(25);
         WriterBuilder::new(&mut cursor)
             .writer_options(options)
             .build(items)

crates/core/src/item.rs

@@ -1,7 +1,10 @@
 //! STAC Items.
 
-use crate::{Asset, Assets, Bbox, Error, Fields, Link, Result, STAC_VERSION, Version};
-use chrono::{DateTime, FixedOffset, NaiveDateTime, Utc};
+use crate::{
+    Asset, Assets, Bbox, Error, Fields, Link, Result, STAC_VERSION, Version,
+    datetime::parse_datetime_permissively,
+};
+use chrono::{DateTime, Utc};
 use cql2::Expr;
 use geojson::{Feature, Geometry, feature::Id};
 use indexmap::IndexMap;
@@ -523,8 +526,8 @@ impl Item {
     /// ```
     pub fn intersects_datetimes(
         &self,
-        start: Option<DateTime<FixedOffset>>,
-        end: Option<DateTime<FixedOffset>>,
+        start: Option<DateTime<Utc>>,
+        end: Option<DateTime<Utc>>,
     ) -> Result<bool> {
         let (item_start, item_end) = self.datetimes();
         let mut intersects = true;
@@ -722,23 +725,9 @@ where
     use serde::de::Error;
 
     if let Some(s) = Option::<String>::deserialize(deserializer)? {
-        match DateTime::parse_from_rfc3339(&s) {
-            Ok(datetime) => Ok(Some(datetime.to_utc())),
-            Err(err) => {
-                log::warn!(
-                    "error when parsing item datetime as rfc3339 ({err}), trying to parse as naive datetime"
-                );
-                let (mut datetime, remainder) =
-                    NaiveDateTime::parse_and_remainder(&s, "%Y-%m-%dT%H:%M:%S")
-                        .map_err(D::Error::custom)?;
-                // This isn't super efficient but we're in a read-invalid-data path, so I think it's fine.
-                if !remainder.is_empty() && remainder.starts_with(".") {
-                    datetime = NaiveDateTime::parse_from_str(&s, "%Y-%m-%dT%H:%M:%S%.f")
-                        .map_err(D::Error::custom)?;
-                }
-                Ok(Some(datetime.and_utc()))
-            }
-        }
+        parse_datetime_permissively(&s)
+            .map(Some)
+            .map_err(D::Error::custom)
     } else {
         Ok(None)
     }

crates/io/src/store.rs

@@ -1,5 +1,5 @@
 use crate::{Format, Readable, Result, Writeable};
-use object_store::{ObjectStore, ObjectStoreScheme, PutResult, path::Path};
+use object_store::{ObjectStore, ObjectStoreExt, ObjectStoreScheme, PutResult, path::Path};
 use std::{fmt::Debug, sync::Arc};
 use tracing::instrument;
 use url::Url;
@@ -206,7 +206,7 @@ impl StacStore {
             }
             #[cfg(feature = "geoparquet")]
             Format::Geoparquet(writer_options) => {
-                let batch_size = writer_options.max_row_group_size;
+                let batch_size = writer_options.max_row_group_row_count;
                 let mut batch = Vec::with_capacity(batch_size);
                 let mut writer: Option<geoparquet::StacGeoparquetObjectWriter> = None;
                 for item in items {
@@ -368,7 +368,7 @@ mod tests {
     #[tokio::test]
     #[cfg(feature = "geoparquet")]
     async fn write_parquet() {
-        use object_store::ObjectStore;
+        use object_store::ObjectStoreExt;
 
         use super::geoparquet::StacGeoparquetObjectWriter;
 
@@ -400,7 +400,7 @@ mod tests {
     #[tokio::test]
     #[cfg(feature = "geoparquet")]
     async fn write_parquet_with_collection() {
-        use object_store::ObjectStore;
+        use object_store::ObjectStoreExt;
         use parquet::file::reader::{FileReader, SerializedFileReader};
 
         use super::geoparquet::StacGeoparquetObjectWriter;