Ensure PARALLEL statements use indexes (#2206)

lib/src/dbs/channel.rs

@@ -4,11 +4,13 @@ use crate::dbs::Operable;
 use crate::dbs::Options;
 use crate::dbs::Statement;
 use crate::err::Error;
+use crate::idx::planner::plan::Plan;
 use crate::key::graph;
 use crate::key::thing;
 use crate::sql::dir::Dir;
 use crate::sql::thing::Thing;
 use crate::sql::value::Value;
+use crate::sql::{Edges, Range, Table};
 use channel::Sender;
 use std::ops::Bound;
 
@@ -22,16 +24,44 @@ impl Iterable {
 		chn: Sender<(Option<Thing>, Operable)>,
 	) -> Result<(), Error> {
 		if ctx.is_ok() {
-			// Clone transaction
-			let txn = ctx.try_clone_transaction()?;
 			match self {
-				Iterable::Value(v) => {
+				Iterable::Value(v) => Self::channel_value(ctx, opt, v, chn).await?,
+				Iterable::Thing(v) => Self::channel_thing(ctx, opt, v, chn).await?,
+				Iterable::Table(v) => Self::channel_table(ctx, opt, v, chn).await?,
+				Iterable::Range(v) => Self::channel_range(ctx, opt, v, chn).await?,
+				Iterable::Edges(e) => Self::channel_edge(ctx, opt, e, chn).await?,
+				Iterable::Index(t, p) => Self::channel_index(ctx, opt, t, p, chn).await?,
+				Iterable::Mergeable(v, o) => Self::channel_mergeable(ctx, opt, v, o, chn).await?,
+				Iterable::Relatable(f, v, w) => {
+					Self::channel_relatable(ctx, opt, f, v, w, chn).await?
+				}
+			}
+		}
+		Ok(())
+	}
+
+	async fn channel_value(
+		_ctx: &Context<'_>,
+		_opt: &Options,
+		v: Value,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
 		// Pass the value through
 		let val = Operable::Value(v);
 		// Process the document record
 		chn.send((None, val)).await?;
+		// Everything ok
+		Ok(())
 	}
-				Iterable::Thing(v) => {
+
+	async fn channel_thing(
+		ctx: &Context<'_>,
+		opt: &Options,
+		v: Thing,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
+		// Clone transaction
+		let txn = ctx.try_clone_transaction()?;
 		// Check that the table exists
 		txn.lock().await.check_ns_db_tb(opt.ns(), opt.db(), &v.tb, opt.strict).await?;
 		// Fetch the data from the store
@@ -42,10 +72,24 @@ impl Iterable {
 			Some(v) => Value::from(v),
 			None => Value::None,
 		});
+		// Get the optional query executor
+		let mut child_ctx = Context::new(ctx);
+		child_ctx.add_thing(&v);
 		// Process the document record
 		chn.send((Some(v), val)).await?;
+		// Everything ok
+		Ok(())
 	}
-				Iterable::Mergeable(v, o) => {
+
+	async fn channel_mergeable(
+		ctx: &Context<'_>,
+		opt: &Options,
+		v: Thing,
+		o: Value,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
+		// Clone transaction
+		let txn = ctx.try_clone_transaction()?;
 		// Check that the table exists
 		txn.lock().await.check_ns_db_tb(opt.ns(), opt.db(), &v.tb, opt.strict).await?;
 		// Fetch the data from the store
@@ -58,10 +102,25 @@ impl Iterable {
 		};
 		// Create a new operable value
 		let val = Operable::Mergeable(x, o);
+		// Create a new context to process the operable
+		let mut child_ctx = Context::new(ctx);
+		child_ctx.add_thing(&v);
 		// Process the document record
 		chn.send((Some(v), val)).await?;
+		// Everything ok
+		Ok(())
 	}
-				Iterable::Relatable(f, v, w) => {
+
+	async fn channel_relatable(
+		ctx: &Context<'_>,
+		opt: &Options,
+		f: Thing,
+		v: Thing,
+		w: Thing,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
+		// Clone transaction
+		let txn = ctx.try_clone_transaction()?;
 		// Check that the table exists
 		txn.lock().await.check_ns_db_tb(opt.ns(), opt.db(), &v.tb, opt.strict).await?;
 		// Fetch the data from the store
@@ -74,10 +133,23 @@ impl Iterable {
 		};
 		// Create a new operable value
 		let val = Operable::Relatable(f, x, w);
+		// Create the child context
+		let mut child_ctx = Context::new(ctx);
+		child_ctx.add_thing(&v);
 		// Process the document record
 		chn.send((Some(v), val)).await?;
+		// Everything ok
+		Ok(())
 	}
-				Iterable::Table(v) => {
+
+	async fn channel_table(
+		ctx: &Context<'_>,
+		opt: &Options,
+		v: Table,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
+		// Clone transaction
+		let txn = ctx.try_clone_transaction()?;
 		// Check that the table exists
 		txn.lock().await.check_ns_db_tb(opt.ns(), opt.db(), &v, opt.strict).await?;
 		// Prepare the start and end keys
@@ -109,10 +181,6 @@ impl Iterable {
 			if !res.is_empty() {
 				// Get total results
 				let n = res.len();
-							// Exit when settled
-							if n == 0 {
-								break;
-							}
 				// Loop over results
 				for (i, (k, v)) in res.into_iter().enumerate() {
 					// Check the context
@@ -129,6 +197,8 @@ impl Iterable {
 					let rid = Thing::from((key.tb, key.id));
 					// Create a new operable value
 					let val = Operable::Value(val);
+					let mut child_ctx = Context::new(ctx);
+					child_ctx.add_thing(&rid);
 					// Process the record
 					chn.send((Some(rid), val)).await?;
 				}
@@ -136,19 +206,26 @@ impl Iterable {
 			}
 			break;
 		}
+		// Everything ok
+		Ok(())
 	}
-				Iterable::Range(v) => {
+
+	async fn channel_range(
+		ctx: &Context<'_>,
+		opt: &Options,
+		v: Range,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
+		// Clone transaction
+		let txn = ctx.try_clone_transaction()?;
 		// Check that the table exists
 		txn.lock().await.check_ns_db_tb(opt.ns(), opt.db(), &v.tb, opt.strict).await?;
 		// Prepare the range start key
 		let beg = match &v.beg {
 			Bound::Unbounded => thing::prefix(opt.ns(), opt.db(), &v.tb),
-						Bound::Included(id) => {
+			Bound::Included(id) => thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap(),
-							thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap()
-						}
 			Bound::Excluded(id) => {
-							let mut key =
+				let mut key = thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap();
-								thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap();
 				key.push(0x00);
 				key
 			}
@@ -156,12 +233,9 @@ impl Iterable {
 		// Prepare the range end key
 		let end = match &v.end {
 			Bound::Unbounded => thing::suffix(opt.ns(), opt.db(), &v.tb),
-						Bound::Excluded(id) => {
+			Bound::Excluded(id) => thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap(),
-							thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap()
-						}
 			Bound::Included(id) => {
-							let mut key =
+				let mut key = thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap();
-								thing::new(opt.ns(), opt.db(), &v.tb, id).encode().unwrap();
 				key.push(0x00);
 				key
 			}
@@ -192,10 +266,6 @@ impl Iterable {
 			if !res.is_empty() {
 				// Get total results
 				let n = res.len();
-							// Exit when settled
-							if n == 0 {
-								break;
-							}
 				// Loop over results
 				for (i, (k, v)) in res.into_iter().enumerate() {
 					// Check the context
@@ -210,6 +280,8 @@ impl Iterable {
 					let key: crate::key::thing::Thing = (&k).into();
 					let val: crate::sql::value::Value = (&v).into();
 					let rid = Thing::from((key.tb, key.id));
+					let mut ctx = Context::new(ctx);
+					ctx.add_thing(&rid);
 					// Create a new operable value
 					let val = Operable::Value(val);
 					// Process the record
@@ -219,8 +291,16 @@ impl Iterable {
 			}
 			break;
 		}
+		// Everything ok
+		Ok(())
 	}
-				Iterable::Edges(e) => {
+
+	async fn channel_edge(
+		ctx: &Context<'_>,
+		opt: &Options,
+		e: Edges,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
 		// Pull out options
 		let ns = opt.ns();
 		let db = opt.db();
@@ -304,13 +384,13 @@ impl Iterable {
 					None => {
 						let min = beg.clone();
 						let max = end.clone();
-									txn.clone().lock().await.scan(min..max, 1000).await?
+						ctx.try_clone_transaction()?.lock().await.scan(min..max, 1000).await?
 					}
 					Some(ref mut beg) => {
 						beg.push(0x00);
 						let min = beg.clone();
 						let max = end.clone();
-									txn.clone().lock().await.scan(min..max, 1000).await?
+						ctx.try_clone_transaction()?.lock().await.scan(min..max, 1000).await?
 					}
 				};
 				// If there are key-value entries then fetch them
@@ -335,8 +415,10 @@ impl Iterable {
 						let gra: crate::key::graph::Graph = (&k).into();
 						// Fetch the data from the store
 						let key = thing::new(opt.ns(), opt.db(), gra.ft, &gra.fk);
-									let val = txn.clone().lock().await.get(key).await?;
+						let val = ctx.try_clone_transaction()?.lock().await.get(key).await?;
 						let rid = Thing::from((gra.ft, gra.fk));
+						let mut ctx = Context::new(ctx);
+						ctx.add_thing(&rid);
 						// Parse the data from the store
 						let val = Operable::Value(match val {
 							Some(v) => Value::from(v),
@@ -350,12 +432,66 @@ impl Iterable {
 				break;
 			}
 		}
-				}
+		// Everything ok
-				Iterable::Index(_t, _p) => {
-					todo!()
-				}
-			}
-		}
 		Ok(())
 	}
+
+	async fn channel_index(
+		ctx: &Context<'_>,
+		opt: &Options,
+		table: Table,
+		plan: Plan,
+		chn: Sender<(Option<Thing>, Operable)>,
+	) -> Result<(), Error> {
+		let txn = ctx.try_clone_transaction()?;
+		// Check that the table exists
+		txn.lock().await.check_ns_db_tb(opt.ns(), opt.db(), &table.0, opt.strict).await?;
+		let exe = ctx.get_query_executor(&table.0);
+		if let Some(exe) = exe {
+			let mut iterator = plan.new_iterator(opt, &txn, exe).await?;
+			let mut things = iterator.next_batch(&txn, 1000).await?;
+			while !things.is_empty() {
+				// Check if the context is finished
+				if ctx.is_done() {
+					break;
+				}
+
+				for (thing, doc_id) in things {
+					// Check the context
+					if ctx.is_done() {
+						break;
+					}
+
+					// If the record is from another table we can skip
+					if !thing.tb.eq(table.as_str()) {
+						continue;
+					}
+
+					// Fetch the data from the store
+					let key = thing::new(opt.ns(), opt.db(), &table.0, &thing.id);
+					let val = txn.lock().await.get(key.clone()).await?;
+					let rid = Thing::from((key.tb, key.id));
+					let mut ctx = Context::new(ctx);
+					ctx.add_thing(&rid);
+					ctx.add_doc_id(doc_id);
+					// Parse the data from the store
+					let val = Operable::Value(match val {
+						Some(v) => Value::from(v),
+						None => Value::None,
+					});
+					// Process the document record
+					chn.send((Some(rid), val)).await?;
+				}
+
+				// Collect the next batch of ids
+				things = iterator.next_batch(&txn, 1000).await?;
+			}
+			// Everything ok
+			Ok(())
+		} else {
+			Err(Error::QueryNotExecutedDetail {
+				message: "The QueryExecutor has not been found.".to_string(),
+			})
+		}
+	}
 }

lib/src/dbs/iterate.rs

@@ -26,17 +26,15 @@ impl Iterable {
 			match self {
 				Iterable::Value(v) => Self::iterate_value(ctx, opt, stm, v, ite).await,
 				Iterable::Thing(v) => Self::iterate_thing(ctx, opt, stm, v, ite).await?,
-				Iterable::Mergeable(v, o) => {
-					Self::iterate_mergeable(ctx, opt, stm, v, o, ite).await?;
-				}
-				Iterable::Relatable(f, v, w) => {
-					Self::iterate_relatable(ctx, opt, stm, f, v, w, ite).await?
-				}
 				Iterable::Table(v) => Self::iterate_table(ctx, opt, stm, v, ite).await?,
 				Iterable::Range(v) => Self::iterate_range(ctx, opt, stm, v, ite).await?,
 				Iterable::Edges(e) => Self::iterate_edge(ctx, opt, stm, e, ite).await?,
-				Iterable::Index(t, p) => {
+				Iterable::Index(t, p) => Self::iterate_index(ctx, opt, stm, t, p, ite).await?,
-					Self::iterate_index(ctx, opt, stm, t, p, ite).await?;
+				Iterable::Mergeable(v, o) => {
+					Self::iterate_mergeable(ctx, opt, stm, v, o, ite).await?
+				}
+				Iterable::Relatable(f, v, w) => {
+					Self::iterate_relatable(ctx, opt, stm, f, v, w, ite).await?
 				}
 			}
 		}
@@ -80,6 +78,7 @@ impl Iterable {
 		child_ctx.add_thing(&v);
 		// Process the document record
 		ite.process(&child_ctx, opt, stm, val).await;
+		// Everything ok
 		Ok(())
 	}
 
@@ -110,6 +109,7 @@ impl Iterable {
 		child_ctx.add_thing(&v);
 		// Process the document record
 		ite.process(&child_ctx, opt, stm, val).await;
+		// Everything ok
 		Ok(())
 	}
 
@@ -141,6 +141,7 @@ impl Iterable {
 		child_ctx.add_thing(&v);
 		// Process the document record
 		ite.process(&child_ctx, opt, stm, val).await;
+		// Everything ok
 		Ok(())
 	}
 
@@ -209,6 +210,7 @@ impl Iterable {
 			}
 			break;
 		}
+		// Everything ok
 		Ok(())
 	}
 
@@ -294,6 +296,7 @@ impl Iterable {
 			}
 			break;
 		}
+		// Everything ok
 		Ok(())
 	}
 
@@ -435,6 +438,7 @@ impl Iterable {
 				break;
 			}
 		}
+		// Everything ok
 		Ok(())
 	}
 
@@ -446,6 +450,7 @@ impl Iterable {
 		plan: Plan,
 		ite: &mut Iterator,
 	) -> Result<(), Error> {
+		// Clone transaction
 		let txn = ctx.try_clone_transaction()?;
 		// Check that the table exists
 		txn.lock().await.check_ns_db_tb(opt.ns(), opt.db(), &table.0, opt.strict).await?;
@@ -489,6 +494,7 @@ impl Iterable {
 				// Collect the next batch of ids
 				things = iterator.next_batch(&txn, 1000).await?;
 			}
+			// Everything ok
 			Ok(())
 		} else {
 			Err(Error::QueryNotExecutedDetail {