Migration improvements

cmd/token_normalization_migrate/main.go

@@ -136,15 +136,15 @@ drop index if exists token_definitions_chain_contract_address_token_idx;
 drop index if exists token_definitions_contract_id_idx;`
 
 const addConstraints = `
-alter table token_definitions set (autovacuum_enabled = true);
-alter table token_definitions set logged;
 alter table token_definitions add primary key(id);
 alter table token_definitions add constraint token_definitions_contract_id_fkey foreign key(contract_id) references contracts(id);
 alter table token_definitions add constraint token_definitions_token_media_id_fkey foreign key(token_media_id) references token_medias(id);
 alter table token_definitions add constraint token_definitions_contract_id_chain_contract_address_fkey foreign key(contract_id, chain, contract_address) references contracts(id, chain, address) on update cascade;
 create unique index if not exists token_definitions_chain_contract_id_token_idx on token_definitions(chain, contract_id, token_id) where not deleted;
 create unique index token_definitions_chain_contract_address_token_idx on token_definitions(chain, contract_address, token_id) where not deleted;
 create index token_definitions_contract_id_idx on token_definitions(contract_id) where not deleted;
+alter table token_definitions set (autovacuum_enabled = true);
+alter table token_definitions set logged;
 analyze token_definitions;`
 
 const insertBatch = `
@@ -268,7 +268,7 @@ func dropTokenDefinitionConstraints() {
 	fmt.Print("dropping constraints")
 	_, err := pq.Exec(dropConstraints)
 	check(err)
-	fmt.Printf("...done")
+	fmt.Println("...done")
 }
 
 func createTokenDefinitions(ctx context.Context, pq *sql.DB) {
@@ -280,9 +280,9 @@ func createTokenDefinitions(ctx context.Context, pq *sql.DB) {
 	requireMustBeEmpty()
 	prepareStatements()
 	analyzeTokens()
-	totalTokens := saveStagingTable()
 	dropTokenDefinitionConstraints()
 	lockTables(tx)
+	totalTokens := saveStagingTable()
 
 	wp := workerpool.New(poolSize)
 
@@ -296,7 +296,6 @@ func createTokenDefinitions(ctx context.Context, pq *sql.DB) {
 		e := end
 		wp.Submit(func() {
 			batchStart := time.Now()
-			fmt.Printf("handling chunk(id=%d) [%d, %d); %d/%d \n", chunkID, s, e, chunkID, totalBatches)
 
 			tokenCh := make(chan bool)
 			mediaCh := make(chan bool)

db/migrations/core/000114_token_norm_add_token_definition_constraints.up.sql

@@ -1,8 +1,13 @@
 /* Migrate to rebuild the tokens table with the token_definition_id column */
 begin;
 
+-- Disable WAL
+alter table token_definitions set unlogged;
+
 -- Temporarily increase work_mem to speed up the migration so prevent data spilling to disk
-set local work_mem = '5500 MB';
+set local work_mem = '4096 MB';
+set maintenance_work_mem = '4096 MB';
+set max_parallel_maintenance_workers = 8;
 
 -- Lock access to table
 lock table tokens in access exclusive mode;
@@ -59,6 +64,7 @@ create index on tokens_with_token_definition_fk(last_updated) where deleted = fa
 create index on tokens_with_token_definition_fk(contract, token_definition_id) where deleted = false;
 
 -- Rename the table, create a backup of the old table
+drop table if exists tokens_backup;
 alter table tokens rename to tokens_backup;
 alter table tokens_with_token_definition_fk rename to tokens;
 
@@ -78,4 +84,7 @@ alter table tokens rename column token_uri to token_uri__deprecated;
 alter table tokens rename column fallback_media to fallback_media__deprecated;
 alter table tokens rename column token_media_id to token_media_id__deprecated;
 
+-- Enable WAL
+alter table token_definitions set logged;
+
 end;