Merge #117499 #137069

117499: opt: add rule to merge GroupBy and Window operators r=DrewKimball a=DrewKimball

#### opt: add method to add strict dependency to FuncDepSet

This commit adds a new method, `AddStrictDependency`, to `FuncDepSet`.
This will be used in the following commit to add a dependency between
a Window operator's partition columns and its functions. Existing
methods don't work for this because the dependency is not a key.

Epic: None

Release note: None

#### opt: infer functional dependencies for window functions

This commit adds logic to infer strict functional dependencies from
a Window operator's partition column(s) to some or all of its window
functions when the following conditions are satisfied:
1. The window function must be an aggregate, or first_value or last_value.
2. The window frame must be unbounded.

The above conditions ensure that the window function always produces the
same result given the same window frame, as well as that every row in a
partition has the same window frame. This means that the window function
produces the same output for every row in the partition, and therefore,
the partition columns functionally determine the output of the window
function.

Epic: None

Release note: None

#### opt: add rule to merge GroupBy and Window

This commit adds a new norm rule, `FoldGroupByAndWindow`, which can
merge a Window operator with a parent GroupBy operator when the grouping
columns are the same as the partition columns. See the rule comment for
the complete list of conditions. In addition to removing a potentially
expensive Window operator, this transformation makes way for other rules
to match.

Fixes #113292

Release note: None

137069: opt: add implicit SELECT FOR UPDATE to initial scan of DELETE r=yuzefovich a=yuzefovich

This commit makes it so that we apply implicit SELECT FOR UPDATE locking behavior to the initial scan of the DELETE operation in some cases. Namely, we do so when the input to the DELETE is either a scan or an index join on top of a scan (with any number of renders on top) - these conditions mean that all filters were pushed down into the scan, so we won't lock any unnecessary rows. I think it's only possible to have at most one render expression on top of the scan, but I chose to be defensive and allowed nested renders too. In such form the conditions are exactly the same as we use for adding SFU to UPDATEs, so the same function is reused. Existing `enable_implicit_select_for_update` session variable is consulted.

Fixes: #50181.

Release note (sql change): DELETE statements now acquire locks using the FOR UPDATE locking mode during their initial row scan in some comes, which improves performance for contended workloads. This behavior is configurable using the `enable_implicit_select_for_update` session variable.

Co-authored-by: Drew Kimball <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>

pkg/bench/rttanalysis/testdata/benchmark_expectations

@@ -74,7 +74,7 @@ exp,benchmark
 3,Jobs/show_job
 3-5,Jobs/show_jobs
 3,ORMQueries/activerecord_type_introspection_query
-0,ORMQueries/asyncpg_types
+4,ORMQueries/asyncpg_types
 6,ORMQueries/column_descriptions_json_agg
 4,ORMQueries/django_column_introspection_1_table
 4,ORMQueries/django_column_introspection_4_tables

pkg/ccl/logictestccl/testdata/logic_test/regional_by_row_query_behavior

@@ -3171,6 +3171,7 @@ vectorized: true
 │             missing stats
 │             table: user_settings_cascades@user_settings_cascades_user_id_idx
 │             spans: [/'ap-southeast-2'/'5ebfedee-0dcf-41e6-a315-5fa0b51b9882' - /'ap-southeast-2'/'5ebfedee-0dcf-41e6-a315-5fa0b51b9882'] [/'ca-central-1'/'5ebfedee-0dcf-41e6-a315-5fa0b51b9882' - /'ca-central-1'/'5ebfedee-0dcf-41e6-a315-5fa0b51b9882'] [/'us-east-1'/'5ebfedee-0dcf-41e6-a315-5fa0b51b9882' - /'us-east-1'/'5ebfedee-0dcf-41e6-a315-5fa0b51b9882']
+│             locking strength: for update
 │
 └── • constraint-check
     │

pkg/sql/colexec/colexecwindow/window_aggregator_tmpl.go

@@ -247,6 +247,11 @@ func (a *slidingWindowAggregator) processBatch(batch coldata.Batch, startIdx, en
 	})
 }
 
