sql: add support for DELETE FROM ... USING to execbuilder

Previously, while the optimizer would generate query plans
for sql statements of the form `DELETE FROM ... USING`
executing the statement in an instance of CockroachDB
may return errors, particularly with statements that
included `RETURNING` clauses. This commit adds support to
the execbuilder to execute statements of the form
`DELETE FROM ... USING`.

Release note (sql change): CockroachDB now supports
executing statements of the form `DELETE FROM ... USING`.

pkg/sql/delete.go

@@ -50,8 +50,8 @@ type deleteRun struct {
 	traceKV bool
 
 	// partialIndexDelValsOffset is the offset of partial index delete
-	// indicators in the source values. It is equal to the number of fetched
-	// columns.
+	// indicators in the source values. It is equal to the sum of the number
+	// of fetched columns and the number of passthrough columns.
 	partialIndexDelValsOffset int
 
 	// rowIdxToRetIdx is the mapping from the columns returned by the deleter
@@ -60,6 +60,11 @@ type deleteRun struct {
 	// of the mutation. Otherwise, the value at the i-th index refers to the
 	// index of the resultRowBuffer where the i-th column is to be returned.
 	rowIdxToRetIdx []int
+
+	// numPassthrough is the number of columns in addition to the set of columns
+	// of the target table being returned, that must be passed through from the
+	// input node.
+	numPassthrough int
 }
 
 var _ mutationPlanNode = &deleteNode{}
@@ -184,12 +189,32 @@ func (d *deleteNode) processSourceRow(params runParams, sourceVals tree.Datums)
 		// d.run.rows.NumCols() is guaranteed to only contain the requested
 		// public columns.
 		resultValues := make(tree.Datums, d.run.td.rows.NumCols())
-		for i, retIdx := range d.run.rowIdxToRetIdx {
+		largestRetIdx := -1
+		for i := range d.run.rowIdxToRetIdx {
+			retIdx := d.run.rowIdxToRetIdx[i]
 			if retIdx >= 0 {
+				if retIdx >= largestRetIdx {
+					largestRetIdx = retIdx
+				}
 				resultValues[retIdx] = sourceVals[i]
 			}
 		}
 
+		// At this point we've extracted all the RETURNING values that are part
+		// of the target table. We must now extract the columns in the RETURNING
+		// clause that refer to other tables (from the USING clause of the delete).
+		if d.run.numPassthrough > 0 {
+			passthroughBegin := len(d.run.td.rd.FetchCols)
+			passthroughEnd := passthroughBegin + d.run.numPassthrough
+			passthroughValues := sourceVals[passthroughBegin:passthroughEnd]
+
+			for i := 0; i < d.run.numPassthrough; i++ {
+				largestRetIdx++
+				resultValues[largestRetIdx] = passthroughValues[i]
+			}
+
+		}
+
 		if _, err := d.run.td.rows.AddRow(params.ctx, resultValues); err != nil {
 			return err
 		}

pkg/sql/distsql_spec_exec_factory.go

@@ -997,6 +997,7 @@ func (e *distSQLSpecExecFactory) ConstructDelete(
 	table cat.Table,
 	fetchCols exec.TableColumnOrdinalSet,
 	returnCols exec.TableColumnOrdinalSet,
+	passthrough colinfo.ResultColumns,
 	autoCommit bool,
 ) (exec.Node, error) {
 	return nil, unimplemented.NewWithIssue(47473, "experimental opt-driven distsql planning: delete")

pkg/sql/logictest/testdata/logic_test/cursor

@@ -598,4 +598,3 @@ FETCH 1 a b;
 
 statement ok
 COMMIT;
-

pkg/sql/logictest/testdata/logic_test/delete

@@ -307,9 +307,6 @@ SELECT x, y, z FROM family
 1  1  NULL
 3  3  NULL
 
-statement error at or near "where": syntax error: unimplemented: this syntax
-DELETE FROM family USING family, other_table WHERE x=2
-
 # Verify that the fast path does its deletes at the expected timestamp.
 statement ok
 CREATE TABLE a (a INT PRIMARY KEY)
@@ -337,3 +334,194 @@ SELECT * FROM a AS OF SYSTEM TIME $ts
 3
 4
 5
+
+# Test that USING works.
+
+statement ok
+CREATE TABLE u_a (
+    a INT NOT NULL PRIMARY KEY,
+    b STRING,
+    c INT
+)
+
+statement ok
+CREATE TABLE u_b (
+  a INT NOT NULL PRIMARY KEY,
+  b STRING
+)
+
+statement ok
+CREATE TABLE u_c (
+  a INT NOT NULL PRIMARY KEY,
+  b STRING,
+  c INT
+)
+
+statement ok
+CREATE TABLE u_d (
+  a INT,
+  b INT
+)
+
+statement ok
+INSERT INTO u_a VALUES (1, 'a', 10), (2, 'b', 20), (3, 'c', 30), (4, 'd', 40)
+
+statement ok
+INSERT INTO u_b VALUES (10, 'a'), (20, 'b'), (30, 'c'), (40, 'd')
+
+statement ok
+INSERT INTO u_c VALUES (1, 'a', 10), (2, 'b', 50), (3, 'c', 50), (4, 'd', 40)
+
+# Test a join with a filter.
+statement ok
+DELETE FROM u_a USING u_b WHERE c = u_b.a AND u_b.b = 'd'
+
+query ITI rowsort
+SELECT * FROM u_a;
+----
+1  a  10
+2  b  20
+3  c  30
+
+# Test a self join.
+statement ok
+INSERT INTO u_a VALUES (5, 'd', 5), (6, 'e', 6)
+
+statement ok
+DELETE FROM u_a USING u_a u_a2 WHERE u_a.a = u_a2.c
+
+query ITI rowsort
+SELECT * FROM u_a;
+----
+1  a  10
+2  b  20
+3  c  30
+
+# Test when USING uses multiple tables.
+
+statement ok
+INSERT INTO u_c VALUES (30, 'a', 1)
+
+statement ok
+DELETE FROM u_a USING u_b, u_c WHERE u_a.c = u_b.a AND u_a.c = u_c.a
+
+query ITI rowsort
+SELECT * FROM u_a;
+----
+1  a  10
+2  b  20
+
+# Test if USING works well with RETURNING expressions that reference
+# the USING table and target table.
+query ITIT colnames,rowsort
+DELETE FROM u_a USING u_b WHERE u_a.c = u_b.a RETURNING u_b.a, u_b.b, u_a.a, u_a.b;
+----
+a   b  a  b
+10  a  1  a
+20  b  2  b
+
+query ITI rowsort
+SELECT * FROM u_a;
+----
+
+statement ok
+INSERT INTO u_a VALUES (1, 'a', 10), (2, 'b', 20), (3, 'c', 30), (4, 'd', 40);
+
+# Test if RETURNING * returns everything.
+query ITIITI colnames,rowsort
+DELETE FROM u_a USING u_c WHERE u_a.c = u_c.c RETURNING *;
+----
+a  b  c   a  b  c
+1  a  10  1  a  10
+4  d  40  4  d  40
+
+# Clean u_a to input a new set of data, and to improve test readability.
+statement ok
+TRUNCATE u_a
+
+statement ok
+INSERT INTO u_a VALUES (1, 'a', 5), (2, 'b', 10), (3, 'c', 15), (4, 'd', 20), (5, 'd', 25), (6, 'd', 30), (7, 'd', 35), (8, 'd', 40), (9, 'd', 45)
+
+# Using ORDER BY and LIMIT with a `DELETE ... USING` where ORDER BY and LIMIT references the USING
+# table is not supported.
+# TODO(#89817): Add support in DELETE ... USING for ORDER BY clauses to reference the USING
+# table. This is not supported in UPDATE ... FROM either: #89817.
+statement error SELECT DISTINCT ON expressions must match initial ORDER BY expressions
+DELETE FROM u_a AS foo USING u_b AS bar WHERE bar.a > foo.c ORDER BY bar.a DESC LIMIT 3 RETURNING *;
+
+# Test aliased table names, ORDER BY and LIMIT where ORDER BY references the target
+# table.
+query ITIIT
+DELETE FROM u_a AS foo USING u_b AS bar WHERE bar.a > foo.c ORDER BY foo.a DESC LIMIT 3 RETURNING *;
+----
+7  d  35  40  d
+6  d  30  40  d
+5  d  25  40  d
+
+query ITI rowsort
+SELECT * FROM u_a;
+----
+1  a  5
+2  b  10
+3  c  15
+4  d  20
+8  d  40
+9  d  45
+
+statement ok
+INSERT INTO u_d VALUES (1, 10), (2, 20), (3, 30), (4, 40)
+
+query IT rowsort
+SELECT * FROM u_b;
+----
+10  a
+20  b
+30  c
+40  d
+
+query ITI rowsort
+SELECT * FROM u_c;
+----
+1   a  10
+2   b  50
+3   c  50
+4   d  40
+30  a  1
+
+# Test if DELETE FROM ... USING works with LATERAL.
+
+statement ok
+DELETE FROM u_a USING u_b, LATERAL (SELECT u_c.a, u_c.b, u_c.c FROM u_c WHERE u_b.b = u_c.b) AS other WHERE other.c = 1 AND u_a.c = 35
+
+query ITI rowsort
+SELECT * FROM u_a
+----
+1  a  5
+2  b  10
+3  c  15
+4  d  20
+8  d  40
+9  d  45
+
+# Test if DELETE FROM ... USING works with partial indexes.
+
+statement ok
+CREATE TABLE pindex (
+  a DECIMAL(10, 2),
+  INDEX (a) WHERE a > 3
+)
+
+statement ok
+INSERT INTO pindex VALUES (1.0), (2.0), (3.0), (4.0), (5.0), (8.0)
+
+statement ok
+DELETE FROM pindex USING (VALUES (5.0), (6.0)) v(b) WHERE pindex.a = v.b
+
+query F rowsort
+SELECT * FROM pindex;
+----
+1.00
+2.00
+3.00
+4.00
+8.00

pkg/sql/logictest/testdata/logic_test/views

@@ -1492,4 +1492,3 @@ SELECT * FROM v;
 
 statement ok
 SET DATABASE = test;
-

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

@@ -493,25 +493,39 @@ func (b *Builder) buildDelete(del *memo.DeleteExpr) (execPlan, error) {
 	//
 	// TODO(andyk): Using ensureColumns here can result in an extra Render.
 	// Upgrade execution engine to not require this.
-	colList := make(opt.ColList, 0, len(del.FetchCols)+len(del.PartialIndexDelCols))
+	colList := make(opt.ColList, 0, len(del.FetchCols)+len(del.PartialIndexDelCols)+len(del.PassthroughCols))
 	colList = appendColsWhenPresent(colList, del.FetchCols)
 	colList = appendColsWhenPresent(colList, del.PartialIndexDelCols)
 
+	if del.NeedResults() {
+		colList = append(colList, del.PassthroughCols...)
+	}
+
 	input, err := b.buildMutationInput(del, del.Input, colList, &del.MutationPrivate)
 	if err != nil {
 		return execPlan{}, err
 	}
-
 	// Construct the Delete node.
 	md := b.mem.Metadata()
 	tab := md.Table(del.Table)
 	fetchColOrds := ordinalSetFromColList(del.FetchCols)
 	returnColOrds := ordinalSetFromColList(del.ReturnCols)
+
+	//Construct the result columns for the passthrough set
+	var passthroughCols colinfo.ResultColumns
+	if del.NeedResults() {
+		for _, passthroughCol := range del.PassthroughCols {
+			colMeta := b.mem.Metadata().ColumnMeta(passthroughCol)
+			passthroughCols = append(passthroughCols, colinfo.ResultColumn{Name: colMeta.Alias, Typ: colMeta.Type})
+		}
+	}
+
 	node, err := b.factory.ConstructDelete(
 		input.root,
 		tab,
 		fetchColOrds,
 		returnColOrds,
+		passthroughCols,
 		b.allowAutoCommit && len(del.FKChecks) == 0 && len(del.FKCascades) == 0,
 	)
 	if err != nil {

pkg/sql/opt/exec/explain/testdata/gists

@@ -699,8 +699,8 @@ explain(gist):
 gist-explain-roundtrip
 DELETE FROM foo
 ----
-hash: 5369057709634423529
-plan-gist: AgFqAgAHAAAAI2oB
+hash: 17378315733259356217
+plan-gist: AgFqAgAHAAAAI2oAAQ==
 explain(shape):
 • delete
 │ from: foo
@@ -722,8 +722,8 @@ explain(gist):
 gist-explain-roundtrip
 DELETE FROM foo WHERE a = 1
 ----
-hash: 7691685103096689151
-plan-gist: AgFqAgAHAgAAI2oB
+hash: 11485970487285265051
+plan-gist: AgFqAgAHAgAAI2oAAQ==
 explain(shape):
 • delete
 │ from: foo

pkg/sql/opt/exec/explain/testdata/gists_tpce

@@ -210,8 +210,8 @@ update_trade_submitted AS (
 )
 SELECT * FROM request_list;
 ----
-hash: 7096273538769246907
-plan-gist: AgGkAQIAHwIAAAcQBRAhpAEAAAcCMAGUAQIAHwAAAAMHCDAxBQIUAJQBAgIBBQgHCAUII5QBAAcCMDEFAgcGBQYwH5IBADEFAhQFkAECAgEqMQUCFAWwAQICASoHAjAxBQIUAJABAgIBBRwHIAUgMCGQAQAAMQUCFAWwAQICASoHAjAxBQgGCA==
+hash: 14329018118666014305
+plan-gist: AgGkAQIAHwIAAAcQBRAhpAEAAAcCMAGUAQIAHwAAAAMHCDAxBQIUAJQBAgIBBQgHCAUII5QBAAAHAjAxBQIHBgUGMB+SAQAxBQIUBZABAgIBKjEFAhQFsAECAgEqBwIwMQUCFACQAQICAQUcByAFIDAhkAEAADEFAhQFsAECAgEqBwIwMQUIBgg=
 explain(shape):
 • root
 │

pkg/sql/opt/exec/factory.opt

@@ -556,11 +556,17 @@ define Upsert {
 # The fetchCols set contains the ordinal positions of the fetch columns in
 # the target table. The input must contain those columns in the same order
 # as they appear in the table schema.
+#
+# The passthrough parameter contains all the result columns that are part of
+# the input node that the update node needs to return (passing through from
+# the input). The pass through columns are used to return any column from the
+# USING tables that are referenced in the RETURNING clause.
 define Delete {
     Input exec.Node
     Table cat.Table
     FetchCols exec.TableColumnOrdinalSet
     ReturnCols exec.TableColumnOrdinalSet
+    Passthrough colinfo.ResultColumns
 
     # If set, the operator will commit the transaction as part of its execution.
     # This is false when executing inside an explicit transaction, or there are