Skip to content

Commit 94a9c06

Browse files
bcomnesbcomnes
committed
fix: add ordered job fetch indexes
1 parent a5a3e71 commit 94a9c06

7 files changed

Lines changed: 261 additions & 26 deletions

File tree

‎examples/index-perf/index.ts‎

Lines changed: 16 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -49,9 +49,9 @@ async function main () {
4949
const fetchIndex = await findFetchIndex(db)
5050
console.log(`\nfetch index on ${SCHEMA}.${TABLE}: ${fetchIndex.name}\n ${fetchIndex.def}\n`)
5151

52-
// Before mangling indexes for the variant comparison, verify the migration-built job_i5 is
53-
// actually reached by every dynamic shape the production fetchNextJob() can emit.
54-
await verifyShapes(db, fetchIndex.name)
52+
// Before mangling indexes for the variant comparison, verify a migration-built fetch index is
53+
// reached by every dynamic shape the production fetchNextJob() can emit.
54+
await verifyShapes(db, [fetchIndex.name, `${TABLE}_i10`, `${TABLE}_i11`])
5555

5656
// Record the real, full fetchNextJob() plan once (UPDATE ... RETURNING, rolled back).
5757
console.log('='.repeat(90))
@@ -162,8 +162,8 @@ async function useOnlyIndex (db: Db, ddls: Array<(name: string) => string>) {
162162
[SCHEMA, TABLE]
163163
)
164164
for (const r of res.rows) {
165-
// Drop anything that could serve the fetch (start_after- or priority-keyed), but never the PK.
166-
if (/start_after|priority/.test(r.indexdef) && !/_pkey/.test(r.indexname)) {
165+
// Drop anything that could serve the fetch (start_after- or ordered-keyed), but never the PK.
166+
if (/start_after|priority|created_on/.test(r.indexdef) && !/_pkey/.test(r.indexname)) {
167167
await db.executeSql(`DROP INDEX ${SCHEMA}.${r.indexname}`)
168168
}
169169
}
@@ -198,13 +198,13 @@ function nextCteBlock (plan: string): string {
198198
}
199199

200200
// Drive the REAL fetchNextJob() across the dynamic option combinations and confirm each still
201-
// reaches job_i5 via the index (not a Seq Scan). This guards against a dynamic WHERE/ORDER BY shape
202-
// silently regressing the hot path to a table scan. Runs against the migration-built index.
203-
async function verifyShapes (db: Db, fetchIndexName: string) {
201+
// reaches a migration-built fetch index (not a Seq Scan). This guards against a dynamic
202+
// WHERE/ORDER BY shape silently regressing the hot path to a table scan.
203+
async function verifyShapes (db: Db, fetchIndexNames: string[]) {
204204
console.log('\n' + '='.repeat(90))
205-
console.log(`SHAPE MATRIX — does the real fetchNextJob() still reach ${fetchIndexName} for every dynamic condition?`)
205+
console.log(`SHAPE MATRIX — does the real fetchNextJob() still reach ${fetchIndexNames.join('/')} for every dynamic condition?`)
206206
console.log('='.repeat(90))
207-
console.log(['shape', 'next-CTE access', `uses ${fetchIndexName}?`, 'sort?'].join('\t'))
207+
console.log(['shape', 'next-CTE access', 'uses fetch index?', 'sort?'].join('\t'))
208208
const base = {
209209
schema: SCHEMA,
210210
table: TABLE,
@@ -231,17 +231,17 @@ async function verifyShapes (db: Db, fetchIndexName: string) {
231231
const plan = await explain(db, s.label, q.text, q.values as unknown[], true)
232232
const block = nextCteBlock(plan)
233233
const seqOnFetch = new RegExp(`Seq Scan on ${SCHEMA}\\.${TABLE}\\b`).test(block)
234-
const usesIdx = block.includes(fetchIndexName)
234+
const usedIndex = fetchIndexNames.find(name => block.includes(name))
235235
const usesPkey = /_pkey/.test(block)
236-
// The only real regression is a Seq Scan. Using job_i5 (Bitmap/Index) is the target; falling to
237-
// the PK (name, id) is an acceptable index plan the planner picks when ORDER BY is id-only.
236+
// The only real regression is a Seq Scan. Using a fetch index (Bitmap/Index) is the target;
237+
// falling to the PK (name, id) is acceptable when ORDER BY is id-only.
238238
const access = seqOnFetch
239239
? 'SEQ SCAN ⚠️'
240-
: usesIdx
241-
? (/Bitmap/.test(block) ? 'Bitmap(job_i5)' : 'Index Scan(job_i5)')
240+
: usedIndex
241+
? (/Bitmap/.test(block) ? `Bitmap(${usedIndex})` : `Index Scan(${usedIndex})`)
242242
: usesPkey ? 'Index Scan(pkey)' : 'other ⚠️'
243243
const sort = /\bSort\b/.test(block)
244-
console.log([s.label, access, usesIdx ? 'yes' : (usesPkey ? 'pkey' : 'NO ⚠️'), sort ? 'SORT' : 'no-sort'].join('\t'))
244+
console.log([s.label, access, usedIndex || (usesPkey ? 'pkey' : 'NO ⚠️'), sort ? 'SORT' : 'no-sort'].join('\t'))
245245
}
246246
}
247247

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@
6565
"docs:readme": "node ./scripts/sync-readme.js"
6666
},
6767
"pgboss": {
68-
"schema": 37
68+
"schema": 38
6969
},
7070
"repository": {
7171
"type": "git",

‎src/migrationStore.ts‎

Lines changed: 136 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ interface MigrateOptions {
1818
function formatJobTable (command: string, table: string) {
1919
// Anchor both rewrites so a schema name that itself contains these substrings (e.g. `job_intake`)
2020
// isn't mangled: `.job\b` only matches the base table reference (`schema.job`, not `schema.job_i5`
21-
// whose `job` is followed by `_`), and `job_iN` only matches the bare index-name tokens (job_i1..9),
21+
// whose `job` is followed by `_`), and `job_iN` only matches bare index-name tokens (job_i1, job_i10),
2222
// never the `job_i` inside an arbitrary schema name.
2323
return command
2424
.replace(/\.job\b/g, `.${table}`)
@@ -731,6 +731,95 @@ const createQueueFn: Record<number, (schema: string) => string> = {
731731
END;
732732
$$
733733
LANGUAGE plpgsql;
734+
`,
735+
736+
38: (schema) => `
737+
CREATE OR REPLACE FUNCTION ${schema}.create_queue(queue_name text, options jsonb)
738+
RETURNS VOID AS
739+
$$
740+
DECLARE
741+
tablename varchar := CASE WHEN options->>'partition' = 'true'
742+
THEN 'j' || encode(sha224(queue_name::bytea), 'hex')
743+
ELSE 'job_common'
744+
END;
745+
queue_created_on timestamptz;
746+
BEGIN
747+
748+
WITH q as (
749+
INSERT INTO ${schema}.queue (
750+
name,
751+
policy,
752+
retry_limit,
753+
retry_delay,
754+
retry_backoff,
755+
retry_delay_max,
756+
expire_seconds,
757+
retention_seconds,
758+
deletion_seconds,
759+
warning_queued,
760+
dead_letter,
761+
partition,
762+
table_name,
763+
heartbeat_seconds,
764+
notify
765+
)
766+
VALUES (
767+
queue_name,
768+
options->>'policy',
769+
COALESCE((options->>'retryLimit')::int, 2),
770+
COALESCE((options->>'retryDelay')::int, 0),
771+
COALESCE((options->>'retryBackoff')::bool, false),
772+
(options->>'retryDelayMax')::int,
773+
COALESCE((options->>'expireInSeconds')::int, 900),
774+
COALESCE((options->>'retentionSeconds')::int, 1209600),
775+
COALESCE((options->>'deleteAfterSeconds')::int, 604800),
776+
COALESCE((options->>'warningQueueSize')::int, 0),
777+
options->>'deadLetter',
778+
COALESCE((options->>'partition')::bool, false),
779+
tablename,
780+
(options->>'heartbeatSeconds')::int,
781+
COALESCE((options->>'notify')::bool, false)
782+
)
783+
ON CONFLICT DO NOTHING
784+
RETURNING created_on
785+
)
786+
SELECT created_on into queue_created_on from q;
787+
788+
IF queue_created_on IS NULL OR options->>'partition' IS DISTINCT FROM 'true' THEN
789+
RETURN;
790+
END IF;
791+
792+
EXECUTE format('CREATE TABLE ${schema}.%I (LIKE ${schema}.job INCLUDING DEFAULTS)', tablename);
793+
794+
EXECUTE ${schema}.job_table_format($cmd$ALTER TABLE ${schema}.job ADD PRIMARY KEY (name, id)$cmd$, tablename);
795+
EXECUTE ${schema}.job_table_format($cmd$ALTER TABLE ${schema}.job ADD CONSTRAINT q_fkey FOREIGN KEY (name) REFERENCES ${schema}.queue (name) ON DELETE RESTRICT DEFERRABLE INITIALLY DEFERRED$cmd$, tablename);
796+
EXECUTE ${schema}.job_table_format($cmd$ALTER TABLE ${schema}.job ADD CONSTRAINT dlq_fkey FOREIGN KEY (dead_letter) REFERENCES ${schema}.queue (name) ON DELETE RESTRICT DEFERRABLE INITIALLY DEFERRED$cmd$, tablename);
797+
798+
EXECUTE ${schema}.job_table_format($cmd$CREATE INDEX job_i5 ON ${schema}.job (name, start_after) WHERE state < 'active' AND NOT blocked$cmd$, tablename);
799+
EXECUTE ${schema}.job_table_format($cmd$CREATE INDEX job_i10 ON ${schema}.job (name, priority DESC, created_on, id) WHERE state < 'active' AND NOT blocked$cmd$, tablename);
800+
EXECUTE ${schema}.job_table_format($cmd$CREATE INDEX job_i11 ON ${schema}.job (name, created_on, id) WHERE state < 'active' AND NOT blocked$cmd$, tablename);
801+
EXECUTE ${schema}.job_table_format($cmd$CREATE UNIQUE INDEX job_i4 ON ${schema}.job (name, singleton_on, COALESCE(singleton_key, '')) WHERE state <> 'cancelled' AND singleton_on IS NOT NULL$cmd$, tablename);
802+
EXECUTE ${schema}.job_table_format($cmd$CREATE INDEX job_i7 ON ${schema}.job (name, group_id) WHERE state = 'active' AND group_id IS NOT NULL$cmd$, tablename);
803+
EXECUTE ${schema}.job_table_format($cmd$CREATE INDEX job_i9 ON ${schema}.job (name, id) WHERE blocking AND state = 'completed'$cmd$, tablename);
804+
805+
IF options->>'policy' = 'short' THEN
806+
EXECUTE ${schema}.job_table_format($cmd$CREATE UNIQUE INDEX job_i1 ON ${schema}.job (name, COALESCE(singleton_key, '')) WHERE state = 'created' AND policy = 'short'$cmd$, tablename);
807+
ELSIF options->>'policy' = 'singleton' THEN
808+
EXECUTE ${schema}.job_table_format($cmd$CREATE UNIQUE INDEX job_i2 ON ${schema}.job (name, COALESCE(singleton_key, '')) WHERE state = 'active' AND policy = 'singleton'$cmd$, tablename);
809+
ELSIF options->>'policy' = 'stately' THEN
810+
EXECUTE ${schema}.job_table_format($cmd$CREATE UNIQUE INDEX job_i3 ON ${schema}.job (name, state, COALESCE(singleton_key, '')) WHERE state <= 'active' AND policy = 'stately'$cmd$, tablename);
811+
ELSIF options->>'policy' = 'exclusive' THEN
812+
EXECUTE ${schema}.job_table_format($cmd$CREATE UNIQUE INDEX job_i6 ON ${schema}.job (name, COALESCE(singleton_key, '')) WHERE state <= 'active' AND policy = 'exclusive'$cmd$, tablename);
813+
ELSIF options->>'policy' = 'key_strict_fifo' THEN
814+
EXECUTE ${schema}.job_table_format($cmd$CREATE UNIQUE INDEX job_i8 ON ${schema}.job (name, singleton_key) WHERE state IN ('active', 'retry', 'failed') AND policy = 'key_strict_fifo'$cmd$, tablename);
815+
EXECUTE ${schema}.job_table_format($cmd$ALTER TABLE ${schema}.job ADD CONSTRAINT job_key_strict_fifo_singleton_key_check CHECK (NOT (policy = 'key_strict_fifo' AND singleton_key IS NULL))$cmd$, tablename);
816+
END IF;
817+
818+
EXECUTE format('ALTER TABLE ${schema}.%I ADD CONSTRAINT cjc CHECK (name=%L)', tablename, queue_name);
819+
EXECUTE format('ALTER TABLE ${schema}.job ATTACH PARTITION ${schema}.%I FOR VALUES IN (%L)', tablename, queue_name);
820+
END;
821+
$$
822+
LANGUAGE plpgsql;
734823
`
735824
}
736825

@@ -1269,6 +1358,52 @@ function getAll (schema: string, noPartitioning = false, noCovering = false): ty
12691358
uninstall: noPartitioning
12701359
? []
12711360
: [jobTableFormatFn[36](schema)]
1361+
},
1362+
{
1363+
release: '12.27.0',
1364+
version: 38,
1365+
previous: 37,
1366+
// Add indexes that match the two ordered fetch shapes. The partial predicate keeps completed,
1367+
// active, and blocked jobs out of both indexes; unlike a covering index, their key columns are
1368+
// useful before the row-locking heap visit because they satisfy ORDER BY and let LIMIT stop the
1369+
// scan early. job_i5 remains in place for queues with a large future-scheduled working set.
1370+
install: noPartitioning
1371+
? [
1372+
`CREATE INDEX job_i10 ON ${schema}.job (name, priority DESC, created_on, id) WHERE state < 'active' AND NOT blocked`,
1373+
`CREATE INDEX job_i11 ON ${schema}.job (name, created_on, id) WHERE state < 'active' AND NOT blocked`
1374+
]
1375+
: [createQueueFn[38](schema)],
1376+
// Partitioned PostgreSQL deployments fan the builds out through BAM so each potentially large
1377+
// job table is indexed CONCURRENTLY, outside the migration transaction. Backends configured
1378+
// without table partitioning use their native online CREATE INDEX behavior in install above.
1379+
async: noPartitioning
1380+
? []
1381+
: [
1382+
`SELECT ${schema}.job_table_run_async(
1383+
'fetch_priority_index',
1384+
$VERSION$,
1385+
$$
1386+
CREATE INDEX CONCURRENTLY IF NOT EXISTS job_i10 ON ${schema}.job (name, priority DESC, created_on, id) WHERE state < 'active' AND NOT blocked
1387+
$$
1388+
)`,
1389+
`SELECT ${schema}.job_table_run_async(
1390+
'fetch_created_index',
1391+
$VERSION$,
1392+
$$
1393+
CREATE INDEX CONCURRENTLY IF NOT EXISTS job_i11 ON ${schema}.job (name, created_on, id) WHERE state < 'active' AND NOT blocked
1394+
$$
1395+
)`
1396+
],
1397+
uninstall: noPartitioning
1398+
? [
1399+
`DROP INDEX IF EXISTS ${schema}.job_i11`,
1400+
`DROP INDEX IF EXISTS ${schema}.job_i10`
1401+
]
1402+
: [
1403+
createQueueFn[33](schema),
1404+
`SELECT ${schema}.job_table_run($cmd$DROP INDEX IF EXISTS ${schema}.job_i11$cmd$)`,
1405+
`SELECT ${schema}.job_table_run($cmd$DROP INDEX IF EXISTS ${schema}.job_i10$cmd$)`
1406+
]
12721407
}
12731408
]
12741409
}

‎src/plans.ts‎

Lines changed: 20 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -277,7 +277,7 @@ export function createIndexJobDependencyParent (schema: string) {
277277
// Anchored so a schema name that itself contains these substrings (e.g. `job_intake`) isn't
278278
// mangled: `\.job\y` matches only the base table reference (`schema.job`, not `schema.job_i5` whose
279279
// `job` is followed by `_`, nor `.job_dependency`), and `\yjob_i(\d+)` matches only the bare
280-
// index-name tokens (job_i1..9), never the `job_i` inside a schema name. Mirrors formatJobTable()
280+
// index-name tokens (job_i1, job_i10), never the `job_i` inside a schema name. Mirrors formatJobTable()
281281
// in migrationStore.ts; the migration that fixed this (v37) carries its own frozen copy.
282282
export function jobTableFormatFunction (schema: string) {
283283
return `
@@ -458,6 +458,8 @@ function createTableJobCommon (schema: string) {
458458
SELECT ${schema}.job_table_run($cmd$${createCheckConstraintKeyStrictFifo(schema)}$cmd$, '${COMMON_JOB_TABLE}');
459459
SELECT ${schema}.job_table_run($cmd$${createIndexJobThrottle(schema)}$cmd$, '${COMMON_JOB_TABLE}');
460460
SELECT ${schema}.job_table_run($cmd$${createIndexJobFetch(schema)}$cmd$, '${COMMON_JOB_TABLE}');
461+
SELECT ${schema}.job_table_run($cmd$${createIndexJobFetchPriority(schema)}$cmd$, '${COMMON_JOB_TABLE}');
462+
SELECT ${schema}.job_table_run($cmd$${createIndexJobFetchCreated(schema)}$cmd$, '${COMMON_JOB_TABLE}');
461463
SELECT ${schema}.job_table_run($cmd$${createIndexJobGroupConcurrency(schema)}$cmd$, '${COMMON_JOB_TABLE}');
462464
SELECT ${schema}.job_table_run($cmd$${createIndexJobBlocking(schema)}$cmd$, '${COMMON_JOB_TABLE}');
463465
@@ -478,6 +480,8 @@ function createTableJobIndexes (schema: string, noDeferrableConstraints = false,
478480
${createCheckConstraintKeyStrictFifo(schema)};
479481
${createIndexJobThrottle(schema)};
480482
${createIndexJobFetch(schema, noCoveringIndex)};
483+
${createIndexJobFetchPriority(schema)};
484+
${createIndexJobFetchCreated(schema)};
481485
${createIndexJobGroupConcurrency(schema)};
482486
${createIndexJobBlocking(schema)};
483487
`
@@ -593,6 +597,8 @@ function createQueueFunction (schema: string, noPartitioning = false) {
593597
EXECUTE ${schema}.job_table_format($cmd$${createQueueForeignKeyJobDeadLetter(schema)}$cmd$, tablename);
594598
595599
EXECUTE ${schema}.job_table_format($cmd$${createIndexJobFetch(schema)}$cmd$, tablename);
600+
EXECUTE ${schema}.job_table_format($cmd$${createIndexJobFetchPriority(schema)}$cmd$, tablename);
601+
EXECUTE ${schema}.job_table_format($cmd$${createIndexJobFetchCreated(schema)}$cmd$, tablename);
596602
EXECUTE ${schema}.job_table_format($cmd$${createIndexJobThrottle(schema)}$cmd$, tablename);
597603
EXECUTE ${schema}.job_table_format($cmd$${createIndexJobGroupConcurrency(schema)}$cmd$, tablename);
598604
EXECUTE ${schema}.job_table_format($cmd$${createIndexJobBlocking(schema)}$cmd$, tablename);
@@ -725,6 +731,17 @@ function createIndexJobFetch (schema: string, noCoveringIndex = false) {
725731
return `CREATE INDEX job_i5 ON ${schema}.job (name, start_after) WHERE state < '${JOB_STATES.active}' AND NOT blocked`
726732
}
727733

734+
// Ordered partial indexes let the fetch LIMIT stop as soon as it has enough candidates instead of
735+
// materializing and sorting every eligible row for the queue. Keep job_i5 as well: its start_after
736+
// key remains useful when a queue contains a large scheduled backlog that is not ready to run yet.
737+
function createIndexJobFetchPriority (schema: string) {
738+
return `CREATE INDEX job_i10 ON ${schema}.job (name, priority DESC, created_on, id) WHERE state < '${JOB_STATES.active}' AND NOT blocked`
739+
}
740+
741+
function createIndexJobFetchCreated (schema: string) {
742+
return `CREATE INDEX job_i11 ON ${schema}.job (name, created_on, id) WHERE state < '${JOB_STATES.active}' AND NOT blocked`
743+
}
744+
728745
function createIndexJobPolicyExclusive (schema: string) {
729746
return `CREATE UNIQUE INDEX job_i6 ON ${schema}.job (name, COALESCE(singleton_key, '')) WHERE state <= '${JOB_STATES.active}' AND policy = '${QUEUE_POLICIES.exclusive}'`
730747
}
@@ -2854,8 +2871,8 @@ const POLICY_JOB_INDEXES: Record<number, string> = {
28542871
8: QUEUE_POLICIES.key_strict_fifo
28552872
}
28562873
// job_iN indexes with no policy gate — created on every job table regardless of policy
2857-
// (throttle i4, fetch i5, group-concurrency i7, blocking i9).
2858-
const BASE_JOB_INDEXES = [4, 5, 7, 9]
2874+
// (throttle i4, fetch i5/i10/i11, group-concurrency i7, blocking i9).
2875+
const BASE_JOB_INDEXES = [4, 5, 7, 9, 10, 11]
28592876

28602877
// The fixed (non-job) managed tables; job/job_common/partitions are handled separately.
28612878
const FIXED_MANAGED_TABLES = ['version', 'queue', 'schedule', 'subscription', 'bam', 'warning', 'queue_stats', 'job_dependency']

0 commit comments

Comments
 (0)