opt: add swap flag to MutationPrivate

Inside the optimizer we can simply use a boolean flag to distinguish
between update and delete, and update swap and delete swap,
respectively. Add this flag to MutationPrivate and use it in execbuild
to construct the swap nodes.

Informs: #144503

Release note: None

Author: michae2

pkg/sql/opt/exec/execbuilder/mutation.go

@@ -463,19 +463,37 @@ func (b *Builder) buildUpdate(upd *memo.UpdateExpr) (_ execPlan, outputCols colO
 		}
 	}
 
-	node, err := b.factory.ConstructUpdate(
-		input.root,
-		tab,
-		fetchColOrds,
-		updateColOrds,
-		returnColOrds,
-		checkOrds,
-		passthroughCols,
-		upd.UniqueWithTombstoneIndexes,
-		lockedIndexes,
-		b.allowAutoCommit && len(upd.UniqueChecks) == 0 &&
-			len(upd.FKChecks) == 0 && len(upd.FKCascades) == 0 && upd.AfterTriggers == nil,
-	)
+	allowAutoCommit := b.allowAutoCommit && len(upd.UniqueChecks) == 0 &&
+		len(upd.FKChecks) == 0 && len(upd.FKCascades) == 0 && upd.AfterTriggers == nil
+	var node exec.Node
+	if upd.Swap {
+		if !checkOrds.Empty() || len(upd.UniqueWithTombstoneIndexes) != 0 {
+			return execPlan{}, colOrdMap{}, errors.AssertionFailedf("update swap does not support checks")
+		}
+		node, err = b.factory.ConstructUpdateSwap(
+			input.root,
+			tab,
+			fetchColOrds,
+			updateColOrds,
+			returnColOrds,
+			passthroughCols,
+			lockedIndexes,
+			allowAutoCommit,
+		)
+	} else {
+		node, err = b.factory.ConstructUpdate(
+			input.root,
+			tab,
+			fetchColOrds,
+			updateColOrds,
+			returnColOrds,
+			checkOrds,
+			passthroughCols,
+			upd.UniqueWithTombstoneIndexes,
+			lockedIndexes,
+			allowAutoCommit,
+		)
+	}
 	if err != nil {
 		return execPlan{}, colOrdMap{}, err
 	}
@@ -624,16 +642,30 @@ func (b *Builder) buildDelete(del *memo.DeleteExpr) (_ execPlan, outputCols colO
 		}
 	}
 
-	node, err := b.factory.ConstructDelete(
-		input.root,
-		tab,
-		fetchColOrds,
-		returnColOrds,
-		passthroughCols,
-		lockedIndexes,
-		b.allowAutoCommit && len(del.FKChecks) == 0 &&
-			len(del.FKCascades) == 0 && del.AfterTriggers == nil,
-	)
+	allowAutoCommit := b.allowAutoCommit && len(del.FKChecks) == 0 &&
+		len(del.FKCascades) == 0 && del.AfterTriggers == nil
+	var node exec.Node
+	if del.Swap {
+		node, err = b.factory.ConstructDeleteSwap(
+			input.root,
+			tab,
+			fetchColOrds,
+			returnColOrds,
+			passthroughCols,
+			lockedIndexes,
+			allowAutoCommit,
+		)
+	} else {
+		node, err = b.factory.ConstructDelete(
+			input.root,
+			tab,
+			fetchColOrds,
+			returnColOrds,
+			passthroughCols,
+			lockedIndexes,
+			allowAutoCommit,
+		)
+	}
 	if err != nil {
 		return execPlan{}, colOrdMap{}, err
 	}

pkg/sql/opt/memo/expr_format.go

@@ -1918,6 +1918,9 @@ func FormatPrivate(f *ExprFmtCtx, private interface{}, physProps *physical.Requi
 		fmt.Fprintf(f.Buffer, " %s", seq.Name())
 
 	case *MutationPrivate:
+		if t.Swap {
+			fmt.Fprint(f.Buffer, " (swap)")
+		}
 		f.formatIndex(t.Table, cat.PrimaryIndex, false /* reverse */)
 
 	case *LockPrivate:

pkg/sql/opt/memo/logical_props_builder.go

@@ -1610,6 +1610,10 @@ func (b *logicalPropsBuilder) buildMutationProps(mutation RelExpr, rel *props.Re
 	// -----------
 	// Inherit cardinality from input.
 	rel.Cardinality = inputProps.Cardinality
+	if private.Swap {
+		// Swap mutations can return zero rows.
+		rel.Cardinality = rel.Cardinality.AsLowAs(0)
+	}
 
 	// Statistics
 	// ----------

pkg/sql/opt/ops/mutation.opt

@@ -211,6 +211,9 @@ define MutationPrivate {
     # VectorInsert indicates that the mutation is an insert with a specialized
     # vectorized implementation used for Copy statements.
     VectorInsert bool
+
+    # Swap indicates that the mutation is an update swap or delete swap.
+    Swap bool
 }
 
 # Update evaluates a relational input expression that fetches existing rows from