Revert "[ABI] Remove the special first element of iterator (#420)"

This reverts commit 469dff6.

Author: cloutiertyler

crates/bindings-csharp/Runtime/Runtime.cs

@@ -140,18 +140,20 @@ public void Reset()
 
     public class RawTableIter : IEnumerable<byte[]>
     {
-        private readonly uint tableId;
-        private readonly byte[]? filterBytes;
+        public readonly byte[] Schema;
+
+        private readonly IEnumerator<byte[]> iter;
 
         public RawTableIter(uint tableId, byte[]? filterBytes = null)
         {
-            this.tableId = tableId;
-            this.filterBytes = filterBytes;
+            iter = new BufferIter(tableId, filterBytes);
+            iter.MoveNext();
+            Schema = iter.Current;
         }
 
         public IEnumerator<byte[]> GetEnumerator()
         {
-            return new BufferIter(tableId, filterBytes);
+            return iter;
         }
 
         IEnumerator IEnumerable.GetEnumerator()

crates/bindings-csharp/Runtime/bindings.c

@@ -100,7 +100,7 @@ static MonoArray* stdb_buffer_consume(Buffer buf);
 //   return out;
 // }
 
-#define STDB_IMPORT_MODULE_MINOR(minor) "spacetime_7." #minor
+#define STDB_IMPORT_MODULE_MINOR(minor) "spacetime_6." #minor
 #define STDB_IMPORT_MODULE STDB_IMPORT_MODULE_MINOR(0)
 
 __attribute__((import_module(STDB_IMPORT_MODULE),

crates/bindings-sys/src/lib.rs

@@ -24,7 +24,7 @@ pub mod raw {
     // on. Any non-breaking additions to the abi surface should be put in a new `extern {}` block
     // with a module identifier with a minor version 1 above the previous highest minor version.
     // For breaking changes, all functions should be moved into one new `spacetime_X.0` block.
-    #[link(wasm_import_module = "spacetime_7.0")]
+    #[link(wasm_import_module = "spacetime_6.0")]
     extern "C" {
         /*
         /// Create a table with `name`, a UTF-8 slice in WASM memory lasting `name_len` bytes,

crates/bindings/src/lib.rs

@@ -198,11 +198,52 @@ pub fn delete_by_col_eq(table_id: TableId, col_id: u8, value: &impl Serialize) -
     })
 }
 
-/// A table iterator which yields values of the `TableType` corresponding to the table.
-type TableTypeTableIter<T> = RawTableIter<TableTypeBufferDeserialize<T>>;
+/*
+pub fn delete_pk(table_id: u32, primary_key: &PrimaryKey) -> Result<()> {
+    with_row_buf(|bytes| {
+        primary_key.encode(bytes);
+        sys::delete_pk(table_id, bytes)
+    })
+}
+
+pub fn delete_filter<F: Fn(&ProductValue) -> bool>(table_id: u32, f: F) -> Result<usize> {
+    with_row_buf(|bytes| {
+        let mut count = 0;
+        for tuple_value in pv_table_iter(table_id, None)? {
+            if f(&tuple_value) {
+                count += 1;
+                bytes.clear();
+                tuple_value.encode(bytes);
+                sys::delete_value(table_id, bytes)?;
+            }
+        }
+        Ok(count)
+    })
+}
+
+pub fn delete_range(table_id: u32, col_id: u8, range: Range<AlgebraicValue>) -> Result<u32> {
+    with_row_buf(|bytes| {
+        range.start.encode(bytes);
+        let mid = bytes.len();
+        range.end.encode(bytes);
+        let (range_start, range_end) = bytes.split_at(mid);
+        sys::delete_range(table_id, col_id.into(), range_start, range_end)
+    })
+}
+*/
 
-// Get the iterator for this table with an optional filter,
-fn table_iter<T: TableType>(table_id: TableId, filter: Option<spacetimedb_lib::filter::Expr>) -> Result<TableIter<T>> {
+//
+// fn page_table(table_id : u32, pager_token : u32, read_entries : u32) {
+//
+// }
+
+// Get the buffer iterator for this table,
+// with an optional filter,
+// and return it and its decoded `ProductType` schema.
+fn buffer_table_iter(
+    table_id: u32,
+    filter: Option<spacetimedb_lib::filter::Expr>,
+) -> Result<(BufferIter, ProductType)> {
     // Decode the filter, if any.
     let filter = filter
         .as_ref()
@@ -211,8 +252,35 @@ fn table_iter<T: TableType>(table_id: TableId, filter: Option<spacetimedb_lib::f
         .expect("Couldn't decode the filter query");
 
     // Create the iterator.
-    let iter = sys::iter(table_id, filter.as_deref())?;
+    let mut iter = sys::iter(table_id, filter.as_deref())?;
+
+    // First item is an encoded schema.
+    let schema_raw = iter
+        .next()
+        .expect("Missing schema")
+        .expect("Failed to get schema")
+        .read();
+    let schema = decode_schema(&mut &schema_raw[..]).expect("Could not decode schema");
+
+    Ok((iter, schema))
+}
+
+/// A table iterator which yields `ProductValue`s.
+// type ProductValueTableIter = RawTableIter<ProductValue, ProductValueBufferDeserialize>;
+
+// fn pv_table_iter(table_id: u32, filter: Option<spacetimedb_lib::filter::Expr>) -> Result<ProductValueTableIter> {
+//     let (iter, schema) = buffer_table_iter(table_id, filter)?;
+//     let deserializer = ProductValueBufferDeserialize::new(schema);
+//     Ok(RawTableIter::new(iter, deserializer))
+// }
+
+/// A table iterator which yields values of the `TableType` corresponding to the table.
+type TableTypeTableIter<T> = RawTableIter<TableTypeBufferDeserialize<T>>;
 
+fn table_iter<T: TableType>(table_id: u32, filter: Option<spacetimedb_lib::filter::Expr>) -> Result<TableIter<T>> {
+    // The TableType deserializer doesn't need the schema, as we have type-directed
+    // dispatch to deserialize any given `TableType`.
+    let (iter, _schema) = buffer_table_iter(table_id, filter)?;
     let deserializer = TableTypeBufferDeserialize::new();
     Ok(RawTableIter::new(iter, deserializer).into())
 }

crates/core/src/host/instance_env.rs

@@ -50,13 +50,9 @@ impl BufWriter for ChunkedWriter {
 }
 
 impl ChunkedWriter {
-    /// Flushes the data collected in the scratch space if it's larger than our
-    /// chunking threshold.
-    pub fn flush(&mut self) {
-        // For now, just send buffers over a certain fixed size.
-        const ITER_CHUNK_SIZE: usize = 64 * 1024;
-
-        if self.scratch_space.len() > ITER_CHUNK_SIZE {
+    /// Flushes the currently populated part of the scratch space as a new chunk.
+    pub fn force_flush(&mut self) {
+        if !self.scratch_space.is_empty() {
             // We intentionally clone here so that our scratch space is not
             // recreated with zero capacity (via `Vec::new`), but instead can
             // be `.clear()`ed in-place and reused.
@@ -70,11 +66,22 @@ impl ChunkedWriter {
         }
     }
 
+    /// Similar to [`Self::force_flush`], but only flushes if the data in the
+    /// scratch space is larger than our chunking threshold.
+    pub fn flush(&mut self) {
+        // For now, just send buffers over a certain fixed size.
+        const ITER_CHUNK_SIZE: usize = 64 * 1024;
+
+        if self.scratch_space.len() > ITER_CHUNK_SIZE {
+            self.force_flush();
+        }
+    }
+
     /// Finalises the writer and returns all the chunks.
     pub fn into_chunks(mut self) -> Vec<Box<[u8]>> {
         if !self.scratch_space.is_empty() {
-            // Avoid extra clone by just shrinking and pushing the scratch space
-            // in-place.
+            // This is equivalent to calling `force_flush`, but we avoid extra
+            // clone by just shrinking and pushing the scratch space in-place.
             self.chunks.push(self.scratch_space.into());
         }
         self.chunks
@@ -274,6 +281,10 @@ impl InstanceEnv {
         let stdb = &*self.dbic.relational_db;
         let tx = &mut *self.tx.get()?;
 
+        stdb.row_schema_for_table(tx, table_id)?.encode(&mut chunked_writer);
+        // initial chunk is expected to be schema itself, so force-flush it as a separate chunk
+        chunked_writer.force_flush();
+
         for row in stdb.iter(ctx, tx, table_id)? {
             row.view().encode(&mut chunked_writer);
             // Flush at row boundaries.
@@ -319,12 +330,18 @@ impl InstanceEnv {
             }
         }
 
+        let mut chunked_writer = ChunkedWriter::default();
+
         let stdb = &self.dbic.relational_db;
         let tx = &mut *self.tx.get()?;
 
         let schema = stdb.schema_for_table(tx, table_id)?;
         let row_type = ProductType::from(&*schema);
 
+        // write and force flush schema as it's expected to be the first individual chunk
+        row_type.encode(&mut chunked_writer);
+        chunked_writer.force_flush();
+
         let filter = filter::Expr::from_bytes(
             // TODO: looks like module typespace is currently not hooked up to instances;
             // use empty typespace for now which should be enough for primitives
@@ -342,8 +359,6 @@ impl InstanceEnv {
             _ => unreachable!("query should always return a table"),
         };
 
-        let mut chunked_writer = ChunkedWriter::default();
-
         // write all rows and flush at row boundaries
         for row in results.data {
             row.data.encode(&mut chunked_writer);

crates/core/src/host/wasmer/wasmer_module.rs

@@ -45,13 +45,13 @@ impl WasmerModule {
         WasmerModule { module, engine }
     }
 
-    pub const IMPLEMENTED_ABI: abi::VersionTuple = abi::VersionTuple::new(7, 0);
+    pub const IMPLEMENTED_ABI: abi::VersionTuple = abi::VersionTuple::new(6, 0);
 
     fn imports(&self, store: &mut Store, env: &FunctionEnv<WasmInstanceEnv>) -> Imports {
         #[allow(clippy::assertions_on_constants)]
         const _: () = assert!(WasmerModule::IMPLEMENTED_ABI.major == spacetimedb_lib::MODULE_ABI_MAJOR_VERSION);
         imports! {
-            "spacetime_7.0" => {
+            "spacetime_6.0" => {
                 "_schedule_reducer" => Function::new_typed_with_env(store, env, WasmInstanceEnv::schedule_reducer),
                 "_cancel_reducer" => Function::new_typed_with_env(store, env, WasmInstanceEnv::cancel_reducer),
                 "_delete_by_col_eq" => Function::new_typed_with_env(

crates/lib/src/lib.rs

@@ -41,7 +41,7 @@ pub use type_value::{AlgebraicValue, ProductValue};
 
 pub use spacetimedb_sats as sats;
 
-pub const MODULE_ABI_MAJOR_VERSION: u16 = 7;
+pub const MODULE_ABI_MAJOR_VERSION: u16 = 6;
 
 // if it ends up we need more fields in the future, we can split one of them in two
 #[derive(PartialEq, Eq, PartialOrd, Ord, Copy, Clone, Debug)]