test(wasm): follow current VGI producer API

Author: rustyconover

test/support/sabtable/Cargo.lock

@@ -21,7 +21,7 @@ path = "src/lib.rs"
 # staticlib builds with no heavy/C deps.
 vgi = { path = "../../../../vgi-rust/vgi", default-features = false }
 # Must match the exact vgi-rpc that `vgi` resolves to (the local checkout via
-# the [patch.crates-io] below — 0.23.0 today) so the `OutputCollector` /
+# the [patch.crates-io] below) so the `OutputCollector` /
 # `Result` / `RpcError` types are one crate instance, not two. A stale version
 # requirement here silently resolves a SECOND vgi-rpc from crates.io and the
 # build fails on mismatched `RpcError`/`TableProducer` types.
@@ -35,8 +35,8 @@ serde = { version = "1", features = ["derive"] }
 [profile.release]
 opt-level = 2
 
-# vgi (via ../../../../vgi-rust) resolves vgi-rpc to the local 0.12.0 checkout;
+# vgi (via ../../../../vgi-rust) resolves vgi-rpc to the local checkout;
 # point this standalone crate at the same source so both unify to ONE vgi-rpc
-# instance. Remove once vgi-rpc 0.12.0 is on crates.io.
+# instance.
 [patch.crates-io]
 vgi-rpc = { path = "../../../../vgi-rpc-rust/vgi-rpc" }

test/support/sabtable/Cargo.toml

@@ -115,6 +115,11 @@ impl TableProducer for CountProducer {
             .map_err(|e| RpcError::runtime_error(e.to_string()))?;
         Ok(Some(batch))
     }
+
+    fn resume_supported(&self) -> bool {
+        // The producer returns at most one batch.
+        false
+    }
 }
 
 impl TableFunction for CountTo {
@@ -167,6 +172,8 @@ impl TableProducer for EmitProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), vec![col])
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    vgi::resume_fields!(batch_idx, counter);
 }
 impl TableFunction for EmitBatches {
     fn name(&self) -> &str {
@@ -204,6 +211,11 @@ impl TableProducer for BoomProducer {
     fn next_batch(&mut self, _out: &mut OutputCollector) -> Result<Option<RecordBatch>> {
         Err(RpcError::runtime_error("boom: intentional worker error"))
     }
+
+    fn resume_supported(&self) -> bool {
+        // This fixture errors before it can emit a batch.
+        false
+    }
 }
 impl TableFunction for Boom {
     fn name(&self) -> &str {
@@ -265,6 +277,8 @@ impl TableProducer for SlowProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), vec![col])
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    vgi::resume_fields!(i);
 }
 impl TableFunction for SlowCount {
     fn name(&self) -> &str {
@@ -309,6 +323,11 @@ impl TableProducer for PeekProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), vec![col])
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    fn resume_supported(&self) -> bool {
+        // The producer returns at most one batch.
+        false
+    }
 }
 impl TableFunction for PeekMaxConcurrency {
     fn name(&self) -> &str {
@@ -353,6 +372,8 @@ impl TableProducer for ParallelProbeProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), vec![col])
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    vgi::resume_fields!(i);
 }
 impl TableFunction for ParallelProbe {
     fn name(&self) -> &str {
@@ -519,6 +540,8 @@ impl TableProducer for SabBigProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), vec![col])
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    vgi::resume_fields!(emitted);
 }
 impl TableFunction for SabBig {
     fn name(&self) -> &str {
@@ -578,6 +601,11 @@ impl TableProducer for SabCachedProducer {
     fn last_metadata(&self) -> Option<HashMap<String, String>> {
         self.meta.clone()
     }
+
+    fn resume_supported(&self) -> bool {
+        // The cache fixture deliberately returns one metadata-bearing batch.
+        false
+    }
 }
 impl TableFunction for SabCached {
     fn name(&self) -> &str {
@@ -636,6 +664,11 @@ impl TableProducer for BrowserInfoProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), cols)
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    fn resume_supported(&self) -> bool {
+        // The browser snapshot is returned in one batch.
+        false
+    }
 }
 #[cfg(target_os = "emscripten")]
 impl TableFunction for BrowserInfo {
@@ -696,6 +729,11 @@ impl TableProducer for ClientRandomProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), vec![col])
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    fn resume_supported(&self) -> bool {
+        // All requested random values are returned in one batch.
+        false
+    }
 }
 #[cfg(target_os = "emscripten")]
 impl TableFunction for ClientRandom {
@@ -760,6 +798,11 @@ impl TableProducer for ClientFetchProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), cols)
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    fn resume_supported(&self) -> bool {
+        // One fetch produces one response batch.
+        false
+    }
 }
 #[cfg(target_os = "emscripten")]
 impl TableFunction for ClientFetch {
@@ -827,6 +870,11 @@ impl TableProducer for ClientGeoProducer {
         Ok(Some(RecordBatch::try_new(self.schema.clone(), cols)
             .map_err(|e| RpcError::runtime_error(e.to_string()))?))
     }
+
+    fn resume_supported(&self) -> bool {
+        // One geolocation lookup produces one snapshot batch.
+        false
+    }
 }
 #[cfg(target_os = "emscripten")]
 impl TableFunction for ClientGeo {