Overhaul force (#3632)

core/src/dbs/executor.rs

@@ -13,6 +13,7 @@ use wasm_bindgen_futures::spawn_local as spawn;
 
 use crate::ctx::Context;
 use crate::dbs::response::Response;
+use crate::dbs::Force;
 use crate::dbs::Notification;
 use crate::dbs::Options;
 use crate::dbs::QueryType;
@@ -248,11 +249,12 @@ impl<'a> Executor<'a> {
 					stm.name.0.make_ascii_uppercase();
 					// Process the option
 					opt = match stm.name.0.as_str() {
-						"FIELDS" => opt.with_fields(stm.what),
-						"EVENTS" => opt.with_events(stm.what),
-						"TABLES" => opt.with_tables(stm.what),
 						"IMPORT" => opt.with_import(stm.what),
-						"FORCE" => opt.with_force(stm.what),
+						"FORCE" => opt.with_force(if stm.what {
+							Force::All
+						} else {
+							Force::None
+						}),
 						_ => break,
 					};
 					// Continue
@@ -492,7 +494,7 @@ mod tests {
             (Session::for_level(("NS", "DB").into(), Role::Editor).with_ns("OTHER_NS").with_db("DB"), false, "editor at database level should not be able to set options on another namespace even if the database name matches"),
             (Session::for_level(("NS", "DB").into(), Role::Viewer).with_ns("NS").with_db("DB"), false, "viewer at database level should not be able to set options on its database"),
         ];
-		let statement = "OPTION FIELDS = false";
+		let statement = "OPTION IMPORT = false";
 
 		for test in tests.iter() {
 			let (session, should_succeed, msg) = test;

core/src/dbs/options.rs

@@ -3,7 +3,9 @@ use crate::cnf;
 use crate::dbs::Notification;
 use crate::err::Error;
 use crate::iam::{Action, Auth, ResourceKind, Role};
-use crate::sql::Base;
+use crate::sql::{
+	statements::define::DefineIndexStatement, statements::define::DefineTableStatement, Base,
+};
 use channel::Sender;
 use std::sync::Arc;
 use uuid::Uuid;
@@ -31,19 +33,13 @@ pub struct Options {
 	/// Whether live queries are allowed?
 	pub live: bool,
 	/// Should we force tables/events to re-run?
-	pub force: bool,
+	pub force: Force,
 	/// Should we run permissions checks?
 	pub perms: bool,
 	/// Should we error if tables don't exist?
 	pub strict: bool,
 	/// Should we process field queries?
-	pub fields: bool,
+	pub import: bool,
-	/// Should we process event queries?
-	pub events: bool,
-	/// Should we process table queries?
-	pub tables: bool,
-	/// Should we process index queries?
-	pub indexes: bool,
 	/// Should we process function futures?
 	pub futures: bool,
 	/// Should we process variable field projections?
@@ -54,6 +50,24 @@ pub struct Options {
 	pub capabilities: Arc<Capabilities>,
 }
 
+#[derive(Clone, Debug)]
+pub enum Force {
+	All,
+	None,
+	Table(Arc<[DefineTableStatement]>),
+	Index(Arc<[DefineIndexStatement]>),
+}
+
+impl Force {
+	pub fn is_none(&self) -> bool {
+		matches!(self, Force::None)
+	}
+
+	pub fn is_forced(&self) -> bool {
+		!matches!(self, Force::None)
+	}
+}
+
 impl Default for Options {
 	fn default() -> Self {
 		Options::new()
@@ -70,12 +84,9 @@ impl Options {
 			dive: 0,
 			live: false,
 			perms: true,
-			force: false,
+			force: Force::None,
 			strict: false,
-			fields: true,
+			import: false,
-			events: true,
-			tables: true,
-			indexes: true,
 			futures: false,
 			projections: false,
 			auth_enabled: true,
@@ -161,7 +172,7 @@ impl Options {
 	}
 
 	/// Specify wether tables/events should re-run
-	pub fn with_force(mut self, force: bool) -> Self {
+	pub fn with_force(mut self, force: Force) -> Self {
 		self.force = force;
 		self
 	}
@@ -172,27 +183,9 @@ impl Options {
 		self
 	}
 
-	/// Specify if we should process fields
+	/// Specify if we are currently importing data
-	pub fn with_fields(mut self, fields: bool) -> Self {
+	pub fn with_import(mut self, import: bool) -> Self {
-		self.fields = fields;
+		self.import = import;
-		self
-	}
-
-	/// Specify if we should process event queries
-	pub fn with_events(mut self, events: bool) -> Self {
-		self.events = events;
-		self
-	}
-
-	/// Specify if we should process table queries
-	pub fn with_tables(mut self, tables: bool) -> Self {
-		self.tables = tables;
-		self
-	}
-
-	/// Specify if we should process index queries
-	pub fn with_indexes(mut self, indexes: bool) -> Self {
-		self.indexes = indexes;
 		self
 	}
 
@@ -208,14 +201,6 @@ impl Options {
 		self
 	}
 
-	/// Create a new Options object for a subquery
-	pub fn with_import(mut self, import: bool) -> Self {
-		self.fields = !import;
-		self.events = !import;
-		self.tables = !import;
-		self
-	}
-
 	/// Create a new Options object with auth enabled
 	pub fn with_auth_enabled(mut self, auth_enabled: bool) -> Self {
 		self.auth_enabled = auth_enabled;
@@ -238,13 +223,14 @@ impl Options {
 			capabilities: self.capabilities.clone(),
 			ns: self.ns.clone(),
 			db: self.db.clone(),
+			force: self.force.clone(),
 			perms,
 			..*self
 		}
 	}
 
 	/// Create a new Options object for a subquery
-	pub fn new_with_force(&self, force: bool) -> Self {
+	pub fn new_with_force(&self, force: Force) -> Self {
 		Self {
 			sender: self.sender.clone(),
 			auth: self.auth.clone(),
@@ -264,59 +250,22 @@ impl Options {
 			capabilities: self.capabilities.clone(),
 			ns: self.ns.clone(),
 			db: self.db.clone(),
+			force: self.force.clone(),
 			strict,
 			..*self
 		}
 	}
 
 	/// Create a new Options object for a subquery
-	pub fn new_with_fields(&self, fields: bool) -> Self {
+	pub fn new_with_import(&self, import: bool) -> Self {
 		Self {
 			sender: self.sender.clone(),
 			auth: self.auth.clone(),
 			capabilities: self.capabilities.clone(),
 			ns: self.ns.clone(),
 			db: self.db.clone(),
-			fields,
+			force: self.force.clone(),
-			..*self
+			import,
-		}
-	}
-
-	/// Create a new Options object for a subquery
-	pub fn new_with_events(&self, events: bool) -> Self {
-		Self {
-			sender: self.sender.clone(),
-			auth: self.auth.clone(),
-			capabilities: self.capabilities.clone(),
-			ns: self.ns.clone(),
-			db: self.db.clone(),
-			events,
-			..*self
-		}
-	}
-
-	/// Create a new Options object for a subquery
-	pub fn new_with_tables(&self, tables: bool) -> Self {
-		Self {
-			sender: self.sender.clone(),
-			auth: self.auth.clone(),
-			capabilities: self.capabilities.clone(),
-			ns: self.ns.clone(),
-			db: self.db.clone(),
-			tables,
-			..*self
-		}
-	}
-
-	/// Create a new Options object for a subquery
-	pub fn new_with_indexes(&self, indexes: bool) -> Self {
-		Self {
-			sender: self.sender.clone(),
-			auth: self.auth.clone(),
-			capabilities: self.capabilities.clone(),
-			ns: self.ns.clone(),
-			db: self.db.clone(),
-			indexes,
 			..*self
 		}
 	}
@@ -329,6 +278,7 @@ impl Options {
 			capabilities: self.capabilities.clone(),
 			ns: self.ns.clone(),
 			db: self.db.clone(),
+			force: self.force.clone(),
 			futures,
 			..*self
 		}
@@ -342,26 +292,12 @@ impl Options {
 			capabilities: self.capabilities.clone(),
 			ns: self.ns.clone(),
 			db: self.db.clone(),
+			force: self.force.clone(),
 			projections,
 			..*self
 		}
 	}
 
-	/// Create a new Options object for a subquery
-	pub fn new_with_import(&self, import: bool) -> Self {
-		Self {
-			sender: self.sender.clone(),
-			auth: self.auth.clone(),
-			capabilities: self.capabilities.clone(),
-			ns: self.ns.clone(),
-			db: self.db.clone(),
-			fields: !import,
-			events: !import,
-			tables: !import,
-			..*self
-		}
-	}
-
 	/// Create a new Options object for a subquery
 	pub fn new_with_sender(&self, sender: Sender<Notification>) -> Self {
 		Self {
@@ -369,6 +305,7 @@ impl Options {
 			capabilities: self.capabilities.clone(),
 			ns: self.ns.clone(),
 			db: self.db.clone(),
+			force: self.force.clone(),
 			sender: Some(sender),
 			..*self
 		}
@@ -397,6 +334,7 @@ impl Options {
 				capabilities: self.capabilities.clone(),
 				ns: self.ns.clone(),
 				db: self.db.clone(),
+				force: self.force.clone(),
 				dive,
 				..*self
 			})

core/src/doc/event.rs

@@ -14,12 +14,12 @@ impl<'a> Document<'a> {
 		txn: &Transaction,
 		stm: &Statement<'_>,
 	) -> Result<(), Error> {
-		// Check events
+		// Check import
-		if !opt.events {
+		if opt.import {
 			return Ok(());
 		}
-		// Check if forced
+		// Check if changed
-		if !opt.force && !self.changed() {
+		if !self.changed() {
 			return Ok(());
 		}
 		// Don't run permissions

core/src/doc/field.rs

@@ -15,8 +15,8 @@ impl<'a> Document<'a> {
 		txn: &Transaction,
 		_stm: &Statement<'_>,
 	) -> Result<(), Error> {
-		// Check fields
+		// Check import
-		if !opt.fields {
+		if opt.import {
 			return Ok(());
 		}
 		// Get the record id

core/src/doc/index.rs

@@ -1,5 +1,5 @@
 use crate::ctx::Context;
-use crate::dbs::Statement;
+use crate::dbs::{Force, Statement};
 use crate::dbs::{Options, Transaction};
 use crate::doc::{CursorDoc, Document};
 use crate::err::Error;
@@ -21,14 +21,23 @@ impl<'a> Document<'a> {
 		txn: &Transaction,
 		_stm: &Statement<'_>,
 	) -> Result<(), Error> {
-		// Check indexes
+		// Check import
-		if !opt.indexes {
+		if opt.import {
-			return Ok(());
-		}
-		// Check if forced
-		if !opt.force && !self.changed() {
 			return Ok(());
 		}
+		// Was this force targeted at a specific index?
+		let targeted_force = matches!(opt.force, Force::Index(_));
+		// Collect indexes or skip
+		let ixs = match &opt.force {
+			Force::Index(ix)
+				if ix.first().is_some_and(|ix| self.id.is_some_and(|id| ix.what.0 == id.tb)) =>
+			{
+				ix.clone()
+			}
+			Force::All => self.ix(opt, txn).await?,
+			_ if self.changed() => self.ix(opt, txn).await?,
+			_ => return Ok(()),
+		};
 		// Check if the table is a view
 		if self.tb(opt, txn).await?.drop {
 			return Ok(());
@@ -36,7 +45,7 @@ impl<'a> Document<'a> {
 		// Get the record id
 		let rid = self.id.as_ref().unwrap();
 		// Loop through all index statements
-		for ix in self.ix(opt, txn).await?.iter() {
+		for ix in ixs.iter() {
 			// Calculate old values
 			let o = build_opt_values(ctx, opt, txn, ix, &self.initial).await?;
 
@@ -44,7 +53,7 @@ impl<'a> Document<'a> {
 			let n = build_opt_values(ctx, opt, txn, ix, &self.current).await?;
 
 			// Update the index entries
-			if opt.force || o != n {
+			if targeted_force || o != n {
 				// Store all the variable and parameters required by the index operation
 				let mut ic = IndexOperation::new(opt, ix, o, n, rid);
 

core/src/doc/lives.rs

@@ -27,8 +27,8 @@ impl<'a> Document<'a> {
 		txn: &Transaction,
 		stm: &Statement<'_>,
 	) -> Result<(), Error> {
-		// Check if forced
+		// Check if changed
-		if !opt.force && !self.changed() {
+		if !self.changed() {
 			return Ok(());
 		}
 		// Under the new mechanism, live query notifications only come from polling the change feed

core/src/doc/purge.rs

@@ -20,8 +20,8 @@ impl<'a> Document<'a> {
 		txn: &Transaction,
 		_stm: &Statement<'_>,
 	) -> Result<(), Error> {
-		// Check if forced
+		// Check if changed
-		if !opt.force && !self.changed() {
+		if !self.changed() {
 			return Ok(());
 		}
 		// Clone transaction

core/src/doc/store.rs

@@ -13,8 +13,8 @@ impl<'a> Document<'a> {
 		txn: &Transaction,
 		stm: &Statement<'_>,
 	) -> Result<(), Error> {
-		// Check if forced
+		// Check if changed
-		if !opt.force && !self.changed() {
+		if !self.changed() {
 			return Ok(());
 		}
 		// Check if the table is a view

core/src/doc/table.rs

@@ -1,5 +1,5 @@
 use crate::ctx::Context;
-use crate::dbs::Statement;
+use crate::dbs::{Force, Statement};
 use crate::dbs::{Options, Transaction};
 use crate::doc::Document;
 use crate::err::Error;
@@ -37,14 +37,27 @@ impl<'a> Document<'a> {
 		txn: &Transaction,
 		stm: &Statement<'_>,
 	) -> Result<(), Error> {
-		// Check tables
+		// Check import
-		if !opt.tables {
+		if opt.import {
-			return Ok(());
-		}
-		// Check if forced
-		if !opt.force && !self.changed() {
 			return Ok(());
 		}
+		// Was this force targeted at a specific foreign table?
+		let targeted_force = matches!(opt.force, Force::Table(_));
+		// Collect foreign tables or skip
+		let fts = match &opt.force {
+			Force::Table(tb)
+				if tb.first().is_some_and(|tb| {
+					tb.view.as_ref().is_some_and(|v| {
+						self.id.is_some_and(|id| v.what.iter().any(|p| p.0 == id.tb))
+					})
+				}) =>
+			{
+				tb.clone()
+			}
+			Force::All => self.ft(opt, txn).await?,
+			_ if self.changed() => self.ft(opt, txn).await?,
+			_ => return Ok(()),
+		};
 		// Don't run permissions
 		let opt = &opt.new_with_perms(false);
 		// Get the record id
@@ -58,7 +71,7 @@ impl<'a> Document<'a> {
 			Action::Update
 		};
 		// Loop through all foreign table statements
-		for ft in self.ft(opt, txn).await?.iter() {
+		for ft in fts.iter() {
 			// Get the table definition
 			let tb = ft.view.as_ref().unwrap();
 			// Check if there is a GROUP BY clause
@@ -93,7 +106,7 @@ impl<'a> Document<'a> {
 						Some(cond) => {
 							match cond.compute(ctx, opt, txn, Some(&self.current)).await? {
 								v if v.is_truthy() => {
-									if !opt.force && act != Action::Create {
+									if !targeted_force && act != Action::Create {
 										// Delete the old value
 										let act = Action::Delete;
 										// Modify the value in the table
@@ -123,7 +136,7 @@ impl<'a> Document<'a> {
 									}
 								}
 								_ => {
-									if !opt.force && act != Action::Create {
+									if !targeted_force && act != Action::Create {
 										// Update the new value
 										let act = Action::Update;
 										// Modify the value in the table
@@ -142,7 +155,7 @@ impl<'a> Document<'a> {
 						}
 						// No WHERE clause is specified
 						None => {
-							if !opt.force && act != Action::Create {
+							if !targeted_force && act != Action::Create {
 								// Delete the old value
 								let act = Action::Delete;
 								// Modify the value in the table

core/src/sql/statements/define/index.rs

@@ -1,5 +1,5 @@
 use crate::ctx::Context;
-use crate::dbs::{Options, Transaction};
+use crate::dbs::{Force, Options, Transaction};
 use crate::doc::CursorDoc;
 use crate::err::Error;
 use crate::iam::{Action, ResourceKind};
@@ -8,6 +8,7 @@ use derive::Store;
 use revision::revisioned;
 use serde::{Deserialize, Serialize};
 use std::fmt::{self, Display};
+use std::sync::Arc;
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Store, Hash)]
 #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
@@ -68,13 +69,7 @@ impl DefineIndexStatement {
 		// Release the transaction
 		drop(run);
 		// Force queries to run
-		let opt = &opt.new_with_force(true);
+		let opt = &opt.new_with_force(Force::Index(Arc::new([self.clone()])));
-		// Don't process field queries
-		let opt = &opt.new_with_fields(false);
-		// Don't process event queries
-		let opt = &opt.new_with_events(false);
-		// Don't process table queries
-		let opt = &opt.new_with_tables(false);
 		// Update the index data
 		let stm = UpdateStatement {
 			what: Values(vec![Value::Table(self.what.clone().into())]),

core/src/sql/statements/define/table.rs

@@ -5,7 +5,7 @@ use revision::revisioned;
 use serde::{Deserialize, Serialize};
 
 use crate::ctx::Context;
-use crate::dbs::{Options, Transaction};
+use crate::dbs::{Force, Options, Transaction};
 use crate::doc::CursorDoc;
 use crate::err::Error;
 use crate::iam::{Action, ResourceKind};
@@ -15,6 +15,7 @@ use crate::sql::{
 	statements::UpdateStatement,
 	Base, Ident, Permissions, Strand, Value, Values, View,
 };
+use std::sync::Arc;
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Store, Hash)]
 #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
@@ -86,13 +87,7 @@ impl DefineTableStatement {
 			// Release the transaction
 			drop(run);
 			// Force queries to run
-			let opt = &opt.new_with_force(true);
+			let opt = &opt.new_with_force(Force::Table(Arc::new([dt])));
-			// Don't process field queries
-			let opt = &opt.new_with_fields(false);
-			// Don't process event queries
-			let opt = &opt.new_with_events(false);
-			// Don't process index queries
-			let opt = &opt.new_with_indexes(false);
 			// Process each foreign table
 			for v in view.what.0.iter() {
 				// Process the view data

lib/tests/table.rs

@@ -106,3 +106,65 @@ async fn define_foreign_table() -> Result<(), Error> {
 	//
 	Ok(())
 }
+
+#[tokio::test]
+async fn define_foreign_table_no_doubles() -> Result<(), Error> {
+	// From: https://github.com/surrealdb/surrealdb/issues/3556
+	let sql = "
+		CREATE happy:1 SET year=2024, month=1, day=1;
+		CREATE happy:2 SET year=2024, month=1, day=1;
+		CREATE happy:3 SET year=2024, month=1, day=1;
+		DEFINE TABLE monthly AS SELECT count() as activeRounds, year, month FROM happy GROUP BY year, month;
+		DEFINE TABLE daily AS SELECT count() as activeRounds, year, month, day FROM happy GROUP BY year, month, day;
+		SELECT * FROM monthly;
+		SELECT * FROM daily;
+	";
+	let dbs = new_ds().await?;
+	let ses = Session::owner().with_ns("test").with_db("test");
+	let res = &mut dbs.execute(sql, &ses, None).await?;
+	assert_eq!(res.len(), 7);
+	//
+	let tmp = res.remove(0).result;
+	assert!(tmp.is_ok());
+	//
+	let tmp = res.remove(0).result;
+	assert!(tmp.is_ok());
+	//
+	let tmp = res.remove(0).result;
+	assert!(tmp.is_ok());
+	//
+	let tmp = res.remove(0).result;
+	assert!(tmp.is_ok());
+	//
+	let tmp = res.remove(0).result;
+	assert!(tmp.is_ok());
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: monthly:[2024, 1],
+				activeRounds: 3,
+				year: 2024,
+				month: 1,
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: daily:[2024, 1, 1],
+				activeRounds: 3,
+				year: 2024,
+				month: 1,
+				day: 1,
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	Ok(())
+}