@@ -96,18 +96,20 @@ export async function rebuildPackageDepsIndexes(): Promise<{
9696 const existing : Array < { indexdef : string } > = await qx . select ( NON_CONSTRAINT_INDEXES_SQL )
9797 const existingDefs = existing . map ( ( r : { indexdef : string } ) => r . indexdef . toLowerCase ( ) )
9898
99+ const toRebuild = SECONDARY_INDEXES . filter (
100+ ( idx ) => ! existingDefs . some ( ( def : string ) => def . includes ( `(${ idx . columns } ) ) ,
101+ )
99102 const rebuilt : string [ ] = [ ]
100103
101- for ( const idx of SECONDARY_INDEXES ) {
102- const alreadyExists = existingDefs . some ( ( def : string ) => def . includes ( `(${ idx . columns } ) )
103- if ( alreadyExists ) {
104- log . info ( { columns : idx . columns } , 'Index already exists, skipping' )
105- continue
106- }
107- log . info ( { columns : idx . columns } , 'Creating index on package_dependencies' )
108- await qx . result ( idx . createSql )
109- rebuilt . push ( idx . columns )
110- }
104+ // Build indexes in parallel — each on its own connection so they run concurrently.
105+ await Promise . all (
106+ toRebuild . map ( async ( idx ) => {
107+ const conn = await getPackagesDb ( )
108+ log . info ( { columns : idx . columns } , 'Creating index on package_dependencies' )
109+ await conn . result ( idx . createSql )
110+ rebuilt . push ( idx . columns )
111+ } ) ,
112+ )
111113
112114 // Remove cross-chunk duplicates before rebuilding the UNIQUE constraint.
113115 // DISTINCT ON deduplicates within a single chunk, but the same (root, dep) pair can appear in
@@ -118,27 +120,43 @@ export async function rebuildPackageDepsIndexes(): Promise<{
118120 // Run per partition (depends_on_id % 64 = p) so each iteration prunes to one of the 64
119121 // partitions instead of scanning all 1.15B rows at once. Same total rows read; each pass
120122 // fits in work_mem and avoids cross-partition sort.
123+ // Run in parallel batches of DEDUP_CONCURRENCY to cut wall-clock from ~10h to ~1-2h.
121124 const NUM_PARTITIONS = 64
125+ const DEDUP_CONCURRENCY = 8
122126 let totalDedupDeleted = 0
123- for ( let p = 0 ; p < NUM_PARTITIONS ; p ++ ) {
124- const result = await qx . result ( `
125- DELETE FROM package_dependencies pd
126- USING (
127- SELECT id, depends_on_id
128- FROM (
129- SELECT id, depends_on_id,
130- ROW_NUMBER() OVER (
131- PARTITION BY version_id, depends_on_id, dependency_kind
132- ORDER BY id
133- ) AS rn
134- FROM package_dependencies
135- WHERE depends_on_id % ${ NUM_PARTITIONS } ${ p }
136- ) sub
137- WHERE rn > 1
138- ) dupes
139- WHERE pd.id = dupes.id AND pd.depends_on_id = dupes.depends_on_id
140- ` )
141- totalDedupDeleted += result
127+ for ( let batch = 0 ; batch < NUM_PARTITIONS ; batch += DEDUP_CONCURRENCY ) {
128+ const partitions = Array . from (
129+ { length : Math . min ( DEDUP_CONCURRENCY , NUM_PARTITIONS - batch ) } ,
130+ ( _ , i ) => batch + i ,
131+ )
132+ const counts = await Promise . all (
133+ partitions . map ( async ( p ) => {
134+ const conn = await getPackagesDb ( )
135+ // work_mem (not maintenance_work_mem) controls sort memory for window functions.
136+ // 2GB avoids disk spill during the ROW_NUMBER() sort on each partition.
137+ return conn . tx ( async ( tx ) => {
138+ await tx . result ( `SET LOCAL work_mem = '2GB'` )
139+ return tx . result ( `
140+ DELETE FROM package_dependencies pd
141+ USING (
142+ SELECT id, depends_on_id
143+ FROM (
144+ SELECT id, depends_on_id,
145+ ROW_NUMBER() OVER (
146+ PARTITION BY version_id, depends_on_id, dependency_kind
147+ ORDER BY id
148+ ) AS rn
149+ FROM package_dependencies
150+ WHERE depends_on_id % ${ NUM_PARTITIONS } ${ p }
151+ ) sub
152+ WHERE rn > 1
153+ ) dupes
154+ WHERE pd.id = dupes.id AND pd.depends_on_id = dupes.depends_on_id
155+ ` )
156+ } )
157+ } ) ,
158+ )
159+ totalDedupDeleted += counts . reduce ( ( sum , n ) => sum + n , 0 )
142160 }
143161 log . info (
144162 { rowsDeleted : totalDedupDeleted } ,
0 commit comments