sql: re-add GC job on schema element deletion

This commit creates GC jobs upon the deletion of an index, table or
database. Similarly to the previous implementation, it considers the
walltime at which the schema change completed to be the drop time of the
schema element.

Release note (sql change): Previously, after deleting an index, table,
or database the relevant schema change job would change its running
status to waiting for GC TTL. The schema change and the GC process are
now decoupled into 2 jobs.

Release justification: This is a follow up to the migration of turning
schema changes into actual jobs. This commit re-adds the ability to
properly GC indexes and tables.

pkg/base/testing_knobs.go

@@ -24,6 +24,7 @@ type TestingKnobs struct {
 	SQLExecutor         ModuleTestingKnobs
 	SQLLeaseManager     ModuleTestingKnobs
 	SQLSchemaChanger    ModuleTestingKnobs
+	GCJob               ModuleTestingKnobs
 	PGWireTestingKnobs  ModuleTestingKnobs
 	SQLMigrationManager ModuleTestingKnobs
 	DistSQL             ModuleTestingKnobs

pkg/ccl/backupccl/backup_test.go

@@ -1347,12 +1347,10 @@ func TestRestoreFailCleanup(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
 	params := base.TestServerArgs{}
-	// Disable external processing of mutations so that the final check of
-	// crdb_internal.tables is guaranteed to not be cleaned up. Although this
-	// was never observed by a stress test, it is here for safety.
-	params.Knobs.SQLSchemaChanger = &sql.SchemaChangerTestingKnobs{
-		// TODO (lucy): Turn on knob to disable GC once the GC job is implemented.
-	}
+	// Disable GC job so that the final check of crdb_internal.tables is
+	// guaranteed to not be cleaned up. Although this was never observed by a
+	// stress test, it is here for safety.
+	params.Knobs.GCJob = &sql.GCJobTestingKnobs{RunBeforeResume: func() { select {} /* blocks forever */ }}
 
 	const numAccounts = 1000
 	_, _, sqlDB, dir, cleanup := backupRestoreTestSetupWithParams(t, singleNode, numAccounts,
@@ -1395,9 +1393,8 @@ func TestRestoreFailDatabaseCleanup(t *testing.T) {
 	// Disable external processing of mutations so that the final check of
 	// crdb_internal.tables is guaranteed to not be cleaned up. Although this
 	// was never observed by a stress test, it is here for safety.
-	params.Knobs.SQLSchemaChanger = &sql.SchemaChangerTestingKnobs{
-		// TODO (lucy): Turn on knob to disable GC once the GC job is implemented.
-	}
+	blockGC := make(chan struct{})
+	params.Knobs.GCJob = &sql.GCJobTestingKnobs{RunBeforeResume: func() { <-blockGC }}
 
 	const numAccounts = 1000
 	_, _, sqlDB, dir, cleanup := backupRestoreTestSetupWithParams(t, singleNode, numAccounts,
@@ -1428,6 +1425,7 @@ func TestRestoreFailDatabaseCleanup(t *testing.T) {
 		t, `database "data" does not exist`,
 		`DROP DATABASE data`,
 	)
+	close(blockGC)
 }
 
 func TestBackupRestoreInterleaved(t *testing.T) {

pkg/ccl/partitionccl/drop_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/gcjob"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -40,15 +41,16 @@ func TestDropIndexWithZoneConfigCCL(t *testing.T) {
 
 	const numRows = 100
 
+	defer gcjob.SetSmallMaxGCIntervalForTest()()
+
 	asyncNotification := make(chan struct{})
 
 	params, _ := tests.CreateTestServerParams()
 	params.Knobs = base.TestingKnobs{
-		SQLSchemaChanger: &sql.SchemaChangerTestingKnobs{
-			// TODO (lucy): Currently there's no index GC job implemented. Eventually
-			// the GC job needs to block until the asyncNotification channel is
-			// closed, which will probably need to be controlled in a schema change
-			// knob.
+		GCJob: &sql.GCJobTestingKnobs{
+			RunBeforeResume: func() {
+				<-asyncNotification
+			},
 		},
 	}
 	s, sqlDBRaw, kvDB := serverutils.StartServer(t, params)
@@ -116,7 +118,6 @@ func TestDropIndexWithZoneConfigCCL(t *testing.T) {
 	}
 	close(asyncNotification)
 
-	t.Skip("skipping last portion of test until schema change GC job is implemented")
 	// Wait for index drop to complete so zone configs are updated.
 	testutils.SucceedsSoon(t, func() error {
 		if kvs, err := kvDB.Scan(context.TODO(), indexSpan.Key, indexSpan.EndKey, 0); err != nil {

pkg/server/server.go

@@ -63,6 +63,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/distsql"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	_ "github.com/cockroachdb/cockroach/pkg/sql/gcjob" // register jobs declared outside of pkg/sql
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire"
 	"github.com/cockroachdb/cockroach/pkg/sql/querycache"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -821,6 +822,11 @@ func NewServer(cfg Config, stopper *stop.Stopper) (*Server, error) {
 	} else {
 		execCfg.SchemaChangerTestingKnobs = new(sql.SchemaChangerTestingKnobs)
 	}
+	if gcJobTestingKnobs := s.cfg.TestingKnobs.GCJob; gcJobTestingKnobs != nil {
+		execCfg.GCJobTestingKnobs = gcJobTestingKnobs.(*sql.GCJobTestingKnobs)
+	} else {
+		execCfg.GCJobTestingKnobs = new(sql.GCJobTestingKnobs)
+	}
 	if distSQLRunTestingKnobs := s.cfg.TestingKnobs.DistSQL; distSQLRunTestingKnobs != nil {
 		execCfg.DistSQLRunTestingKnobs = distSQLRunTestingKnobs.(*execinfra.TestingKnobs)
 	} else {

pkg/sql/as_of_test.go

@@ -33,9 +33,7 @@ func TestAsOfTime(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
 	params, _ := tests.CreateTestServerParams()
-	params.Knobs.SQLSchemaChanger = &sql.SchemaChangerTestingKnobs{
-		// TODO (lucy): Turn on knob to disable GC once the GC job is implemented.
-	}
+	params.Knobs.GCJob = &sql.GCJobTestingKnobs{RunBeforeResume: func() { select {} }}
 	s, db, _ := serverutils.StartServer(t, params)
 	defer s.Stopper().Stop(context.TODO())