chore: vaccum status tables if they cross threshold #3434

BonapartePC · 2023-06-02T07:27:43Z

Description

Vacuum status table if they cross threshold

Notion Ticket

https://www.notion.so/rudderstacks/Vacuum-and-Vacuum-full-during-jobsdb-compaction-6f78b37caea64aef9161c1732fdd76a2

Security

The code changed/added as part of this pull request won't create any security issues with how the software is being used.

atzoum · 2023-06-02T09:30:11Z

jobsdb/jobsdb.go

@@ -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")


Value should be in bytes so we don't need to include the InMB prefix. You may use 500*bytesize.MB for the default value.

atzoum · 2023-06-02T09:31:37Z

jobsdb/migration.go

 		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


I don't think this is necessary, as a delete statement returns the number of rows affected already

atzoum · 2023-06-02T09:33:23Z

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) {


We would need to have some tests verifying this method's behaviour

atzoum · 2023-06-02T09:39:08Z

jobsdb/migration.go

@@ -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)


A table that we know is going to be vacuumed full doesn't need to be vacuumed if it is in the cleanup list

codecov · 2023-06-05T11:20:27Z

Codecov Report

Patch coverage: 67.94% and project coverage change: -0.06 ⚠️

Comparison is base (3d03653) 68.53% compared to head (498da6c) 68.48%.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #3434      +/-   ##
==========================================
- Coverage   68.53%   68.48%   -0.06%     
==========================================
  Files         330      330              
  Lines       52905    52973      +68     
==========================================
+ Hits        36259    36277      +18     
- Misses      14296    14335      +39     
- Partials     2350     2361      +11

Impacted Files	Coverage Δ
jobsdb/migration.go	`74.66% <63.23%> (-3.08%)`	⬇️
jobsdb/backup.go	`75.00% <100.00%> (ø)`
jobsdb/jobsdb.go	`73.97% <100.00%> (+0.02%)`	⬆️
warehouse/schema.go	`100.00% <100.00%> (ø)`

... and 12 files with indirect coverage changes

☔ View full report in Codecov by Sentry.
📢 Do you have feedback about the report comment? Let us know in this issue.

Sidddddarth · 2023-06-02T09:30:51Z

jobsdb/migration.go

+		return nil, err
+	}
+
+	datasets := lo.Filter(dsList,


can do this comparison above too, eliminating the need for tableSizes map.

True, but the query gives me all the tables and their sizes. I'm doing the filtering since we are only interested in a few status tables that are going to be cleaned

I meant something like this:

. . . defer func() { _ = rows.Close() }() toVacuum := map[string]struct{}{} for rows.Next() { var ( tableSize int64 tableName string ) err = rows.Scan(&tableSize, &tableName) if err != nil { return nil, err } if tableSize > vacuumFullStatusTableThreshold { toVacuum[tableName] = struct{}{} } } err = rows.Err() if err != nil { return nil, err } return toVacuum, nil }

Why do we need to pass dsList to this function..?

I just read your comment again and understand now that you were wanting to vacuum only tables that were marked for compaction.
Might as well vacuum all..?

or if you intend to vacuum only a subset of dsList, you may use it as part of the query.
Maybe the threshold as well.

I think we can vacuum all. WDYT @atzoum ?

Sidddddarth · 2023-06-02T09:34:52Z

jobsdb/migration.go

@@ -223,29 +274,53 @@ func (jd *HandleT) cleanupStatusTables(ctx context.Context, dsList []dataSetT) e
 				return err
 			}
 		}
