Skip to content

Commit 8d620e7

Browse files
chrisbbreuerchrisbbreuer
committed
perf(database): batch query log writes
1 parent 881d9e6 commit 8d620e7

2 files changed

Lines changed: 117 additions & 23 deletions

File tree

‎storage/framework/core/database/src/query-logger.ts‎

Lines changed: 86 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,71 @@ interface QueryLogRecord {
7070
// drops unrelated requests while an asynchronous INSERT is still pending.
7171
const queryLogContext = new AsyncLocalStorage<boolean>()
7272

73+
interface PendingQueryLog {
74+
record: QueryLogRecord
75+
resolve: () => void
76+
}
77+
78+
const QUERY_LOG_BATCH_SIZE = 100
79+
const pendingQueryLogs: PendingQueryLog[] = []
80+
let queryLogFlushScheduled = false
81+
let queryLogFlushInFlight = false
82+
83+
/** Persist query diagnostics in bounded multi-row inserts. */
84+
function enqueueQueryLog(record: QueryLogRecord): Promise<void> {
85+
const settled = new Promise<void>((resolve) => {
86+
pendingQueryLogs.push({ record, resolve })
87+
})
88+
89+
if (pendingQueryLogs.length >= QUERY_LOG_BATCH_SIZE) {
90+
void flushQueuedQueryLogs()
91+
}
92+
else if (!queryLogFlushScheduled) {
93+
queryLogFlushScheduled = true
94+
setImmediate(() => {
95+
queryLogFlushScheduled = false
96+
void flushQueuedQueryLogs()
97+
})
98+
}
99+
100+
return settled
101+
}
102+
103+
async function flushQueuedQueryLogs(): Promise<void> {
104+
if (queryLogFlushInFlight)
105+
return
106+
107+
queryLogFlushInFlight = true
108+
try {
109+
while (pendingQueryLogs.length > 0) {
110+
const batch = pendingQueryLogs.splice(0, QUERY_LOG_BATCH_SIZE)
111+
const recordsByShape = Map.groupBy(
112+
batch.map(item => item.record),
113+
record => Object.keys(record).join('\0'),
114+
)
115+
for (const records of recordsByShape.values()) {
116+
const stored = await queryLogContext.run(true, () => storeQueryLogs(records, records.length === 1))
117+
if (!stored) {
118+
for (const record of records)
119+
await queryLogContext.run(true, () => storeQueryLogs([record]))
120+
}
121+
}
122+
for (const item of batch)
123+
item.resolve()
124+
}
125+
}
126+
finally {
127+
queryLogFlushInFlight = false
128+
if (pendingQueryLogs.length > 0 && !queryLogFlushScheduled) {
129+
queryLogFlushScheduled = true
130+
setImmediate(() => {
131+
queryLogFlushScheduled = false
132+
void flushQueuedQueryLogs()
133+
})
134+
}
135+
}
136+
}
137+
73138
/**
74139
* Process an executed query and store it in the database
75140
*/
@@ -114,7 +179,7 @@ export async function logQuery(event: LogEvent): Promise<void> {
114179

115180
// Deferred query hooks inherit this context, so the INSERT cannot log
116181
// itself while other requests remain free to record their own queries.
117-
await queryLogContext.run(true, () => storeQueryLog(logRecord))
182+
await enqueueQueryLog(logRecord)
118183

119184
// Log slow or failed queries to the application log
120185
if (status !== 'completed') {
@@ -505,15 +570,28 @@ function generateOptimizationSuggestions(explainResult: any, logRecord: QueryLog
505570
/**
506571
* Store the query log in the database
507572
*/
508-
async function storeQueryLog(logRecord: QueryLogRecord): Promise<void> {
573+
async function storeQueryLogs(logRecords: QueryLogRecord[], reportFailure = true): Promise<boolean> {
509574
try {
510-
await db.insertInto('query_logs').values(logRecord as unknown as Record<string, unknown>).execute()
575+
const values = logRecords as unknown as Record<string, unknown>[]
576+
if (values.length === 1) {
577+
await db.insertInto('query_logs').values(values).execute()
578+
}
579+
else {
580+
await db.transaction(async (rawTrx) => {
581+
const trx = rawTrx as unknown as typeof db
582+
await trx.insertInto('query_logs').values(values).execute()
583+
})
584+
}
585+
return true
511586
}
512587
catch (error) {
513-
const message = error instanceof Error ? error.message : String(error)
514-
if (/no such table|does not exist|doesn't exist/i.test(message) && /query_logs/i.test(message))
515-
log.debug('Query logging will start after the query_logs table is migrated.')
516-
else
517-
log.error('Failed to store query log:', error)
588+
if (reportFailure) {
589+
const message = error instanceof Error ? error.message : String(error)
590+
if (/no such table|does not exist|doesn't exist/i.test(message) && /query_logs/i.test(message))
591+
log.debug('Query logging will start after the query_logs table is migrated.')
592+
else
593+
log.error('Failed to store query log:', error)
594+
}
595+
return false
518596
}
519597
}

‎storage/framework/core/database/tests/fixtures/query-logger-dispatch.ts‎

Lines changed: 31 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
const { config, overridesReady } = await import('@stacksjs/config')
22
const { db, ensureDatabaseConfigLoaded, initializeDbConfig, resetDatabaseConnection } = await import('../../src/utils')
3+
const { registerPersistentQueryHooks } = await import('@stacksjs/query-builder')
34

45
await overridesReady
56
await ensureDatabaseConfigLoaded()
@@ -80,24 +81,39 @@ try {
8081
await queryWithDiagnostics('query_logger_missing_table', true)
8182
await queryWithDiagnostics('query_logger_after_error')
8283
if (persistence) {
84+
let queryLogInsertStatements = 0
85+
const stopCountingLogInserts = registerPersistentQueryHooks({
86+
onQueryEnd(event) {
87+
if (/insert\s+into\s+[`"]?query_logs/i.test(event.sql))
88+
queryLogInsertStatements++
89+
},
90+
})
8391
// The logger is fully loaded here. A process-wide recursion guard can
8492
// drop every other real query while its preceding log INSERT is pending.
85-
for (const mode of ['sequential', 'concurrent']) {
86-
const marker = `query_logger_burst_${mode}`
87-
const query = () => db.selectFrom('query_logger_fixture').select([`id as ${marker}`]).execute()
88-
if (mode === 'sequential') {
89-
for (let i = 0; i < 200; i++) await query()
90-
}
91-
else {
92-
await Promise.all(Array.from({ length: 200 }, query))
93+
try {
94+
for (const mode of ['sequential', 'concurrent']) {
95+
if (mode === 'concurrent') queryLogInsertStatements = 0
96+
const marker = `query_logger_burst_${mode}`
97+
const query = () => db.selectFrom('query_logger_fixture').select([`id as ${marker}`]).execute()
98+
if (mode === 'sequential') {
99+
for (let i = 0; i < 200; i++) await query()
100+
}
101+
else {
102+
await Promise.all(Array.from({ length: 200 }, query))
103+
}
104+
await Bun.sleep(0)
105+
const records = await db.unsafe('SELECT query FROM query_logs').execute()
106+
const matching = records.filter(row => String(row.query).includes(marker))
107+
if (matching.length !== 200)
108+
throw new Error(`Lost ${mode} query logs: expected 200, received ${matching.length}`)
109+
if (records.some(row => /insert\s+into/i.test(String(row.query))))
110+
throw new Error('Query logging recursively persisted its own INSERT')
111+
if (mode === 'concurrent' && queryLogInsertStatements >= matching.length)
112+
throw new Error(`Concurrent query logs were not batched: ${queryLogInsertStatements} INSERTs for ${matching.length} rows`)
93113
}
94-
await Bun.sleep(0)
95-
const records = await db.unsafe('SELECT query FROM query_logs').execute()
96-
const matching = records.filter(row => String(row.query).includes(marker))
97-
if (matching.length !== 200)
98-
throw new Error(`Lost ${mode} query logs: expected 200, received ${matching.length}`)
99-
if (records.some(row => /insert\s+into/i.test(String(row.query))))
100-
throw new Error('Query logging recursively persisted its own INSERT')
114+
}
115+
finally {
116+
stopCountingLogInserts()
101117
}
102118

103119
await db.unsafe('DROP TABLE query_logs').execute()

0 commit comments

Comments
 (0)