wire: drop hand-rolled flatbuffer parsing for upstream RecordBatch::custom_metadata

Pin arrow-rs to rustyconover/arrow-rs#feat/recordbatch-custom-metadata
(apache/arrow-rs#9445) and rewrite vgi-rpc/src/wire.rs as a thin wrapper
around arrow_ipc::reader::StreamReader / writer::StreamWriter. Per-batch
metadata now travels on RecordBatch directly via with_custom_metadata()
/ custom_metadata(); the Metadata alias becomes HashMap<String, String>
and the ReadBatch wrapper is gone. relax_nullability flips
with_skip_validation(true) on the inner reader since upstream validates
before our schema rewrap.

Also bundles in-progress conformance worker, http, and arrow_type
changes that were already pending on the branch.

Conformance: 723/723 across pipe/subprocess/http/unix/externalize.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rustyconover

Cargo.lock

@@ -35,6 +35,17 @@ bincode = "1.3"
 syn = { version = "2", features = ["full", "extra-traits"] }
 quote = "1"
 proc-macro2 = "1"
+chrono = { version = "0.4", default-features = false, features = ["std"] }
+rust_decimal = { version = "1", default-features = false }
+
+[patch.crates-io]
+arrow-array = { git = "https://github.com/rustyconover/arrow-rs.git", branch = "feat/recordbatch-custom-metadata" }
+arrow-buffer = { git = "https://github.com/rustyconover/arrow-rs.git", branch = "feat/recordbatch-custom-metadata" }
+arrow-data = { git = "https://github.com/rustyconover/arrow-rs.git", branch = "feat/recordbatch-custom-metadata" }
+arrow-ipc = { git = "https://github.com/rustyconover/arrow-rs.git", branch = "feat/recordbatch-custom-metadata" }
+arrow-schema = { git = "https://github.com/rustyconover/arrow-rs.git", branch = "feat/recordbatch-custom-metadata" }
+arrow-select = { git = "https://github.com/rustyconover/arrow-rs.git", branch = "feat/recordbatch-custom-metadata" }
+arrow-cast = { git = "https://github.com/rustyconover/arrow-rs.git", branch = "feat/recordbatch-custom-metadata" }
 
 [profile.release]
 opt-level = 3

Cargo.toml

@@ -11,6 +11,8 @@ path = "src/main.rs"
 
 [dependencies]
 vgi-rpc = { path = "../vgi-rpc", features = ["http"] }
+vgi-rpc-s3 = { path = "../vgi-rpc-s3" }
+reqwest = { version = "0.12", default-features = false, features = ["blocking", "rustls-tls", "json"] }
 arrow-array.workspace = true
 arrow-buffer.workspace = true
 arrow-data.workspace = true
@@ -26,3 +28,4 @@ parking_lot = "0.12"
 serde.workspace = true
 serde_json = "1"
 ctrlc = "3"
+chrono.workspace = true

conformance-worker/Cargo.toml

@@ -12,19 +12,32 @@ mod params;
 mod streams;
 mod types;
 mod unary;
+mod wide_types;
 
 use std::sync::Arc;
 
 use vgi_rpc::RpcServer;
 
 /// Build an `RpcServer` with all conformance methods registered.
 pub fn build_server() -> RpcServer {
+    build_server_with_external(None)
+}
+
+/// Build an `RpcServer` with all conformance methods registered, optionally
+/// wired to an external-location config (used by `TestExternalLocation`).
+pub fn build_server_with_external(
+    external: Option<vgi_rpc::external::ExternalLocationConfig>,
+) -> RpcServer {
     let mut builder = RpcServer::builder()
         .server_id("rust-conf-0001")
         .protocol_name("ConformanceService")
         .server_version("rust-conformance-0.2.0")
         .enable_describe(true);
 
+    if let Some(cfg) = external {
+        builder = builder.with_external_location(cfg);
+    }
+
     // When VGI_ACCESS_LOG is set, emit JSON-per-call access records to that
     // file. Used for manual validation against vgi_rpc.access_log_conformance.
     if let Ok(path) = std::env::var("VGI_ACCESS_LOG") {

conformance-worker/src/conformance/mod.rs

@@ -57,18 +57,17 @@ impl Point {
         let mut buf = Vec::new();
         {
             let mut w = StreamWriter::new(&mut buf, &schema)?;
-            w.write(&batch, None)?;
+            w.write(&batch)?;
             w.finish()?;
         }
         Ok(buf)
     }
 
     pub fn deserialize_ipc(bytes: &[u8]) -> Result<Self> {
         let mut r = StreamReader::new(bytes)?;
-        let rb = r
+        let batch = r
             .read_next()?
             .ok_or_else(|| RpcError::type_error("empty Point IPC stream"))?;
-        let batch = rb.batch;
         let schema = batch.schema();
         let x_idx = schema
             .index_of("x")
@@ -338,18 +337,17 @@ impl AllTypes {
         let mut buf = Vec::new();
         {
             let mut w = StreamWriter::new(&mut buf, &all_types_schema())?;
-            w.write(&batch, None)?;
+            w.write(&batch)?;
             w.finish()?;
         }
         Ok(buf)
     }
 
     pub fn deserialize_ipc(bytes: &[u8]) -> Result<Self> {
         let mut r = StreamReader::new(bytes)?;
-        let rb = r
+        let batch = r
             .read_next()?
             .ok_or_else(|| RpcError::type_error("empty AllTypes IPC"))?;
-        let batch = rb.batch;
         AllTypes::from_record_batch(&batch)
     }
 
@@ -679,7 +677,7 @@ pub fn serialize_bounding_box_ipc(top: &Point, bot: &Point, label: &str) -> Resu
     let mut buf = Vec::new();
     {
         let mut w = StreamWriter::new(&mut buf, &schema)?;
-        w.write(&batch, None)?;
+        w.write(&batch)?;
         w.finish()?;
     }
     Ok(buf)
@@ -690,5 +688,5 @@ pub fn deserialize_bounding_box_ipc(bytes: &[u8]) -> Result<(Point, Point, Strin
     let rb = r
         .read_next()?
         .ok_or_else(|| RpcError::type_error("empty BoundingBox IPC"))?;
-    bounding_box_from_batch(&rb.batch)
+    bounding_box_from_batch(&rb)
 }