feat: add geoparquet writer encoder and object writing (#863)

## Closes

- Closes #847

## Related to

We'll want to add reading eventually as well, but that's a follow-on.

## Checklist

Delete any checklist items that do not apply (e.g. if your change is
minor, it may not require documentation updates).

- [x] Unit tests
- [x] Documentation, including doctests
- [x] Git history is linear
- [x] Commit messages are descriptive
- [x] (optional) Git commit messages follow [conventional
commits](https://www.conventionalcommits.org/en/v1.0.0/)
- [x] Code is formatted (`cargo fmt`)
- [x] `cargo test`
- [x] Changes are added to the CHANGELOG

<!-- markdownlint-disable-file MD041 -->

Author: gadomski

crates/core/CHANGELOG.md

@@ -9,6 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Added
 
 - Add `max_row_group_size` parameter to `geoparquet::WriterBuilder` with default value of `150_000` ([#846](https://github.com/stac-utils/rustac/pull/846))
+- An intermediate `WriterEncoder` for writing **stac-geoparquet** ([#863](https://github.com/stac-utils/rustac/pull/863))
 
 ## [0.14.0] - 2025-11-14
 

crates/core/src/geoparquet.rs

@@ -4,6 +4,7 @@ use crate::{
     Catalog, Collection, Error, Item, ItemCollection, Result, Value,
     geoarrow::{Encoder, Options, VERSION, VERSION_KEY},
 };
+use arrow_array::RecordBatch;
 use bytes::Bytes;
 use geoparquet::{
     reader::{GeoParquetReaderBuilder, GeoParquetRecordBatchReader},
@@ -35,6 +36,13 @@ pub struct WriterOptions {
     pub max_row_group_size: usize,
 }
 
+/// An encoder for writing stac-geoparquet
+#[allow(missing_debug_implementations)]
+pub struct WriterEncoder {
+    geoarrow_encoder: Encoder,
+    encoder: GeoParquetRecordBatchEncoder,
+}
+
 impl WriterOptions {
     /// Creates a new WriterOptions with default values.
     ///
@@ -178,10 +186,8 @@ pub struct WriterBuilder<W: Write + Send> {
 /// Write items to stac-geoparquet.
 #[allow(missing_debug_implementations)]
 pub struct Writer<W: Write + Send> {
-    geoarrow_encoder: Encoder,
-    // We make this an option so we can consume it during write but keep write
-    // as only requiring a mutable reference.
-    encoder: Option<GeoParquetRecordBatchEncoder>,
+    // We make this an option so we can consume the encoder without dropping it.
+    encoder: Option<WriterEncoder>,
     arrow_writer: ArrowWriter<W>,
 }
 
@@ -268,30 +274,86 @@ impl<W: Write + Send> WriterBuilder<W> {
     }
 }
 
+impl WriterEncoder {
+    /// Creates a new writer encoder.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use stac::{Item, geoarrow::Options, geoparquet::WriterEncoder};
+    ///
+    /// let item: Item = stac::read("examples/simple-item.json").unwrap();
+    /// let options = Options::default();
+    /// let (encoder, record_batch) = WriterEncoder::new(options, vec![item]).unwrap();
+    /// assert_eq!(record_batch.num_rows(), 1);
+    /// ```
+    pub fn new(options: Options, items: Vec<Item>) -> Result<(WriterEncoder, RecordBatch)> {
+        let (geoarrow_encoder, record_batch) = Encoder::new(items, options)?;
+        let options = GeoParquetWriterOptionsBuilder::default()
+            .set_primary_column("geometry".to_string())
+            .build();
+        let mut encoder = GeoParquetRecordBatchEncoder::try_new(&record_batch.schema(), &options)?;
+        let record_batch = encoder.encode_record_batch(&record_batch)?;
+        Ok((
+            WriterEncoder {
+                geoarrow_encoder,
+                encoder,
+            },
+            record_batch,
+        ))
+    }
+
+    /// Encodes items into a record batch.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use stac::{Item, geoarrow::Options, geoparquet::WriterEncoder};
+    ///
+    /// let item1: Item = stac::read("examples/simple-item.json").unwrap();
+    /// let item2 = item1.clone();
+    /// let options = Options::default();
+    /// let (mut encoder, record_batch) = WriterEncoder::new(options, vec![item1]).unwrap();
+    /// let record_batch = encoder.encode(vec![item2]).unwrap();
+    /// assert_eq!(record_batch.num_rows(), 1);
+    /// ```
+    pub fn encode(&mut self, items: Vec<Item>) -> Result<RecordBatch> {
+        let record_batch = self.geoarrow_encoder.encode(items)?;
+        let record_batch = self.encoder.encode_record_batch(&record_batch)?;
+        Ok(record_batch)
+    }
+
+    /// Consumes this encoder and returns the keys and values.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use stac::{Item, geoarrow::Options, geoparquet::WriterEncoder};
+    ///
+    /// let item: Item = stac::read("examples/simple-item.json").unwrap();
+    /// let options = Options::default();
+    /// let (encoder, _) = WriterEncoder::new(options, vec![item]).unwrap();
+    /// let key_value = encoder.into_keyvalue().unwrap();
+    /// assert_eq!(key_value.key, "geo");
+    /// ```
+    pub fn into_keyvalue(self) -> Result<KeyValue> {
+        let keyvalue = self.encoder.into_keyvalue()?;
+        Ok(keyvalue)
+    }
+}
+
 impl<W: Write + Send> Writer<W> {
     fn new(
         writer: W,
         options: Options,
         writer_options: WriterOptions,
         items: Vec<Item>,
     ) -> Result<Self> {
-        let (geoarrow_encoder, record_batch) = Encoder::new(items, options)?;
-        let options = GeoParquetWriterOptionsBuilder::default()
-            .set_primary_column("geometry".to_string())
-            .build();
-        let mut encoder = GeoParquetRecordBatchEncoder::try_new(&record_batch.schema(), &options)?;
-        let mut builder = WriterProperties::builder();
-        if let Some(compression) = writer_options.compression {
-            builder = builder.set_compression(compression);
-        }
-        builder = builder.set_max_row_group_size(writer_options.max_row_group_size);
-        let properties = builder.build();
+        let (encoder, record_batch) = WriterEncoder::new(options, items)?;
         let mut arrow_writer =
-            ArrowWriter::try_new(writer, encoder.target_schema(), Some(properties))?;
-        let record_batch = encoder.encode_record_batch(&record_batch)?;
+            ArrowWriter::try_new(writer, record_batch.schema(), Some(writer_options.into()))?;
         arrow_writer.write(&record_batch)?;
         Ok(Writer {
-            geoarrow_encoder,
             encoder: Some(encoder),
             arrow_writer,
         })
@@ -315,9 +377,8 @@ impl<W: Write + Send> Writer<W> {
     /// writer.finish().unwrap();
     /// ```
     pub fn write(&mut self, items: Vec<Item>) -> Result<()> {
-        let record_batch = self.geoarrow_encoder.encode(items)?;
         let record_batch = if let Some(encoder) = self.encoder.as_mut() {
-            encoder.encode_record_batch(&record_batch)?
+            encoder.encode(items)?
         } else {
             return Err(Error::ClosedGeoparquetWriter);
         };
@@ -491,6 +552,17 @@ impl IntoGeoparquet for serde_json::Value {
     }
 }
 
+impl From<WriterOptions> for WriterProperties {
+    fn from(value: WriterOptions) -> Self {
+        let mut builder = WriterProperties::builder();
+        if let Some(compression) = value.compression {
+            builder = builder.set_compression(compression);
+        }
+        builder = builder.set_max_row_group_size(value.max_row_group_size);
+        builder.build()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::{FromGeoparquet, Item, ItemCollection, SelfHref, Value, geoparquet::WriterBuilder};

crates/io/CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ### Added
 
+- Writing **stac-geoparquet** to an object store ([#863](https://github.com/stac-utils/rustac/pull/863))
 - API client ([#864](https://github.com/stac-utils/rustac/pull/864))
 
 ## [0.1.2] - 2025-11-14

crates/io/Cargo.toml

@@ -27,7 +27,7 @@ futures.workspace = true
 http.workspace = true
 jsonschema = { workspace = true, optional = true }
 object_store = { workspace = true, optional = true }
-parquet = { workspace = true, optional = true }
+parquet = { workspace = true, optional = true, features = ["arrow", "async", "object_store"] }
 reqwest = { workspace = true, features = ["json", "blocking"] }
 serde.workspace = true
 serde_json = { workspace = true, features = ["preserve_order"] }

crates/io/src/lib.rs

@@ -8,7 +8,7 @@ mod ndjson;
 mod read;
 mod realized_href;
 #[cfg(feature = "store")]
-mod store;
+pub mod store;
 mod write;
 
 #[cfg(feature = "geoparquet")]

crates/io/src/store.rs

@@ -166,9 +166,61 @@ where
     }
 }
 
+#[cfg(feature = "geoparquet")]
+pub mod geoparquet {
+    use crate::Result;
+    use object_store::{ObjectStore, path::Path};
+    use parquet::arrow::async_writer::{AsyncArrowWriter, ParquetObjectWriter};
+    use stac::Item;
+    use stac::geoarrow::Options;
+    use stac::geoparquet::{WriterEncoder, WriterOptions};
+    use std::sync::Arc;
+
+    /// Writes stac-geoparquet to an object store.
+    pub struct StacGeoparquetObjectWriter {
+        encoder: WriterEncoder,
+        writer: AsyncArrowWriter<ParquetObjectWriter>,
+    }
+
+    impl StacGeoparquetObjectWriter {
+        pub async fn new(
+            store: Arc<dyn ObjectStore>,
+            path: Path,
+            items: Vec<Item>,
+            options: Options,
+            writer_options: WriterOptions,
+        ) -> Result<StacGeoparquetObjectWriter> {
+            let (encoder, record_batch) = WriterEncoder::new(options, items)?;
+            let object_store_writer = ParquetObjectWriter::new(store.clone(), path);
+            let mut writer = AsyncArrowWriter::try_new(
+                object_store_writer,
+                record_batch.schema(),
+                Some(writer_options.into()),
+            )?;
+            writer.write(&record_batch).await?;
+            Ok(StacGeoparquetObjectWriter { encoder, writer })
+        }
+
+        pub async fn write(&mut self, items: Vec<Item>) -> Result<()> {
+            let record_batch = self.encoder.encode(items)?;
+            self.writer.write(&record_batch).await?;
+            Ok(())
+        }
+
+        pub async fn close(mut self) -> Result<()> {
+            self.writer
+                .append_key_value_metadata(self.encoder.into_keyvalue()?);
+            self.writer.close().await?;
+            Ok(())
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
+    use object_store::{memory::InMemory, path::Path};
     use stac::{Item, SelfHref};
+    use std::sync::Arc;
 
     #[tokio::test]
     async fn get_local() {
@@ -186,4 +238,36 @@ mod tests {
         let href = format!("file:///{path}");
         let _: Item = store.get(href).await.unwrap();
     }
+
+    #[tokio::test]
+    #[cfg(feature = "geoparquet")]
+    async fn write_parquet() {
+        use object_store::ObjectStore;
+
+        use super::geoparquet::StacGeoparquetObjectWriter;
+
+        let store = Arc::new(InMemory::new());
+        let item: Item = stac::read("examples/simple-item.json").unwrap();
+        let mut writer = StacGeoparquetObjectWriter::new(
+            store.clone(),
+            Path::from("test"),
+            vec![item.clone()],
+            Default::default(),
+            Default::default(),
+        )
+        .await
+        .unwrap();
+        writer.write(vec![item]).await.unwrap();
+        writer.close().await.unwrap();
+
+        let bytes = store
+            .get(&Path::from("test"))
+            .await
+            .unwrap()
+            .bytes()
+            .await
+            .unwrap();
+        let item_collection = stac::geoparquet::from_reader(bytes).unwrap();
+        assert_eq!(item_collection.items.len(), 2);
+    }
 }