diff --git a/src/AsyncMysql/AsyncMysqlQueryResult.php b/src/AsyncMysql/AsyncMysqlQueryResult.php
index f78f53c..e4bf072 100644
--- a/src/AsyncMysql/AsyncMysqlQueryResult.php
+++ b/src/AsyncMysql/AsyncMysqlQueryResult.php
@@ -10,9 +10,9 @@
 final class AsyncMysqlQueryResult extends \AsyncMysqlQueryResult {
 
 	/* HH_IGNORE_ERROR[3012] I don't want to call parent::construct */
-	public function __construct(private dataset $rows, private int $rows_affected = 0, private int $last_insert_id = 0) {}
+	public function __construct(private vec<dict<string, mixed>> $rows, private int $rows_affected = 0, private int $last_insert_id = 0) {}
 
-	public function rows(): dataset {
+	public function rows(): vec<dict<string, mixed>> {
 		return $this->rows;
 	}
 
diff --git a/src/DataIntegrity.php b/src/DataIntegrity.php
index c1d24bd..dc4e348 100644
--- a/src/DataIntegrity.php
+++ b/src/DataIntegrity.php
@@ -78,13 +78,8 @@ public static function ensureFieldsPresent(dict<string, mixed> $row, TableSchema
 			$field_unsigned = $field->unsigned ?? false;
 
 			if (!C\contains_key($row, $field_name)) {
-				$row[$field_name] = self::getDefaultValueForField(
-					$field_type,
-					$field_nullable,
-					$field_default,
-					$field_name,
-					$schema->name,
-				);
+				$row[$field_name] =
+					self::getDefaultValueForField($field_type, $field_nullable, $field_default, $field_name, $schema->name);
 			} else if ($row[$field_name] === null) {
 				if ($field_nullable) {
 					// explicit null value and nulls are allowed, let it through
@@ -92,17 +87,10 @@ public static function ensureFieldsPresent(dict<string, mixed> $row, TableSchema
 				} else if (QueryContext::$strictSQLMode) {
 					// if we got this far the column has no default and isn't nullable, strict would throw
 					// but default MySQL mode would coerce to a valid value
-					throw new SQLFakeRuntimeException(
-						"Column '{$field_name}' on '{$schema->name}' does not allow null values",
-					);
+					throw new SQLFakeRuntimeException("Column '{$field_name}' on '{$schema->name}' does not allow null values");
 				} else {
-					$row[$field_name] = self::getDefaultValueForField(
-						$field_type,
-						$field_nullable,
-						$field_default,
-						$field_name,
-						$schema->name,
-					);
+					$row[$field_name] =
+						self::getDefaultValueForField($field_type, $field_nullable, $field_default, $field_name, $schema->name);
 				}
 			} else {
 				// TODO more integrity constraints, check field length for varchars, check timestamps
@@ -315,7 +303,7 @@ public static function checkUniqueConstraints(
 		dict<string, mixed> $row,
 		TableSchema $schema,
 		?arraykey $update_row_id = null,
-	): ?(string, int) {
+	): ?(string, arraykey) {
 
 		// gather all unique keys
 		$unique_keys = dict[];
@@ -343,10 +331,8 @@ public static function checkUniqueConstraints(
 				if (C\every($unique_key, $field ==> $r[$field] === $row[$field])) {
 					$dupe_unique_key_value = Vec\map($unique_key, $field ==> (string)$row[$field])
 						|> Str\join($$, ', ');
-					return tuple(
-						"Duplicate entry '{$dupe_unique_key_value}' for key '{$name}' in table '{$schema->name}'",
-						$row_id,
-					);
+					return
+						tuple("Duplicate entry '{$dupe_unique_key_value}' for key '{$name}' in table '{$schema->name}'", $row_id);
 				}
 			}
 		}
diff --git a/src/Expressions/BinaryOperatorExpression.php b/src/Expressions/BinaryOperatorExpression.php
index beb49cb..ff60575 100644
--- a/src/Expressions/BinaryOperatorExpression.php
+++ b/src/Expressions/BinaryOperatorExpression.php
@@ -4,7 +4,7 @@
 
 namespace Slack\SQLFake;
 
-use namespace HH\Lib\{C, Regex, Str, Vec};
+use namespace HH\Lib\{C, Dict, Regex, Str, Vec};
 
 /**
  * any operator that takes arguments on the left and right side, like +, -, *, AND, OR...
@@ -303,6 +303,28 @@ public function evaluateImpl(row $row, AsyncMysqlConnection $conn): mixed {
 		}
 	}
 
+	private static function getColumnNamesFromBinop(BinaryOperatorExpression $expr): dict<string, mixed> {
+		$column_names = dict[];
+
+		if ($expr->operator === Operator::EQUALS) {
+			if ($expr->left is ColumnExpression && $expr->left->name !== '*' && $expr->right is ConstantExpression) {
+				$column_names[$expr->left->name] = $expr->right->value;
+			}
+		}
+
+		if ($expr->operator === Operator::AND) {
+			if ($expr->left is BinaryOperatorExpression) {
+				$column_names = self::getColumnNamesFromBinop($expr->left);
+			}
+
+			if ($expr->right is BinaryOperatorExpression) {
+				$column_names = Dict\merge($column_names, self::getColumnNamesFromBinop($expr->right));
+			}
+		}
+
+		return $column_names;
+	}
+
 	/**
 	 * Coerce a mixed value to a num,
 	 * but also handle sub-expressions that return a dataset containing a num
diff --git a/src/Expressions/JSONFunctionExpression.hack b/src/Expressions/JSONFunctionExpression.hack
index fb47f14..1be52ff 100644
--- a/src/Expressions/JSONFunctionExpression.hack
+++ b/src/Expressions/JSONFunctionExpression.hack
@@ -331,9 +331,8 @@ final class JSONFunctionExpression extends BaseFunctionExpression {
 
 		$argCount = C\count($args);
 		if ($argCount !== 1) {
-			throw new SQLFakeRuntimeException(
-				'MySQL JSON_DEPTH() function must be called with 1 JSON document argument',
-			);
+			throw
+				new SQLFakeRuntimeException('MySQL JSON_DEPTH() function must be called with 1 JSON document argument');
 		}
 
 		$json = $args[0]->evaluate($row, $conn);
@@ -403,43 +402,43 @@ final class JSONFunctionExpression extends BaseFunctionExpression {
 			}
 
 			$term = (new JSONPath\JSONObject($term))->get('$');
-			 if ($term is null || $term->value is null || !($term->value is vec<_>)) {
+			if ($term is null || $term->value is null || !($term->value is vec<_>)) {
 				throw new SQLFakeRuntimeException('MySQL JSON_CONTAINS() function given invalid json');
 			}
 			$term = $term->value[0];
 
 			if ($json is vec<_>) {
 				// If $json is a vec then we have an array and will test if the array contains the given value
-				if ($term is dict<_,_>) {
+				if ($term is dict<_, _>) {
 					return C\count(Vec\filter($json, $val ==> {
-						if ($val is dict<_,_>) {
+						if ($val is dict<_, _>) {
 							return Dict\equal($val, $term);
 						}
 						return false;
-					})) > 0;
-				}
-				else {
+					})) >
+						0;
+				} else {
 					return C\contains($json, $term);
 				}
-			}
-			else if ($json is dict<_,_>) {
+			} else if ($json is dict<_, _>) {
 				// If $json is a dict then we have an object and will test that either (1) $json and $term are the same or
 				// (2) one of $json's members is the same as $term
-				if ($term is dict<_,_>) {
-					if (Dict\equal($json, $term)) { return true; }
+				if ($term is dict<_, _>) {
+					if (Dict\equal($json, $term)) {
+						return true;
+					}
 
 					return C\count(Dict\filter($json, $val ==> {
-						if ($val is dict<_,_>) {
+						if ($val is dict<_, _>) {
 							return Dict\equal($val, $term);
 						}
 						return false;
-					})) > 0;
-				}
-				else {
+					})) >
+						0;
+				} else {
 					return C\count(Dict\filter($json, $val ==> $term == $val)) > 0;
 				}
-			}
-			else {
+			} else {
 				return $json == $term;
 			}
 
diff --git a/src/Logger.php b/src/Logger.php
index 6d20a89..631141b 100644
--- a/src/Logger.php
+++ b/src/Logger.php
@@ -43,7 +43,7 @@ protected static function write(string $message): void {
 	 *   1 row from cluster1
 	 *
 	 */
-	public static function logResult(string $server, dataset $data, int $rows_affected): void {
+	public static function logResult(string $server, vec<dict<string, mixed>> $data, int $rows_affected): void {
 		if (QueryContext::$verbosity >= Verbosity::RESULTS) {
 			if ($rows_affected > 0) {
 				self::write("{$rows_affected} rows affected\n");
@@ -56,7 +56,7 @@ public static function logResult(string $server, dataset $data, int $rows_affect
 		}
 	}
 
-	private static function formatData(dataset $rows, string $server): string {
+	private static function formatData(vec<dict<string, mixed>> $rows, string $server): string {
 		$count = C\count($rows);
 
 		$tbl_columns = static::formatColumns($rows);
@@ -83,7 +83,7 @@ private static function formatData(dataset $rows, string $server): string {
 	/**
 	 * Determine maximum string length of column names or values
 	 */
-	protected static function formatColumns(dataset $data): dict<string, int> {
+	protected static function formatColumns(vec<dict<string, mixed>> $data): dict<string, int> {
 
 		$columns = dict[];
 		foreach ($data as $row) {
diff --git a/src/Parser/CreateTableParser.php b/src/Parser/CreateTableParser.php
index 7b2f947..2e6c92b 100644
--- a/src/Parser/CreateTableParser.php
+++ b/src/Parser/CreateTableParser.php
@@ -310,13 +310,13 @@ private function parseCreateTable(vec<string> $tokens, string $sql): parsed_tabl
 		//
 
 		$fields = vec[];
-		$indexes = vec[];
+		$index_refs = vec[];
 
 		if ($this->nextTokenIs($tokens, '(')) {
 			$tokens = Vec\drop($tokens, 1);
 			$ret = $this->parseCreateDefinition(inout $tokens);
 			$fields = $ret['fields'];
-			$indexes = $ret['indexes'];
+			$index_refs = $ret['indexes'];
 		}
 
 		$props = $this->parseTableProps(inout $tokens);
@@ -324,7 +324,7 @@ private function parseCreateTable(vec<string> $tokens, string $sql): parsed_tabl
 		$table = shape(
 			'name' => $name,
 			'fields' => $fields,
-			'indexes' => $indexes,
+			'indexes' => $index_refs,
 			'props' => $props,
 			'sql' => $sql,
 		);
@@ -342,26 +342,26 @@ private function parseCreateDefinition(inout vec<string> $tokens): shape(
 	) {
 
 		$fields = vec[];
-		$indexes = vec[];
+		$index_refs = vec[];
 
 		while ($tokens[0] !== ')') {
 			$these_tokens = $this->sliceUntilNextField(inout $tokens);
 
-			$this->parseFieldOrKey(inout $these_tokens, inout $fields, inout $indexes);
+			$this->parseFieldOrKey(inout $these_tokens, inout $fields, inout $index_refs);
 		}
 
 		$tokens = Vec\drop($tokens, 1); // closing paren
 
 		return shape(
 			'fields' => $fields,
-			'indexes' => $indexes,
+			'indexes' => $index_refs,
 		);
 	}
 
 	private function parseFieldOrKey(
 		inout vec<string> $tokens,
 		inout vec<parsed_field> $fields,
-		inout vec<parsed_index> $indexes,
+		inout vec<parsed_index> $index_refs,
 	): void {
 
 		//
@@ -424,7 +424,7 @@ private function parseFieldOrKey(
 				if (C\count($tokens)) {
 					$index['more'] = $tokens;
 				}
-				$indexes[] = $index;
+				$index_refs[] = $index;
 				return;
 
 			//
@@ -447,7 +447,7 @@ private function parseFieldOrKey(
 				if (C\count($tokens)) {
 					$index['more'] = $tokens;
 				}
-				$indexes[] = $index;
+				$index_refs[] = $index;
 				return;
 
 			// FULLTEXT		[index_name] (index_col_name,...) [index_option] ...
@@ -487,7 +487,7 @@ private function parseFieldOrKey(
 				if (C\count($tokens)) {
 					$index['more'] = $tokens;
 				}
-				$indexes[] = $index;
+				$index_refs[] = $index;
 				return;
 
 			// older stuff
diff --git a/src/Query/DeleteQuery.php b/src/Query/DeleteQuery.php
index 5f8b648..8f3a609 100644
--- a/src/Query/DeleteQuery.php
+++ b/src/Query/DeleteQuery.php
@@ -2,7 +2,7 @@
 
 namespace Slack\SQLFake;
 
-use namespace HH\Lib\{C, Keyset, Vec};
+use namespace HH\Lib\{C, Dict, Keyset};
 
 final class DeleteQuery extends Query {
 	public ?from_table $fromClause = null;
@@ -12,13 +12,15 @@ public function __construct(public string $sql) {}
 	public function execute(AsyncMysqlConnection $conn): int {
 		$this->fromClause as nonnull;
 		list($database, $table_name) = Query::parseTableName($conn, $this->fromClause['name']);
-		$data = $conn->getServer()->getTable($database, $table_name) ?? vec[];
+		$data = $conn->getServer()->getTableData($database, $table_name) ?? tuple(dict[], dict[], dict[]);
+		$schema = QueryContext::getSchema($database, $table_name);
+
 		Metrics::trackQuery(QueryType::DELETE, $conn->getServer()->name, $table_name, $this->sql);
 
-		return $this->applyWhere($conn, $data)
+		return $this->applyWhere($conn, $data[0])
 			|> $this->applyOrderBy($conn, $$)
 			|> $this->applyLimit($$)
-			|> $this->applyDelete($conn, $database, $table_name, $$, $data);
+			|> $this->applyDelete($conn, $database, $table_name, $$, $data[0], $data[1], $data[2], $schema);
 	}
 
 	/**
@@ -30,18 +32,37 @@ protected function applyDelete(
 		string $table_name,
 		dataset $filtered_rows,
 		dataset $original_table,
+		unique_index_refs $unique_index_refs,
+		index_refs $index_refs,
+		?TableSchema $table_schema,
 	): int {
-
-		// if this isn't a dict keyed by the original ids in the row, it could delete the wrong rows
-		$filtered_rows as dict<_, _>;
-
 		$rows_to_delete = Keyset\keys($filtered_rows);
-		$remaining_rows =
-			Vec\filter_with_key($original_table, ($row_num, $_) ==> !C\contains_key($rows_to_delete, $row_num));
+		$remaining_rows = Dict\filter_with_key(
+			$original_table,
+			($row_num, $_) ==> !C\contains_key($rows_to_delete, $row_num),
+		);
 		$rows_affected = C\count($original_table) - C\count($remaining_rows);
 
+		if ($table_schema is nonnull) {
+			foreach ($filtered_rows as $row_id => $row_to_delete) {
+				list($unique_index_ref_deletes, $index_ref_deletes) = self::getIndexRemovalsForRow(
+					$table_schema->indexes,
+					$row_id,
+					$row_to_delete,
+				);
+
+				foreach ($unique_index_ref_deletes as list($index_name, $index_key)) {
+					unset($unique_index_refs[$index_name][$index_key]);
+				}
+
+				foreach ($index_ref_deletes as list($index_name, $index_key, $_)) {
+					unset($index_refs[$index_name][$index_key][$row_id]);
+				}
+			}
+		}
+
 		// write it back to the database
-		$conn->getServer()->saveTable($database, $table_name, $remaining_rows);
+		$conn->getServer()->saveTable($database, $table_name, $remaining_rows, $unique_index_refs, $index_refs);
 		return $rows_affected;
 	}
 }
diff --git a/src/Query/FromClause.php b/src/Query/FromClause.php
index f09d77a..2e1fec4 100644
--- a/src/Query/FromClause.php
+++ b/src/Query/FromClause.php
@@ -2,7 +2,7 @@
 
 namespace Slack\SQLFake;
 
-use namespace HH\Lib\C;
+use namespace HH\Lib\{C, Vec};
 
 /**
  * Represents the entire FROM clause of a query,
@@ -31,18 +31,25 @@ public function aliasRecentExpression(string $name): void {
 
 	/**
 	 * The FROM clause of the query gets processed first, retrieving data from tables, executing subqueries, and handling joins
-	 * This is also where we build up the $columns list which is commonly used throughout the entire library to map column references to indexes in this dataset
+	 * This is also where we build up the $columns list which is commonly used throughout the entire library to map column references to index_refs in this dataset
 	 * @reviewer, we don't build up the $columns, since the variable is unused...
 	 */
-	public function process(AsyncMysqlConnection $conn, string $sql): dataset {
+	public function process(
+		AsyncMysqlConnection $conn,
+		string $sql,
+	): (dataset, unique_index_refs, index_refs, vec<Index>) {
 
-		$data = vec[];
+		$data = dict[];
 		$is_first_table = true;
+		$unique_index_refs = dict[];
+		$index_refs = dict[];
+		$indexes = vec[];
 
 		foreach ($this->tables as $table) {
 			$schema = null;
 			if (Shapes::keyExists($table, 'subquery')) {
 				$res = $table['subquery']->evaluate(dict[], $conn);
+				invariant($res is KeyedContainer<_, _>, 'evaluated result of SubqueryExpression must be dataset');
 				$name = $table['name'];
 			} else {
 				$table_name = $table['name'];
@@ -56,24 +63,22 @@ public function process(AsyncMysqlConnection $conn, string $sql): dataset {
 					throw new SQLFakeRuntimeException("Table $table_name not found in schema and strict mode is enabled");
 				}
 
-				$res = $conn->getServer()->getTable($database, $table_name);
-
-				if ($res === null) {
-					$res = vec[];
+				list($res, $unique_index_refs, $index_refs) =
+					$conn->getServer()->getTableData($database, $table_name) ?: tuple(dict[], dict[], dict[]);
+				if ($schema is nonnull) {
+					$indexes = Vec\concat($indexes, $schema->indexes);
 				}
 			}
 
-			invariant($res is KeyedContainer<_, _>, 'evaluated result of SubqueryExpression must be dataset');
-
-			$new_dataset = vec[];
+			$new_dataset = dict[];
 			if ($schema is nonnull && QueryContext::$strictSchemaMode) {
-				foreach ($res as $row) {
+				foreach ($res as $key => $row) {
 					$row as dict<_, _>;
 					$m = dict[];
 					foreach ($row as $field => $val) {
 						$m["{$name}.{$field}"] = $val;
 					}
-					$new_dataset[] = $m;
+					$new_dataset[$key] = $m;
 				}
 			} else if ($schema is nonnull) {
 				// if schema is set, order the fields in the right order on each row
@@ -82,7 +87,7 @@ public function process(AsyncMysqlConnection $conn, string $sql): dataset {
 					$ordered_fields[] = $field->name;
 				}
 
-				foreach ($res as $row) {
+				foreach ($res as $key => $row) {
 					invariant($row is dict<_, _>, 'each item in evaluated result of SubqueryExpression must be row');
 
 					$m = dict[];
@@ -92,17 +97,17 @@ public function process(AsyncMysqlConnection $conn, string $sql): dataset {
 						}
 						$m["{$name}.{$field}"] = $row[$field];
 					}
-					$new_dataset[] = $m;
+					$new_dataset[$key] = $m;
 				}
 			} else {
-				foreach ($res as $row) {
+				foreach ($res as $key => $row) {
 					invariant($row is dict<_, _>, 'each item in evaluated result of SubqueryExpression must be row');
 
 					$m = dict[];
-					foreach ($row as $key => $val) {
-						$m["{$name}.{$key}"] = $val;
+					foreach ($row as $column => $val) {
+						$m["{$name}.{$column}"] = $val;
 					}
-					$new_dataset[] = $m;
+					$new_dataset[$key] = $m;
 				}
 			}
 
@@ -128,6 +133,6 @@ public function process(AsyncMysqlConnection $conn, string $sql): dataset {
 			}
 		}
 
-		return $data;
+		return tuple($data, $unique_index_refs, $index_refs, $indexes);
 	}
 }
diff --git a/src/Query/InsertQuery.php b/src/Query/InsertQuery.php
index 0b072ed..3bc5bad 100644
--- a/src/Query/InsertQuery.php
+++ b/src/Query/InsertQuery.php
@@ -18,12 +18,13 @@ public function __construct(public string $table, public string $sql, public boo
 	 */
 	public function execute(AsyncMysqlConnection $conn): int {
 		list($database, $table_name) = Query::parseTableName($conn, $this->table);
-		$table = $conn->getServer()->getTable($database, $table_name) ?? vec[];
+		list($table, $unique_index_refs, $index_refs) =
+			$conn->getServer()->getTableData($database, $table_name) ?? tuple(dict[], dict[], dict[]);
 
 		Metrics::trackQuery(QueryType::INSERT, $conn->getServer()->name, $table_name, $this->sql);
 
-		$schema = QueryContext::getSchema($database, $table_name);
-		if ($schema === null && QueryContext::$strictSchemaMode) {
+		$table_schema = QueryContext::getSchema($database, $table_name);
+		if ($table_schema === null && QueryContext::$strictSchemaMode) {
 			throw new SQLFakeRuntimeException("Table $table_name not found in schema and strict mode is enabled");
 		}
 
@@ -35,52 +36,110 @@ public function execute(AsyncMysqlConnection $conn): int {
 			}
 
 			// can't enforce uniqueness or defaults if there is no schema available
-			if ($schema === null) {
-				$table[] = $row;
+			if ($table_schema === null) {
+				$table[C\count($table)] = $row;
 				$rows_affected++;
 				continue;
 			}
 
 			// ensure all fields are present with appropriate types and default values
 			// throw for nonexistent fields
-			$row = DataIntegrity::coerceToSchema($row, $schema);
+			$row = DataIntegrity::coerceToSchema($row, $table_schema);
+
+			$primary_key_columns = $table_schema->getPrimaryKeyColumns();
+
+			if ($primary_key_columns is nonnull) {
+				if (C\count($primary_key_columns) === 1) {
+					$primary_key = $row[C\firstx($primary_key_columns)] as arraykey;
+				} else {
+					$primary_key = '';
+					foreach ($primary_key_columns as $primary_key_column) {
+						$primary_key .= (string)$row[$primary_key_column].'||';
+					}
+				}
+			} else {
+				$primary_key = C\count($table);
+			}
+
+			$unique_index_ref_additions = vec[];
+			$index_ref_additions = vec[];
+
+			list($unique_index_ref_additions, $index_ref_additions) =
+				self::getIndexAdditionsForRow($table_schema->indexes, $row);
+
+			$key_violation = false;
+
+			if (isset($table[$primary_key])) {
+				$key_violation = true;
+			} else {
+				foreach ($unique_index_ref_additions as list($index_name, $index_key)) {
+					if (
+						isset($unique_index_refs[$index_name][$index_key]) &&
+						$unique_index_refs[$index_name][$index_key] !== $primary_key
+					) {
+						$key_violation = true;
+					}
+				}
+			}
+
+			$unique_key_violation = null;
+			if ($key_violation) {
+				$unique_key_violation = DataIntegrity::checkUniqueConstraints($table, $row, $table_schema);
+			}
 
-			// check for unique key violations
-			$unique_key_violation = DataIntegrity::checkUniqueConstraints($table, $row, $schema);
 			if ($unique_key_violation is nonnull) {
 				list($msg, $row_id) = $unique_key_violation;
 				// is this an "INSERT ... ON DUPLICATE KEY UPDATE?"
 				// if so, this is where we apply the updates
 				if (!C\is_empty($this->updateExpressions)) {
 					$existing_row = $table[$row_id];
-					list($affected, $table) = $this->applySet(
+					list($affected, $table, $unique_index_refs, $index_refs) = $this->applySet(
 						$conn,
 						$database,
 						$table_name,
 						dict[$row_id => $existing_row],
 						$table,
+						$unique_index_refs,
+						$index_refs,
 						$this->updateExpressions,
-						$schema,
+						$table_schema,
 						$row,
 					);
 					// MySQL always counts dupe inserts twice intentionally
 					$rows_affected += $affected * 2;
 					continue;
-				} else if ($this->ignoreDupes) {
-					// silently continue if INSERT IGNORE was specified
-					continue;
-				} else if (!QueryContext::$relaxUniqueConstraints) {
+				}
+
+				if (!$this->ignoreDupes && !QueryContext::$relaxUniqueConstraints) {
 					throw new SQLFakeUniqueKeyViolation($msg);
-				} else {
-					continue;
 				}
+
+				continue;
+			}
+
+			foreach ($unique_index_ref_additions as list($index_name, $index_key)) {
+				if (!C\contains_key($unique_index_refs, $index_name)) {
+					$unique_index_refs[$index_name] = dict[];
+				}
+				$unique_index_refs[$index_name][$index_key] = $primary_key;
 			}
-			$table[] = $row;
+
+			foreach ($index_ref_additions as list($index_name, $index_key)) {
+				if (!C\contains_key($index_refs, $index_name)) {
+					$index_refs[$index_name] = dict[];
+				}
+				if (!C\contains_key($index_refs[$index_name], $index_key)) {
+					$index_refs[$index_name][$index_key] = keyset[];
+				}
+				$index_refs[$index_name][$index_key][] = $primary_key;
+			}
+
+			$table[$primary_key] = $row;
 			$rows_affected++;
 		}
 
 		// write it back to the database
-		$conn->getServer()->saveTable($database, $table_name, $table);
+		$conn->getServer()->saveTable($database, $table_name, $table, $unique_index_refs, $index_refs);
 		return $rows_affected;
 	}
 }
diff --git a/src/Query/JoinProcessor.php b/src/Query/JoinProcessor.php
index 3736a24..9c6a777 100644
--- a/src/Query/JoinProcessor.php
+++ b/src/Query/JoinProcessor.php
@@ -154,7 +154,7 @@ public static function process(
 				break;
 		}
 
-		return $out;
+		return dict($out);
 	}
 
 	/**
@@ -180,9 +180,7 @@ protected static function buildNaturalJoinFilter(dataset $left_dataset, dataset
 
 		// MySQL actually doesn't throw if there's no matching columns, but I think we can take the liberty to assume it's not what you meant to do and throw here
 		if ($filter === null) {
-			throw new SQLFakeParseException(
-				'NATURAL join keyword was used with tables that do not share any column names',
-			);
+			throw new SQLFakeParseException('NATURAL join keyword was used with tables that do not share any column names');
 		}
 
 		return $filter;
@@ -305,6 +303,6 @@ private static function processHashJoin(
 			default:
 				invariant_violation('unreachable');
 		}
-		return $out;
+		return dict($out);
 	}
 }
diff --git a/src/Query/Query.php b/src/Query/Query.php
index 8e7a0ea..502cdb0 100644
--- a/src/Query/Query.php
+++ b/src/Query/Query.php
@@ -23,7 +23,10 @@ abstract class Query {
 	public string $sql;
 	public bool $ignoreDupes = false;
 
-	protected function applyWhere(AsyncMysqlConnection $conn, dataset $data): dataset {
+	protected function applyWhere(
+		AsyncMysqlConnection $conn,
+		dataset $data,
+	): dataset {
 		$where = $this->whereClause;
 		if ($where === null) {
 			// no where clause? cool! just return the given data
@@ -61,7 +64,8 @@ protected function applyOrderBy(AsyncMysqlConnection $_conn, dataset $data): dat
 				if ($value_a != $value_b) {
 					if ($value_a is num && $value_b is num) {
 						return (
-							((float)$value_a < (float)$value_b ? 1 : 0) ^ (($rule['direction'] === SortDirection::DESC) ? 1 : 0)
+							((float)$value_a < (float)$value_b ? 1 : 0) ^
+							(($rule['direction'] === SortDirection::DESC) ? 1 : 0)
 						)
 							? -1
 							: 1;
@@ -83,14 +87,26 @@ protected function applyOrderBy(AsyncMysqlConnection $_conn, dataset $data): dat
 		// Work around default sorting behavior to provide a usort that looks like MySQL, where equal values are ordered deterministically
 		// record the keys in a dict for usort
 		$data_temp = dict[];
+		$offset = 0;
 		foreach ($data as $i => $item) {
-			$data_temp[$i] = tuple($i, $item);
+			$data_temp[$i] = tuple($i, $offset, $item);
+			$offset++;
 		}
 
-		$data_temp = Dict\sort($data_temp, ((int, dict<string, mixed>) $a, (int, dict<string, mixed>) $b): int ==> {
-			$result = $sort_fun($a[1], $b[1]);
+		$data_temp = Dict\sort($data_temp, (
+			(arraykey, int, dict<string, mixed>) $a,
+			(arraykey, int, dict<string, mixed>) $b,
+		): int ==> {
+			$result = $sort_fun($a[2], $b[2]);
+
+			if ($result !== 0) {
+				return $result;
+			}
+
+			$a_index = $a[1];
+			$b_index = $b[1];
 
-			return $result === 0 ? $b[0] - $a[0] : $result;
+			return $b_index > $a_index ? 1 : -1;
 		});
 
 		// re-key the input dataset
@@ -99,7 +115,7 @@ protected function applyOrderBy(AsyncMysqlConnection $_conn, dataset $data): dat
 		// keys for updates/deletes to be able to delete the right rows
 		$data = dict[];
 		foreach ($data_temp as $item) {
-			$data[$item[0]] = $item[1];
+			$data[$item[0]] = $item[2];
 		}
 
 		return $data;
@@ -148,25 +164,26 @@ protected function applySet(
 		string $table_name,
 		dataset $filtered_rows,
 		dataset $original_table,
+		unique_index_refs $unique_index_refs,
+		index_refs $index_refs,
 		vec<BinaryOperatorExpression> $set_clause,
 		?TableSchema $table_schema,
 		/* for dupe inserts only */
 		?row $values = null,
-	): (int, vec<dict<string, mixed>>) {
-
-		$original_table as vec<_>;
-
+	): (int, dataset, unique_index_refs, index_refs) {
 		$valid_fields = null;
 		if ($table_schema !== null) {
 			$valid_fields = Keyset\map($table_schema->fields, $field ==> $field->name);
 		}
 
+		$columns = keyset[];
 		$set_clauses = vec[];
 		foreach ($set_clause as $expression) {
 			// the parser already asserts this at parse time
 			$left = $expression->left as ColumnExpression;
 			$right = $expression->right as nonnull;
 			$column = $left->name;
+			$columns[] = $column;
 
 			// If we know the valid fields for this table, only allow setting those
 			if ($valid_fields !== null) {
@@ -178,6 +195,16 @@ protected function applySet(
 			$set_clauses[] = shape('column' => $column, 'expression' => $right);
 		}
 
+		$applicable_indexes = vec[];
+
+		if ($table_schema is nonnull) {
+			foreach ($table_schema->indexes as $index) {
+				if (Keyset\intersect($index->fields, $columns) !== keyset[]) {
+					$applicable_indexes[] = $index;
+				}
+			}
+		}
+
 		$update_count = 0;
 
 		foreach ($filtered_rows as $row_id => $row) {
@@ -194,6 +221,13 @@ protected function applySet(
 					$update_row['sql_fake_values.'.$col] = $val;
 				}
 			}
+
+			list($unique_index_ref_deletes, $index_ref_deletes) = self::getIndexRemovalsForRow(
+				$applicable_indexes,
+				$row_id,
+				$row,
+			);
+
 			foreach ($set_clauses as $clause) {
 				$existing_value = $row[$clause['column']] ?? null;
 				$expr = $clause['expression'];
@@ -205,11 +239,60 @@ protected function applySet(
 				}
 			}
 
+			$new_row_id = $row_id;
+			$unique_index_ref_additions = vec[];
+			$index_ref_additions = vec[];
+
 			if ($changes_found) {
 				if ($table_schema is nonnull) {
 					// throw on invalid data types if strict mode
 					$row = DataIntegrity::coerceToSchema($row, $table_schema);
-					$result = DataIntegrity::checkUniqueConstraints($original_table, $row, $table_schema, $row_id);
+				}
+
+				foreach ($applicable_indexes as $index) {
+					if ($index->type === 'PRIMARY') {
+						if (C\count($index->fields) === 1) {
+							$index_key = $row[C\firstx($index->fields)] as arraykey;
+						} else {
+							$index_key = '';
+							foreach ($index->fields as $field) {
+								$index_key .= $row[$field] as arraykey.'||';
+							}
+						}
+
+						$new_row_id = $index_key;
+					}
+				}
+
+				list($unique_index_ref_additions, $index_ref_additions) = self::getIndexAdditionsForRow(
+					$applicable_indexes,
+					$row,
+				);
+			}
+
+			if ($changes_found) {
+				if ($table_schema is nonnull) {
+					$key_violation = false;
+
+					if (C\contains_key($original_table, $new_row_id)) {
+						$key_violation = true;
+					} else {
+						foreach ($unique_index_ref_deletes as list($index_name, $index_key)) {
+							if (
+								isset($unique_index_refs[$index_name][$index_key]) &&
+								$unique_index_refs[$index_name][$index_key] !== $row_id
+							) {
+								$key_violation = true;
+								break;
+							}
+						}
+					}
+
+					$result = null;
+					if ($key_violation) {
+						$result = DataIntegrity::checkUniqueConstraints($original_table, $row, $table_schema, $row_id);
+					}
+
 					if ($result is nonnull) {
 						if ($this->ignoreDupes) {
 							continue;
@@ -219,13 +302,138 @@ protected function applySet(
 						}
 					}
 				}
-				$original_table[$row_id] = $row;
+
+				foreach ($unique_index_ref_deletes as list($table, $key)) {
+					unset($unique_index_refs[$table][$key]);
+				}
+
+				foreach ($index_ref_deletes as list($table, $key, $index_row)) {
+					unset($index_refs[$table][$key][$index_row]);
+				}
+
+				foreach ($unique_index_ref_additions as list($index_name, $index_key)) {
+					if (!C\contains_key($unique_index_refs, $index_name)) {
+						$unique_index_refs[$index_name] = dict[];
+					}
+					$unique_index_refs[$index_name][$index_key] = $new_row_id;
+				}
+
+				foreach ($index_ref_additions as list($index_name, $index_key)) {
+					if (!C\contains_key($index_refs, $index_name)) {
+						$index_refs[$index_name] = dict[];
+					}
+					if (!C\contains_key($index_refs[$index_name], $index_key)) {
+						$index_refs[$index_name][$index_key] = keyset[];
+					}
+					$index_refs[$index_name][$index_key][] = $new_row_id;
+				}
+
+				if ($new_row_id !== $row_id) {
+					// Remap keys to preserve insertion order when primary key has changed
+					$original_table = Dict\pull_with_key(
+						$original_table,
+						($k, $v) ==> $k === $row_id ? $row : $v,
+						($k, $_) ==> $k === $row_id ? $new_row_id : $k,
+					);
+				} else {
+					$original_table[$row_id] = $row;
+				}
+
 				$update_count++;
 			}
 		}
 
 		// write it back to the database
-		$conn->getServer()->saveTable($database, $table_name, $original_table);
-		return tuple($update_count, $original_table);
+		$conn->getServer()->saveTable($database, $table_name, $original_table, $unique_index_refs, $index_refs);
+		return tuple($update_count, $original_table, $unique_index_refs, $index_refs);
+	}
+
+	public static function getIndexRemovalsForRow(
+		vec<Index> $applicable_indexes,
+		arraykey $row_id,
+		row $row,
+	): (vec<(string, arraykey)>, vec<(string, arraykey, arraykey)>) {
+		$unique_index_ref_deletes = vec[];
+		$index_ref_deletes = vec[];
+
+		foreach ($applicable_indexes as $index) {
+			if (C\count($index->fields) === 1) {
+				$index_key = $row[C\firstx($index->fields)] as ?arraykey;
+			} else {
+				$index_key = '';
+				$saw_null = false;
+
+				foreach ($index->fields as $field) {
+					$index_part = $row[$field] as ?arraykey;
+
+					if ($index_part is null) {
+						$saw_null = true;
+						break;
+					}
+
+					$index_key .= $row[$field] as arraykey.'||';
+				}
+
+				if ($saw_null) {
+					$index_key = null;
+				}
+			}
+
+			if ($index_key is null) {
+				continue;
+			}
+
+			if ($index->type === 'UNIQUE') {
+				$unique_index_ref_deletes[] = tuple($index->name, $index_key);
+			} else if ($index->type === 'INDEX') {
+				$index_ref_deletes[] = tuple($index->name, $index_key, $row_id);
+			}
+		}
+
+		return tuple($unique_index_ref_deletes, $index_ref_deletes);
+	}
+
+	public static function getIndexAdditionsForRow(
+		vec<Index> $applicable_indexes,
+		row $row,
+	): (vec<(string, arraykey)>, vec<(string, arraykey)>) {
+		$unique_index_ref_additions = vec[];
+		$index_ref_additions = vec[];
+
+		foreach ($applicable_indexes as $index) {
+			if (C\count($index->fields) === 1) {
+				$index_key = $row[C\firstx($index->fields)] as ?arraykey;
+			} else {
+				$index_key = '';
+				$saw_null = false;
+
+				foreach ($index->fields as $field) {
+					$index_part = $row[$field] as ?arraykey;
+
+					if ($index_part is null) {
+						$saw_null = true;
+						break;
+					}
+
+					$index_key .= $row[$field] as arraykey.'||';
+				}
+
+				if ($saw_null) {
+					$index_key = null;
+				}
+			}
+
+			if ($index_key is null) {
+				continue;
+			}
+
+			if ($index->type === 'UNIQUE') {
+				$unique_index_ref_additions[] = tuple($index->name, $index_key);
+			} else if ($index->type === 'INDEX') {
+				$index_ref_additions[] = tuple($index->name, $index_key);
+			}
+		}
+
+		return tuple($unique_index_ref_additions, $index_ref_additions);
 	}
 }
diff --git a/src/Query/SelectQuery.php b/src/Query/SelectQuery.php
index 2c72071..a0e48d3 100644
--- a/src/Query/SelectQuery.php
+++ b/src/Query/SelectQuery.php
@@ -55,7 +55,7 @@ public function execute(AsyncMysqlConnection $conn, ?row $_ = null): dataset {
 			// FROM clause handling - builds a data set including extracting rows from tables, applying joins
 			$this->applyFrom($conn)
 			// WHERE caluse - filter out any rows that don't match it
-			|> $this->applyWhere($conn, $$)
+			|> $this->applyWhere($conn, $$[0])
 			// GROUP BY clause - may group the rows if necessary. all clauses after this need to know how to handled both grouped and ungrouped inputs
 			|> $this->applyGroupBy($conn, $$)
 			// HAVING clause, filter out any rows not matching it
@@ -74,14 +74,14 @@ public function execute(AsyncMysqlConnection $conn, ?row $_ = null): dataset {
 
 	/**
 	 * The FROM clause of the query gets processed first, retrieving data from tables, executing subqueries, and handling joins
-	 * This is also where we build up the $columns list which is commonly used throughout the entire library to map column references to indexes in this dataset
+	 * This is also where we build up the $columns list which is commonly used throughout the entire library to map column references to index_refs in this dataset
 	 */
-	protected function applyFrom(AsyncMysqlConnection $conn): dataset {
+	protected function applyFrom(AsyncMysqlConnection $conn): (dataset, unique_index_refs, index_refs, vec<Index>) {
 
 		$from = $this->fromClause;
 		if ($from === null) {
 			// we put one empty row when there is no FROM so that queries like "SELECT 1" will return a row
-			return vec[dict[]];
+			return tuple(dict[0 => dict[]], dict[], dict[], vec[]);
 		}
 
 		return $from->process($conn, $this->sql);
@@ -109,7 +109,7 @@ protected function applyGroupBy(AsyncMysqlConnection $conn, dataset $data): data
 				$grouped_data[$hash][(string)$count] = $row;
 			}
 
-			$data = vec($grouped_data);
+			$data = dict($grouped_data);
 		} else {
 			$found_aggregate = false;
 			foreach ($select_expressions as $expr) {
@@ -122,7 +122,7 @@ protected function applyGroupBy(AsyncMysqlConnection $conn, dataset $data): data
 			// if we have an aggregate function in the select clause but no group by, do an implicit group that puts all rows in one grouping
 			// this makes things like "SELECT COUNT(*) FROM mytable" work
 			if ($found_aggregate) {
-				return vec[Dict\map_keys($data, $k ==> (string)$k)];
+				return dict[0 => Dict\map_keys($data, $k ==> (string)$k)];
 			}
 		}
 
@@ -137,7 +137,7 @@ protected function applyGroupBy(AsyncMysqlConnection $conn, dataset $data): data
 	protected function applyHaving(AsyncMysqlConnection $conn, dataset $data): dataset {
 		$havingClause = $this->havingClause;
 		if ($havingClause is nonnull) {
-			return Vec\filter($data, $row ==> (bool)$havingClause->evaluate($row, $conn));
+			return Dict\filter($data, $row ==> (bool)$havingClause->evaluate($row, $conn));
 		}
 
 		return $data;
@@ -157,10 +157,10 @@ protected function applySelect(AsyncMysqlConnection $conn, dataset $data): datas
 
 		$order_by_expressions = $this->orderBy ?? vec[];
 
-		$out = vec[];
+		$out = dict[];
 
 		// ok now you got that filter, let's do the formatting
-		foreach ($data as $row) {
+		foreach ($data as $key => $row) {
 			$formatted_row = dict[];
 
 			foreach ($this->selectExpressions as $expr) {
@@ -221,11 +221,11 @@ protected function applySelect(AsyncMysqlConnection $conn, dataset $data): datas
 				$formatted_row[$name] ??= $val;
 			}
 
-			$out[] = $formatted_row;
+			$out[$key] = $formatted_row;
 		}
 
 		if (C\contains_key($this->options, 'DISTINCT')) {
-			return Vec\unique_by($out, (row $row): string ==> Str\join(Vec\map($row, $col ==> (string)$col), '-'));
+			return Dict\unique_by($out, (row $row): string ==> Str\join(Vec\map($row, $col ==> (string)$col), '-'));
 		}
 
 		return $out;
@@ -268,7 +268,7 @@ protected function removeOrderByExtras(AsyncMysqlConnection $_conn, dataset $dat
 		}
 
 		// remove the fields we don't want from each row
-		return Vec\map($data, $row ==> Dict\filter_keys($row, $field ==> !C\contains_key($remove_fields, $field)));
+		return Dict\map($data, $row ==> Dict\filter_keys($row, $field ==> !C\contains_key($remove_fields, $field)));
 	}
 
 	/**
@@ -287,19 +287,19 @@ protected function processMultiQuery(AsyncMysqlConnection $conn, dataset $data):
 			switch ($sub['type']) {
 				case MultiOperand::UNION:
 					// contact the results, then get unique rows by converting all fields to string and comparing a joined-up representation
-					$data = Vec\concat($data, $subquery_results) |> Vec\unique_by($$, $row_encoder);
+					$data = Vec\concat($data, $subquery_results) |> Dict\unique_by($$, $row_encoder);
 					break;
 				case MultiOperand::UNION_ALL:
 					// just concatenate with no uniqueness
-					$data = Vec\concat($data, $subquery_results);
+					$data = Vec\concat($data, $subquery_results) |> dict($$);
 					break;
 				case MultiOperand::INTERSECT:
 					// there's no Vec\intersect_by currently
 					$encoded_data = Keyset\map($data, $row_encoder);
-					$data = Vec\filter($subquery_results, $row ==> C\contains_key($encoded_data, $row_encoder($row)));
+					$data = Dict\filter($subquery_results, $row ==> C\contains_key($encoded_data, $row_encoder($row)));
 					break;
 				case MultiOperand::EXCEPT:
-					$data = Vec\diff_by($data, $subquery_results, $row_encoder);
+					$data = dict(Vec\diff_by($data, $subquery_results, $row_encoder));
 					break;
 			}
 		}
diff --git a/src/Query/UpdateQuery.php b/src/Query/UpdateQuery.php
index cca5b57..ef48d15 100644
--- a/src/Query/UpdateQuery.php
+++ b/src/Query/UpdateQuery.php
@@ -9,14 +9,25 @@ public function __construct(public from_table $updateClause, public string $sql,
 	public vec<BinaryOperatorExpression> $setClause = vec[];
 
 	public function execute(AsyncMysqlConnection $conn): int {
-		list($tableName, $database, $data) = $this->processUpdateClause($conn);
+		list($tableName, $database, $data, $unique_index_refs, $index_refs) = $this->processUpdateClause($conn);
 		Metrics::trackQuery(QueryType::UPDATE, $conn->getServer()->name, $tableName, $this->sql);
 		$schema = QueryContext::getSchema($database, $tableName);
 
-		list($rows_affected, $_) = $this->applyWhere($conn, $data)
+		list($rows_affected, $_, $_, $_) =
+			$this->applyWhere($conn, $data)
 			|> $this->applyOrderBy($conn, $$)
 			|> $this->applyLimit($$)
-			|> $this->applySet($conn, $database, $tableName, $$, $data, $this->setClause, $schema);
+			|> $this->applySet(
+				$conn,
+				$database,
+				$tableName,
+				$$,
+				$data,
+				$unique_index_refs,
+				$index_refs,
+				$this->setClause,
+				$schema,
+			);
 
 		return $rows_affected;
 	}
@@ -25,9 +36,12 @@ public function execute(AsyncMysqlConnection $conn): int {
 	 * process the UPDATE clause to retrieve the table
 	 * add a row identifier to each element in the result which we can later use to update the underlying table
 	 */
-	protected function processUpdateClause(AsyncMysqlConnection $conn): (string, string, dataset) {
+	protected function processUpdateClause(
+		AsyncMysqlConnection $conn,
+	): (string, string, dataset, unique_index_refs, index_refs) {
 		list($database, $tableName) = Query::parseTableName($conn, $this->updateClause['name']);
-		$table = $conn->getServer()->getTable($database, $tableName) ?? vec[];
-		return tuple($tableName, $database, $table);
+		list($table_data, $unique_index_refs, $index_refs) =
+			$conn->getServer()->getTableData($database, $tableName) ?? tuple(dict[], dict[], dict[]);
+		return tuple($tableName, $database, $table_data, $unique_index_refs, $index_refs);
 	}
 }
diff --git a/src/SQLCommandProcessor.php b/src/SQLCommandProcessor.php
index e2bfdce..e76a12c 100644
--- a/src/SQLCommandProcessor.php
+++ b/src/SQLCommandProcessor.php
@@ -10,7 +10,7 @@
  */
 abstract final class SQLCommandProcessor {
 
-	public static function execute(string $sql, AsyncMysqlConnection $conn): (dataset, int) {
+	public static function execute(string $sql, AsyncMysqlConnection $conn): (vec<dict<string, mixed>>, int) {
 
 		// Check for unsupported statements
 		if (Str\starts_with_ci($sql, 'SET') || Str\starts_with_ci($sql, 'BEGIN') || Str\starts_with_ci($sql, 'COMMIT')) {
@@ -33,7 +33,7 @@ public static function execute(string $sql, AsyncMysqlConnection $conn): (datase
 		}
 
 		if ($query is SelectQuery) {
-			return tuple($query->execute($conn), 0);
+			return tuple(vec($query->execute($conn)), 0);
 		} else if ($query is UpdateQuery) {
 			return tuple(vec[], $query->execute($conn));
 		} else if ($query is DeleteQuery) {
diff --git a/src/Server.php b/src/Server.php
index 4a12cc3..8b9d324 100644
--- a/src/Server.php
+++ b/src/Server.php
@@ -10,12 +10,25 @@ public function __construct(public string $name, public ?server_config $config =
 
 	private static dict<string, Server> $instances = dict[];
 	private static keyset<string> $snapshot_names = keyset[];
+	/**
+	 * The main storage mechanism
+	 * dict of strings (database schema names)
+	 * -> dict of string table names to tables
+	 * -> vec of rows
+	 * -> dict of string column names to columns
+	 *
+	 * While a structure based on objects all the way down the stack may be more powerful and readable,
+	 * This structure uses value types intentionally, to enable a relatively efficient reset/snapshot logic
+	 * which is often used frequently between test cases
+	 */
+	public dict<string, database> $databases = dict[];
+	private dict<string, dict<string, database>> $snapshots = dict[];
 
 	public static function getAll(): dict<string, this> {
 		return static::$instances;
 	}
 
-	public static function getAllTables(): dict<string, dict<string, dict<string, vec<dict<string, mixed>>>>> {
+	public static function getAllTables(): dict<string, dict<string, database>> {
 		return Dict\map(static::getAll(), ($server) ==> {
 			return $server->databases;
 		});
@@ -91,24 +104,10 @@ protected function doReset(): void {
 		$this->databases = dict[];
 	}
 
-	/**
-	 * The main storage mechanism
-	 * dict of strings (database schema names)
-	 * -> dict of string table names to tables
-	 * -> vec of rows
-	 * -> dict of string column names to columns
-	 *
-	 * While a structure based on objects all the way down the stack may be more powerful and readable,
-	 * This structure uses value types intentionally, to enable a relatively efficient reset/snapshot logic
-	 * which is often used frequently between test cases
-	 */
-	public dict<string, dict<string, vec<dict<string, mixed>>>> $databases = dict[];
-	private dict<string, dict<string, dict<string, vec<dict<string, mixed>>>>> $snapshots = dict[];
-
 	/**
 	 * Retrieve a table from the specified database, if it exists, by value
 	 */
-	public function getTable(string $dbname, string $name): ?vec<dict<string, mixed>> {
+	public function getTableData(string $dbname, string $name): ?table_data {
 		return $this->databases[$dbname][$name] ?? null;
 	}
 
@@ -117,13 +116,19 @@ public function getTable(string $dbname, string $name): ?vec<dict<string, mixed>
 	 * note, because insert and update operations already grab the full table for checking constraints,
 	 * we don't bother providing an insert or update helper here.
 	 */
-	public function saveTable(string $dbname, string $name, vec<dict<string, mixed>> $rows): void {
+	public function saveTable(
+		string $dbname,
+		string $name,
+		dataset $rows,
+		unique_index_refs $unique_index_refs,
+		index_refs $index_refs,
+	): void {
 		// create table if not exists
 		if (!C\contains_key($this->databases, $dbname)) {
 			$this->databases[$dbname] = dict[];
 		}
 
 		// save rows
-		$this->databases[$dbname][$name] = $rows;
+		$this->databases[$dbname][$name] = tuple($rows, $unique_index_refs, $index_refs);
 	}
 }
diff --git a/src/TableSchema.hack b/src/TableSchema.hack
index 4890707..41b9f59 100644
--- a/src/TableSchema.hack
+++ b/src/TableSchema.hack
@@ -14,11 +14,11 @@ final class TableSchema implements IMemoizeParam {
 	) {}
 
 	public function getInstanceKey(): string {
-		return $this->name;
+		return $this->name.\implode(', ', \HH\Lib\Vec\map($this->fields, $field ==> $field->name));
 	}
 
-	public function getPrimaryKeyColumns(): keyset<string> {
-		$primary = \HH\Lib\Vec\filter($this->indexes, $index ==> $index->name === 'PRIMARY')[0];
-		return $primary->fields;
+	public function getPrimaryKeyColumns(): ?keyset<string> {
+		$primary = \HH\Lib\Vec\filter($this->indexes, $index ==> $index->name === 'PRIMARY')[0] ?? null;
+		return $primary?->fields;
 	}
 }
diff --git a/src/Types.php b/src/Types.php
index 7d880e5..1729043 100644
--- a/src/Types.php
+++ b/src/Types.php
@@ -9,9 +9,12 @@
 // a single DB row
 type row = dict<string, mixed>;
 // vec of rows can be a stored table, a query result set, or an intermediate state for either of those
-type dataset = KeyedContainer<int, row>;
+type dataset = dict<arraykey, row>;
+type unique_index_refs = dict<string, dict<arraykey, arraykey>>;
+type index_refs = dict<string, dict<arraykey, keyset<arraykey>>>;
+type table_data = (dataset, unique_index_refs, index_refs);
 // a database is a collection of named tables
-type database = dict<string, dataset>;
+type database = dict<string, table_data>;
 
 //
 // Parser
diff --git a/tests/DeleteQueryTest.php b/tests/DeleteQueryTest.php
index 8baba96..587ac7b 100644
--- a/tests/DeleteQueryTest.php
+++ b/tests/DeleteQueryTest.php
@@ -15,8 +15,8 @@ final class DeleteQueryTest extends HackTest {
 		expect($results->rows())->toBeSame(vec[
 			dict['id' => 2, 'group_id' => 12345, 'name' => 'name2'],
 			dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
@@ -25,8 +25,8 @@ final class DeleteQueryTest extends HackTest {
 		await $conn->query('DELETE FROM table3 WHERE group_id=12345');
 		$results = await $conn->query('SELECT * FROM table3');
 		expect($results->rows())->toBeSame(vec[
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
@@ -36,8 +36,8 @@ final class DeleteQueryTest extends HackTest {
 		$results = await $conn->query('SELECT * FROM table3');
 		expect($results->rows())->toBeSame(vec[
 			dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
@@ -47,8 +47,8 @@ final class DeleteQueryTest extends HackTest {
 		$results = await $conn->query('SELECT * FROM table3');
 		expect($results->rows())->toBeSame(vec[
 			dict['id' => 1, 'group_id' => 12345, 'name' => 'name1'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
@@ -56,24 +56,24 @@ final class DeleteQueryTest extends HackTest {
 		$conn = static::$conn as nonnull;
 		await $conn->query('DELETE FROM db2.table3 WHERE group_id=12345');
 		$expected = vec[
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		];
 		$results = await $conn->query('SELECT * FROM table3');
 		expect($results->rows())->toBeSame($expected, 'with no backticks');
 
 		await $conn->query('DELETE FROM `db2`.`table3` WHERE group_id=12345');
 		$expected = vec[
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		];
 		$results = await $conn->query('SELECT * FROM table3');
 		expect($results->rows())->toBeSame($expected, 'with backticks');
 
 		await $conn->query('DELETE FROM `db2`.table3 WHERE group_id=12345');
 		$expected = vec[
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		];
 		$results = await $conn->query('SELECT * FROM table3');
 		expect($results->rows())->toBeSame($expected, 'with partial backticks because why not');
@@ -86,8 +86,8 @@ final class DeleteQueryTest extends HackTest {
 		expect($results->rows())->toBeSame(vec[
 			dict['id' => 2, 'group_id' => 12345, 'name' => 'name2'],
 			dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
diff --git a/tests/MultiQueryTest.php b/tests/MultiQueryTest.php
index 9d26f30..f20cf18 100644
--- a/tests/MultiQueryTest.php
+++ b/tests/MultiQueryTest.php
@@ -81,8 +81,8 @@ final class MultiQueryTest extends HackTest {
 			dict['id' => 1, 'group_id' => 12345, 'name' => 'name1'],
 			// no dedupe with union all
 			dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
diff --git a/tests/SelectClauseTest.php b/tests/SelectClauseTest.php
index 06cadb7..19f0715 100644
--- a/tests/SelectClauseTest.php
+++ b/tests/SelectClauseTest.php
@@ -271,8 +271,6 @@ final class SelectClauseTest extends HackTest {
 			dict['id' => 1, 'group_id' => 12345],
 			dict['id' => 2, 'group_id' => 12345],
 			dict['id' => 3, 'group_id' => 12345],
-			dict['id' => 4, 'group_id' => 6],
-			dict['id' => 6, 'group_id' => 6],
 		]);
 	}
 }
diff --git a/tests/SelectExpressionTest.php b/tests/SelectExpressionTest.php
index 7a317f9..c997e4c 100644
--- a/tests/SelectExpressionTest.php
+++ b/tests/SelectExpressionTest.php
@@ -39,8 +39,8 @@ final class SelectExpressionTest extends HackTest {
 			dict['id' => 1, 'group_id' => 12345, 'name' => 'name1'],
 			dict['id' => 2, 'group_id' => 12345, 'name' => 'name2'],
 			dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
@@ -285,8 +285,8 @@ final class SelectExpressionTest extends HackTest {
 		$expected = vec[
 			dict['id' => 1, 'group_id' => 12345, 'name' => 'name1'],
 			dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		];
 		expect($results->rows())->toBeSame($expected, 'not regexp');
 		$results = await $conn->query("SELECT * FROM table3 WHERE name NOT REGEXP BINARY('[a-z]2')");
@@ -371,8 +371,8 @@ final class SelectExpressionTest extends HackTest {
 		expect($results->rows())->toBeSame(
 			vec[
 				dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-				dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-				dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+				dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+				dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 			],
 			'with actual rows',
 		);
diff --git a/tests/SharedSetup.php b/tests/SharedSetup.php
index 43a29ea..cc44bd6 100644
--- a/tests/SharedSetup.php
+++ b/tests/SharedSetup.php
@@ -10,42 +10,117 @@ final class SharedSetup {
 		$pool = new AsyncMysqlConnectionPool(darray[]);
 		$conn = await $pool->connect('example', 1, 'db2', '', '');
 
-		// populate database state
-		$database = dict[
-			'table3' => vec[
-				dict['id' => 1, 'group_id' => 12345, 'name' => 'name1'],
-				dict['id' => 2, 'group_id' => 12345, 'name' => 'name2'],
-				dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-				dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-				dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+		$table3_data = tuple(
+			dict[
+				1 => dict['id' => 1, 'group_id' => 12345, 'name' => 'name1'],
+				2 => dict['id' => 2, 'group_id' => 12345, 'name' => 'name2'],
+				3 => dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
+				4 => dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+				6 => dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
+			],
+			dict[
+				'name_uniq' => dict[
+					'name1' => 1,
+					'name2' => 2,
+					'name3' => 3,
+					'name4' => 4,
+					'name6' => 6,
+				],
+			],
+			dict[
+				'group_id' => dict[
+					12345 => keyset[1, 2, 3],
+					6 => keyset[4, 6],
+				],
+			],
+		);
+
+		$table4_data = tuple(
+			dict[
+				1000 => dict['id' => 1000, 'group_id' => 12345, 'description' => 'desc1'],
+				1001 => dict['id' => 1001, 'group_id' => 12345, 'description' => 'desc2'],
+				1002 => dict['id' => 1002, 'group_id' => 12345, 'description' => 'desc3'],
+				1003 => dict['id' => 1003, 'group_id' => 7, 'description' => 'desc1'],
+				1004 => dict['id' => 1004, 'group_id' => 7, 'description' => 'desc2'],
+			],
+			dict[],
+			dict[
+				'group_id' => dict[
+					12345 => keyset[1000, 1001, 1002],
+					7 => keyset[1003, 1004],
+				],
 			],
-			'table4' => vec[
-				dict['id' => 1000, 'group_id' => 12345, 'description' => 'desc1'],
-				dict['id' => 1001, 'group_id' => 12345, 'description' => 'desc2'],
-				dict['id' => 1002, 'group_id' => 12345, 'description' => 'desc3'],
-				dict['id' => 1003, 'group_id' => 7, 'description' => 'desc1'],
-				dict['id' => 1004, 'group_id' => 7, 'description' => 'desc2'],
+		);
+
+		$table5_data = tuple(
+			dict[
+				1000 => dict['id' => 1000, 'test_type' => 0x0, 'description' => 'desc0'],
+				1001 => dict['id' => 1001, 'test_type' => 0x1, 'description' => 'desc1'],
+				1002 => dict['id' => 1002, 'test_type' => 0x1, 'description' => 'desc2'],
+				1003 => dict['id' => 1003, 'test_type' => 0x2, 'description' => 'desc3'],
+				1004 => dict['id' => 1004, 'test_type' => 0x1, 'description' => 'desc4'],
 			],
-			'table5' => vec[
-				dict['id' => 1000, 'test_type' => 0x0, 'description' => 'desc0'],
-				dict['id' => 1001, 'test_type' => 0x1, 'description' => 'desc1'],
-				dict['id' => 1002, 'test_type' => 0x1, 'description' => 'desc2'],
-				dict['id' => 1003, 'test_type' => 0x2, 'description' => 'desc3'],
-				dict['id' => 1004, 'test_type' => 0x1, 'description' => 'desc4'],
+			dict[],
+			dict[
+				'test_type' => dict[
+					0x0 => keyset[1000],
+					0x1 => keyset[1001, 1002, 1004],
+					0x2 => keyset[1003],
+				],
+			],
+		);
+
+		$association_table_data = tuple(
+			dict[
+				'1||1000||' => dict[
+					'table_3_id' => 1,
+					'table_4_id' => 1000,
+					'group_id' => 12345,
+					'description' => 'association 1',
+				],
+				'1||1001||' => dict[
+					'table_3_id' => 1,
+					'table_4_id' => 1001,
+					'group_id' => 12345,
+					'description' => 'association 2',
+				],
+				'2||1000||' => dict[
+					'table_3_id' => 2,
+					'table_4_id' => 1000,
+					'group_id' => 12345,
+					'description' => 'association 3',
+				],
+				'3||1003||' => dict['table_3_id' => 3, 'table_4_id' => 1003, 'group_id' => 0, 'description' => 'association 4'],
 			],
-			'association_table' => vec[
-				dict['table_3_id' => 1, 'table_4_id' => 1000, 'group_id' => 12345, 'description' => 'association 1'],
-				dict['table_3_id' => 1, 'table_4_id' => 1001, 'group_id' => 12345, 'description' => 'association 2'],
-				dict['table_3_id' => 2, 'table_4_id' => 1000, 'group_id' => 12345, 'description' => 'association 3'],
-				dict['table_3_id' => 3, 'table_4_id' => 1003, 'group_id' => 0, 'description' => 'association 4'],
+			dict[],
+			dict[
+				'table_4_id' => dict[
+					1000 => keyset['1||1000||', '2||1000||'],
+					1001 => keyset['1||1001||'],
+					1003 => keyset['3||1003||'],
+				],
 			],
-			'table6' => vec[
-				dict['id' => 1000, 'position' => '5'],
-				dict['id' => 1001, 'position' => '125'],
-				dict['id' => 1002, 'position' => '75'],
-				dict['id' => 1003, 'position' => '625'],
-				dict['id' => 1004, 'position' => '25'],
+		);
+
+		$table6_data = tuple(
+			dict[
+				1000 => dict['id' => 1000, 'position' => '5'],
+				1001 => dict['id' => 1001, 'position' => '125'],
+				1002 => dict['id' => 1002, 'position' => '75'],
+				1003 => dict['id' => 1003, 'position' => '625'],
+				1004 => dict['id' => 1004, 'position' => '25'],
 			],
+			dict[],
+			dict[],
+		);
+
+		// populate database state
+		$database = dict[
+			'table3' => $table3_data,
+			'table4' => $table4_data,
+			'table5' => $table5_data,
+			'association_table' => $association_table_data,
+			'table6' => $table6_data,
 		];
 
 		$conn->getServer()->databases['db2'] = $database;
@@ -62,18 +137,26 @@ final class SharedSetup {
 		$vitess_conn = await $pool->connect('example2', 2, 'vitess', '', '');
 
 		$vitess_dbs = dict[
-			'vt_table1' => vec[
-				dict['id' => 1, 'name' => 'Pallettown Chickenstrips'],
-				dict['id' => 2, 'name' => 'Brewery Cuttlefish'],
-				dict['id' => 3, 'name' => 'Blasphemy Chowderpants'],
-				dict['id' => 4, 'name' => 'Benjamin Ampersand'],
-			],
-			'vt_table2' => vec[
-				dict['id' => 11, 'vt_table1_id' => 1, 'description' => 'no'],
-				dict['id' => 12, 'vt_table1_id' => 2, 'description' => 'no'],
-				dict['id' => 13, 'vt_table1_id' => 3, 'description' => 'no'],
-				dict['id' => 14, 'vt_table1_id' => 4, 'description' => 'no'],
-			],
+			'vt_table1' => tuple(
+				dict[
+					1 => dict['id' => 1, 'name' => 'Pallettown Chickenstrips'],
+					2 => dict['id' => 2, 'name' => 'Brewery Cuttlefish'],
+					3 => dict['id' => 3, 'name' => 'Blasphemy Chowderpants'],
+					4 => dict['id' => 4, 'name' => 'Benjamin Ampersand'],
+				],
+				dict[],
+				dict[],
+			),
+			'vt_table2' => tuple(
+				dict[
+					11 => dict['id' => 11, 'vt_table1_id' => 1, 'description' => 'no'],
+					12 => dict['id' => 12, 'vt_table1_id' => 2, 'description' => 'no'],
+					13 => dict['id' => 13, 'vt_table1_id' => 3, 'description' => 'no'],
+					14 => dict['id' => 14, 'vt_table1_id' => 4, 'description' => 'no'],
+				],
+				dict[],
+				dict[],
+			),
 		];
 
 		$vitess_conn->getServer()->databases['vitess'] = $vitess_dbs;
diff --git a/tests/UpdateQueryTest.php b/tests/UpdateQueryTest.php
index 8624834..2e70e23 100644
--- a/tests/UpdateQueryTest.php
+++ b/tests/UpdateQueryTest.php
@@ -16,8 +16,8 @@ final class UpdateQueryTest extends HackTest {
 			dict['id' => 1, 'group_id' => 12345, 'name' => 'updated'],
 			dict['id' => 2, 'group_id' => 12345, 'name' => 'name2'],
 			dict['id' => 3, 'group_id' => 12345, 'name' => 'name3'],
-			dict['id' => 4, 'group_id' => 6, 'name' => 'name3'],
-			dict['id' => 6, 'group_id' => 6, 'name' => 'name3'],
+			dict['id' => 4, 'group_id' => 6, 'name' => 'name4'],
+			dict['id' => 6, 'group_id' => 6, 'name' => 'name5'],
 		]);
 	}
 
@@ -26,8 +26,8 @@ final class UpdateQueryTest extends HackTest {
 		await $conn->query("UPDATE table3 set name=CONCAT(name, id, 'updated'), group_id = 13 WHERE group_id=6");
 		$results = await $conn->query('SELECT * FROM table3 WHERE group_id=13');
 		expect($results->rows())->toBeSame(vec[
-			dict['id' => 4, 'group_id' => 13, 'name' => 'name34updated'],
-			dict['id' => 6, 'group_id' => 13, 'name' => 'name36updated'],
+			dict['id' => 4, 'group_id' => 13, 'name' => 'name44updated'],
+			dict['id' => 6, 'group_id' => 13, 'name' => 'name56updated'],
 		]);
 	}
 
@@ -57,8 +57,8 @@ final class UpdateQueryTest extends HackTest {
 	public async function testQualifiedTable(): Awaitable<void> {
 		$conn = static::$conn as nonnull;
 		$expected = vec[
-			dict['id' => 4, 'group_id' => 13, 'name' => 'name34updated'],
-			dict['id' => 6, 'group_id' => 13, 'name' => 'name36updated'],
+			dict['id' => 4, 'group_id' => 13, 'name' => 'name44updated'],
+			dict['id' => 6, 'group_id' => 13, 'name' => 'name56updated'],
 		];
 		await $conn->query("UPDATE db2.table3 set name=CONCAT(name, id, 'updated'), group_id = 13 WHERE group_id=6");
 		$results = await $conn->query('SELECT * FROM table3 WHERE group_id=13');
@@ -68,8 +68,8 @@ final class UpdateQueryTest extends HackTest {
 	public async function testQualifiedTableBackticks(): Awaitable<void> {
 		$conn = static::$conn as nonnull;
 		$expected = vec[
-			dict['id' => 4, 'group_id' => 13, 'name' => 'name34updated'],
-			dict['id' => 6, 'group_id' => 13, 'name' => 'name36updated'],
+			dict['id' => 4, 'group_id' => 13, 'name' => 'name44updated'],
+			dict['id' => 6, 'group_id' => 13, 'name' => 'name56updated'],
 		];
 		await $conn->query("UPDATE `db2`.`table3` set name=CONCAT(name, id, 'updated'), group_id = 13 WHERE group_id=6");
 		$results = await $conn->query('SELECT * FROM table3 WHERE group_id=13');