feat(gis): add spatial/GIS query support with GeoParquet output

Cargo.toml

@@ -11,6 +11,8 @@ path = "src/bin/server.rs"
 datafusion = "51.0"
 datafusion-tracing = "51.0.0"
 instrumented-object-store = "52.0.0"
+geodatafusion = "0.2"
+hex = "0.4"
 duckdb = { version = "1.4.4", features = ["bundled"] }
 sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "sqlite", "mysql", "chrono", "tls-rustls", "bigdecimal"] }
 bigdecimal = "0.4"

src/datafetch/batch_writer.rs

@@ -1,6 +1,8 @@
 use datafusion::arrow::datatypes::Schema;
 use datafusion::arrow::record_batch::RecordBatch;
+use std::collections::HashMap;
 
+use super::types::GeometryColumnInfo;
 use super::DataFetchError;
 
 /// Summary returned when a BatchWriter is closed.
@@ -15,16 +17,24 @@ pub struct BatchWriteResult {
 /// A trait for writing Arrow RecordBatches to storage.
 ///
 /// Implementors must follow this lifecycle:
-/// 1. `init(schema)` - Initialize with the Arrow schema (must be called first)
-/// 2. `write_batch(batch)` - Write batches (can be called zero or more times)
-/// 3. `close()` - Finalize and return metadata (consumes the writer)
+/// 1. Optionally call `set_geometry_columns()` to enable GeoParquet metadata
+/// 2. `init(schema)` - Initialize with the Arrow schema
+/// 3. `write_batch(batch)` - Write batches (can be called zero or more times)
+/// 4. `close()` - Finalize and return metadata (consumes the writer)
 ///
 /// All methods are synchronous. When used in async contexts, callers should
 /// ensure writes are batched to minimize blocking time.
 pub trait BatchWriter: Send {
     /// Initialize the writer with the schema for the data to be written.
     fn init(&mut self, schema: &Schema) -> Result<(), DataFetchError>;
 
+    /// Set geometry column metadata for GeoParquet support.
+    /// Must be called before `init()` for the metadata to be included.
+    /// The map key is the column name.
+    fn set_geometry_columns(&mut self, _columns: HashMap<String, GeometryColumnInfo>) {
+        // Default implementation does nothing - non-GeoParquet writers can ignore this
+    }
+
     /// Write a single RecordBatch. May be called multiple times.
     fn write_batch(&mut self, batch: &RecordBatch) -> Result<(), DataFetchError>;
 

src/datafetch/mod.rs

@@ -3,11 +3,14 @@ mod error;
 mod fetcher;
 pub mod native;
 mod orchestrator;
-mod types;
+pub(crate) mod types;
 
 pub use batch_writer::{BatchWriteResult, BatchWriter};
 pub use error::DataFetchError;
 pub use fetcher::DataFetcher;
-pub use native::{NativeFetcher, StreamingParquetWriter};
+pub use native::{parse_geoparquet_metadata, NativeFetcher, StreamingParquetWriter};
 pub use orchestrator::FetchOrchestrator;
-pub use types::{deserialize_arrow_schema, ColumnMetadata, TableMetadata};
+pub use types::{
+    deserialize_arrow_schema, extract_geometry_columns, ColumnMetadata, GeometryColumnInfo,
+    TableMetadata, GEOMETRY_COLUMNS_METADATA_KEY,
+};

src/datafetch/native/bigquery.rs

@@ -194,6 +194,7 @@ pub async fn discover_tables(
                 table_name,
                 table_type,
                 columns: vec![column],
+                geometry_columns: std::collections::HashMap::new(),
             });
         }
     }