upgrades: update system.web_sessions migration to handle orphaned rows

This patch updates the system.web_sessions user ID migration to delete
orphaned rows (i.e. rows corresponding to users that have been dropped)
before attempting to backfill.

Release note: None

pkg/upgrade/upgrades/web_sessions_table_user_id_migration.go

@@ -45,6 +45,12 @@ func alterWebSessionsTableAddUserIDColumn(
 	return nil
 }
 
+const deleteRowsForDroppedUsersWebSessionsTableStmt = `
+DELETE FROM system.web_sessions
+WHERE username NOT IN (SELECT username FROM system.users)
+LIMIT 1000
+`
+
 const backfillUserIDColumnWebSessionsTableStmt = `
 UPDATE system.web_sessions AS w
 SET user_id = u.user_id
@@ -62,16 +68,27 @@ func backfillWebSessionsTableUserIDColumn(
 	ctx context.Context, cs clusterversion.ClusterVersion, d upgrade.TenantDeps,
 ) error {
 	ie := d.DB.Executor()
-	for {
-		rowsAffected, err := ie.ExecEx(ctx, "backfill-user-id-col-web-sessions-table", nil, /* txn */
-			sessiondata.NodeUserSessionDataOverride,
-			backfillUserIDColumnWebSessionsTableStmt,
-		)
-		if err != nil {
-			return err
-		}
-		if rowsAffected == 0 {
-			break
+	for _, op := range []struct {
+		name  string
+		query string
+	}{
+		{
+			name:  "delete-rows-for-dropped-users-web-sessions-table",
+			query: deleteRowsForDroppedUsersWebSessionsTableStmt,
+		},
+		{
+			name:  "backfill-user-id-col-web-sessions-table",
+			query: backfillUserIDColumnWebSessionsTableStmt,
+		},
+	} {
+		for {
+			rowsAffected, err := ie.ExecEx(ctx, op.name, nil /* txn */, sessiondata.NodeUserSessionDataOverride, op.query)
+			if err != nil {
+				return err
+			}
+			if rowsAffected == 0 {
+				break
+			}
 		}
 	}
 

pkg/upgrade/upgrades/web_sessions_table_user_id_migration_test.go

@@ -101,6 +101,12 @@ VALUES (
 	})
 	tdb.CheckQueryResults(t, "SELECT count(*) FROM system.web_sessions", [][]string{{strconv.Itoa(numUsers)}})
 
+	// Drop a user to test that migration deletes orphaned rows.
+	if numUsers > 0 {
+		tdb.Exec(t, "DROP USER testuser0")
+		numUsers -= 1
+	}
+
 	// Run migrations.
 	_, err := tc.Conns[0].ExecContext(ctx, `SET CLUSTER SETTING version = $1`,
 		clusterversion.ByKey(clusterversion.V23_1WebSessionsTableHasUserIDColumn).String())