+		for _, statusTable := range toVaccum {


if a status table belongs to both toCompact and toVaccum, it'll undergo vacuum twice(analyze and full).
Is such a situation possible..?

That shouldn't be possible because of the condition here:

rudder-server/jobsdb/migration.go

Line 319 in 16b5c34

if numJobStatusDeleted > vacuumAnalyzeStatusTableThreshold && canBeVacuumed {

atzoum · 2023-06-06T05:41:33Z

jobsdb/migration.go

+				WHERE schemaname NOT IN ('pg_catalog','information_schema')
+				AND tablename like $1
+		)`,
+		jd.tablePrefix+"_job%",


we are only interested on job status tables, correct?

Yes, made the change in query

atzoum · 2023-06-06T05:46:36Z

jobsdb/migration.go

+	datasets := lo.Filter(dsList,
+		func(ds dataSetT, idx int) bool {
+			tableSize := tableSizes[ds.JobStatusTable]
+			return tableSize > vacuumFullStatusTableThreshold
+		})
+	toVacuum := map[string]struct{}{}
+	for _, ds := range datasets {
+		toVacuum[ds.JobStatusTable] = struct{}{}
+	}


The code below is less and more efficient :)

Suggested change

datasets := lo.Filter(dsList,

func(ds dataSetT, idx int) bool {

tableSize := tableSizes[ds.JobStatusTable]

return tableSize > vacuumFullStatusTableThreshold

})

toVacuum := map[string]struct{}{}

for _, ds := range datasets {

toVacuum[ds.JobStatusTable] = struct{}{}

}

toVacuum := map[string]struct{}{}

for _, ds := range dsList {

if tableSizes[ds.JobStatusTable] > vacuumFullStatusTableThreshold {

toVacuum[ds.JobStatusTable] = struct{}{}

}

}

atzoum · 2023-06-06T05:50:56Z

jobsdb/migration.go

@@ -206,6 +256,11 @@ func (jd *HandleT) cleanupStatusTables(ctx context.Context, dsList []dataSetT) e
 	if err != nil {
 		return err
 	}
+	// vacuum status table if total size exceeds vacuumFullStatusTableThreshold in the toCompact list
+	toVacuum, err := jd.getVacuumCandidates(ctx, toCompact)


I don't believe we should only expect to be doing full vacuuming for tables that need to be compacted. A table might reach the full vacuuming threshold and still not need a compaction yet.

Correct. Should we just vacuum all the tables or just tables in dsList? Tagged you in the above message as well

Define all tables. Which additional tables are you proposing to vacuum full and why?

I mean the tables which the query returns. Do we need to pass dsList? We can just return all the tables that the query returns which cross the threshold right?

Not sure what else is going to be included now or in the future if we remove this filter. Limiting the tables to those contained in the dsList can easily guarantee that we are not performing an undesirable full vaccuming (costly operation) to another table.

makes sense
cc @Sidddddarth

so a subset of the compacted tables will be vacuumed?

No, a subset of dsList will be vacuumed. compacted tables and tables that we're going to vacuum are independent at this point

atzoum · 2023-06-06T06:14:38Z

jobsdb/migration.go

+	var rows *sql.Rows
+	rows, err := jd.dbHandle.QueryContext(
+		ctx,
+		`SELECT pg_relation_size(oid) AS size, relname


Suggested change

`SELECT pg_relation_size(oid) AS size, relname

`SELECT pg_table_size(oid) AS size, relname

…er-server into chore.vaccumStatusTables

atzoum · 2023-06-06T07:08:28Z

jobsdb/migration.go

@@ -153,6 +153,53 @@ func (jd *HandleT) doMigrateDS(ctx context.Context) error {
 	return err
 }

+// based on size of given DSs, gives a list of DSs for us to vacuum status tables
+func (jd *HandleT) getVacuumCandidates(ctx context.Context, dsList []dataSetT) (map[string]struct{}, error) {


Suggested change

func (jd *HandleT) getVacuumCandidates(ctx context.Context, dsList []dataSetT) (map[string]struct{}, error) {

func (jd *HandleT) getVacuumFullCandidates(ctx context.Context, dsList []dataSetT) (map[string]struct{}, error) {

atzoum · 2023-06-06T07:10:17Z

jobsdb/migration.go

+func (*HandleT) vacuumFullStatusTable(ctx context.Context, tx *Tx, table string) error {
+	if _, err := tx.ExecContext(ctx, fmt.Sprintf(`VACUUM FULL %[1]q`, table)); err != nil {
+		return err
+	}
+	return nil
+}
+


I don't believe we need this as a separate method

…er-server into chore.vaccumStatusTables

atzoum · 2023-06-06T07:16:13Z

jobsdb/migration.go

+			// vacuum full if present in toVacuum
+			if ok {
+				if _, err := tx.ExecContext(ctx, fmt.Sprintf(`VACUUM FULL %[1]q`, table)); err != nil {
+					return err
+				}
+			}


Full vacuuming shall be performed on its own loop, following but not coupled with the compaction loop.

Yeah, my bad 🤦
Made the changes

atzoum · 2023-06-06T07:17:35Z

jobsdb/migration_test.go

@@ -196,4 +196,10 @@ func TestMigration(t *testing.T) {
 	).Scan(&count)
 	require.NoError(t, err)
 	require.EqualValues(t, 10, count)
+
+	vacuumFullStatusTableThreshold = 8 * 1024 // 8KB is common value for toast size in postgres


Suggested change

vacuumFullStatusTableThreshold = 8 * 1024 // 8KB is common value for toast size in postgres

vacuumFullStatusTableThreshold = 8 * bytesize.KB // 8KB is common value for toast size in postgres

atzoum · 2023-06-06T07:50:48Z

jobsdb/migration.go

+		for table := range toVacuum {
+			if _, err := tx.ExecContext(ctx, fmt.Sprintf(`VACUUM FULL %[1]q`, table)); err != nil {
+				return err
+			}
+		}


If more than 1 status tables need full vacuuming, it would be preferable for it to happen in ascending dataset order instead of randomly (map iteration order is random).

That being said, getVacuumFullCandidates could return a slice instead of a map & cleanupStatusTables could prepare a map from it for its own purposes

atzoum · 2023-06-06T08:21:28Z

jobsdb/migration.go

+	toVacuumFullMap := map[string]struct{}{}
+	for _, table := range toVacuumFull {
+		toVacuumFullMap[table] = struct{}{}
+	}


there is 1 line we can get rid of!

Suggested change

toVacuumFullMap := map[string]struct{}{}

for _, table := range toVacuumFull {

toVacuumFullMap[table] = struct{}{}

}

toVacuumFullMap := lo.Associate(toVacuumFull, func(k string) (string, struct{}) {

return k, struct{}{}

})

chore: vaccum status tables if they cross threshold

47e3925

BonapartePC force-pushed the chore.vaccumStatusTables branch from 9cc9b70 to 47e3925 Compare June 2, 2023 07:28

github-actions bot added the server-team label Jun 2, 2023

add limit

860a081

atzoum reviewed Jun 2, 2023

View reviewed changes

atzoum requested a review from Sidddddarth June 2, 2023 09:39

github-actions bot added the with tests label Jun 5, 2023

BonapartePC requested a review from atzoum June 5, 2023 10:55

address comments

28d0291

BonapartePC force-pushed the chore.vaccumStatusTables branch from 4582bf2 to 28d0291 Compare June 5, 2023 11:08

Merge branch 'master' into chore.vaccumStatusTables

2d34a67

Merge branch 'master' into chore.vaccumStatusTables

16b5c34

Sidddddarth reviewed Jun 6, 2023

View reviewed changes

BonapartePC requested a review from Sidddddarth June 6, 2023 06:12