+// INVARIANT: the rows within a window frame are always processed in the same
+// order, regardless of whether the user specified an ordering. This means that
+// two rows with the exact same frame will produce the same result for a given
+// aggregation.
+//
 // execgen:inline
 // execgen:template<removeRows>
 func aggregateOverIntervals(intervals []windowInterval, removeRows bool) {

pkg/sql/logictest/testdata/logic_test/pg_catalog

@@ -4490,6 +4490,7 @@ FROM (
     WHERE c.relname = 'indexes_table'
 ) s2
 GROUP BY indexname, indisunique, indisprimary, amname, exprdef, attoptions
+ORDER BY indexname
 ----
 indexname            array_agg  indisunique  indisprimary  array_agg      amname  exprdef  attoptions
 indexes_include_idx  {a,c,d}    false        false         {ASC,ASC,ASC}  prefix  NULL     NULL

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

@@ -1131,6 +1131,10 @@ var forUpdateLocking = opt.Locking{
 func (b *Builder) shouldApplyImplicitLockingToMutationInput(
 	mutExpr memo.RelExpr,
 ) (opt.TableID, error) {
+	if !b.evalCtx.SessionData().ImplicitSelectForUpdate {
+		return 0, nil
+	}
+
 	switch t := mutExpr.(type) {
 	case *memo.InsertExpr:
 		// Unlike with the other three mutation expressions, it never makes
@@ -1140,23 +1144,23 @@ func (b *Builder) shouldApplyImplicitLockingToMutationInput(
 		return 0, nil
 
 	case *memo.UpdateExpr:
-		return b.shouldApplyImplicitLockingToUpdateInput(t), nil
+		return shouldApplyImplicitLockingToUpdateOrDeleteInput(t), nil
 
 	case *memo.UpsertExpr:
-		return b.shouldApplyImplicitLockingToUpsertInput(t), nil
+		return shouldApplyImplicitLockingToUpsertInput(t), nil
 
 	case *memo.DeleteExpr:
-		return b.shouldApplyImplicitLockingToDeleteInput(t), nil
+		return shouldApplyImplicitLockingToUpdateOrDeleteInput(t), nil
 
 	default:
 		return 0, errors.AssertionFailedf("unexpected mutation expression %T", t)
 	}
 }
 
-// shouldApplyImplicitLockingToUpdateInput determines whether or not the builder
-// should apply a FOR UPDATE row-level locking mode to the initial row scan of
-// an UPDATE statement. If the builder should lock the initial row scan, it
-// returns the TableID of the scan, otherwise it returns 0.
+// shouldApplyImplicitLockingToUpdateOrDeleteInput determines whether the
+// builder should apply a FOR UPDATE row-level locking mode to the initial row
+// scan of an UPDATE statement or a DELETE. If the builder should lock the
+// initial row scan, it returns the TableID of the scan, otherwise it returns 0.
 //
 // Conceptually, if we picture an UPDATE statement as the composition of a
 // SELECT statement and an INSERT statement (with loosened semantics around
@@ -1176,16 +1180,15 @@ func (b *Builder) shouldApplyImplicitLockingToMutationInput(
 // is strictly a performance optimization for contended writes. Therefore, it is
 // not worth risking the transformation being a pessimization, so it is only
 // applied when doing so does not risk creating artificial contention.
-func (b *Builder) shouldApplyImplicitLockingToUpdateInput(upd *memo.UpdateExpr) opt.TableID {
-	if !b.evalCtx.SessionData().ImplicitSelectForUpdate {
-		return 0
-	}
-
-	// Try to match the Update's input expression against the pattern:
+//
+// UPDATEs and DELETEs happen to have exactly the same matching pattern, so we
+// reuse this function for both.
+func shouldApplyImplicitLockingToUpdateOrDeleteInput(mutExpr memo.RelExpr) opt.TableID {
+	// Try to match the mutation's input expression against the pattern:
 	//
 	//   [Project]* [IndexJoin] Scan
 	//
-	input := upd.Input
+	input := mutExpr.Child(0).(memo.RelExpr)
 	input = unwrapProjectExprs(input)
 	if idxJoin, ok := input.(*memo.IndexJoinExpr); ok {
 		input = idxJoin.Input
@@ -1200,11 +1203,7 @@ func (b *Builder) shouldApplyImplicitLockingToUpdateInput(upd *memo.UpdateExpr)
 // should apply a FOR UPDATE row-level locking mode to the initial row scan of
 // an UPSERT statement. If the builder should lock the initial row scan, it
 // returns the TableID of the scan, otherwise it returns 0.
-func (b *Builder) shouldApplyImplicitLockingToUpsertInput(ups *memo.UpsertExpr) opt.TableID {
-	if !b.evalCtx.SessionData().ImplicitSelectForUpdate {
-		return 0
-	}
-
+func shouldApplyImplicitLockingToUpsertInput(ups *memo.UpsertExpr) opt.TableID {
 	// Try to match the Upsert's input expression against the pattern:
 	//
 	//   [Project]* (LeftJoin Scan | LookupJoin) [Project]* Values
@@ -1235,17 +1234,6 @@ func (b *Builder) shouldApplyImplicitLockingToUpsertInput(ups *memo.UpsertExpr)
 	return 0
 }
 
-// tryApplyImplicitLockingToDeleteInput determines whether or not the builder
-// should apply a FOR UPDATE row-level locking mode to the initial row scan of a
-// DELETE statement. If the builder should lock the initial row scan, it returns
-// the TableID of the scan, otherwise it returns 0.
-//
-// TODO(nvanbenschoten): implement this method to match on appropriate Delete
-// expression trees and apply a row-level locking mode.
-func (b *Builder) shouldApplyImplicitLockingToDeleteInput(del *memo.DeleteExpr) opt.TableID {
-	return 0
-}
-
 // unwrapProjectExprs unwraps zero or more nested ProjectExprs. It returns the
 // first non-ProjectExpr in the chain, or the input if it is not a ProjectExpr.
 func unwrapProjectExprs(input memo.RelExpr) memo.RelExpr {

pkg/sql/opt/exec/execbuilder/testdata/cascade

@@ -900,6 +900,7 @@ vectorized: true
 │             missing stats
 │             table: self@self_pkey
 │             spans: [/4 - /4]
+│             locking strength: for update
 │
 └── • fk-cascade
     │ fk: self_b_fkey
@@ -984,6 +985,7 @@ vectorized: true
 │             missing stats
 │             table: loop_a@loop_a_pkey
 │             spans: [/'loop_a-pk1' - /'loop_a-pk1']
+│             locking strength: for update
 │
 └── • fk-cascade
     │ fk: loop_b_cascade_delete_fkey
@@ -1085,6 +1087,7 @@ quality of service: regular
 │             missing stats
 │             table: loop_a@loop_a_pkey
 │             spans: [/'loop_a-pk1' - /'loop_a-pk1']
+│             locking strength: for update
 │
 └── • fk-cascade
     │ fk: loop_b_cascade_delete_fkey
@@ -1235,6 +1238,7 @@ vectorized: true
 │             missing stats
 │             table: t3@t3_pkey
 │             spans: [/1 - /1]
+│             locking strength: for update
 │
 ├── • fk-cascade
 │   │ fk: fk1

pkg/sql/opt/exec/execbuilder/testdata/delete

@@ -141,6 +141,7 @@ vectorized: true
       table: unindexed@unindexed_pkey
       spans: [/1 - ]
       limit: 1
+      locking strength: for update
 
 query T
 EXPLAIN DELETE FROM indexed WHERE value = 5 LIMIT 10
@@ -157,6 +158,7 @@ vectorized: true
       table: indexed@indexed_value_idx
       spans: [/5 - /5]
       limit: 10
+      locking strength: for update
 
 query T
 EXPLAIN DELETE FROM indexed LIMIT 10
@@ -173,6 +175,7 @@ vectorized: true
       table: indexed@indexed_value_idx
       spans: LIMITED SCAN
       limit: 10
+      locking strength: for update
 
 # TODO(andyk): Prune columns so that index-join is not necessary.
 query T
@@ -190,6 +193,7 @@ vectorized: true
       table: indexed@indexed_value_idx
       spans: [/5 - /5]
       limit: 10
+      locking strength: for update
 
 # Ensure that index hints in DELETE statements force the choice of a specific index
 # as described in #38799.
@@ -213,6 +217,7 @@ vectorized: true
       estimated row count: 1,000 (missing stats)
       table: t38799@foo
       spans: FULL SCAN
+      locking strength: for update
 
 # Tracing tests for fast delete.
 statement ok
@@ -331,12 +336,14 @@ vectorized: true
         │ estimated row count: 990 (missing stats)
         │ table: xyz@xyz_pkey
         │ key columns: x
+        │ locking strength: for update
         │
         └── • scan
               columns: (x, y)
               estimated row count: 990 (missing stats)
               table: xyz@xyz_y_idx
               spans: /1-/1000 /2001-/3000
+              locking strength: for update
 
 # Testcase for issue 105803.
 

pkg/sql/opt/exec/execbuilder/testdata/explain

@@ -560,7 +560,8 @@ vectorized: true
     └── • group (hash)
         │ group by: column_name, ordinal_position, column_default, is_nullable, generation_expression, is_hidden, crdb_sql_type
         │
-        └── • window
+        └── • sort
+            │ order: +index_name
             │
             └── • hash join (left outer)
                 │ equality: (column_name) = (column_name)
@@ -734,39 +735,41 @@ vectorized: true
             │ estimated row count: 3
             │ order: +"role"
             │
-            └── • hash join (left outer)
+            └── • merge join (right outer)
                 │ estimated row count: 3
-                │ equality: (username) = (member)
-                │ left cols are key
+                │ equality: (member) = (username)
+                │ right cols are key
                 │
-                ├── • group (hash)
-                │   │ estimated row count: 3
-                │   │ group by: username
-                │   │
-                │   └── • window
-                │       │ estimated row count: 3
-                │       │
-                │       └── • render
-                │           │
-                │           └── • hash join (left outer)
-                │               │ estimated row count: 3
-                │               │ equality: (username) = (username)
-                │               │ left cols are key
-                │               │
-                │               ├── • scan
-                │               │     estimated row count: 3 (100% of the table; stats collected <hidden> ago)
-                │               │     table: users@users_user_id_idx
-                │               │     spans: FULL SCAN
-                │               │
-                │               └── • scan
-                │                     estimated row count: 1 (100% of the table; stats collected <hidden> ago)
-                │                     table: role_options@primary
-                │                     spans: FULL SCAN
+                ├── • scan
+                │     estimated row count: 1 (100% of the table; stats collected <hidden> ago)
+                │     table: role_members@role_members_member_idx
+                │     spans: FULL SCAN
                 │
-                └── • scan
-                      estimated row count: 1 (100% of the table; stats collected <hidden> ago)
-                      table: role_members@role_members_role_idx
-                      spans: FULL SCAN
+                └── • group (streaming)
+                    │ estimated row count: 3
+                    │ group by: username
+                    │ ordered: +username
+                    │
+                    └── • sort
+                        │ estimated row count: 3
+                        │ order: +username,+option
+                        │
+                        └── • render
+                            │
+                            └── • hash join (left outer)
+                                │ estimated row count: 3
+                                │ equality: (username) = (username)
+                                │ left cols are key
+                                │
+                                ├── • scan
+                                │     estimated row count: 3 (100% of the table; stats collected <hidden> ago)
+                                │     table: users@users_user_id_idx
+                                │     spans: FULL SCAN
+                                │
+                                └── • scan
+                                      estimated row count: 1 (100% of the table; stats collected <hidden> ago)
+                                      table: role_options@primary
+                                      spans: FULL SCAN
 
 # EXPLAIN selecting from a sequence.
 statement ok

pkg/sql/opt/exec/execbuilder/testdata/explain_redact

@@ -1458,6 +1458,7 @@ vectorized: true
           missing stats
           table: f@f_f_idx (partial index)
           spans: 1 span
+          locking strength: for update
 
 query T
 EXPLAIN (VERBOSE, REDACT) DELETE FROM f WHERE f = 8.5
@@ -1482,6 +1483,7 @@ vectorized: true
           estimated row count: 10 (missing stats)
           table: f@f_f_idx (partial index)
           spans: 1 span
+          locking strength: for update
 
 query T
 EXPLAIN (OPT, REDACT) DELETE FROM f WHERE f = 8.5