Add UPSERT statement (#4163)

core/src/dbs/processor.rs

@@ -123,6 +123,8 @@ impl<'a> Processor<'a> {
 				Iterable::Value(v) => self.process_value(stk, ctx, opt, stm, v).await?,
 				Iterable::Thing(v) => self.process_thing(stk, ctx, opt, stm, v).await?,
 				Iterable::Defer(v) => self.process_defer(stk, ctx, opt, stm, v).await?,
+				Iterable::Range(v) => self.process_range(stk, ctx, opt, stm, v).await?,
+				Iterable::Edges(e) => self.process_edge(stk, ctx, opt, stm, e).await?,
 				Iterable::Table(v) => {
 					if let Some(qp) = ctx.get_query_planner() {
 						if let Some(exe) = qp.get_query_executor(&v.0) {
@@ -135,8 +137,6 @@ impl<'a> Processor<'a> {
 					}
 					self.process_table(stk, ctx, opt, stm, &v).await?
 				}
-				Iterable::Range(v) => self.process_range(stk, ctx, opt, stm, v).await?,
-				Iterable::Edges(e) => self.process_edge(stk, ctx, opt, stm, e).await?,
 				Iterable::Index(t, irf) => {
 					if let Some(qp) = ctx.get_query_planner() {
 						if let Some(exe) = qp.get_query_executor(&t.0) {

core/src/dbs/statement.rs

@@ -17,6 +17,7 @@ use crate::sql::statements::relate::RelateStatement;
 use crate::sql::statements::select::SelectStatement;
 use crate::sql::statements::show::ShowStatement;
 use crate::sql::statements::update::UpdateStatement;
+use crate::sql::statements::upsert::UpsertStatement;
 use crate::sql::Explain;
 use std::fmt;
 
@@ -26,6 +27,7 @@ pub(crate) enum Statement<'a> {
 	Show(&'a ShowStatement),
 	Select(&'a SelectStatement),
 	Create(&'a CreateStatement),
+	Upsert(&'a UpsertStatement),
 	Update(&'a UpdateStatement),
 	Relate(&'a RelateStatement),
 	Delete(&'a DeleteStatement),
@@ -56,6 +58,12 @@ impl<'a> From<&'a CreateStatement> for Statement<'a> {
 	}
 }
 
+impl<'a> From<&'a UpsertStatement> for Statement<'a> {
+	fn from(v: &'a UpsertStatement) -> Self {
+		Statement::Upsert(v)
+	}
+}
+
 impl<'a> From<&'a UpdateStatement> for Statement<'a> {
 	fn from(v: &'a UpdateStatement) -> Self {
 		Statement::Update(v)
@@ -87,6 +95,7 @@ impl<'a> fmt::Display for Statement<'a> {
 			Statement::Show(v) => write!(f, "{v}"),
 			Statement::Select(v) => write!(f, "{v}"),
 			Statement::Create(v) => write!(f, "{v}"),
+			Statement::Upsert(v) => write!(f, "{v}"),
 			Statement::Update(v) => write!(f, "{v}"),
 			Statement::Relate(v) => write!(f, "{v}"),
 			Statement::Delete(v) => write!(f, "{v}"),
@@ -128,6 +137,7 @@ impl<'a> Statement<'a> {
 	pub fn data(&self) -> Option<&Data> {
 		match self {
 			Statement::Create(v) => v.data.as_ref(),
+			Statement::Upsert(v) => v.data.as_ref(),
 			Statement::Update(v) => v.data.as_ref(),
 			Statement::Relate(v) => v.data.as_ref(),
 			Statement::Insert(v) => v.update.as_ref(),
@@ -140,6 +150,7 @@ impl<'a> Statement<'a> {
 		match self {
 			Statement::Live(v) => v.cond.as_ref(),
 			Statement::Select(v) => v.cond.as_ref(),
+			Statement::Upsert(v) => v.cond.as_ref(),
 			Statement::Update(v) => v.cond.as_ref(),
 			Statement::Delete(v) => v.cond.as_ref(),
 			_ => None,
@@ -198,6 +209,7 @@ impl<'a> Statement<'a> {
 	pub fn output(&self) -> Option<&Output> {
 		match self {
 			Statement::Create(v) => v.output.as_ref(),
+			Statement::Upsert(v) => v.output.as_ref(),
 			Statement::Update(v) => v.output.as_ref(),
 			Statement::Relate(v) => v.output.as_ref(),
 			Statement::Delete(v) => v.output.as_ref(),
@@ -212,6 +224,7 @@ impl<'a> Statement<'a> {
 		match self {
 			Statement::Select(v) => v.parallel,
 			Statement::Create(v) => v.parallel,
+			Statement::Upsert(v) => v.parallel,
 			Statement::Update(v) => v.parallel,
 			Statement::Relate(v) => v.parallel,
 			Statement::Delete(v) => v.parallel,

core/src/doc/alter.rs

@@ -40,6 +40,11 @@ impl<'a> Document<'a> {
 					let data = data.compute(stk, ctx, opt, Some(&self.current)).await?;
 					self.current.doc.to_mut().replace(data)?
 				}
+				Data::UnsetExpression(i) => {
+					for i in i.iter() {
+						self.current.doc.to_mut().del(stk, ctx, opt, i).await?
+					}
+				}
 				Data::SetExpression(x) => {
 					for x in x.iter() {
 						let v = x.2.compute(stk, ctx, opt, Some(&self.current)).await?;
@@ -63,11 +68,6 @@ impl<'a> Document<'a> {
 						}
 					}
 				}
-				Data::UnsetExpression(i) => {
-					for i in i.iter() {
-						self.current.doc.to_mut().del(stk, ctx, opt, i).await?
-					}
-				}
 				Data::UpdateExpression(x) => {
 					// Duplicate context
 					let mut ctx = Context::new(ctx);

core/src/doc/compute.rs

@@ -37,6 +37,7 @@ impl<'a> Document<'a> {
 			let res = match stm {
 				Statement::Select(_) => doc.select(stk, ctx, opt, stm).await,
 				Statement::Create(_) => doc.create(stk, ctx, opt, stm).await,
+				Statement::Upsert(_) => doc.upsert(stk, ctx, opt, stm).await,
 				Statement::Update(_) => doc.update(stk, ctx, opt, stm).await,
 				Statement::Relate(_) => doc.relate(stk, ctx, opt, stm).await,
 				Statement::Delete(_) => doc.delete(stk, ctx, opt, stm).await,

core/src/doc/event.rs

@@ -28,7 +28,7 @@ impl<'a> Document<'a> {
 		// Loop through all event statements
 		for ev in self.ev(ctx, opt).await?.iter() {
 			// Get the event action
-			let met = if stm.is_delete() {
+			let evt = if stm.is_delete() {
 				Value::from("DELETE")
 			} else if self.is_new() {
 				Value::from("CREATE")
@@ -42,7 +42,7 @@ impl<'a> Document<'a> {
 			};
 			// Configure the context
 			let mut ctx = Context::new(ctx);
-			ctx.add_value("event", met);
+			ctx.add_value("event", evt);
 			ctx.add_value("value", doc.doc.deref());
 			ctx.add_value("after", self.current.doc.deref());
 			ctx.add_value("before", self.initial.doc.deref());

core/src/doc/insert.rs

@@ -44,6 +44,7 @@ impl<'a> Document<'a> {
 		}
 	}
 	// Attempt to run an INSERT clause
+	#[inline(always)]
 	async fn insert_create(
 		&mut self,
 		stk: &mut Stk,
@@ -81,6 +82,7 @@ impl<'a> Document<'a> {
 		self.pluck(stk, ctx, opt, stm).await
 	}
 	// Attempt to run an UPDATE clause
+	#[inline(always)]
 	async fn insert_update(
 		&mut self,
 		stk: &mut Stk,

core/src/doc/lives.rs

@@ -123,7 +123,7 @@ impl<'a> Document<'a> {
 			// Create a new statement
 			let lq = Statement::from(*lv);
 			// Get the event action
-			let met = if is_delete {
+			let evt = if stm.is_delete() {
 				Value::from("DELETE")
 			} else if self.is_new() {
 				Value::from("CREATE")
@@ -168,7 +168,7 @@ impl<'a> Document<'a> {
 			// Add $before, $after, $value, and $event params
 			// to this LIVE query so that user can use these
 			// within field projections and WHERE clauses.
-			lqctx.add_value("event", met);
+			lqctx.add_value("event", evt);
 			lqctx.add_value("value", self.current.doc.deref());
 			lqctx.add_value("after", self.current.doc.deref());
 			lqctx.add_value("before", self.initial.doc.deref());

core/src/doc/mod.rs

@@ -18,6 +18,7 @@ mod insert; // Processes a INSERT statement for this document
 mod relate; // Processes a RELATE statement for this document
 mod select; // Processes a SELECT statement for this document
 mod update; // Processes a UPDATE statement for this document
+mod upsert; // Processes a UPSERT statement for this document
 
 mod allow; // Checks whether the query can access this document
 mod alter; // Modifies and updates the fields in this document

core/src/doc/pluck.rs

@@ -61,6 +61,9 @@ impl<'a> Document<'a> {
 				Statement::Create(_) => {
 					self.current.doc.compute(stk, ctx, opt, Some(&self.current)).await
 				}
+				Statement::Upsert(_) => {
+					self.current.doc.compute(stk, ctx, opt, Some(&self.current)).await
+				}
 				Statement::Update(_) => {
 					self.current.doc.compute(stk, ctx, opt, Some(&self.current)).await
 				}

core/src/doc/process.rs

@@ -30,8 +30,9 @@ impl<'a> Document<'a> {
 			// Process the statement
 			let res = match stm {
 				Statement::Select(_) => doc.select(stk, ctx, opt, stm).await,
-				Statement::Update(_) => doc.update(stk, ctx, opt, stm).await,
 				Statement::Create(_) => doc.create(stk, ctx, opt, stm).await,
+				Statement::Upsert(_) => doc.upsert(stk, ctx, opt, stm).await,
+				Statement::Update(_) => doc.update(stk, ctx, opt, stm).await,
 				Statement::Relate(_) => doc.relate(stk, ctx, opt, stm).await,
 				Statement::Delete(_) => doc.delete(stk, ctx, opt, stm).await,
 				Statement::Insert(_) => doc.insert(stk, ctx, opt, stm).await,

core/src/doc/relate.rs

@@ -16,10 +16,27 @@ impl<'a> Document<'a> {
 	) -> Result<Value, Error> {
 		// Check if table has correct relation status
 		self.relation(ctx, opt, stm).await?;
-		// Check current record
+		// Check whether current record exists
 		match self.current.doc.is_some() {
-			// Create new edge
+			// We attempted to RELATE a document with an ID,
-			false => {
+			// and this ID already exists in the database,
+			// so we need to update the record instead.
+			true => self.relate_update(stk, ctx, opt, stm).await,
+			// We attempted to RELATE a document with an ID,
+			// which does not exist in the database, or we
+			// are creating a new record with a new ID.
+			false => self.relate_create(stk, ctx, opt, stm).await,
+		}
+	}
+	// Attempt to run an INSERT clause
+	#[inline(always)]
+	async fn relate_create(
+		&mut self,
+		stk: &mut Stk,
+		ctx: &Context<'_>,
+		opt: &Options,
+		stm: &Statement<'_>,
+	) -> Result<Value, Error> {
 		// Store record edges
 		self.edges(ctx, opt, stm).await?;
 		// Alter record data
@@ -47,8 +64,15 @@ impl<'a> Document<'a> {
 		// Yield document
 		self.pluck(stk, ctx, opt, stm).await
 	}
-			// Update old edge
+	// Attempt to run an UPDATE clause
-			true => {
+	#[inline(always)]
+	async fn relate_update(
+		&mut self,
+		stk: &mut Stk,
+		ctx: &Context<'_>,
+		opt: &Options,
+		stm: &Statement<'_>,
+	) -> Result<Value, Error> {
 		// Check if allowed
 		self.allow(stk, ctx, opt, stm).await?;
 		// Store record edges
@@ -78,6 +102,4 @@ impl<'a> Document<'a> {
 		// Yield document
 		self.pluck(stk, ctx, opt, stm).await
 	}
-		}
-	}
 }

core/src/doc/relation.rs

@@ -24,6 +24,24 @@ impl<'a> Document<'a> {
 					});
 				}
 			}
+			Statement::Upsert(_) => {
+				if !tb.allows_normal() {
+					return Err(Error::TableCheck {
+						thing: rid.to_string(),
+						relation: false,
+						target_type: tb.kind.clone(),
+					});
+				}
+			}
+			Statement::Relate(_) => {
+				if !tb.allows_relation() {
+					return Err(Error::TableCheck {
+						thing: rid.to_string(),
+						relation: true,
+						target_type: tb.kind.clone(),
+					});
+				}
+			}
 			Statement::Insert(_) => match self.extras {
 				Workable::Relate(_, _, _) => {
 					if !tb.allows_relation() {
@@ -44,15 +62,6 @@ impl<'a> Document<'a> {
 					}
 				}
 			},
-			Statement::Relate(_) => {
-				if !tb.allows_relation() {
-					return Err(Error::TableCheck {
-						thing: rid.to_string(),
-						relation: true,
-						target_type: tb.kind.clone(),
-					});
-				}
-			}
 			_ => {}
 		}
 		// Carry on

core/src/doc/table.rs

@@ -13,7 +13,7 @@ use crate::sql::part::Part;
 use crate::sql::paths::ID;
 use crate::sql::statements::delete::DeleteStatement;
 use crate::sql::statements::ifelse::IfelseStatement;
-use crate::sql::statements::update::UpdateStatement;
+use crate::sql::statements::upsert::UpsertStatement;
 use crate::sql::statements::{DefineTableStatement, SelectStatement};
 use crate::sql::subquery::Subquery;
 use crate::sql::thing::Thing;
@@ -219,12 +219,12 @@ impl<'a> Document<'a> {
 										}
 										// Update the value in the table
 										_ => {
-											let stm = UpdateStatement {
+											let stm = UpsertStatement {
 												what: Values(vec![Value::from(rid)]),
 												data: Some(
 													self.full(stk, ctx, opt, &tb.expr).await?,
 												),
-												..UpdateStatement::default()
+												..UpsertStatement::default()
 											};
 											// Execute the statement
 											stm.compute(stk, ctx, opt, None).await?;
@@ -257,10 +257,10 @@ impl<'a> Document<'a> {
 								}
 								// Update the value in the table
 								_ => {
-									let stm = UpdateStatement {
+									let stm = UpsertStatement {
 										what: Values(vec![Value::from(rid)]),
 										data: Some(self.full(stk, ctx, opt, &tb.expr).await?),
-										..UpdateStatement::default()
+										..UpsertStatement::default()
 									};
 									// Execute the statement
 									stm.compute(stk, ctx, opt, None).await?;
@@ -321,10 +321,10 @@ impl<'a> Document<'a> {
 			id: fdc.group_ids.into(),
 		};
 		let what = Values(vec![Value::from(thg.clone())]);
-		let stm = UpdateStatement {
+		let stm = UpsertStatement {
 			what,
 			data: Some(Data::SetExpression(set_ops)),
-			..UpdateStatement::default()
+			..UpsertStatement::default()
 		};
 		stm.compute(stk, ctx, opt, None).await?;
 

core/src/doc/update.rs

@@ -14,6 +14,8 @@ impl<'a> Document<'a> {
 		opt: &Options,
 		stm: &Statement<'_>,
 	) -> Result<Value, Error> {
+		// Check if record exists
+		self.empty(ctx, opt, stm).await?;
 		// Check where clause
 		self.check(stk, ctx, opt, stm).await?;
 		// Check if allowed

core/src/doc/upsert.rs

@@ -0,0 +1,46 @@
+use crate::ctx::Context;
+use crate::dbs::Options;
+use crate::dbs::Statement;
+use crate::doc::Document;
+use crate::err::Error;
+use crate::sql::value::Value;
+use reblessive::tree::Stk;
+
+impl<'a> Document<'a> {
+	pub async fn upsert(
+		&mut self,
+		stk: &mut Stk,
+		ctx: &Context<'_>,
+		opt: &Options,
+		stm: &Statement<'_>,
+	) -> Result<Value, Error> {
+		// Check where clause
+		self.check(stk, ctx, opt, stm).await?;
+		// Check if allowed
+		self.allow(stk, ctx, opt, stm).await?;
+		// Alter record data
+		self.alter(stk, ctx, opt, stm).await?;
+		// Merge fields data
+		self.field(stk, ctx, opt, stm).await?;
+		// Reset fields data
+		self.reset(ctx, opt, stm).await?;
+		// Clean fields data
+		self.clean(stk, ctx, opt, stm).await?;
+		// Check if allowed
+		self.allow(stk, ctx, opt, stm).await?;
+		// Store record data
+		self.store(ctx, opt, stm).await?;
+		// Store index data
+		self.index(stk, ctx, opt, stm).await?;
+		// Run table queries
+		self.table(stk, ctx, opt, stm).await?;
+		// Run lives queries
+		self.lives(stk, ctx, opt, stm).await?;
+		// Run change feeds queries
+		self.changefeeds(ctx, opt, stm).await?;
+		// Run event queries
+		self.event(stk, ctx, opt, stm).await?;
+		// Yield document
+		self.pluck(stk, ctx, opt, stm).await
+	}
+}

core/src/err/mod.rs

@@ -446,6 +446,12 @@ pub enum Error {
 		value: String,
 	},
 
+	/// Can not execute UPSERT statement using the specified value
+	#[error("Can not execute UPSERT statement using value '{value}'")]
+	UpsertStatement {
+		value: String,
+	},
+
 	/// Can not execute UPDATE statement using the specified value
 	#[error("Can not execute UPDATE statement using value '{value}'")]
 	UpdateStatement {

core/src/iam/entities/action.rs

@@ -49,6 +49,7 @@ impl From<&Statement<'_>> for Action {
 			Statement::Select(_) => Action::View,
 			Statement::Show(_) => Action::View,
 			Statement::Create(_) => Action::Edit,
+			Statement::Upsert(_) => Action::Edit,
 			Statement::Update(_) => Action::Edit,
 			Statement::Relate(_) => Action::Edit,
 			Statement::Delete(_) => Action::Edit,

core/src/kvs/tests/lq.rs

@@ -1,4 +1,3 @@
-use crate::fflags::FFLAGS;
 use crate::kvs::lq_structs::{LqIndexKey, LqIndexValue, LqSelector};
 use uuid::Uuid;
 

core/src/rpc/method.rs

@@ -15,6 +15,7 @@ pub enum Method {
 	Select,
 	Insert,
 	Create,
+	Upsert,
 	Update,
 	Merge,
 	Patch,
@@ -45,6 +46,7 @@ impl Method {
 			"select" => Self::Select,
 			"insert" => Self::Insert,
 			"create" => Self::Create,
+			"upsert" => Self::Upsert,
 			"update" => Self::Update,
 			"merge" => Self::Merge,
 			"patch" => Self::Patch,
@@ -76,6 +78,7 @@ impl Method {
 			Self::Select => "select",
 			Self::Insert => "insert",
 			Self::Create => "create",
+			Self::Upsert => "upsert",
 			Self::Update => "update",
 			Self::Merge => "merge",
 			Self::Patch => "patch",
@@ -104,9 +107,9 @@ impl Method {
 			Method::Ping
 				| Method::Info | Method::Select
 				| Method::Insert | Method::Create
-				| Method::Update | Method::Merge
+				| Method::Update | Method::Upsert
-				| Method::Patch | Method::Delete
+				| Method::Merge | Method::Patch
-				| Method::Version
+				| Method::Delete | Method::Version
 				| Method::Query | Method::Relate
 				| Method::Run | Method::Unknown
 		)

core/src/rpc/rpc_context.rs

@@ -53,6 +53,7 @@ pub trait RpcContext {
 			Method::Select => self.select(params).await.map(Into::into).map_err(Into::into),
 			Method::Insert => self.insert(params).await.map(Into::into).map_err(Into::into),
 			Method::Create => self.create(params).await.map(Into::into).map_err(Into::into),
+			Method::Upsert => self.upsert(params).await.map(Into::into).map_err(Into::into),
 			Method::Update => self.update(params).await.map(Into::into).map_err(Into::into),
 			Method::Merge => self.merge(params).await.map(Into::into).map_err(Into::into),
 			Method::Patch => self.patch(params).await.map(Into::into).map_err(Into::into),
@@ -72,6 +73,7 @@ pub trait RpcContext {
 			Method::Select => self.select(params).await.map(Into::into).map_err(Into::into),
 			Method::Insert => self.insert(params).await.map(Into::into).map_err(Into::into),
 			Method::Create => self.create(params).await.map(Into::into).map_err(Into::into),
+			Method::Upsert => self.upsert(params).await.map(Into::into).map_err(Into::into),
 			Method::Update => self.update(params).await.map(Into::into).map_err(Into::into),
 			Method::Merge => self.merge(params).await.map(Into::into).map_err(Into::into),
 			Method::Patch => self.patch(params).await.map(Into::into).map_err(Into::into),
@@ -320,6 +322,39 @@ pub trait RpcContext {
 		Ok(res)
 	}
 
+	// ------------------------------
+	// Methods for upserting
+	// ------------------------------
+
+	async fn upsert(&self, params: Array) -> Result<impl Into<Data>, RpcError> {
+		let Ok((what, data)) = params.needs_one_or_two() else {
+			return Err(RpcError::InvalidParams);
+		};
+		// Return a single result?
+		let one = what.is_thing();
+		// Specify the SQL query string
+		let sql = if data.is_none_or_null() {
+			"UPSERT $what RETURN AFTER"
+		} else {
+			"UPSERT $what CONTENT $data RETURN AFTER"
+		};
+		// Specify the query parameters
+		let var = Some(map! {
+			String::from("what") => what.could_be_table(),
+			String::from("data") => data,
+			=> &self.vars()
+		});
+		// Execute the query on the database
+		let mut res = self.kvs().execute(sql, self.session(), var).await?;
+		// Extract the first query result
+		let res = match one {
+			true => res.remove(0).result?.first(),
+			false => res.remove(0).result?,
+		};
+		// Return the result to the client
+		Ok(res)
+	}
+
 	// ------------------------------
 	// Methods for updating
 	// ------------------------------

core/src/sql/block.rs

@@ -9,6 +9,7 @@ use crate::sql::statements::{
 	BreakStatement, ContinueStatement, CreateStatement, DefineStatement, DeleteStatement,
 	ForeachStatement, IfelseStatement, InsertStatement, OutputStatement, RelateStatement,
 	RemoveStatement, SelectStatement, SetStatement, ThrowStatement, UpdateStatement,
+	UpsertStatement,
 };
 use crate::sql::value::Value;
 use reblessive::tree::Stk;
@@ -86,6 +87,9 @@ impl Block {
 				Entry::Create(v) => {
 					v.compute(stk, &ctx, opt, doc).await?;
 				}
+				Entry::Upsert(v) => {
+					v.compute(stk, &ctx, opt, doc).await?;
+				}
 				Entry::Update(v) => {
 					v.compute(stk, &ctx, opt, doc).await?;
 				}
@@ -178,7 +182,7 @@ impl InfoStructure for Block {
 	}
 }
 
-#[revisioned(revision = 2)]
+#[revisioned(revision = 3)]
 #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, Hash)]
 #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
 #[non_exhaustive]
@@ -201,6 +205,8 @@ pub enum Entry {
 	Foreach(ForeachStatement),
 	#[revision(start = 2)]
 	Rebuild(RebuildStatement),
+	#[revision(start = 3)]
+	Upsert(UpsertStatement),
 }
 
 impl PartialOrd for Entry {
@@ -219,6 +225,7 @@ impl Entry {
 			Self::Ifelse(v) => v.writeable(),
 			Self::Select(v) => v.writeable(),
 			Self::Create(v) => v.writeable(),
+			Self::Upsert(v) => v.writeable(),
 			Self::Update(v) => v.writeable(),
 			Self::Delete(v) => v.writeable(),
 			Self::Relate(v) => v.writeable(),
@@ -243,6 +250,7 @@ impl Display for Entry {
 			Self::Ifelse(v) => write!(f, "{v}"),
 			Self::Select(v) => write!(f, "{v}"),
 			Self::Create(v) => write!(f, "{v}"),
+			Self::Upsert(v) => write!(f, "{v}"),
 			Self::Update(v) => write!(f, "{v}"),
 			Self::Delete(v) => write!(f, "{v}"),
 			Self::Relate(v) => write!(f, "{v}"),

core/src/sql/statement.rs

@@ -10,7 +10,8 @@ use crate::sql::{
 		ContinueStatement, CreateStatement, DefineStatement, DeleteStatement, ForeachStatement,
 		IfelseStatement, InfoStatement, InsertStatement, KillStatement, LiveStatement,
 		OptionStatement, OutputStatement, RelateStatement, RemoveStatement, SelectStatement,
-		SetStatement, ShowStatement, SleepStatement, ThrowStatement, UpdateStatement, UseStatement,
+		SetStatement, ShowStatement, SleepStatement, ThrowStatement, UpdateStatement,
+		UpsertStatement, UseStatement,
 	},
 	value::Value,
 };
@@ -54,7 +55,7 @@ impl Display for Statements {
 	}
 }
 
-#[revisioned(revision = 2)]
+#[revisioned(revision = 3)]
 #[derive(Clone, Debug, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Store, Hash)]
 #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
 #[non_exhaustive]
@@ -88,6 +89,8 @@ pub enum Statement {
 	Use(UseStatement),
 	#[revision(start = 2)]
 	Rebuild(RebuildStatement),
+	#[revision(start = 3)]
+	Upsert(UpsertStatement),
 }
 
 impl Statement {
@@ -99,6 +102,7 @@ impl Statement {
 			Self::Insert(v) => v.timeout.as_ref().map(|v| *v.0),
 			Self::Relate(v) => v.timeout.as_ref().map(|v| *v.0),
 			Self::Select(v) => v.timeout.as_ref().map(|v| *v.0),
+			Self::Upsert(v) => v.timeout.as_ref().map(|v| *v.0),
 			Self::Update(v) => v.timeout.as_ref().map(|v| *v.0),
 			_ => None,
 		}
@@ -129,6 +133,7 @@ impl Statement {
 			Self::Show(_) => false,
 			Self::Sleep(_) => false,
 			Self::Throw(_) => false,
+			Self::Upsert(v) => v.writeable(),
 			Self::Update(v) => v.writeable(),
 			Self::Use(_) => false,
 			_ => unreachable!(),
@@ -165,6 +170,7 @@ impl Statement {
 			Self::Sleep(v) => v.compute(ctx, opt, doc).await,
 			Self::Throw(v) => v.compute(stk, ctx, opt, doc).await,
 			Self::Update(v) => v.compute(stk, ctx, opt, doc).await,
+			Self::Upsert(v) => v.compute(stk, ctx, opt, doc).await,
 			Self::Value(v) => {
 				// Ensure futures are processed
 				let opt = &opt.new_with_futures(true);
@@ -206,6 +212,7 @@ impl Display for Statement {
 			Self::Sleep(v) => write!(Pretty::from(f), "{v}"),
 			Self::Throw(v) => write!(Pretty::from(f), "{v}"),
 			Self::Update(v) => write!(Pretty::from(f), "{v}"),
+			Self::Upsert(v) => write!(Pretty::from(f), "{v}"),
 			Self::Use(v) => write!(Pretty::from(f), "{v}"),
 		}
 	}

core/src/sql/statements/foreach.rs

@@ -63,6 +63,7 @@ impl ForeachStatement {
 							Entry::Ifelse(v) => stk.run(|stk| v.compute(stk, &ctx, opt, doc)).await,
 							Entry::Select(v) => stk.run(|stk| v.compute(stk, &ctx, opt, doc)).await,
 							Entry::Create(v) => stk.run(|stk| v.compute(stk, &ctx, opt, doc)).await,
+							Entry::Upsert(v) => stk.run(|stk| v.compute(stk, &ctx, opt, doc)).await,
 							Entry::Update(v) => stk.run(|stk| v.compute(stk, &ctx, opt, doc)).await,
 							Entry::Delete(v) => stk.run(|stk| v.compute(stk, &ctx, opt, doc)).await,
 							Entry::Relate(v) => stk.run(|stk| v.compute(stk, &ctx, opt, doc)).await,

core/src/sql/statements/mod.rs

@@ -24,6 +24,7 @@ pub(crate) mod show;
 pub(crate) mod sleep;
 pub(crate) mod throw;
 pub(crate) mod update;
+pub(crate) mod upsert;
 pub(crate) mod r#use;
 
 pub use self::analyze::AnalyzeStatement;
@@ -50,6 +51,7 @@ pub use self::show::ShowStatement;
 pub use self::sleep::SleepStatement;
 pub use self::throw::ThrowStatement;
 pub use self::update::UpdateStatement;
+pub use self::upsert::UpsertStatement;
 
 pub use self::define::{
 	DefineAccessStatement, DefineAnalyzerStatement, DefineDatabaseStatement, DefineEventStatement,

core/src/sql/statements/upsert.rs

@@ -0,0 +1,98 @@
+use crate::ctx::Context;
+use crate::dbs::{Iterator, Options, Statement};
+use crate::doc::CursorDoc;
+use crate::err::Error;
+use crate::sql::{Cond, Data, Output, Timeout, Value, Values};
+use derive::Store;
+use reblessive::tree::Stk;
+use revision::revisioned;
+use serde::{Deserialize, Serialize};
+use std::fmt;
+
+#[revisioned(revision = 1)]
+#[derive(Clone, Debug, Default, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Store, Hash)]
+#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
+#[non_exhaustive]
+pub struct UpsertStatement {
+	pub only: bool,
+	pub what: Values,
+	pub data: Option<Data>,
+	pub cond: Option<Cond>,
+	pub output: Option<Output>,
+	pub timeout: Option<Timeout>,
+	pub parallel: bool,
+}
+
+impl UpsertStatement {
+	/// Check if we require a writeable transaction
+	pub(crate) fn writeable(&self) -> bool {
+		true
+	}
+	/// Process this type returning a computed simple Value
+	pub(crate) async fn compute(
+		&self,
+		stk: &mut Stk,
+		ctx: &Context<'_>,
+		opt: &Options,
+		doc: Option<&CursorDoc<'_>>,
+	) -> Result<Value, Error> {
+		// Valid options?
+		opt.valid_for_db()?;
+		// Create a new iterator
+		let mut i = Iterator::new();
+		// Assign the statement
+		let stm = Statement::from(self);
+		// Ensure futures are stored
+		let opt = &opt.new_with_futures(false).with_projections(false);
+		// Loop over the upsert targets
+		for w in self.what.0.iter() {
+			let v = w.compute(stk, ctx, opt, doc).await?;
+			i.prepare(stk, ctx, opt, &stm, v).await.map_err(|e| match e {
+				Error::InvalidStatementTarget {
+					value: v,
+				} => Error::UpsertStatement {
+					value: v,
+				},
+				e => e,
+			})?;
+		}
+		// Output the results
+		match i.output(stk, ctx, opt, &stm).await? {
+			// This is a single record result
+			Value::Array(mut a) if self.only => match a.len() {
+				// There was exactly one result
+				1 => Ok(a.remove(0)),
+				// There were no results
+				_ => Err(Error::SingleOnlyOutput),
+			},
+			// This is standard query result
+			v => Ok(v),
+		}
+	}
+}
+
+impl fmt::Display for UpsertStatement {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		write!(f, "UPSERT")?;
+		if self.only {
+			f.write_str(" ONLY")?
+		}
+		write!(f, " {}", self.what)?;
+		if let Some(ref v) = self.data {
+			write!(f, " {v}")?
+		}
+		if let Some(ref v) = self.cond {
+			write!(f, " {v}")?
+		}
+		if let Some(ref v) = self.output {
+			write!(f, " {v}")?
+		}
+		if let Some(ref v) = self.timeout {
+			write!(f, " {v}")?
+		}
+		if self.parallel {
+			f.write_str(" PARALLEL")?
+		}
+		Ok(())
+	}
+}

core/src/sql/subquery.rs

@@ -6,6 +6,7 @@ use crate::sql::statements::rebuild::RebuildStatement;
 use crate::sql::statements::{
 	CreateStatement, DefineStatement, DeleteStatement, IfelseStatement, InsertStatement,
 	OutputStatement, RelateStatement, RemoveStatement, SelectStatement, UpdateStatement,
+	UpsertStatement,
 };
 use crate::sql::value::Value;
 use reblessive::tree::Stk;
@@ -16,7 +17,7 @@ use std::fmt::{self, Display, Formatter};
 
 pub(crate) const TOKEN: &str = "$surrealdb::private::sql::Subquery";
 
-#[revisioned(revision = 2)]
+#[revisioned(revision = 3)]
 #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, Hash)]
 #[serde(rename = "$surrealdb::private::sql::Subquery")]
 #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
@@ -35,7 +36,8 @@ pub enum Subquery {
 	Remove(RemoveStatement),
 	#[revision(start = 2)]
 	Rebuild(RebuildStatement),
-	// Add new variants here
+	#[revision(start = 3)]
+	Upsert(UpsertStatement),
 }
 
 impl PartialOrd for Subquery {
@@ -54,6 +56,7 @@ impl Subquery {
 			Self::Output(v) => v.writeable(),
 			Self::Select(v) => v.writeable(),
 			Self::Create(v) => v.writeable(),
+			Self::Upsert(v) => v.writeable(),
 			Self::Update(v) => v.writeable(),
 			Self::Delete(v) => v.writeable(),
 			Self::Relate(v) => v.writeable(),
@@ -87,6 +90,7 @@ impl Subquery {
 			Self::Remove(ref v) => v.compute(&ctx, opt, doc).await,
 			Self::Select(ref v) => v.compute(stk, &ctx, opt, doc).await,
 			Self::Create(ref v) => v.compute(stk, &ctx, opt, doc).await,
+			Self::Upsert(ref v) => v.compute(stk, &ctx, opt, doc).await,
 			Self::Update(ref v) => v.compute(stk, &ctx, opt, doc).await,
 			Self::Delete(ref v) => v.compute(stk, &ctx, opt, doc).await,
 			Self::Relate(ref v) => v.compute(stk, &ctx, opt, doc).await,
@@ -102,6 +106,7 @@ impl Display for Subquery {
 			Self::Output(v) => write!(f, "({v})"),
 			Self::Select(v) => write!(f, "({v})"),
 			Self::Create(v) => write!(f, "({v})"),
+			Self::Upsert(v) => write!(f, "({v})"),
 			Self::Update(v) => write!(f, "({v})"),
 			Self::Delete(v) => write!(f, "({v})"),
 			Self::Relate(v) => write!(f, "({v})"),

core/src/sql/value/serde/ser/block/entry/mod.rs

@@ -50,6 +50,9 @@ impl ser::Serializer for Serializer {
 			"Create" => {
 				Ok(Entry::Create(value.serialize(ser::statement::create::Serializer.wrap())?))
 			}
+			"Upsert" => {
+				Ok(Entry::Upsert(value.serialize(ser::statement::upsert::Serializer.wrap())?))
+			}
 			"Update" => {
 				Ok(Entry::Update(value.serialize(ser::statement::update::Serializer.wrap())?))
 			}
@@ -120,6 +123,13 @@ mod tests {
 		assert_eq!(entry, serialized);
 	}
 
+	#[test]
+	fn upsert() {
+		let entry = Entry::Upsert(Default::default());
+		let serialized = entry.serialize(Serializer.wrap()).unwrap();
+		assert_eq!(entry, serialized);
+	}
+
 	#[test]
 	fn update() {
 		let entry = Entry::Update(Default::default());

core/src/sql/value/serde/ser/statement/mod.rs

@@ -23,6 +23,7 @@ pub mod show;
 pub mod sleep;
 pub mod throw;
 pub mod update;
+pub mod upsert;
 pub mod vec;
 pub mod yuse;
 
@@ -86,6 +87,7 @@ impl ser::Serializer for Serializer {
 			"Sleep" => Ok(Statement::Sleep(value.serialize(sleep::Serializer.wrap())?)),
 			"Throw" => Ok(Statement::Throw(value.serialize(throw::Serializer.wrap())?)),
 			"Update" => Ok(Statement::Update(value.serialize(update::Serializer.wrap())?)),
+			"Upsert" => Ok(Statement::Upsert(value.serialize(upsert::Serializer.wrap())?)),
 			"Use" => Ok(Statement::Use(value.serialize(yuse::Serializer.wrap())?)),
 			variant => {
 				Err(Error::custom(format!("unexpected newtype variant `{name}::{variant}`")))
@@ -252,6 +254,13 @@ mod tests {
 		assert_eq!(statement, serialized);
 	}
 
+	#[test]
+	fn upsert() {
+		let statement = Statement::Upsert(Default::default());
+		let serialized = statement.serialize(Serializer.wrap()).unwrap();
+		assert_eq!(statement, serialized);
+	}
+
 	#[test]
 	fn yuse() {
 		let statement = Statement::Use(Default::default());

core/src/sql/value/serde/ser/statement/upsert.rs

@@ -0,0 +1,159 @@
+use crate::err::Error;
+use crate::sql::statements::UpsertStatement;
+use crate::sql::value::serde::ser;
+use crate::sql::Cond;
+use crate::sql::Data;
+use crate::sql::Duration;
+use crate::sql::Output;
+use crate::sql::Timeout;
+use crate::sql::Values;
+use ser::Serializer as _;
+use serde::ser::Error as _;
+use serde::ser::Impossible;
+use serde::ser::Serialize;
+
+#[non_exhaustive]
+pub struct Serializer;
+
+impl ser::Serializer for Serializer {
+	type Ok = UpsertStatement;
+	type Error = Error;
+
+	type SerializeSeq = Impossible<UpsertStatement, Error>;
+	type SerializeTuple = Impossible<UpsertStatement, Error>;
+	type SerializeTupleStruct = Impossible<UpsertStatement, Error>;
+	type SerializeTupleVariant = Impossible<UpsertStatement, Error>;
+	type SerializeMap = Impossible<UpsertStatement, Error>;
+	type SerializeStruct = SerializeUpsertStatement;
+	type SerializeStructVariant = Impossible<UpsertStatement, Error>;
+
+	const EXPECTED: &'static str = "a struct `UpsertStatement`";
+
+	#[inline]
+	fn serialize_struct(
+		self,
+		_name: &'static str,
+		_len: usize,
+	) -> Result<Self::SerializeStruct, Error> {
+		Ok(SerializeUpsertStatement::default())
+	}
+}
+
+#[derive(Default)]
+#[non_exhaustive]
+pub struct SerializeUpsertStatement {
+	only: Option<bool>,
+	what: Option<Values>,
+	data: Option<Data>,
+	cond: Option<Cond>,
+	output: Option<Output>,
+	timeout: Option<Timeout>,
+	parallel: Option<bool>,
+}
+
+impl serde::ser::SerializeStruct for SerializeUpsertStatement {
+	type Ok = UpsertStatement;
+	type Error = Error;
+
+	fn serialize_field<T>(&mut self, key: &'static str, value: &T) -> Result<(), Error>
+	where
+		T: ?Sized + Serialize,
+	{
+		match key {
+			"only" => {
+				self.only = Some(value.serialize(ser::primitive::bool::Serializer.wrap())?);
+			}
+			"what" => {
+				self.what = Some(Values(value.serialize(ser::value::vec::Serializer.wrap())?));
+			}
+			"data" => {
+				self.data = value.serialize(ser::data::opt::Serializer.wrap())?;
+			}
+			"cond" => {
+				self.cond = value.serialize(ser::cond::opt::Serializer.wrap())?;
+			}
+			"output" => {
+				self.output = value.serialize(ser::output::opt::Serializer.wrap())?;
+			}
+			"timeout" => {
+				if let Some(duration) = value.serialize(ser::duration::opt::Serializer.wrap())? {
+					self.timeout = Some(Timeout(Duration(duration)));
+				}
+			}
+			"parallel" => {
+				self.parallel = Some(value.serialize(ser::primitive::bool::Serializer.wrap())?);
+			}
+			key => {
+				return Err(Error::custom(format!("unexpected field `UpsertStatement::{key}`")));
+			}
+		}
+		Ok(())
+	}
+
+	fn end(self) -> Result<Self::Ok, Error> {
+		match (self.what, self.parallel) {
+			(Some(what), Some(parallel)) => Ok(UpsertStatement {
+				only: self.only.is_some_and(|v| v),
+				what,
+				parallel,
+				data: self.data,
+				cond: self.cond,
+				output: self.output,
+				timeout: self.timeout,
+			}),
+			_ => Err(Error::custom("`UpsertStatement` missing required field(s)")),
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn default() {
+		let stmt = UpsertStatement::default();
+		let value: UpsertStatement = stmt.serialize(Serializer.wrap()).unwrap();
+		assert_eq!(value, stmt);
+	}
+
+	#[test]
+	fn with_data() {
+		let stmt = UpsertStatement {
+			data: Some(Default::default()),
+			..Default::default()
+		};
+		let value: UpsertStatement = stmt.serialize(Serializer.wrap()).unwrap();
+		assert_eq!(value, stmt);
+	}
+
+	#[test]
+	fn with_cond() {
+		let stmt = UpsertStatement {
+			cond: Some(Default::default()),
+			..Default::default()
+		};
+		let value: UpsertStatement = stmt.serialize(Serializer.wrap()).unwrap();
+		assert_eq!(value, stmt);
+	}
+
+	#[test]
+	fn with_output() {
+		let stmt = UpsertStatement {
+			output: Some(Default::default()),
+			..Default::default()
+		};
+		let value: UpsertStatement = stmt.serialize(Serializer.wrap()).unwrap();
+		assert_eq!(value, stmt);
+	}
+
+	#[test]
+	fn with_timeout() {
+		let stmt = UpsertStatement {
+			timeout: Some(Default::default()),
+			..Default::default()
+		};
+		let value: UpsertStatement = stmt.serialize(Serializer.wrap()).unwrap();
+		assert_eq!(value, stmt);
+	}
+}

core/src/syn/lexer/keywords.rs

@@ -34,6 +34,7 @@ pub static RESERVED_KEYWORD: phf::Set<UniCase<&'static str>> = phf_set! {
 	UniCase::ascii("SLEEP"),
 	UniCase::ascii("THROW"),
 	UniCase::ascii("UPDATE"),
+	UniCase::ascii("UPSERT"),
 	UniCase::ascii("USE"),
 	UniCase::ascii("DIFF"),
 	UniCase::ascii("RAND"),
@@ -220,6 +221,7 @@ pub(crate) static KEYWORDS: phf::Map<UniCase<&'static str>, TokenKind> = phf_map
 	UniCase::ascii("UNIQUE") => TokenKind::Keyword(Keyword::Unique),
 	UniCase::ascii("UNSET") => TokenKind::Keyword(Keyword::Unset),
 	UniCase::ascii("UPDATE") => TokenKind::Keyword(Keyword::Update),
+	UniCase::ascii("UPSERT") => TokenKind::Keyword(Keyword::Upsert),
 	UniCase::ascii("UPPERCASE") => TokenKind::Keyword(Keyword::Uppercase),
 	UniCase::ascii("URL") => TokenKind::Keyword(Keyword::Url),
 	UniCase::ascii("USE") => TokenKind::Keyword(Keyword::Use),

core/src/syn/parser/prime.rs

@@ -81,6 +81,7 @@ impl Parser<'_> {
 			t!("RETURN")
 			| t!("SELECT")
 			| t!("CREATE")
+			| t!("UPSERT")
 			| t!("UPDATE")
 			| t!("DELETE")
 			| t!("RELATE")
@@ -241,6 +242,7 @@ impl Parser<'_> {
 			t!("RETURN")
 			| t!("SELECT")
 			| t!("CREATE")
+			| t!("UPSERT")
 			| t!("UPDATE")
 			| t!("DELETE")
 			| t!("RELATE")
@@ -389,6 +391,11 @@ impl Parser<'_> {
 				let stmt = ctx.run(|ctx| self.parse_create_stmt(ctx)).await?;
 				Subquery::Create(stmt)
 			}
+			t!("UPSERT") => {
+				self.pop_peek();
+				let stmt = ctx.run(|ctx| self.parse_upsert_stmt(ctx)).await?;
+				Subquery::Upsert(stmt)
+			}
 			t!("UPDATE") => {
 				self.pop_peek();
 				let stmt = ctx.run(|ctx| self.parse_update_stmt(ctx)).await?;
@@ -540,6 +547,11 @@ impl Parser<'_> {
 				let stmt = ctx.run(|ctx| self.parse_create_stmt(ctx)).await?;
 				Subquery::Create(stmt)
 			}
+			t!("UPSERT") => {
+				self.pop_peek();
+				let stmt = ctx.run(|ctx| self.parse_upsert_stmt(ctx)).await?;
+				Subquery::Upsert(stmt)
+			}
 			t!("UPDATE") => {
 				self.pop_peek();
 				let stmt = ctx.run(|ctx| self.parse_update_stmt(ctx)).await?;

core/src/syn/parser/stmt/mod.rs

@@ -35,6 +35,7 @@ mod relate;
 mod remove;
 mod select;
 mod update;
+mod upsert;
 
 impl Parser<'_> {
 	pub async fn parse_stmt_list(&mut self, ctx: &mut Stk) -> ParseResult<Statements> {
@@ -92,7 +93,8 @@ impl Parser<'_> {
 				| t!("REMOVE") | t!("SELECT")
 				| t!("LET") | t!("SHOW")
 				| t!("SLEEP") | t!("THROW")
-				| t!("UPDATE") | t!("USE")
+				| t!("UPDATE") | t!("UPSERT")
+				| t!("USE")
 		)
 	}
 
@@ -209,6 +211,10 @@ impl Parser<'_> {
 				self.pop_peek();
 				ctx.run(|ctx| self.parse_update_stmt(ctx)).await.map(Statement::Update)
 			}
+			t!("UPSERT") => {
+				self.pop_peek();
+				ctx.run(|ctx| self.parse_upsert_stmt(ctx)).await.map(Statement::Upsert)
+			}
 			t!("USE") => {
 				self.pop_peek();
 				self.parse_use_stmt().map(Statement::Use)
@@ -294,6 +300,10 @@ impl Parser<'_> {
 				self.pop_peek();
 				self.parse_update_stmt(ctx).await.map(Entry::Update)
 			}
+			t!("UPSERT") => {
+				self.pop_peek();
+				self.parse_upsert_stmt(ctx).await.map(Entry::Upsert)
+			}
 			_ => {
 				// TODO: Provide information about keywords.
 				let v = ctx.run(|ctx| self.parse_value_field(ctx)).await?;

core/src/syn/parser/stmt/relate.rs

@@ -79,6 +79,7 @@ impl Parser<'_> {
 			t!("RETURN")
 			| t!("SELECT")
 			| t!("CREATE")
+			| t!("UPSERT")
 			| t!("UPDATE")
 			| t!("DELETE")
 			| t!("RELATE")

core/src/syn/parser/stmt/upsert.rs

@@ -0,0 +1,31 @@
+use reblessive::Stk;
+
+use crate::{
+	sql::{statements::UpsertStatement, Values},
+	syn::{
+		parser::{ParseResult, Parser},
+		token::t,
+	},
+};
+
+impl Parser<'_> {
+	pub async fn parse_upsert_stmt(&mut self, stk: &mut Stk) -> ParseResult<UpsertStatement> {
+		let only = self.eat(t!("ONLY"));
+		let what = Values(self.parse_what_list(stk).await?);
+		let data = self.try_parse_data(stk).await?;
+		let cond = self.try_parse_condition(stk).await?;
+		let output = self.try_parse_output(stk).await?;
+		let timeout = self.try_parse_timeout()?;
+		let parallel = self.eat(t!("PARALLEL"));
+
+		Ok(UpsertStatement {
+			only,
+			what,
+			data,
+			cond,
+			output,
+			timeout,
+			parallel,
+		})
+	}
+}

core/src/syn/parser/test/stmt.rs

@@ -25,7 +25,7 @@ use crate::{
 			RemoveFieldStatement, RemoveFunctionStatement, RemoveIndexStatement,
 			RemoveNamespaceStatement, RemoveParamStatement, RemoveStatement, RemoveTableStatement,
 			RemoveUserStatement, SelectStatement, SetStatement, ThrowStatement, UpdateStatement,
-			UseStatement,
+			UpsertStatement, UseStatement,
 		},
 		tokenizer::Tokenizer,
 		user::UserDuration,
@@ -2160,3 +2160,49 @@ fn parse_update() {
 		})
 	);
 }
+
+#[test]
+fn parse_upsert() {
+	let res = test_parse!(
+		parse_stmt,
+		r#"UPSERT ONLY <future> { "text" }, a->b UNSET foo... , a->b, c[*] WHERE true RETURN DIFF TIMEOUT 1s PARALLEL"#
+	)
+	.unwrap();
+	assert_eq!(
+		res,
+		Statement::Upsert(UpsertStatement {
+			only: true,
+			what: Values(vec![
+				Value::Future(Box::new(Future(Block(vec![Entry::Value(Value::Strand(Strand(
+					"text".to_string()
+				)))])))),
+				Value::Idiom(Idiom(vec![
+					Part::Field(Ident("a".to_string())),
+					Part::Graph(Graph {
+						dir: Dir::Out,
+						what: Tables(vec![Table("b".to_string())]),
+						expr: Fields::all(),
+						..Default::default()
+					})
+				]))
+			]),
+			cond: Some(Cond(Value::Bool(true))),
+			data: Some(Data::UnsetExpression(vec![
+				Idiom(vec![Part::Field(Ident("foo".to_string())), Part::Flatten]),
+				Idiom(vec![
+					Part::Field(Ident("a".to_string())),
+					Part::Graph(Graph {
+						dir: Dir::Out,
+						what: Tables(vec![Table("b".to_string())]),
+						expr: Fields::all(),
+						..Default::default()
+					})
+				]),
+				Idiom(vec![Part::Field(Ident("c".to_string())), Part::All])
+			])),
+			output: Some(Output::Diff),
+			timeout: Some(Timeout(Duration(std::time::Duration::from_secs(1)))),
+			parallel: true,
+		})
+	);
+}

core/src/syn/parser/test/streaming.rs

@@ -17,6 +17,7 @@ use crate::{
 			ForeachStatement, IfelseStatement, InfoStatement, InsertStatement, KillStatement,
 			OutputStatement, RelateStatement, RemoveFieldStatement, RemoveFunctionStatement,
 			RemoveStatement, SelectStatement, SetStatement, ThrowStatement, UpdateStatement,
+			UpsertStatement,
 		},
 		tokenizer::Tokenizer,
 		Algorithm, Array, Base, Block, Cond, Data, Datetime, Dir, Duration, Edges, Explain,
@@ -99,6 +100,7 @@ static SOURCE: &str = r#"
 	REMOVE FUNCTION fn::foo::bar();
 	REMOVE FIELD foo.bar[10] ON bar;
 	UPDATE ONLY <future> { "text" }, a->b UNSET foo... , a->b, c[*] WHERE true RETURN DIFF TIMEOUT 1s PARALLEL;
+	UPSERT ONLY <future> { "text" }, a->b UNSET foo... , a->b, c[*] WHERE true RETURN DIFF TIMEOUT 1s PARALLEL;
 "#;
 
 fn statements() -> Vec<Statement> {
@@ -667,6 +669,40 @@ fn statements() -> Vec<Statement> {
 			timeout: Some(Timeout(Duration(std::time::Duration::from_secs(1)))),
 			parallel: true,
 		}),
+		Statement::Upsert(UpsertStatement {
+			only: true,
+			what: Values(vec![
+				Value::Future(Box::new(Future(Block(vec![Entry::Value(Value::Strand(Strand(
+					"text".to_string(),
+				)))])))),
+				Value::Idiom(Idiom(vec![
+					Part::Field(Ident("a".to_string())),
+					Part::Graph(Graph {
+						dir: Dir::Out,
+						what: Tables(vec![Table("b".to_string())]),
+						expr: Fields::all(),
+						..Default::default()
+					}),
+				])),
+			]),
+			cond: Some(Cond(Value::Bool(true))),
+			data: Some(Data::UnsetExpression(vec![
+				Idiom(vec![Part::Field(Ident("foo".to_string())), Part::Flatten]),
+				Idiom(vec![
+					Part::Field(Ident("a".to_string())),
+					Part::Graph(Graph {
+						dir: Dir::Out,
+						what: Tables(vec![Table("b".to_string())]),
+						expr: Fields::all(),
+						..Default::default()
+					}),
+				]),
+				Idiom(vec![Part::Field(Ident("c".to_string())), Part::All]),
+			])),
+			output: Some(Output::Diff),
+			timeout: Some(Timeout(Duration(std::time::Duration::from_secs(1)))),
+			parallel: true,
+		}),
 	]
 }
 

core/src/syn/token/keyword.rs

@@ -175,6 +175,7 @@ keyword! {
 	Unique => "UNIQUE",
 	Unset => "UNSET",
 	Update => "UPDATE",
+	Upsert => "UPSERT",
 	Uppercase => "UPPERCASE",
 	Url => "URL",
 	Use => "USE",

lib/examples/rocket/src/person.rs

@@ -62,7 +62,7 @@ pub async fn update(
 	id: String,
 	person_data: Json<Person>,
 ) -> Result<Json<Option<Person>>, Custom<String>> {
-	db.update((PERSON, &*id))
+	db.upsert((PERSON, &*id))
 		.content(person_data.into_inner())
 		.await
 		.map_err(|e| Custom(Status::InternalServerError, e.to_string()))

lib/examples/rocket/tests/integration_test.rs

@@ -25,6 +25,7 @@ async fn read_person(client: &Client, id: i32) -> Person {
 	let response = client.get(format!("/person/{}", id)).dispatch().await;
 	assert_eq!(response.status(), Status::Ok);
 	let body = response.into_string().await.unwrap();
+	println!("{body}");
 	serde_json::from_str(&body).unwrap()
 }
 

lib/src/api/conn.rs

@@ -92,6 +92,8 @@ pub enum Method {
 	Unset,
 	/// Performs an update operation
 	Update,
+	/// Performs an upsert operation
+	Upsert,
 	/// Selects a namespace and database to use
 	Use,
 	/// Queries the version of the server

lib/src/api/engine/local/mod.rs

@@ -38,6 +38,7 @@ use crate::api::engine::merge_statement;
 use crate::api::engine::patch_statement;
 use crate::api::engine::select_statement;
 use crate::api::engine::update_statement;
+use crate::api::engine::upsert_statement;
 #[cfg(not(target_arch = "wasm32"))]
 use crate::api::err::Error;
 use crate::api::Connect;
@@ -572,6 +573,14 @@ async fn router(
 			let value = take(true, response).await?;
 			Ok(DbResponse::Other(value))
 		}
+		Method::Upsert => {
+			let mut query = Query::default();
+			let (one, statement) = upsert_statement(&mut params);
+			query.0 .0 = vec![Statement::Upsert(statement)];
+			let response = kvs.process(query, &*session, Some(vars.clone())).await?;
+			let value = take(one, response).await?;
+			Ok(DbResponse::Other(value))
+		}
 		Method::Update => {
 			let mut query = Query::default();
 			let (one, statement) = update_statement(&mut params);

lib/src/api/engine/mod.rs

@@ -21,6 +21,7 @@ use crate::sql::statements::DeleteStatement;
 use crate::sql::statements::InsertStatement;
 use crate::sql::statements::SelectStatement;
 use crate::sql::statements::UpdateStatement;
+use crate::sql::statements::UpsertStatement;
 use crate::sql::Data;
 use crate::sql::Field;
 use crate::sql::Output;
@@ -77,6 +78,20 @@ fn create_statement(params: &mut [Value]) -> CreateStatement {
 	stmt
 }
 
+#[allow(dead_code)] // used by the the embedded database and `http`
+fn upsert_statement(params: &mut [Value]) -> (bool, UpsertStatement) {
+	let (one, what, data) = split_params(params);
+	let data = match data {
+		Value::None | Value::Null => None,
+		value => Some(Data::ContentExpression(value)),
+	};
+	let mut stmt = UpsertStatement::default();
+	stmt.what = what;
+	stmt.data = data;
+	stmt.output = Some(Output::After);
+	(one, stmt)
+}
+
 #[allow(dead_code)] // used by the the embedded database and `http`
 fn update_statement(params: &mut [Value]) -> (bool, UpdateStatement) {
 	let (one, what, data) = split_params(params);

lib/src/api/engine/remote/http/mod.rs

@@ -19,6 +19,7 @@ use crate::api::engine::patch_statement;
 use crate::api::engine::remote::duration_from_str;
 use crate::api::engine::select_statement;
 use crate::api::engine::update_statement;
+use crate::api::engine::upsert_statement;
 use crate::api::err::Error;
 use crate::api::method::query::QueryResult;
 use crate::api::Connect;
@@ -465,6 +466,14 @@ async fn router(
 			let value = take(true, request).await?;
 			Ok(DbResponse::Other(value))
 		}
+		Method::Upsert => {
+			let path = base_url.join(SQL_PATH)?;
+			let (one, statement) = upsert_statement(&mut params);
+			let request =
+				client.post(path).headers(headers.clone()).auth(auth).body(statement.to_string());
+			let value = take(one, request).await?;
+			Ok(DbResponse::Other(value))
+		}
 		Method::Update => {
 			let path = base_url.join(SQL_PATH)?;
 			let (one, statement) = update_statement(&mut params);

lib/src/api/method/mod.rs

@@ -23,6 +23,7 @@ mod signin;
 mod signup;
 mod unset;
 mod update;
+mod upsert;
 mod use_db;
 mod use_ns;
 mod version;
@@ -60,6 +61,7 @@ pub use signup::Signup;
 use tokio::sync::watch;
 pub use unset::Unset;
 pub use update::Update;
+pub use upsert::Upsert;
 pub use use_db::UseDb;
 pub use use_ns::UseNs;
 pub use version::Version;
@@ -129,6 +131,7 @@ impl Method {
 			Method::Signup => "signup",
 			Method::Unset => "unset",
 			Method::Update => "update",
+			Method::Upsert => "upsert",
 			Method::Use => "use",
 			Method::Version => "version",
 		}
@@ -879,6 +882,165 @@ where
 		}
 	}
 
+	/// Updates all records in a table, or a specific record
+	///
+	/// # Examples
+	///
+	/// Replace the current document / record data with the specified data.
+	///
+	/// ```no_run
+	/// use serde::Serialize;
+	///
+	/// # #[derive(serde::Deserialize)]
+	/// # struct Person;
+	/// #
+	/// #[derive(Serialize)]
+	/// struct Settings {
+	///     active: bool,
+	///     marketing: bool,
+	/// }
+	///
+	/// #[derive(Serialize)]
+	/// struct User {
+	///     name: &'static str,
+	///     settings: Settings,
+	/// }
+	///
+	/// # #[tokio::main]
+	/// # async fn main() -> surrealdb::Result<()> {
+	/// # let db = surrealdb::engine::any::connect("mem://").await?;
+	/// #
+	/// // Select the namespace/database to use
+	/// db.use_ns("namespace").use_db("database").await?;
+	///
+	/// // Update all records in a table
+	/// let people: Vec<Person> = db.upsert("person").await?;
+	///
+	/// // Update a record with a specific ID
+	/// let person: Option<Person> = db.upsert(("person", "tobie"))
+	///     .content(User {
+	///         name: "Tobie",
+	///         settings: Settings {
+	///             active: true,
+	///             marketing: true,
+	///         },
+	///     })
+	///     .await?;
+	/// #
+	/// # Ok(())
+	/// # }
+	/// ```
+	///
+	/// Merge the current document / record data with the specified data.
+	///
+	/// ```no_run
+	/// use serde::Serialize;
+	/// use time::OffsetDateTime;
+	///
+	/// # #[derive(serde::Deserialize)]
+	/// # struct Person;
+	/// #
+	/// #[derive(Serialize)]
+	/// struct UpdatedAt {
+	///     updated_at: OffsetDateTime,
+	/// }
+	///
+	/// #[derive(Serialize)]
+	/// struct Settings {
+	///     active: bool,
+	/// }
+	///
+	/// #[derive(Serialize)]
+	/// struct User {
+	///     updated_at: OffsetDateTime,
+	///     settings: Settings,
+	/// }
+	///
+	/// # #[tokio::main]
+	/// # async fn main() -> surrealdb::Result<()> {
+	/// # let db = surrealdb::engine::any::connect("mem://").await?;
+	/// #
+	/// // Select the namespace/database to use
+	/// db.use_ns("namespace").use_db("database").await?;
+	///
+	/// // Update all records in a table
+	/// let people: Vec<Person> = db.upsert("person")
+	///     .merge(UpdatedAt {
+	///         updated_at: OffsetDateTime::now_utc(),
+	///     })
+	///     .await?;
+	///
+	/// // Update a record with a specific ID
+	/// let person: Option<Person> = db.upsert(("person", "tobie"))
+	///     .merge(User {
+	///         updated_at: OffsetDateTime::now_utc(),
+	///         settings: Settings {
+	///             active: true,
+	///         },
+	///     })
+	///     .await?;
+	/// #
+	/// # Ok(())
+	/// # }
+	/// ```
+	///
+	/// Apply [JSON Patch](https://jsonpatch.com) changes to all records, or a specific record, in the database.
+	///
+	/// ```no_run
+	/// use serde::Serialize;
+	/// use surrealdb::opt::PatchOp;
+	/// use time::OffsetDateTime;
+	///
+	/// # #[derive(serde::Deserialize)]
+	/// # struct Person;
+	/// #
+	/// #[derive(Serialize)]
+	/// struct UpdatedAt {
+	///     updated_at: OffsetDateTime,
+	/// }
+	///
+	/// #[derive(Serialize)]
+	/// struct Settings {
+	///     active: bool,
+	/// }
+	///
+	/// #[derive(Serialize)]
+	/// struct User {
+	///     updated_at: OffsetDateTime,
+	///     settings: Settings,
+	/// }
+	///
+	/// # #[tokio::main]
+	/// # async fn main() -> surrealdb::Result<()> {
+	/// # let db = surrealdb::engine::any::connect("mem://").await?;
+	/// #
+	/// // Select the namespace/database to use
+	/// db.use_ns("namespace").use_db("database").await?;
+	///
+	/// // Update all records in a table
+	/// let people: Vec<Person> = db.upsert("person")
+	///     .patch(PatchOp::replace("/created_at", OffsetDateTime::now_utc()))
+	///     .await?;
+	///
+	/// // Update a record with a specific ID
+	/// let person: Option<Person> = db.upsert(("person", "tobie"))
+	///     .patch(PatchOp::replace("/settings/active", false))
+	///     .patch(PatchOp::add("/tags", ["developer", "engineer"]))
+	///     .patch(PatchOp::remove("/temp"))
+	///     .await?;
+	/// #
+	/// # Ok(())
+	/// # }
+	/// ```
+	pub fn upsert<R>(&self, resource: impl opt::IntoResource<R>) -> Upsert<C, R> {
+		Upsert {
+			client: Cow::Borrowed(self),
+			resource: resource.into_resource(),
+			range: None,
+			response_type: PhantomData,
+		}
+	}
+
 	/// Updates all records in a table, or a specific record
 	///
 	/// # Examples

lib/src/api/method/tests/server.rs

@@ -67,7 +67,8 @@ pub(super) fn mock(route_rx: Receiver<Option<Route>>) {
 					}
 					_ => unreachable!(),
 				},
-				Method::Update | Method::Merge | Method::Patch => match &params[..] {
+				Method::Upsert | Method::Update | Method::Merge | Method::Patch => {
+					match &params[..] {
 						[Value::Thing(..)] | [Value::Thing(..), _] => {
 							Ok(DbResponse::Other(to_value(User::default()).unwrap()))
 						}
@@ -76,7 +77,8 @@ pub(super) fn mock(route_rx: Receiver<Option<Route>>) {
 							Ok(DbResponse::Other(Value::Array(Default::default())))
 						}
 						_ => unreachable!(),
-				},
+					}
+				}
 				Method::Insert => match &params[..] {
 					[Value::Table(..), Value::Array(..)] => {
 						Ok(DbResponse::Other(Value::Array(Default::default())))

lib/src/api/method/upsert.rs

@@ -0,0 +1,165 @@
+use crate::api::conn::Method;
+use crate::api::conn::Param;
+use crate::api::method::Content;
+use crate::api::method::Merge;
+use crate::api::method::Patch;
+use crate::api::opt::PatchOp;
+use crate::api::opt::Range;
+use crate::api::opt::Resource;
+use crate::api::Connection;
+use crate::api::Result;
+use crate::method::OnceLockExt;
+use crate::sql::Id;
+use crate::sql::Value;
+use crate::Surreal;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use std::borrow::Cow;
+use std::future::Future;
+use std::future::IntoFuture;
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+/// An upsert future
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct Upsert<'r, C: Connection, R> {
+	pub(super) client: Cow<'r, Surreal<C>>,
+	pub(super) resource: Result<Resource>,
+	pub(super) range: Option<Range<Id>>,
+	pub(super) response_type: PhantomData<R>,
+}
+
+impl<C, R> Upsert<'_, C, R>
+where
+	C: Connection,
+{
+	/// Converts to an owned type which can easily be moved to a different thread
+	pub fn into_owned(self) -> Upsert<'static, C, R> {
+		Upsert {
+			client: Cow::Owned(self.client.into_owned()),
+			..self
+		}
+	}
+}
+
+macro_rules! into_future {
+	($method:ident) => {
+		fn into_future(self) -> Self::IntoFuture {
+			let Upsert {
+				client,
+				resource,
+				range,
+				..
+			} = self;
+			Box::pin(async move {
+				let param = match range {
+					Some(range) => resource?.with_range(range)?.into(),
+					None => resource?.into(),
+				};
+				let mut conn = Client::new(Method::Upsert);
+				conn.$method(client.router.extract()?, Param::new(vec![param])).await
+			})
+		}
+	};
+}
+
+impl<'r, Client> IntoFuture for Upsert<'r, Client, Value>
+where
+	Client: Connection,
+{
+	type Output = Result<Value>;
+	type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + Sync + 'r>>;
+
+	into_future! {execute_value}
+}
+
+impl<'r, Client, R> IntoFuture for Upsert<'r, Client, Option<R>>
+where
+	Client: Connection,
+	R: DeserializeOwned,
+{
+	type Output = Result<Option<R>>;
+	type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + Sync + 'r>>;
+
+	into_future! {execute_opt}
+}
+
+impl<'r, Client, R> IntoFuture for Upsert<'r, Client, Vec<R>>
+where
+	Client: Connection,
+	R: DeserializeOwned,
+{
+	type Output = Result<Vec<R>>;
+	type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send + Sync + 'r>>;
+
+	into_future! {execute_vec}
+}
+
+impl<C> Upsert<'_, C, Value>
+where
+	C: Connection,
+{
+	/// Restricts the records to upsert to those in the specified range
+	pub fn range(mut self, bounds: impl Into<Range<Id>>) -> Self {
+		self.range = Some(bounds.into());
+		self
+	}
+}
+
+impl<C, R> Upsert<'_, C, Vec<R>>
+where
+	C: Connection,
+{
+	/// Restricts the records to upsert to those in the specified range
+	pub fn range(mut self, bounds: impl Into<Range<Id>>) -> Self {
+		self.range = Some(bounds.into());
+		self
+	}
+}
+
+impl<'r, C, R> Upsert<'r, C, R>
+where
+	C: Connection,
+	R: DeserializeOwned,
+{
+	/// Replaces the current document / record data with the specified data
+	pub fn content<D>(self, data: D) -> Content<'r, C, D, R>
+	where
+		D: Serialize,
+	{
+		Content {
+			client: self.client,
+			method: Method::Upsert,
+			resource: self.resource,
+			range: self.range,
+			content: data,
+			response_type: PhantomData,
+		}
+	}
+
+	/// Merges the current document / record data with the specified data
+	pub fn merge<D>(self, data: D) -> Merge<'r, C, D, R>
+	where
+		D: Serialize,
+	{
+		Merge {
+			client: self.client,
+			resource: self.resource,
+			range: self.range,
+			content: data,
+			response_type: PhantomData,
+		}
+	}
+
+	/// Patches the current document / record data with the specified JSON Patch data
+	pub fn patch(self, PatchOp(patch): PatchOp) -> Patch<'r, C, R> {
+		Patch {
+			client: self.client,
+			resource: self.resource,
+			range: self.range,
+			patches: vec![patch],
+			response_type: PhantomData,
+		}
+	}
+}

lib/tests/cache.rs

@@ -65,10 +65,10 @@ async fn clear_transaction_cache_field() -> Result<(), Error> {
 	let sql = "
 		DEFINE FIELD test ON person TYPE option<string> VALUE 'test';
 		BEGIN;
-		UPDATE person:one CONTENT { x: 0 };
+		UPSERT person:one CONTENT { x: 0 };
 		SELECT * FROM person;
 		REMOVE FIELD test ON person;
-		UPDATE person:two CONTENT { x: 0 };
+		UPSERT person:two CONTENT { x: 0 };
 		SELECT * FROM person;
 		COMMIT;
 	";

lib/tests/changefeeds.rs

@@ -42,7 +42,7 @@ async fn database_change_feeds() -> Result<(), Error> {
 	"
 	);
 	let sql2 = "
-		UPDATE person:test CONTENT { name: 'Tobie' };
+		UPSERT person:test CONTENT { name: 'Tobie' };
 		DELETE person:test;
         SHOW CHANGES FOR TABLE person SINCE 0;
 	";
@@ -217,11 +217,11 @@ async fn table_change_feeds() -> Result<(), Error> {
 					$value
 				END
 		;
-		UPDATE person:test CONTENT { name: 'Tobie' };
+		UPSERT person:test CONTENT { name: 'Tobie' };
-		UPDATE person:test REPLACE { name: 'jaime' };
+		UPSERT person:test REPLACE { name: 'jaime' };
-		UPDATE person:test MERGE { name: 'Jaime' };
+		UPSERT person:test MERGE { name: 'Jaime' };
-		UPDATE person:test SET name = 'tobie';
+		UPSERT person:test SET name = 'tobie';
-		UPDATE person:test SET name = 'Tobie';
+		UPSERT person:test SET name = 'Tobie';
 		DELETE person:test;
 		CREATE person:1000 SET name = 'Yusuke';
         SHOW CHANGES FOR TABLE person SINCE 0;

lib/tests/define.rs

@@ -311,9 +311,9 @@ async fn define_statement_event() -> Result<(), Error> {
 			CREATE activity SET user = $this, value = $after.email, action = $event
 		);
 		INFO FOR TABLE user;
-		UPDATE user:test SET email = 'info@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'info@surrealdb.com', updated_at = time::now();
-		UPDATE user:test SET email = 'info@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'info@surrealdb.com', updated_at = time::now();
-		UPDATE user:test SET email = 'test@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'test@surrealdb.com', updated_at = time::now();
 		SELECT count() FROM activity GROUP ALL;
 	";
 	let mut t = Test::new(sql).await?;
@@ -349,9 +349,9 @@ async fn define_statement_event_when_event() -> Result<(), Error> {
 			CREATE activity SET user = $this, value = $after.email, action = $event
 		);
 		INFO FOR TABLE user;
-		UPDATE user:test SET email = 'info@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'info@surrealdb.com', updated_at = time::now();
-		UPDATE user:test SET email = 'info@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'info@surrealdb.com', updated_at = time::now();
-		UPDATE user:test SET email = 'test@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'test@surrealdb.com', updated_at = time::now();
 		SELECT count() FROM activity GROUP ALL;
 	";
 	let mut t = Test::new(sql).await?;
@@ -384,7 +384,7 @@ async fn define_statement_event_check_doc_always_populated() -> Result<(), Error
 			CREATE type::thing('log', $event) SET this = $doc, value = $value, before = $before, after = $after;
 		};
 		CREATE test:1 SET num = 1;
-		UPDATE test:1 set num = 2;
+		UPSERT test:1 set num = 2;
 		DELETE test:1;
 		SELECT * FROM log;
 	";
@@ -445,9 +445,9 @@ async fn define_statement_event_when_logic() -> Result<(), Error> {
 			CREATE activity SET user = $this, value = $after.email, action = $event
 		);
 		INFO FOR TABLE user;
-		UPDATE user:test SET email = 'info@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'info@surrealdb.com', updated_at = time::now();
-		UPDATE user:test SET email = 'info@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'info@surrealdb.com', updated_at = time::now();
-		UPDATE user:test SET email = 'test@surrealdb.com', updated_at = time::now();
+		UPSERT user:test SET email = 'test@surrealdb.com', updated_at = time::now();
 		SELECT count() FROM activity GROUP ALL;
 	";
 	let mut t = Test::new(sql).await?;
@@ -637,8 +637,8 @@ async fn define_statement_index_single_simple() -> Result<(), Error> {
 		REMOVE INDEX test ON user;
 		DEFINE INDEX test ON user COLUMNS age;
 		INFO FOR TABLE user;
-		UPDATE user:1 SET age = 24;
+		UPSERT user:1 SET age = 24;
-		UPDATE user:2 SET age = 11;
+		UPSERT user:2 SET age = 11;
 	";
 	let mut t = Test::new(sql).await?;
 	t.skip_ok(5)?;

lib/tests/field.rs

@@ -143,7 +143,7 @@ async fn field_definition_empty_nested_objects() -> Result<(), Error> {
 	let sql = "
 		DEFINE TABLE person SCHEMAFULL;
 		DEFINE FIELD settings on person TYPE object;
-		UPDATE person:test CONTENT {
+		UPSERT person:test CONTENT {
 		    settings: {
 		        nested: {
 		            object: {
@@ -195,7 +195,7 @@ async fn field_definition_empty_nested_arrays() -> Result<(), Error> {
 	let sql = "
 		DEFINE TABLE person SCHEMAFULL;
 		DEFINE FIELD settings on person TYPE object;
-		UPDATE person:test CONTENT {
+		UPSERT person:test CONTENT {
 		    settings: {
 		        nested: [
 					1,
@@ -249,7 +249,7 @@ async fn field_definition_empty_nested_flexible() -> Result<(), Error> {
 	let sql = "
 		DEFINE TABLE person SCHEMAFULL;
 		DEFINE FIELD settings on person FLEXIBLE TYPE object;
-		UPDATE person:test CONTENT {
+		UPSERT person:test CONTENT {
 		    settings: {
 				nested: {
 		            object: {
@@ -465,9 +465,9 @@ async fn field_definition_default_value() -> Result<(), Error> {
 		CREATE product:test SET tertiary = 123;
 		CREATE product:test;
 		--
-		UPDATE product:test SET primary = 654.321;
+		UPSERT product:test SET primary = 654.321;
-		UPDATE product:test SET secondary = false;
+		UPSERT product:test SET secondary = false;
-		UPDATE product:test SET tertiary = 'something';
+		UPSERT product:test SET tertiary = 'something';
 	";
 	let dbs = new_ds().await?;
 	let ses = Session::owner().with_ns("test").with_db("test");
@@ -872,8 +872,8 @@ async fn field_definition_readonly() -> Result<(), Error> {
 		DEFINE TABLE person SCHEMAFULL;
 		DEFINE FIELD birthdate ON person TYPE datetime READONLY;
 		CREATE person:test SET birthdate = d'2023-12-13T21:27:55.632Z';
-		UPDATE person:test SET birthdate = d'2023-12-13T21:27:55.632Z';
+		UPSERT person:test SET birthdate = d'2023-12-13T21:27:55.632Z';
-		UPDATE person:test SET birthdate = d'2024-12-13T21:27:55.632Z';
+		UPSERT person:test SET birthdate = d'2024-12-13T21:27:55.632Z';
 	";
 	let dbs = new_ds().await?;
 	let ses = Session::owner().with_ns("test").with_db("test");

lib/tests/foreach.rs

@@ -13,21 +13,21 @@ async fn foreach_simple() -> Result<(), Error> {
 			IF $test == 2 {
 				BREAK;
 			};
-			UPDATE type::thing('person', $test) SET test = $test;
+			UPSERT type::thing('person', $test) SET test = $test;
 		};
 		SELECT * FROM person;
 		FOR $test IN [4, 5, 6] {
 			IF $test == 5 {
 				CONTINUE;
 			};
-			UPDATE type::thing('person', $test) SET test = $test;
+			UPSERT type::thing('person', $test) SET test = $test;
 		};
 		SELECT * FROM person;
 		FOR $test IN <future> { [7, 8, 9] } {
 			IF $test > 8 {
 				THROW 'This is an error';
 			};
-			UPDATE type::thing('person', $test) SET test = $test;
+			UPSERT type::thing('person', $test) SET test = $test;
 		};
 		SELECT * FROM person;
 	";

lib/tests/future.rs

@@ -9,9 +9,9 @@ use surrealdb::sql::Value;
 #[tokio::test]
 async fn future_function_simple() -> Result<(), Error> {
 	let sql = "
-		UPDATE person:test SET can_drive = <future> { birthday && time::now() > birthday + 18y };
+		UPSERT person:test SET can_drive = <future> { birthday && time::now() > birthday + 18y };
-		UPDATE person:test SET birthday = <datetime> '2007-06-22';
+		UPSERT person:test SET birthday = <datetime> '2007-06-22';
-		UPDATE person:test SET birthday = <datetime> '2001-06-22';
+		UPSERT person:test SET birthday = <datetime> '2001-06-22';
 	";
 	let dbs = new_ds().await?;
 	let ses = Session::owner().with_ns("test").with_db("test");
@@ -38,7 +38,7 @@ async fn future_function_simple() -> Result<(), Error> {
 #[tokio::test]
 async fn future_function_arguments() -> Result<(), Error> {
 	let sql = "
-		UPDATE future:test SET
+		UPSERT future:test SET
 			a = 'test@surrealdb.com',
 			b = <future> { 'test@surrealdb.com' },
 			x = 'a-' + parse::email::user(a),

lib/tests/geometry.rs

@@ -9,7 +9,7 @@ use surrealdb::sql::Value;
 #[tokio::test]
 async fn geometry_point() -> Result<(), Error> {
 	let sql = "
-		UPDATE city:london SET centre = (-0.118092, 51.509865);
+		UPSERT city:london SET centre = (-0.118092, 51.509865);
 		SELECT * FROM city:london;
 	";
 	let dbs = new_ds().await?;
@@ -51,7 +51,7 @@ async fn geometry_point() -> Result<(), Error> {
 #[tokio::test]
 async fn geometry_polygon() -> Result<(), Error> {
 	let sql = "
-		UPDATE city:london SET area = {
+		UPSERT city:london SET area = {
 			type: 'Polygon',
 			coordinates: [[
 				[-0.38314819, 51.37692386], [0.1785278, 51.37692386],
@@ -59,7 +59,7 @@ async fn geometry_polygon() -> Result<(), Error> {
 				[-0.38314819, 51.37692386]
 			]]
 		};
-		UPDATE city:london SET area = {
+		UPSERT city:london SET area = {
 			type: 'Polygon',
 			coordinates: [[
 				[-0.38314819, 51.37692386], [0.1785278, 51.37692386],
@@ -146,14 +146,14 @@ async fn geometry_polygon() -> Result<(), Error> {
 #[tokio::test]
 async fn geometry_multipoint() -> Result<(), Error> {
 	let sql = "
-		UPDATE city:london SET points = {
+		UPSERT city:london SET points = {
 			type: 'MultiPoint',
 			coordinates: [
 				[-0.118092, 51.509865],
 				[-0.118092, 51.509865]
 			]
 		};
-		UPDATE city:london SET points = {
+		UPSERT city:london SET points = {
 			type: 'MultiPoint',
 			coordinates: [
 				[-0.118092, 51.509865],
@@ -224,14 +224,14 @@ async fn geometry_multipoint() -> Result<(), Error> {
 #[tokio::test]
 async fn geometry_multipolygon() -> Result<(), Error> {
 	let sql = "
-		UPDATE university:oxford SET area = {
+		UPSERT university:oxford SET area = {
 			type: 'MultiPolygon',
 			coordinates: [
 				[[ [10.0, 11.2], [10.5, 11.9], [10.8, 12.0], [10.0, 11.2] ]],
 				[[ [9.0, 11.2], [10.5, 11.9], [10.3, 13.0], [9.0, 11.2] ]]
 			]
 		};
-		UPDATE university:oxford SET area = {
+		UPSERT university:oxford SET area = {
 			type: 'MultiPolygon',
 			coordinates: [
 				[[ [10.0, 11.2], [10.5, 11.9], [10.8, 12.0], [10.0, 11.2] ]],

lib/tests/merge.rs

@@ -9,8 +9,8 @@ use surrealdb::sql::Value;
 #[tokio::test]
 async fn merge_record() -> Result<(), Error> {
 	let sql = "
-		UPDATE person:test SET name.initials = 'TMH', name.first = 'Tobie', name.last = 'Morgan Hitchcock';
+		UPSERT person:test SET name.initials = 'TMH', name.first = 'Tobie', name.last = 'Morgan Hitchcock';
-		UPDATE person:test MERGE {
+		UPSERT person:test MERGE {
 			name: {
 				title: 'Mr',
 				initials: NONE,

lib/tests/query.rs

@@ -105,8 +105,8 @@ async fn query_root_function() -> Result<(), Error> {
 #[tokio::test]
 async fn query_root_record() -> Result<(), Error> {
 	let sql = "
-		UPDATE person:tobie SET name = 'Tobie';
+		UPSERT person:tobie SET name = 'Tobie';
-		UPDATE person:jaime SET name = 'Jaime';
+		UPSERT person:jaime SET name = 'Jaime';
 		RELATE person:tobie->knows->person:jaime SET id = 'test', brother = true;
 		<future> { person:tobie->knows->person.name };
 		person:tobie->knows->person.name;

lib/tests/select.rs

@@ -241,7 +241,7 @@ async fn select_expression_value() -> Result<(), Error> {
 async fn select_dynamic_array_keys_and_object_keys() -> Result<(), Error> {
 	let sql = "
 		LET $lang = 'en';
-		UPDATE documentation:test CONTENT {
+		UPSERT documentation:test CONTENT {
 			primarylang: 'en',
 			languages: {
 				'en': 'this is english',
@@ -259,11 +259,11 @@ async fn select_dynamic_array_keys_and_object_keys() -> Result<(), Error> {
 		-- Selecting an object value or array index using a string as a key
 		SELECT languages['en'] AS content FROM documentation:test;
 		-- Updating an object value or array index using a string as a key
-		UPDATE documentation:test SET languages['en'] = 'my primary text';
+		UPSERT documentation:test SET languages['en'] = 'my primary text';
 		-- Selecting an object value or array index using a parameter as a key
 		SELECT languages[$lang] AS content FROM documentation:test;
 		-- Updating an object value or array index using a parameter as a key
-		UPDATE documentation:test SET languages[$lang] = 'my secondary text';
+		UPSERT documentation:test SET languages[$lang] = 'my secondary text';
 		-- Selecting an object or array index value using the value of another document field as a key
 		SELECT languages[primarylang] AS content FROM documentation;
 	";
@@ -359,11 +359,11 @@ async fn select_dynamic_array_keys_and_object_keys() -> Result<(), Error> {
 #[tokio::test]
 async fn select_writeable_subqueries() -> Result<(), Error> {
 	let sql = "
-		LET $id = (UPDATE tester:test);
+		LET $id = (UPSERT tester:test);
 		RETURN $id;
-		LET $id = (UPDATE tester:test).id;
+		LET $id = (UPSERT tester:test).id;
 		RETURN $id;
-		LET $id = (SELECT VALUE id FROM (UPDATE tester:test))[0];
+		LET $id = (SELECT VALUE id FROM (UPSERT tester:test))[0];
 		RETURN $id;
 	";
 	let dbs = new_ds().await?;

lib/tests/subquery.rs

@@ -113,9 +113,9 @@ async fn subquery_ifelse_set() -> Result<(), Error> {
 		RETURN $record;
 		-- Update the record field if it exists
 		IF $record.count THEN
-			(UPDATE person:test SET sport +?= 'football' RETURN sport)
+			(UPSERT person:test SET sport +?= 'football' RETURN sport)
 		ELSE
-			(UPDATE person:test SET sport = ['basketball'] RETURN sport)
+			(UPSERT person:test SET sport = ['basketball'] RETURN sport)
 		END;
 		-- Check if the record exists
 		LET $record = SELECT *, count() AS count FROM person:test;
@@ -123,9 +123,9 @@ async fn subquery_ifelse_set() -> Result<(), Error> {
 		RETURN $record;
 		-- Update the record field if it exists
 		IF $record.count THEN
-			UPDATE person:test SET sport +?= 'football' RETURN sport
+			UPSERT person:test SET sport +?= 'football' RETURN sport
 		ELSE
-			UPDATE person:test SET sport = ['basketball'] RETURN sport
+			UPSERT person:test SET sport = ['basketball'] RETURN sport
 		END;
 		-- Check if the record exists
 		LET $record = SELECT *, count() AS count FROM person:test;
@@ -133,9 +133,9 @@ async fn subquery_ifelse_set() -> Result<(), Error> {
 		RETURN $record;
 		-- Update the record field if it exists
 		IF $record.count THEN
-			UPDATE person:test SET sport +?= 'football' RETURN sport;
+			UPSERT person:test SET sport +?= 'football' RETURN sport;
 		ELSE
-			UPDATE person:test SET sport = ['basketball'] RETURN sport;
+			UPSERT person:test SET sport = ['basketball'] RETURN sport;
 		END;
 	";
 	let dbs = new_ds().await?;
@@ -238,9 +238,9 @@ async fn subquery_ifelse_array() -> Result<(), Error> {
 		RETURN $record;
 		-- Update the record field if it exists
 		IF $record.count THEN
-			(UPDATE person:test SET sport += 'football' RETURN sport)
+			(UPSERT person:test SET sport += 'football' RETURN sport)
 		ELSE
-			(UPDATE person:test SET sport = ['basketball'] RETURN sport)
+			(UPSERT person:test SET sport = ['basketball'] RETURN sport)
 		END;
 		-- Check if the record exists
 		LET $record = SELECT *, count() AS count FROM person:test;
@@ -248,9 +248,9 @@ async fn subquery_ifelse_array() -> Result<(), Error> {
 		RETURN $record;
 		-- Update the record field if it exists
 		IF $record.count THEN
-			UPDATE person:test SET sport += 'football' RETURN sport
+			UPSERT person:test SET sport += 'football' RETURN sport
 		ELSE
-			UPDATE person:test SET sport = ['basketball'] RETURN sport
+			UPSERT person:test SET sport = ['basketball'] RETURN sport
 		END;
 		-- Check if the record exists
 		LET $record = SELECT *, count() AS count FROM person:test;
@@ -258,9 +258,9 @@ async fn subquery_ifelse_array() -> Result<(), Error> {
 		RETURN $record;
 		-- Update the record field if it exists
 		IF $record.count THEN
-			UPDATE person:test SET sport += 'football' RETURN sport;
+			UPSERT person:test SET sport += 'football' RETURN sport;
 		ELSE
-			UPDATE person:test SET sport = ['basketball'] RETURN sport;
+			UPSERT person:test SET sport = ['basketball'] RETURN sport;
 		END;
 	";
 	let dbs = new_ds().await?;

lib/tests/table.rs

@@ -23,11 +23,11 @@ async fn define_foreign_table() -> Result<(), Error> {
 			GROUP BY age
 		;
 		INFO FOR TABLE person;
-		UPDATE person:one SET age = 39, score = 70;
+		UPSERT person:one SET age = 39, score = 70;
 		SELECT * FROM person_by_age;
-		UPDATE person:two SET age = 39, score = 80;
+		UPSERT person:two SET age = 39, score = 80;
 		SELECT * FROM person_by_age;
-		UPDATE person:two SET age = 39, score = 90;
+		UPSERT person:two SET age = 39, score = 90;
 		SELECT * FROM person_by_age;
 	";
 	let dbs = new_ds().await?;

lib/tests/typing.rs

@@ -10,15 +10,15 @@ use surrealdb::sql::Value;
 #[tokio::test]
 async fn strict_typing_inline() -> Result<(), Error> {
 	let sql = "
-		UPDATE person:test SET age = <int> NONE;
+		UPSERT person:test SET age = <int> NONE;
-		UPDATE person:test SET age = <int> '18';
+		UPSERT person:test SET age = <int> '18';
-		UPDATE person:test SET enabled = <bool | int> NONE;
+		UPSERT person:test SET enabled = <bool | int> NONE;
-		UPDATE person:test SET enabled = <bool | int> true;
+		UPSERT person:test SET enabled = <bool | int> true;
-		UPDATE person:test SET name = <string> 'Tobie Morgan Hitchcock';
+		UPSERT person:test SET name = <string> 'Tobie Morgan Hitchcock';
-		UPDATE person:test SET scores = <set<float>> [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET scores = <set<float>> [1,1,2,2,3,3,4,4,5,5];
-		UPDATE person:test SET scores = <array<float>> [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET scores = <array<float>> [1,1,2,2,3,3,4,4,5,5];
-		UPDATE person:test SET scores = <set<float, 5>> [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET scores = <set<float, 5>> [1,1,2,2,3,3,4,4,5,5];
-		UPDATE person:test SET scores = <array<float, 5>> [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET scores = <array<float, 5>> [1,1,2,2,3,3,4,4,5,5];
 	";
 	let dbs = new_ds().await?;
 	let ses = Session::owner().with_ns("test").with_db("test");
@@ -131,10 +131,10 @@ async fn strict_typing_defined() -> Result<(), Error> {
 		DEFINE FIELD enabled ON person TYPE bool | int;
 		DEFINE FIELD name ON person TYPE string;
 		DEFINE FIELD scores ON person TYPE set<float, 5>;
-		UPDATE person:test SET age = NONE, enabled = NONE, name = NONE, scored = [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET age = NONE, enabled = NONE, name = NONE, scored = [1,1,2,2,3,3,4,4,5,5];
-		UPDATE person:test SET age = 18, enabled = NONE, name = NONE, scored = [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET age = 18, enabled = NONE, name = NONE, scored = [1,1,2,2,3,3,4,4,5,5];
-		UPDATE person:test SET age = 18, enabled = true, name = NONE, scored = [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET age = 18, enabled = true, name = NONE, scored = [1,1,2,2,3,3,4,4,5,5];
-		UPDATE person:test SET age = 18, enabled = true, name = 'Tobie Morgan Hitchcock', scores = [1,1,2,2,3,3,4,4,5,5];
+		UPSERT person:test SET age = 18, enabled = true, name = 'Tobie Morgan Hitchcock', scores = [1,1,2,2,3,3,4,4,5,5];
 	";
 	let dbs = new_ds().await?;
 	let ses = Session::owner().with_ns("test").with_db("test");
@@ -193,23 +193,23 @@ async fn strict_typing_none_null() -> Result<(), Error> {
 	let sql = "
 		DEFINE TABLE person SCHEMAFULL;
 		DEFINE FIELD name ON TABLE person TYPE option<string>;
-		UPDATE person:test SET name = 'Tobie';
+		UPSERT person:test SET name = 'Tobie';
-		UPDATE person:test SET name = NULL;
+		UPSERT person:test SET name = NULL;
-		UPDATE person:test SET name = NONE;
+		UPSERT person:test SET name = NONE;
 		--
 		REMOVE TABLE person;
 		DEFINE TABLE person SCHEMAFULL;
 		DEFINE FIELD name ON TABLE person TYPE option<string | null>;
-		UPDATE person:test SET name = 'Tobie';
+		UPSERT person:test SET name = 'Tobie';
-		UPDATE person:test SET name = NULL;
+		UPSERT person:test SET name = NULL;
-		UPDATE person:test SET name = NONE;
+		UPSERT person:test SET name = NONE;
 		--
 		REMOVE TABLE person;
 		DEFINE TABLE person SCHEMAFULL;
 		DEFINE FIELD name ON TABLE person TYPE string | null;
-		UPDATE person:test SET name = 'Tobie';
+		UPSERT person:test SET name = 'Tobie';
-		UPDATE person:test SET name = NULL;
+		UPSERT person:test SET name = NULL;
-		UPDATE person:test SET name = NONE;
+		UPSERT person:test SET name = NONE;
 	";
 	let mut t = Test::new(sql).await?;
 	//

lib/tests/update.rs

@@ -95,6 +95,7 @@ async fn update_simple_with_input() -> Result<(), Error> {
 					$value
 				END
 		;
+		CREATE person:test;
 		UPDATE person:test CONTENT { name: 'Tobie' };
 		UPDATE person:test REPLACE { name: 'jaime' };
 		UPDATE person:test MERGE { name: 'Jaime' };
@@ -105,12 +106,22 @@ async fn update_simple_with_input() -> Result<(), Error> {
 	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);
+	assert_eq!(res.len(), 8);
 	//
 	let tmp = res.remove(0).result;
 	assert!(tmp.is_ok());
 	//
 	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
 	let val = Value::parse(
 		"[
 			{
@@ -306,30 +317,6 @@ async fn common_permissions_checks(auth_enabled: bool) {
 	for ((level, role), (ns, db), should_succeed, msg) in tests.into_iter() {
 		let sess = Session::for_level(level, role).with_ns(ns).with_db(db);
 
-		// Test the statement when the table has to be created
-
-		{
-			let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
-
-			let mut resp = ds.execute(statement, &sess, None).await.unwrap();
-			let res = resp.remove(0).output();
-
-			if should_succeed {
-				assert!(res.is_ok() && res.unwrap() != Value::parse("[]"), "{}", msg);
-			} else if res.is_ok() {
-				assert!(res.unwrap() == Value::parse("[]"), "{}", msg);
-			} else {
-				// Not allowed to create a table
-				let err = res.unwrap_err().to_string();
-				assert!(
-					err.contains("Not enough permissions to perform this action"),
-					"{}: {}",
-					msg,
-					err
-				)
-			}
-		}
-
 		// Test the statement when the table already exists
 		{
 			let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
@@ -428,24 +415,6 @@ async fn check_permissions_auth_enabled() {
 
 	let statement = "UPDATE person:test CONTENT { name: 'Name' };";
 
-	// When the table doesn't exist
-	{
-		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
-
-		let mut resp = ds
-			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
-			.await
-			.unwrap();
-		let res = resp.remove(0).output();
-
-		let err = res.unwrap_err().to_string();
-		assert!(
-			err.contains("Not enough permissions to perform this action"),
-			"anonymous user should not be able to create the table: {}",
-			err
-		);
-	}
-
 	// When the table grants no permissions
 	{
 		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
@@ -500,7 +469,7 @@ async fn check_permissions_auth_enabled() {
 
 		let mut resp = ds
 			.execute(
-				"DEFINE TABLE person PERMISSIONS FULL; CREATE person;",
+				"DEFINE TABLE person PERMISSIONS FULL; CREATE person:test;",
 				&Session::owner().with_ns("NS").with_db("DB"),
 				None,
 			)
@@ -558,30 +527,13 @@ async fn check_permissions_auth_disabled() {
 
 	let statement = "UPDATE person:test CONTENT { name: 'Name' };";
 
-	// When the table doesn't exist
-	{
-		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
-
-		let mut resp = ds
-			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
-			.await
-			.unwrap();
-		let res = resp.remove(0).output();
-
-		assert!(
-			res.unwrap() != Value::parse("[]"),
-			"{}",
-			"anonymous user should be able to create the table"
-		);
-	}
-
 	// When the table grants no permissions
 	{
 		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
 
 		let mut resp = ds
 			.execute(
-				"DEFINE TABLE person PERMISSIONS NONE; CREATE person;",
+				"DEFINE TABLE person PERMISSIONS NONE; CREATE person:test;",
 				&Session::owner().with_ns("NS").with_db("DB"),
 				None,
 			)
@@ -629,7 +581,7 @@ async fn check_permissions_auth_disabled() {
 
 		let mut resp = ds
 			.execute(
-				"DEFINE TABLE person PERMISSIONS FULL; CREATE person;",
+				"DEFINE TABLE person PERMISSIONS FULL; CREATE person:test;",
 				&Session::owner().with_ns("NS").with_db("DB"),
 				None,
 			)

lib/tests/upsert.rs

@@ -0,0 +1,684 @@
+mod parse;
+use parse::Parse;
+mod helpers;
+use crate::helpers::Test;
+use helpers::new_ds;
+use surrealdb::dbs::Session;
+use surrealdb::err::Error;
+use surrealdb::iam::Role;
+use surrealdb::sql::Value;
+
+#[tokio::test]
+async fn upsert_merge_and_content() -> Result<(), Error> {
+	let sql = "
+		CREATE person:test CONTENT { name: 'Tobie' };
+		UPSERT person:test CONTENT { name: 'Jaime' };
+		UPSERT person:test CONTENT 'some content';
+		UPSERT person:test REPLACE 'some content';
+		UPSERT person:test MERGE { age: 50 };
+		UPSERT person:test MERGE 'some content';
+	";
+	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(), 6);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+				name: 'Tobie',
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+				name: 'Jaime',
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result;
+	assert!(matches!(
+		tmp.err(),
+		Some(e) if e.to_string() == r#"Can not use 'some content' in a CONTENT clause"#
+	));
+	//
+	let tmp = res.remove(0).result;
+	assert!(matches!(
+		tmp.err(),
+		Some(e) if e.to_string() == r#"Can not use 'some content' in a CONTENT clause"#
+	));
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+				name: 'Jaime',
+				age: 50,
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result;
+	assert!(matches!(
+		tmp.err(),
+		Some(e) if e.to_string() == r#"Can not use 'some content' in a MERGE clause"#
+	));
+	//
+	Ok(())
+}
+
+#[tokio::test]
+async fn upsert_simple_with_input() -> Result<(), Error> {
+	let sql = "
+		DEFINE FIELD name ON TABLE person
+			ASSERT
+				IF $input THEN
+					$input = /^[A-Z]{1}[a-z]+$/
+				ELSE
+					true
+				END
+			VALUE
+				IF $input THEN
+					'Name: ' + $input
+				ELSE
+					$value
+				END
+		;
+		UPSERT person:test;
+		UPSERT person:test CONTENT { name: 'Tobie' };
+		UPSERT person:test REPLACE { name: 'jaime' };
+		UPSERT person:test MERGE { name: 'Jaime' };
+		UPSERT person:test SET name = 'tobie';
+		UPSERT person:test SET name = 'Tobie';
+		SELECT * FROM person:test;
+	";
+	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(), 8);
+	//
+	let tmp = res.remove(0).result;
+	assert!(tmp.is_ok());
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+				name: 'Name: Tobie',
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result;
+	assert!(matches!(
+		tmp.err(),
+		Some(e) if e.to_string() == r#"Found 'Name: jaime' for field `name`, with record `person:test`, but field must conform to: IF $input THEN $input = /^[A-Z]{1}[a-z]+$/ ELSE true END"#
+	));
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+				name: 'Name: Jaime',
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result;
+	assert!(matches!(
+		tmp.err(),
+		Some(e) if e.to_string() == r#"Found 'Name: tobie' for field `name`, with record `person:test`, but field must conform to: IF $input THEN $input = /^[A-Z]{1}[a-z]+$/ ELSE true END"#
+	));
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+				name: 'Name: Tobie',
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				id: person:test,
+				name: 'Name: Tobie',
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	Ok(())
+}
+
+#[tokio::test]
+async fn update_complex_with_input() -> Result<(), Error> {
+	let sql = "
+		DEFINE FIELD images ON product
+			TYPE array
+			ASSERT array::len($value) > 0
+		;
+		REMOVE FIELD images.* ON product;
+		DEFINE FIELD images.* ON product TYPE string
+			VALUE string::trim($input)
+			ASSERT $input AND string::len($value) > 0
+		;
+		CREATE product:test SET images = [' test.png '];
+	";
+	let mut t = Test::new(sql).await?;
+	t.skip_ok(3)?;
+	t.expect_val(
+		"[
+			{
+				id: product:test,
+				images: ['test.png'],
+			}
+		]",
+	)?;
+	Ok(())
+}
+
+#[tokio::test]
+async fn upsert_with_return_clause() -> Result<(), Error> {
+	let sql = "
+		CREATE person:test SET age = 18, name = 'John';
+		UPSERT person:test SET age = 25 RETURN VALUE $before;
+		UPSERT person:test SET age = 30 RETURN VALUE { old_age: $before.age, new_age: $after.age };
+		UPSERT person:test SET age = 35 RETURN age, name;
+		DELETE person:test RETURN VALUE $before;
+	";
+	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(), 5);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				age: 18,
+				id: person:test,
+				name: 'John'
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				age: 18,
+				id: person:test,
+				name: 'John'
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				new_age: 30,
+				old_age: 25
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				age: 35,
+				name: 'John'
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	let tmp = res.remove(0).result?;
+	let val = Value::parse(
+		"[
+			{
+				age: 35,
+				id: person:test,
+				name: 'John'
+			}
+		]",
+	);
+	assert_eq!(tmp, val);
+	//
+	Ok(())
+}
+
+//
+// Permissions
+//
+
+async fn common_permissions_checks(auth_enabled: bool) {
+	let tests = vec![
+		// Root level
+		((().into(), Role::Owner), ("NS", "DB"), true, "owner at root level should be able to update a record"),
+		((().into(), Role::Editor), ("NS", "DB"), true, "editor at root level should be able to update a record"),
+		((().into(), Role::Viewer), ("NS", "DB"), false, "viewer at root level should not be able to update a record"),
+
+		// Namespace level
+		((("NS",).into(), Role::Owner), ("NS", "DB"), true, "owner at namespace level should be able to update a record on its namespace"),
+		((("NS",).into(), Role::Owner), ("OTHER_NS", "DB"), false, "owner at namespace level should not be able to update a record on another namespace"),
+		((("NS",).into(), Role::Editor), ("NS", "DB"), true, "editor at namespace level should be able to update a record on its namespace"),
+		((("NS",).into(), Role::Editor), ("OTHER_NS", "DB"), false, "editor at namespace level should not be able to update a record on another namespace"),
+		((("NS",).into(), Role::Viewer), ("NS", "DB"), false, "viewer at namespace level should not be able to update a record on its namespace"),
+		((("NS",).into(), Role::Viewer), ("OTHER_NS", "DB"), false, "viewer at namespace level should not be able to update a record on another namespace"),
+
+		// Database level
+		((("NS", "DB").into(), Role::Owner), ("NS", "DB"), true, "owner at database level should be able to update a record on its database"),
+		((("NS", "DB").into(), Role::Owner), ("NS", "OTHER_DB"), false, "owner at database level should not be able to update a record on another database"),
+		((("NS", "DB").into(), Role::Owner), ("OTHER_NS", "DB"), false, "owner at database level should not be able to update a record on another namespace even if the database name matches"),
+		((("NS", "DB").into(), Role::Editor), ("NS", "DB"), true, "editor at database level should be able to update a record on its database"),
+		((("NS", "DB").into(), Role::Editor), ("NS", "OTHER_DB"), false, "editor at database level should not be able to update a record on another database"),
+		((("NS", "DB").into(), Role::Editor), ("OTHER_NS", "DB"), false, "editor at database level should not be able to update a record on another namespace even if the database name matches"),
+		((("NS", "DB").into(), Role::Viewer), ("NS", "DB"), false, "viewer at database level should not be able to update a record on its database"),
+		((("NS", "DB").into(), Role::Viewer), ("NS", "OTHER_DB"), false, "viewer at database level should not be able to update a record on another database"),
+		((("NS", "DB").into(), Role::Viewer), ("OTHER_NS", "DB"), false, "viewer at database level should not be able to update a record on another namespace even if the database name matches"),
+	];
+	let statement = "UPSERT person:test CONTENT { name: 'Name' };";
+
+	for ((level, role), (ns, db), should_succeed, msg) in tests.into_iter() {
+		let sess = Session::for_level(level, role).with_ns(ns).with_db(db);
+
+		// Test the statement when the table has to be created
+
+		{
+			let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+			let mut resp = ds.execute(statement, &sess, None).await.unwrap();
+			let res = resp.remove(0).output();
+
+			if should_succeed {
+				assert!(res.is_ok() && res.unwrap() != Value::parse("[]"), "{}", msg);
+			} else if res.is_ok() {
+				assert!(res.unwrap() == Value::parse("[]"), "{}", msg);
+			} else {
+				// Not allowed to create a table
+				let err = res.unwrap_err().to_string();
+				assert!(
+					err.contains("Not enough permissions to perform this action"),
+					"{}: {}",
+					msg,
+					err
+				)
+			}
+		}
+
+		// Test the statement when the table already exists
+		{
+			let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+			// Prepare datastore
+			let mut resp = ds
+				.execute("CREATE person:test", &Session::owner().with_ns("NS").with_db("DB"), None)
+				.await
+				.unwrap();
+			let res = resp.remove(0).output();
+			assert!(
+				res.is_ok() && res.unwrap() != Value::parse("[]"),
+				"unexpected error creating person record"
+			);
+			let mut resp = ds
+				.execute(
+					"CREATE person:test",
+					&Session::owner().with_ns("OTHER_NS").with_db("DB"),
+					None,
+				)
+				.await
+				.unwrap();
+			let res = resp.remove(0).output();
+			assert!(
+				res.is_ok() && res.unwrap() != Value::parse("[]"),
+				"unexpected error creating person record"
+			);
+			let mut resp = ds
+				.execute(
+					"CREATE person:test",
+					&Session::owner().with_ns("NS").with_db("OTHER_DB"),
+					None,
+				)
+				.await
+				.unwrap();
+			let res = resp.remove(0).output();
+			assert!(
+				res.is_ok() && res.unwrap() != Value::parse("[]"),
+				"unexpected error creating person record"
+			);
+
+			// Run the test
+			let mut resp = ds.execute(statement, &sess, None).await.unwrap();
+			let res = resp.remove(0).output();
+
+			// Select always succeeds, but the result may be empty
+			assert!(res.is_ok());
+
+			if should_succeed {
+				assert!(res.unwrap() != Value::parse("[]"), "{}", msg);
+
+				// Verify the update was persisted
+				let mut resp = ds
+					.execute(
+						"SELECT name FROM person:test",
+						&Session::owner().with_ns("NS").with_db("DB"),
+						None,
+					)
+					.await
+					.unwrap();
+				let res = resp.remove(0).output();
+				let res = res.unwrap().to_string();
+				assert!(res.contains("Name"), "{}: {:?}", msg, res);
+			} else {
+				assert!(res.unwrap() == Value::parse("[]"), "{}", msg);
+
+				// Verify the update was not persisted
+				let mut resp = ds
+					.execute(
+						"SELECT name FROM person:test",
+						&Session::owner().with_ns("NS").with_db("DB"),
+						None,
+					)
+					.await
+					.unwrap();
+				let res = resp.remove(0).output();
+				let res = res.unwrap().to_string();
+				assert!(!res.contains("Name"), "{}: {:?}", msg, res);
+			}
+		}
+	}
+}
+
+#[tokio::test]
+async fn check_permissions_auth_enabled() {
+	let auth_enabled = true;
+	//
+	// Test common scenarios
+	//
+
+	common_permissions_checks(auth_enabled).await;
+
+	//
+	// Test Anonymous user
+	//
+
+	let statement = "UPSERT person:test CONTENT { name: 'Name' };";
+
+	// When the table doesn't exist
+	{
+		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+		let mut resp = ds
+			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+
+		let err = res.unwrap_err().to_string();
+		assert!(
+			err.contains("Not enough permissions to perform this action"),
+			"anonymous user should not be able to create the table: {}",
+			err
+		);
+	}
+
+	// When the table grants no permissions
+	{
+		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+		let mut resp = ds
+			.execute(
+				"DEFINE TABLE person PERMISSIONS NONE; CREATE person:test;",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		assert!(res.is_ok(), "failed to create table: {:?}", res);
+		let res = resp.remove(0).output();
+		assert!(res.is_ok() && res.unwrap() != Value::parse("[]"), "{}", "failed to create record");
+
+		let mut resp = ds
+			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+
+		assert!(
+			res.unwrap() == Value::parse("[]"),
+			"{}",
+			"anonymous user should not be able to select if the table has no permissions"
+		);
+
+		// Verify the update was not persisted
+		let mut resp = ds
+			.execute(
+				"SELECT name FROM person:test",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		let res = res.unwrap().to_string();
+		assert!(
+			!res.contains("Name"),
+			"{}: {:?}",
+			"anonymous user should not be able to update a record if the table has no permissions",
+			res
+		);
+	}
+
+	// When the table exists and grants full permissions
+	{
+		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+		let mut resp = ds
+			.execute(
+				"DEFINE TABLE person PERMISSIONS FULL; CREATE person:test;",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		assert!(res.is_ok(), "failed to create table: {:?}", res);
+		let res = resp.remove(0).output();
+		assert!(res.is_ok() && res.unwrap() != Value::parse("[]"), "{}", "failed to create record");
+
+		let mut resp = ds
+			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+
+		assert!(
+			res.unwrap() != Value::parse("[]"),
+			"{}",
+			"anonymous user should be able to select if the table has full permissions"
+		);
+
+		// Verify the update was persisted
+		let mut resp = ds
+			.execute(
+				"SELECT name FROM person:test",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		let res = res.unwrap().to_string();
+		assert!(
+			res.contains("Name"),
+			"{}: {:?}",
+			"anonymous user should be able to update a record if the table has full permissions",
+			res
+		);
+	}
+}
+
+#[tokio::test]
+async fn check_permissions_auth_disabled() {
+	let auth_enabled = false;
+	//
+	// Test common scenarios
+	//
+
+	common_permissions_checks(auth_enabled).await;
+
+	//
+	// Test Anonymous user
+	//
+
+	let statement = "UPSERT person:test CONTENT { name: 'Name' };";
+
+	// When the table doesn't exist
+	{
+		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+		let mut resp = ds
+			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+
+		assert!(
+			res.unwrap() != Value::parse("[]"),
+			"{}",
+			"anonymous user should be able to create the table"
+		);
+	}
+
+	// When the table grants no permissions
+	{
+		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+		let mut resp = ds
+			.execute(
+				"DEFINE TABLE person PERMISSIONS NONE; CREATE person:test;",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		assert!(res.is_ok(), "failed to create table: {:?}", res);
+		let res = resp.remove(0).output();
+		assert!(res.is_ok() && res.unwrap() != Value::parse("[]"), "{}", "failed to create record");
+
+		let mut resp = ds
+			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+
+		assert!(
+			res.unwrap() != Value::parse("[]"),
+			"{}",
+			"anonymous user should be able to update a record if the table has no permissions"
+		);
+
+		// Verify the update was persisted
+		let mut resp = ds
+			.execute(
+				"SELECT name FROM person:test",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		let res = res.unwrap().to_string();
+		assert!(
+			res.contains("Name"),
+			"{}: {:?}",
+			"anonymous user should be able to update a record if the table has no permissions",
+			res
+		);
+	}
+
+	// When the table exists and grants full permissions
+	{
+		let ds = new_ds().await.unwrap().with_auth_enabled(auth_enabled);
+
+		let mut resp = ds
+			.execute(
+				"DEFINE TABLE person PERMISSIONS FULL; CREATE person:test;",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		assert!(res.is_ok(), "failed to create table: {:?}", res);
+		let res = resp.remove(0).output();
+		assert!(res.is_ok() && res.unwrap() != Value::parse("[]"), "{}", "failed to create record");
+
+		let mut resp = ds
+			.execute(statement, &Session::default().with_ns("NS").with_db("DB"), None)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+
+		assert!(
+			res.unwrap() != Value::parse("[]"),
+			"{}",
+			"anonymous user should be able to select if the table has full permissions"
+		);
+
+		// Verify the update was persisted
+		let mut resp = ds
+			.execute(
+				"SELECT name FROM person:test",
+				&Session::owner().with_ns("NS").with_db("DB"),
+				None,
+			)
+			.await
+			.unwrap();
+		let res = resp.remove(0).output();
+		let res = res.unwrap().to_string();
+		assert!(
+			res.contains("Name"),
+			"{}: {:?}",
+			"anonymous user should be able to update a record if the table has full permissions",
+			res
+		);
+	}
+}

src/net/key.rs

@@ -168,7 +168,7 @@ async fn update_all(
 	match surrealdb::sql::value(data) {
 		Ok(data) => {
 			// Specify the request statement
-			let sql = "UPDATE type::table($table) CONTENT $data";
+			let sql = "UPSERT type::table($table) CONTENT $data";
 			// Specify the request variables
 			let vars = map! {
 				String::from("table") => Value::from(table),
@@ -212,7 +212,7 @@ async fn modify_all(
 	match surrealdb::sql::value(data) {
 		Ok(data) => {
 			// Specify the request statement
-			let sql = "UPDATE type::table($table) MERGE $data";
+			let sql = "UPSERT type::table($table) MERGE $data";
 			// Specify the request variables
 			let vars = map! {
 				String::from("table") => Value::from(table),
@@ -392,7 +392,7 @@ async fn update_one(
 	match surrealdb::sql::value(data) {
 		Ok(data) => {
 			// Specify the request statement
-			let sql = "UPDATE type::thing($table, $id) CONTENT $data";
+			let sql = "UPSERT type::thing($table, $id) CONTENT $data";
 			// Specify the request variables
 			let vars = map! {
 				String::from("table") => Value::from(table),
@@ -442,7 +442,7 @@ async fn modify_one(
 	match surrealdb::sql::value(data) {
 		Ok(data) => {
 			// Specify the request statement
-			let sql = "UPDATE type::thing($table, $id) MERGE $data";
+			let sql = "UPSERT type::thing($table, $id) MERGE $data";
 			// Specify the request variables
 			let vars = map! {
 				String::from("table") => Value::from(table),

tests/http_integration.rs

@@ -425,7 +425,7 @@ mod http_integration {
 				-- TABLE DATA: foo
 				-- ------------------------------
 
-				UPDATE foo:bvklxkhtxumyrfzqoc5i CONTENT { id: foo:bvklxkhtxumyrfzqoc5i };
+				INSERT { id: foo:bvklxkhtxumyrfzqoc5i };
 
 				-- ------------------------------
 				-- TRANSACTION