sql: grant/revoke on a large number of objects can create a lot of jobs

Previously we would create multiple jobs for granting on multiple
tables and types within one transaction.  This caused a performance
slowdown, these code changes skip the making of multiple jobs and
just execute the grants in a batch.

Fixes: #117643
Release note (performance improvement): multiple grants on tables and
types within one transaction now run faster.

pkg/sql/grant_revoke.go

@@ -393,15 +393,6 @@ func (n *changeDescriptorBackedPrivilegesNode) startExec(params runParams) error
 			}
 
 		case *tabledesc.Mutable:
-			// TODO (lucy): This should probably have a single consolidated job like
-			// DROP DATABASE.
-			if err := p.createOrUpdateSchemaChangeJob(
-				ctx, d,
-				fmt.Sprintf("updating privileges for table %d", d.ID),
-				descpb.InvalidMutationID,
-			); err != nil {
-				return err
-			}
 			if !d.Dropped() {
 				if err := p.writeSchemaChangeToBatch(ctx, d, b); err != nil {
 					return err
@@ -419,7 +410,7 @@ func (n *changeDescriptorBackedPrivilegesNode) startExec(params runParams) error
 				})
 			}
 		case *typedesc.Mutable:
-			err := p.writeTypeSchemaChange(ctx, d, fmt.Sprintf("updating privileges for type %d", d.ID))
+			err := p.writeDescToBatch(ctx, d, b)
 			if err != nil {
 				return err
 			}

pkg/sql/logictest/testdata/logic_test/grant_type

@@ -126,3 +126,26 @@ test           public       owner_grant_option  other_owner               ALL
 test           public       owner_grant_option  owner_grant_option_child  USAGE           false
 test           public       owner_grant_option  public                    USAGE           false
 test           public       owner_grant_option  root                      ALL             true
+
+statement ok
+CREATE USER roach;
+CREATE TYPE custom_type1 AS ENUM ('roach1', 'roach2', 'roach3');
+CREATE TYPE custom_type2 AS ENUM ('roachA', 'roachB', 'roachC');
+CREATE TYPE custom_type3 AS ENUM ('roachI', 'roachII', 'roachIII');
+BEGIN;
+GRANT ALL ON TYPE custom_type1 TO roach;
+GRANT ALL ON TYPE custom_type2 TO roach;
+GRANT ALL ON TYPE custom_type3 TO roach;
+COMMIT
+
+query T
+select description from [show jobs] where description LIKE '%type%'
+----
+
+query TTTTTTB colnames,rowsort
+SHOW GRANTS FOR roach
+----
+database_name  schema_name  object_name   object_type  grantee  privilege_type  is_grantable
+test           public       custom_type1  type         roach    ALL             false
+test           public       custom_type2  type         roach    ALL             false
+test           public       custom_type3  type         roach    ALL             false

pkg/sql/schema_changer_test.go

@@ -7821,3 +7821,56 @@ func TestMemoryMonitorErrorsDuringBackfillAreRetried(t *testing.T) {
 		require.Equalf(t, shouldFail.Load(), int64(2), "not all failure conditions were hit %d", shouldFail.Load())
 	})
 }
+
+func TestGrantAndSelectsWithinTransaction(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+
+	s, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(ctx)
+
+	connection1 := sqlutils.MakeSQLRunner(sqlDB)
+	connection2 := sqlutils.MakeSQLRunner(sqlDB)
+
+	connection1.Exec(t, `CREATE USER ROACHMIN;`)
+	connection1.Exec(t, `GRANT ADMIN TO ROACHMIN;`)
+	connection1.Exec(t, `CREATE TABLE PROMO_CODES (my_int INT);`)
+	connection1.Exec(t, `CREATE TABLE RIDES (my_int INT);`)
+
+	transaction1 := connection1.Begin(t)
+	transaction1.Exec(`SELECT * FROM PROMO_CODES;`)
+
+	messages := make(chan struct{})
+
+	group := ctxgroup.WithContext(ctx)
+	group.Go(func() error {
+		// wait for channel before starting the sleep
+		<-messages
+		time.Sleep(2 * time.Second)
+		err := transaction1.Commit()
+		return err
+	})
+
+	resBefore := connection2.QueryStr(t, `select count(*) from [show jobs];`)
+
+	transaction2 := connection2.Begin(t)
+	transaction2.Exec(`GRANT ALL ON TABLE PROMO_CODES TO ROACHMIN;`)
+	transaction2.Exec(`GRANT ALL ON TABLE RIDES TO ROACHMIN;`)
+
+	// post to channel to start sleep timer in go routine
+	messages <- struct{}{}
+	startTime := timeutil.Now()
+
+	err := transaction2.Commit()
+	duration := timeutil.Since(startTime)
+
+	err = group.Wait()
+	require.True(t, duration > 1500 * time.Millisecond)
+	require.NoError(t, err)
+
+	resAfter := connection2.QueryStr(t, `select count(*) from [show jobs];`)
+	// no new jobs should have been created between resBefore to resAfter
+	require.True(t, resBefore[0][0] == resAfter[0][0])
+
+}

pkg/sql/type_change.go

@@ -165,6 +165,12 @@ func (p *planner) writeTypeDesc(ctx context.Context, typeDesc *typedesc.Mutable)
 	return p.txn.Run(ctx, b)
 }
 
+func (p *planner) writeDescToBatch(ctx context.Context, typeDesc catalog.MutableDescriptor, b *kv.Batch) error {
+	return p.Descriptors().WriteDescToBatch(
+		ctx, p.extendedEvalCtx.Tracing.KVTracingEnabled(), typeDesc, b,
+	)
+}
+
 // typeSchemaChanger is the struct that actually runs the type schema change.
 type typeSchemaChanger struct {
 	typeID descpb.ID