Implement record storing and fetching

Cargo.lock

@@ -460,12 +460,12 @@ dependencies = [
 
 [[package]]
 name = "echodb"
-version = "0.2.0"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3b15c1b33dc9bbe3d1dc8beadc8b477d524ad2534c6cebbddb1b1b0ba882825"
+checksum = "e7717db5a8a5af149e6652cce851908dbfe05bdd4183c6d2f382e312ef3dc523"
 dependencies = [
  "arc-swap",
- "im",
+ "imbl",
  "thiserror",
  "tokio",
 ]
@@ -1012,13 +1012,13 @@ dependencies = [
 ]
 
 [[package]]
-name = "im"
-version = "15.0.0"
+name = "imbl"
+version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "111c1983f3c5bb72732df25cddacee9b546d08325fb584b5ebd38148be7b0246"
+checksum = "543682c9082b25e63d03b5acbd65ad111fd49dd93e70843e5175db4ff81d606b"
 dependencies = [
  "bitmaps",
- "rand_core 0.5.1",
+ "rand_core 0.6.3",
  "rand_xoshiro",
  "sized-chunks",
  "typenum",
@@ -1795,11 +1795,11 @@ dependencies = [
 
 [[package]]
 name = "rand_xoshiro"
-version = "0.4.0"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9fcdd2e881d02f1d9390ae47ad8e5696a9e4be7b547a1da2afbc61973217004"
+checksum = "6f97cdb2a36ed4183de61b2f824cc45c9f1037f28afe0a322e9fff4c108b5aaa"
 dependencies = [
- "rand_core 0.5.1",
+ "rand_core 0.6.3",
 ]
 
 [[package]]
@@ -2250,9 +2250,9 @@ dependencies = [
 
 [[package]]
 name = "surrealdb-derive"
-version = "0.1.1"
+version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "71202396222f43d24f42963f6f8e8b6626c90c56ac53a912257ea88707876978"
+checksum = "6045ae5b030a2485f0862dd735eff304b72a7cef3d4d541b3a46a45565737548"
 dependencies = [
  "quote",
  "syn",

lib/Cargo.toml

@@ -19,9 +19,9 @@ async-recursion = "1.0.0"
 byteorder = "1.4.3"
 chrono = { version = "0.4.19", features = ["serde"] }
 dec = { version = "1.21.0", package = "rust_decimal", features = ["maths", "serde-float"] }
-derive = { version = "0.1.1", package = "surrealdb-derive" }
+derive = { version = "0.1.2", package = "surrealdb-derive" }
 dmp = "0.1.1"
-echodb = { version = "0.2.0", optional = true }
+echodb = { version = "0.2.1", optional = true }
 futures = "0.3.21"
 fuzzy-matcher = "0.3.7"
 geo = { version = "0.18.0", features = ["use-serde"] }

lib/src/dbs/channel.rs

@@ -3,6 +3,8 @@ use crate::dbs::Options;
 use crate::dbs::Runtime;
 use crate::dbs::Transaction;
 use crate::err::Error;
+use crate::key;
+use crate::key::thing;
 use crate::sql::array::Array;
 use crate::sql::model::Model;
 use crate::sql::table::Table;
@@ -94,11 +96,20 @@ impl Model {
 impl Thing {
 	pub async fn process(
 		self,
-		_ctx: &Runtime,
-		_opt: &Options,
-		_txn: &Transaction,
-		_chn: &Sender<(Option<Thing>, Value)>,
+		ctx: &Runtime,
+		opt: &Options,
+		txn: &Transaction,
+		chn: &Sender<(Option<Thing>, Value)>,
 	) -> Result<(), Error> {
+		if ctx.is_ok() {
+			let key = thing::new(opt.ns(), opt.db(), &self.tb, &self.id);
+			let val = txn.clone().lock().await.get(key).await?;
+			let val = match val {
+				Some(v) => Value::from(v),
+				None => Value::None,
+			};
+			chn.send((Some(self), val)).await?;
+		}
 		Ok(())
 	}
 }
@@ -106,11 +117,54 @@ impl Thing {
 impl Table {
 	pub async fn process(
 		self,
-		_ctx: &Runtime,
-		_opt: &Options,
-		_txn: &Transaction,
-		_chn: &Sender<(Option<Thing>, Value)>,
+		ctx: &Runtime,
+		opt: &Options,
+		txn: &Transaction,
+		chn: &Sender<(Option<Thing>, Value)>,
 	) -> Result<(), Error> {
+		if ctx.is_ok() {
+			let beg = thing::new(opt.ns(), opt.db(), &self.name, key::PREFIX);
+			let end = thing::new(opt.ns(), opt.db(), &self.name, key::SUFFIX);
+			let mut nxt: Option<Vec<u8>> = None;
+			loop {
+				if ctx.is_ok() {
+					let res = match nxt {
+						None => {
+							let min = beg.encode()?;
+							let max = end.encode()?;
+							txn.clone().lock().await.scan(min..max, 1000).await?
+						}
+						Some(ref mut beg) => {
+							beg.push(0);
+							let min = beg.clone();
+							let max = end.encode()?;
+							txn.clone().lock().await.scan(min..max, 1000).await?
+						}
+					};
+					if !res.is_empty() {
+						// Get total results
+						let n = res.len() - 1;
+						// Loop over results
+						for (i, (k, v)) in res.into_iter().enumerate() {
+							if ctx.is_ok() {
+								// Ready the next
+								if i == n {
+									nxt = Some(k.clone());
+								}
+								// Parse the key-value
+								let k: crate::key::thing::Thing = (&k).into();
+								let v: crate::sql::value::Value = (&v).into();
+								let t = Thing::from((k.tb, k.id));
+								// Process the record
+								chn.send((Some(t), v)).await?;
+							}
+						}
+						continue;
+					}
+				}
+				break;
+			}
+		}
 		Ok(())
 	}
 }

lib/src/dbs/iterate.rs

@@ -4,6 +4,8 @@ use crate::dbs::Options;
 use crate::dbs::Runtime;
 use crate::dbs::Transaction;
 use crate::err::Error;
+use crate::key;
+use crate::key::thing;
 use crate::sql::array::Array;
 use crate::sql::model::Model;
 use crate::sql::table::Table;
@@ -97,11 +99,20 @@ impl Model {
 impl Thing {
 	pub async fn iterate(
 		self,
-		_ctx: &Runtime,
-		_opt: &Options,
-		_txn: &Transaction,
-		_ite: &mut Iterator,
+		ctx: &Runtime,
+		opt: &Options,
+		txn: &Transaction,
+		ite: &mut Iterator,
 	) -> Result<(), Error> {
+		if ctx.is_ok() {
+			let key = thing::new(opt.ns(), opt.db(), &self.tb, &self.id);
+			let val = txn.clone().lock().await.get(key).await?;
+			let val = match val {
+				Some(v) => Value::from(v),
+				None => Value::None,
+			};
+			ite.process(ctx, opt, txn, Some(self), val).await;
+		}
 		Ok(())
 	}
 }
@@ -109,11 +120,54 @@ impl Thing {
 impl Table {
 	pub async fn iterate(
 		self,
-		_ctx: &Runtime,
-		_opt: &Options,
-		_txn: &Transaction,
-		_ite: &mut Iterator,
+		ctx: &Runtime,
+		opt: &Options,
+		txn: &Transaction,
+		ite: &mut Iterator,
 	) -> Result<(), Error> {
+		if ctx.is_ok() {
+			let beg = thing::new(opt.ns(), opt.db(), &self.name, key::PREFIX);
+			let end = thing::new(opt.ns(), opt.db(), &self.name, key::SUFFIX);
+			let mut nxt: Option<Vec<u8>> = None;
+			loop {
+				if ctx.is_ok() {
+					let res = match nxt {
+						None => {
+							let min = beg.encode()?;
+							let max = end.encode()?;
+							txn.clone().lock().await.scan(min..max, 1000).await?
+						}
+						Some(ref mut beg) => {
+							beg.push(0);
+							let min = beg.clone();
+							let max = end.encode()?;
+							txn.clone().lock().await.scan(min..max, 1000).await?
+						}
+					};
+					if !res.is_empty() {
+						// Get total results
+						let n = res.len() - 1;
+						// Loop over results
+						for (i, (k, v)) in res.into_iter().enumerate() {
+							if ctx.is_ok() {
+								// Ready the next
+								if i == n {
+									nxt = Some(k.clone());
+								}
+								// Parse the key-value
+								let k: crate::key::thing::Thing = (&k).into();
+								let v: crate::sql::value::Value = (&v).into();
+								let t = Thing::from((k.tb, k.id));
+								// Process the record
+								ite.process(ctx, opt, txn, Some(t), v).await;
+							}
+						}
+						continue;
+					}
+				}
+				break;
+			}
+		}
 		Ok(())
 	}
 }

lib/src/doc/store.rs

@@ -4,15 +4,19 @@ use crate::dbs::Statement;
 use crate::dbs::Transaction;
 use crate::doc::Document;
 use crate::err::Error;
+use crate::key::thing;
 
 impl<'a> Document<'a> {
 	pub async fn store(
 		&self,
 		_ctx: &Runtime,
-		_opt: &Options,
-		_txn: &Transaction,
+		opt: &Options,
+		txn: &Transaction,
 		_stm: &Statement,
 	) -> Result<(), Error> {
+		let md = self.id.as_ref().unwrap();
+		let key = thing::new(opt.ns(), opt.db(), &md.tb, &md.id);
+		txn.clone().lock().await.set(key, self).await?;
 		Ok(())
 	}
 }

lib/src/sql/thing.rs

@@ -13,6 +13,15 @@ pub struct Thing {
 	pub id: String,
 }
 
+impl From<(String, String)> for Thing {
+	fn from(v: (String, String)) -> Self {
+		Thing {
+			tb: v.0,
+			id: v.1,
+		}
+	}
+}
+
 impl fmt::Display for Thing {
 	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
 		let t = escape(&self.tb, &val_char, "`");