Rework execution

src/executor/alter_row/auto_increment.rs

@@ -1,40 +1,43 @@
 #![cfg(feature = "auto-increment")]
-use crate::{Column, Result, Row, StorageInner, Value, ValueDefault};
+use crate::{Column, Glue, Result, Row, Value, ValueDefault};
 
-pub async fn auto_increment(
-	storage: &mut StorageInner,
-	table_name: &str,
-	columns: &[Column],
-	rows: &mut [Row],
-) -> Result<()> {
-	let auto_increment_columns = columns
-		.iter()
-		.enumerate()
-		.filter(|(_, column)| matches!(column.default, Some(ValueDefault::AutoIncrement(_))))
-		.map(|(index, column)| {
-			(
-				index,
-				column.name.clone(),
-				rows.iter()
-					.filter(|row| matches!(row.0.get(index), Some(Value::Null)))
-					.count() as i64,
-			)
-		})
-		.collect();
+impl Glue {
+	pub async fn auto_increment(
+		&mut self,
+		table_name: &str,
+		columns: &[Column],
+		rows: &mut [Row],
+	) -> Result<()> {
+		let auto_increment_columns = columns
+			.iter()
+			.enumerate()
+			.filter(|(_, column)| matches!(column.default, Some(ValueDefault::AutoIncrement(_))))
+			.map(|(index, column)| {
+				(
+					index,
+					column.name.clone(),
+					rows.iter()
+						.filter(|row| matches!(row.0.get(index), Some(Value::Null)))
+						.count() as i64,
+				)
+			})
+			.collect();
 
-	let column_values = storage
-		.generate_increment_values(table_name.to_string(), auto_increment_columns)
-		.await?;
+		let column_values = self
+			.get_mut_database(&None)?
+			.generate_increment_values(table_name.to_string(), auto_increment_columns)
+			.await?;
 
-	let mut column_values = column_values;
-	for row in rows.iter_mut() {
-		for ((index, _name), value) in &mut column_values {
-			let cell = row.0.get_mut(*index).unwrap();
-			if matches!(cell, Value::Null) {
-				*cell = Value::I64(*value);
-				*value += 1;
+		let mut column_values = column_values;
+		for row in rows.iter_mut() {
+			for ((index, _name), value) in &mut column_values {
+				let cell = row.0.get_mut(*index).unwrap();
+				if matches!(cell, Value::Null) {
+					*cell = Value::I64(*value);
+					*value += 1;
+				}
 			}
 		}
+		Ok(())
 	}
-	Ok(())
 }

src/executor/alter_row/delete.rs

@@ -2,75 +2,75 @@ use {
 	crate::{
 		data::{get_name, Schema},
 		executor::types::ColumnInfo,
-		Column, Context, ExecuteError, MetaRecipe, Payload, PlannedRecipe, Result, StorageInner,
-		Value,
+		Column, ExecuteError, Glue, MetaRecipe, Payload, PlannedRecipe, Result, Value,
 	},
 	sqlparser::ast::{Expr, ObjectName},
 };
 
-pub async fn delete(
-	storages: &mut Vec<(String, &mut StorageInner)>,
-	context: &mut Context,
-	table_name: &ObjectName,
-	selection: &Option<Expr>,
-) -> Result<Payload> {
-	let table_name = get_name(table_name)?;
-	let Schema {
-		column_defs,
-		indexes,
-		..
-	} = storages[0]
-		.1
-		.fetch_schema(table_name)
-		.await?
-		.ok_or(ExecuteError::TableNotExists)?;
+impl Glue {
+	pub async fn delete(
+		&mut self,
+		table_name: &ObjectName,
+		selection: &Option<Expr>,
+	) -> Result<Payload> {
+		let table_name = get_name(table_name)?;
+		let Schema {
+			column_defs,
+			indexes,
+			..
+		} = self
+			.get_database(&None)?
+			.fetch_schema(table_name)
+			.await?
+			.ok_or(ExecuteError::TableNotExists)?;
 
-	let columns = column_defs
-		.clone()
-		.into_iter()
-		.map(|Column { name, .. }| ColumnInfo::of_name(name))
-		.collect::<Vec<ColumnInfo>>();
-	let filter = selection
-		.clone()
-		.map(|selection| {
-			PlannedRecipe::new(
-				MetaRecipe::new(selection)?.simplify_by_context(context)?,
-				&columns,
-			)
-		})
-		.unwrap_or(Ok(PlannedRecipe::TRUE))?;
+		let columns = column_defs
+			.clone()
+			.into_iter()
+			.map(|Column { name, .. }| ColumnInfo::of_name(name))
+			.collect::<Vec<ColumnInfo>>();
+		let filter = selection
+			.clone()
+			.map(|selection| {
+				PlannedRecipe::new(
+					MetaRecipe::new(selection)?.simplify_by_context(&*self.get_context()?)?,
+					&columns,
+				)
+			})
+			.unwrap_or(Ok(PlannedRecipe::TRUE))?;
 
-	let keys = storages[0]
-		.1
-		.scan_data(table_name)
-		.await?
-		.filter_map(|row_result| {
-			let (key, row) = match row_result {
-				Ok(keyed_row) => keyed_row,
-				Err(error) => return Some(Err(error)),
-			};
+		let keys = self
+			.get_database(&None)?
+			.scan_data(table_name)
+			.await?
+			.filter_map(|row_result| {
+				let (key, row) = match row_result {
+					Ok(keyed_row) => keyed_row,
+					Err(error) => return Some(Err(error)),
+				};
 
-			let row = row.0;
+				let row = row.0;
 
-			let confirm_constraint = filter.confirm_constraint(&row);
-			match confirm_constraint {
-				Ok(true) => Some(Ok(key)),
-				Ok(false) => None,
-				Err(error) => Some(Err(error)),
-			}
-		})
-		.collect::<Result<Vec<Value>>>()?;
+				let confirm_constraint = filter.confirm_constraint(&row);
+				match confirm_constraint {
+					Ok(true) => Some(Ok(key)),
+					Ok(false) => None,
+					Err(error) => Some(Err(error)),
+				}
+			})
+			.collect::<Result<Vec<Value>>>()?;
 
-	let num_keys = keys.len();
+		let num_keys = keys.len();
 
-	let result = storages[0]
-		.1
-		.delete_data(table_name, keys)
-		.await
-		.map(|_| Payload::Delete(num_keys))?;
+		let database = &mut **self.get_mut_database(&None)?;
+		let result = database
+			.delete_data(table_name, keys)
+			.await
+			.map(|_| Payload::Delete(num_keys))?;
 
-	for index in indexes.iter() {
-		index.reset(storages[0].1, table_name, &column_defs).await?; // TODO: Not this; optimise
+		for index in indexes.iter() {
+			index.reset(database, table_name, &column_defs).await?; // TODO: Not this; optimise
+		}
+		Ok(result)
 	}
-	Ok(result)
 }

src/executor/alter_row/insert.rs

@@ -1,58 +1,62 @@
 use {
-	super::{auto_increment, columns_to_positions, validate, validate_unique},
+	super::{columns_to_positions, validate},
 	crate::{
 		data::{get_name, Schema},
-		executor::query::query,
-		Context, ExecuteError, Payload, Result, Row, StorageInner,
+		ExecuteError, Glue, Payload, Result, Row,
 	},
 	sqlparser::ast::{Ident, ObjectName, Query},
 };
 
-pub async fn insert(
-	storages: &mut Vec<(String, &mut StorageInner)>,
-	context: &mut Context,
-	table_name: &ObjectName,
-	columns: &[Ident],
-	source: &Query,
-	expect_data: bool,
-) -> Result<Payload> {
-	let table_name = get_name(table_name)?;
-	let Schema {
-		column_defs,
-		indexes,
-		..
-	} = storages[0]
-		.1
-		.fetch_schema(table_name)
-		.await?
-		.ok_or(ExecuteError::TableNotExists)?;
-
-	// TODO: Multi storage
-	let (labels, mut rows) = query(storages, context, source.clone()).await?;
-	let column_positions = columns_to_positions(&column_defs, columns)?;
-
-	validate(&column_defs, &column_positions, &mut rows)?;
-	let mut rows: Vec<Row> = rows.into_iter().map(Row).collect();
-	#[cfg(feature = "auto-increment")]
-	auto_increment(storages[0].1, table_name, &column_defs, &mut rows).await?;
-	validate_unique(storages[0].1, table_name, &column_defs, &rows, None).await?;
-
-	let num_rows = rows.len();
-
-	let result = storages[0].1.insert_data(table_name, rows.clone()).await;
-
-	let result = result.map(|_| {
-		if expect_data {
-			Payload::Select { labels, rows }
-		} else {
-			Payload::Insert(num_rows)
+impl Glue {
+	pub async fn insert(
+		&mut self,
+		table_name: &ObjectName,
+		columns: &[Ident],
+		source: &Query,
+		expect_data: bool,
+	) -> Result<Payload> {
+		let table_name = get_name(table_name)?;
+		let Schema {
+			column_defs,
+			indexes,
+			..
+		} = self
+			.get_database(&None)?
+			.fetch_schema(table_name)
+			.await?
+			.ok_or(ExecuteError::TableNotExists)?;
+
+		// TODO: Multi storage
+		let (labels, mut rows) = self.query(source.clone()).await?;
+		let column_positions = columns_to_positions(&column_defs, columns)?;
+
+		validate(&column_defs, &column_positions, &mut rows)?;
+		let mut rows: Vec<Row> = rows.into_iter().map(Row).collect();
+		#[cfg(feature = "auto-increment")]
+		self.auto_increment(table_name, &column_defs, &mut rows)
+			.await?;
+		self.validate_unique(table_name, &column_defs, &rows, None)
+			.await?;
+
+		let num_rows = rows.len();
+
+		let database = &mut **self.get_mut_database(&None)?;
+
+		let result = database.insert_data(table_name, rows.clone()).await;
+
+		let result = result.map(|_| {
+			if expect_data {
+				Payload::Select { labels, rows }
+			} else {
+				Payload::Insert(num_rows)
+			}
+		})?;
+
+		for index in indexes.iter() {
+			// TODO: Should definitely be just inserting an index record
+			index.reset(database, table_name, &column_defs).await?; // TODO: Not this; optimise
 		}
-	})?;
 
-	for index in indexes.iter() {
-		// TODO: Should definitely be just inserting an index record
-		index.reset(storages[0].1, table_name, &column_defs).await?; // TODO: Not this; optimise
+		Ok(result)
 	}
-
-	Ok(result)
 }

src/executor/alter_row/mod.rs

@@ -5,13 +5,4 @@ mod update;
 mod validate;
 mod validate_unique;
 
-pub use {
-	delete::delete,
-	insert::insert,
-	update::update,
-	validate::{columns_to_positions, validate, ValidateError},
-	validate_unique::validate_unique,
-};
-
-#[cfg(feature = "auto-increment")]
-pub use auto_increment::auto_increment;
+pub use validate::{columns_to_positions, validate, ValidateError};