Use arrow-csv for CSV I/O

Cargo.toml

@@ -10,7 +10,7 @@ categories = ["science::geo"]
 rust-version = "1.80"
 
 [features]
-csv = ["geozero/with-csv"]
+csv = ["dep:arrow-csv", "geozero/with-csv"]
 flatgeobuf = ["dep:flatgeobuf"]
 flatgeobuf_async = [
   "flatgeobuf/http",
@@ -48,6 +48,7 @@ arrow = { version = "53", features = ["ffi"] }
 arrow-array = { version = "53", features = ["chrono-tz"] }
 arrow-buffer = "53"
 arrow-cast = { version = "53" }
+arrow-csv = { version = "53", optional = true }
 arrow-data = "53"
 arrow-ipc = "53"
 arrow-schema = "53"

src/io/csv/writer.rs

@@ -1,16 +1,52 @@
+use crate::array::NativeArrayDyn;
 use crate::error::Result;
 use crate::io::stream::RecordBatchReader;
-use geozero::csv::CsvWriter;
-use geozero::GeozeroDatasource;
+use crate::io::wkt::ToWKT;
+use crate::{ArrayBase, NativeArray};
+use arrow_array::RecordBatch;
+use arrow_schema::Schema;
 use std::io::Write;
+use std::sync::Arc;
+
+// TODO: add CSV writer options
 
 /// Write a Table to CSV
 pub fn write_csv<W: Write, S: Into<RecordBatchReader>>(stream: S, writer: W) -> Result<()> {
-    let mut csv_writer = CsvWriter::new(writer);
-    stream.into().process(&mut csv_writer)?;
+    let mut stream: RecordBatchReader = stream.into();
+    let reader = stream.take().unwrap();
+
+    let mut csv_writer = arrow_csv::Writer::new(writer);
+    for batch in reader {
+        csv_writer.write(&encode_batch(batch?)?)?;
+    }
+
     Ok(())
 }
 
+fn encode_batch(batch: RecordBatch) -> Result<RecordBatch> {
+    let schema = batch.schema();
+    let fields = schema.fields();
+
+    let mut new_fields = Vec::with_capacity(fields.len());
+    let mut new_columns = Vec::with_capacity(fields.len());
+
+    for (field, column) in schema.fields().iter().zip(batch.columns()) {
+        if let Ok(arr) = NativeArrayDyn::from_arrow_array(&column, field) {
+            let wkt_arr = arr.as_ref().to_wkt::<i32>();
+            new_fields.push(wkt_arr.extension_field());
+            new_columns.push(wkt_arr.into_array_ref());
+        } else {
+            new_fields.push(field.clone());
+            new_columns.push(column.clone());
+        }
+    }
+
+    Ok(RecordBatch::try_new(
+        Arc::new(Schema::new(new_fields).with_metadata(schema.metadata().clone())),
+        new_columns,
+    )?)
+}
+
 #[cfg(test)]
 mod test {
     use super::*;