chore: vaccum status tables if they cross threshold

jobsdb/jobsdb.go

@@ -604,6 +604,8 @@ func init() {
 var (
 	maxDSSize, maxMigrateOnce, maxMigrateDSProbe int
 	maxTableSize                                 int64
+	vaccumFullStatusTableThreshold               int64
+	vaccumAnalyzeStatusTableThreshold            int64
 	jobDoneMigrateThres, jobStatusMigrateThres   float64
 	jobMinRowsMigrateThres                       float64
 	migrateDSLoopSleepDuration                   time.Duration
@@ -642,6 +644,8 @@ func loadConfig() {
 	config.RegisterIntConfigVariable(10, &maxMigrateDSProbe, true, 1, "JobsDB.maxMigrateDSProbe")
 	config.RegisterInt64ConfigVariable(300, &maxTableSize, true, 1000000, "JobsDB.maxTableSizeInMB")
 	config.RegisterInt64ConfigVariable(10000, &backupRowsBatchSize, true, 1, "JobsDB.backupRowsBatchSize")
+	config.RegisterInt64ConfigVariable(500, &vaccumFullStatusTableThreshold, true, 1000000, "JobsDB.vaccumFullStatusTableThresholdInMB")
+	config.RegisterInt64ConfigVariable(30000, &vaccumAnalyzeStatusTableThreshold, true, 1, "JobsDB.vaccumAnalyzeStatusTableThreshold")
 	config.RegisterInt64ConfigVariable(64*bytesize.MB, &backupMaxTotalPayloadSize, true, 1, "JobsDB.maxBackupTotalPayloadSize")
 	config.RegisterDurationConfigVariable(30, &migrateDSLoopSleepDuration, true, time.Second, []string{"JobsDB.migrateDSLoopSleepDuration", "JobsDB.migrateDSLoopSleepDurationInS"}...)
 	config.RegisterDurationConfigVariable(5, &addNewDSLoopSleepDuration, true, time.Second, []string{"JobsDB.addNewDSLoopSleepDuration", "JobsDB.addNewDSLoopSleepDurationInS"}...)

jobsdb/migration.go

@@ -153,6 +153,52 @@ func (jd *HandleT) doMigrateDS(ctx context.Context) error {
 	return err
 }
 
+// based on size of given DSs, gives a list of DSs for us to vaccum status tables
+func (jd *HandleT) getVacuumCandidates(ctx context.Context, dsList []dataSetT) ([]dataSetT, error) {
+	// get name and it's size of all tables
+	var rows *sql.Rows
+	rows, err := jd.dbHandle.QueryContext(
+		ctx,
+		`SELECT pg_total_relation_size(oid) AS size, relname
+		FROM pg_class
+		where relname = ANY(
+			SELECT tablename
+				FROM pg_catalog.pg_tables
+				WHERE schemaname NOT IN ('pg_catalog','information_schema')
+				AND tablename like $1
+		)`,
+		jd.tablePrefix+"_job%",
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+
+	tableSizes := map[string]int64{}
+	for rows.Next() {
+		var (
+			tableSize int64
+			tableName string
+		)
+		err = rows.Scan(&tableSize, &tableName)
+		if err != nil {
+			return nil, err
+		}
+		tableSizes[tableName] = tableSize
+	}
+	err = rows.Err()
+	if err != nil {
+		return nil, err
+	}
+
+	datasets := lo.Filter(dsList,
+		func(ds dataSetT, idx int) bool {
+			tableSize := tableSizes[ds.JobStatusTable]
+			return tableSize > vaccumFullStatusTableThreshold
+		})
+	return datasets, nil
+}
+
 // based on an estimate of the rows in DSs, gives a list of DSs for us to cleanup status tables
 func (jd *HandleT) getCleanUpCandidates(ctx context.Context, dsList []dataSetT) ([]dataSetT, error) {
 	// get analyzer estimates for the number of rows(jobs, statuses) in each DS
@@ -206,6 +252,11 @@ func (jd *HandleT) cleanupStatusTables(ctx context.Context, dsList []dataSetT) e
 	if err != nil {
 		return err
 	}
+	// vaccum status table if total size exceeds vaccumFullStatusTableThreshold in the toCompact list
+	toVaccum, err := jd.getVacuumCandidates(ctx, toCompact)
+	if err != nil {
+		return err
+	}
 	start := time.Now()
 	defer stats.Default.NewTaggedStat(
 		"jobsdb_compact_status_tables",
@@ -223,29 +274,53 @@ func (jd *HandleT) cleanupStatusTables(ctx context.Context, dsList []dataSetT) e
 				return err
 			}
 		}
+		for _, statusTable := range toVaccum {
+			if err := jd.vaccumStatusTable(
+				ctx,
+				tx,
+				statusTable.JobStatusTable,
+			); err != nil {
+				return err
+			}
+		}
 		return nil
 	})
 }
 
 // cleanStatusTable deletes all rows except for the latest status for each job
 func (*HandleT) cleanStatusTable(ctx context.Context, tx *Tx, table string) error {
-	_, err := tx.ExecContext(
+	deletedJobsQuery := fmt.Sprintf(`WITH deleted AS (DELETE FROM %[1]q
+		WHERE NOT id = ANY(
+		   SELECT DISTINCT ON (job_id) id from "%[1]s" ORDER BY job_id ASC, id DESC
+	   ) RETURNING *) SELECT count(*) FROM deleted;`, table)
+
+	var numJobStatusDeleted int64
+	var err error
+	if err = tx.QueryRowContext(
 		ctx,
-		fmt.Sprintf(`DELETE FROM %[1]q
-			 			WHERE NOT id = ANY(
-							SELECT DISTINCT ON (job_id) id from "%[1]s" ORDER BY job_id ASC, id DESC
-						)`, table),
-	)
-	if err != nil {
+		deletedJobsQuery,
+	).Scan(&numJobStatusDeleted); err != nil {
 		return err
 	}
+
+	query := fmt.Sprintf(`ANALYZE %q`, table)
+	if numJobStatusDeleted > vaccumAnalyzeStatusTableThreshold {
+		query = fmt.Sprintf(`VACCUM ANALYZE %q`, table)
+	}
 	_, err = tx.ExecContext(
 		ctx,
-		fmt.Sprintf(`ANALYZE %q`, table),
+		query,
 	)
 	return err
 }
 
+func (*HandleT) vaccumStatusTable(ctx context.Context, tx *Tx, table string) error {
+	if _, err := tx.ExecContext(ctx, fmt.Sprintf(`VACCUM FULL %[1]q`, table)); err != nil {
+		return err
+	}
+	return nil
+}
+
 // getMigrationList returns the list of datasets to migrate from,
 // the number of unfinished jobs contained in these datasets
 // and the dataset before which the new (migrated) dataset that will hold these jobs needs to be created