-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- createPartitioned() - showPartitions() - runMaintenance()
- Loading branch information
1 parent
233f2db
commit b50dc6d
Showing
21 changed files
with
608 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,8 @@ | ||
|
|
||
| -- create the extension in the "pgmq" schema | ||
| -- create the extension | ||
| CREATE EXTENSION pgmq; | ||
| CREATE EXTENSION pg_partman; | ||
|
|
||
| -- CREATE EXTENSION dblink; | ||
| -- CREATE EXTENSION pg_jobmon; | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| /* c8 ignore start */ | ||
|
|
||
|
|
||
| export interface ShowPartitionsRecord { | ||
| partition_schemaname: string | ||
| partition_tablename: string | ||
| } | ||
|
|
||
|
|
||
| /* c8 ignore stop */ |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
|
|
||
| export * from './part.js' | ||
| export * from './part.types.js' |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| import type { ShowPartitionsRecord } from './db.types.js' | ||
| import type { ShowPartitions } from './part.types.js' | ||
|
|
||
|
|
||
| export function parseShowPartitionRecord(input: ShowPartitionsRecord): ShowPartitions { | ||
| const ret: ShowPartitions = { | ||
| partitionSchemaname: input.partition_schemaname, | ||
| partitionTablename: input.partition_tablename, | ||
| } | ||
| return ret | ||
| } | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
|
|
||
| export enum PartSql { | ||
| /** | ||
| * @link https://github.com/pgpartman/pg_partman/blob/development/doc/pg_partman.md#show_partitions | ||
| */ | ||
| showPartitions = 'SELECT * FROM show_partitions(?::text, ?, ?)', | ||
| /** | ||
| * @link https://tembo-io.github.io/pgmq/api/sql/functions/#create_partitioned | ||
| */ | ||
| createPartitioned = 'SELECT pgmq.create_partitioned(?, ?, ?)', | ||
| runMaintenance = 'SELECT run_maintenance()', | ||
| runMaintenance2 = 'SELECT run_maintenance(?::text, ?, ?)', | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| import assert from 'node:assert' | ||
|
|
||
| import type { Knex, QueryResponse, Transaction } from '../knex.types.js' | ||
| import type { QueueMetaManager } from '../queue-meta-manager/index.queue-meta.js' | ||
|
|
||
| import type { ShowPartitionsRecord } from './db.types.js' | ||
| import { parseShowPartitionRecord } from './part.helpers.js' | ||
| import { PartSql } from './part.sql.js' | ||
| import type { CreatePartitionedQueueMetaOptions, RunMaintenanceOptions, ShowPartitions, ShowPartitionsOptions } from './part.types.js' | ||
|
|
||
|
|
||
| export class Partition { | ||
|
|
||
| constructor( | ||
| protected readonly dbh: Knex, | ||
| protected readonly queueMeta: QueueMetaManager, | ||
| ) { } | ||
|
|
||
| /** | ||
| * @link https://tembo-io.github.io/pgmq/#partitioned-queues | ||
| */ | ||
| async createPartitioned(options: CreatePartitionedQueueMetaOptions): Promise<string> { | ||
| const opts: CreatePartitionedQueueMetaOptions = { | ||
| ...options, | ||
| trx: options.trx ?? await this.startTransaction(), | ||
| queue: options.queue.toLowerCase(), | ||
| } | ||
| const { trx } = opts | ||
| assert(trx, 'Transaction is required') | ||
|
|
||
| await this._createPartitioned(opts) | ||
| const queueId = await this.queueMeta.create(opts) | ||
|
|
||
| if (! options.trx) { | ||
| await trx.commit() | ||
| } | ||
| return queueId | ||
| } | ||
|
|
||
|
|
||
| // #region showPartitions() | ||
|
|
||
| /** | ||
| * show partitions of a partitioned table | ||
| * @CAUTION queue must prefix with schema name, like 'public.my_queue' | ||
| * @description will throw error if | ||
| * - Given parent table not managed by pg_partman | ||
| */ | ||
| async showPartitions(options: ShowPartitionsOptions): Promise<ShowPartitions[]> { | ||
| const list = await this._showPartitions(options) | ||
| return list | ||
| } | ||
|
|
||
| // #region run_maintenance() | ||
|
|
||
| async runMaintenance(options?: RunMaintenanceOptions): Promise<void> { | ||
| const name = options?.queue?.toLowerCase() ?? null | ||
| const analyze = options?.analyze ?? false | ||
| const jobmon = options?.jobmon ?? true | ||
|
|
||
| const sql = name ? PartSql.runMaintenance2 : PartSql.runMaintenance | ||
| const data = name ? [name, analyze, jobmon] : null | ||
| await this.execute(sql, data, null) | ||
| } | ||
|
|
||
|
|
||
|
|
||
| private async _showPartitions(options: ShowPartitionsOptions): Promise<ShowPartitions[]> { | ||
| const { trx } = options | ||
| const name = options.queue.toLowerCase() | ||
| const ord = options.order ?? 'ASC' | ||
| const includeDefault = options.includeDefault ?? false | ||
|
|
||
| const sql = PartSql.showPartitions | ||
| const res = await this.execute<QueryResponse<ShowPartitionsRecord>>(sql, [name, ord, includeDefault], trx) | ||
|
|
||
| const ret = res.rows.map(parseShowPartitionRecord) | ||
| return ret | ||
| } | ||
|
|
||
|
|
||
| protected async _createPartitioned(options: CreatePartitionedQueueMetaOptions): Promise<void> { | ||
| const { queue, partitionInterval, retentionInterval, trx } = options | ||
| const sql = PartSql.createPartitioned | ||
| const args = [ | ||
| queue, | ||
| partitionInterval ?? '1 month', | ||
| retentionInterval ?? '1 year', | ||
| ] | ||
| await this.execute(sql, args, trx) | ||
| } | ||
|
|
||
|
|
||
| // #region common ------------------------------ | ||
|
|
||
| protected async execute<T = unknown>(sql: string, params: unknown[] | null, trx: Transaction | undefined | null): Promise<T> { | ||
| if (trx) { | ||
| assert(! trx.isCompleted(), 'parameter trx is completed already') | ||
| } | ||
| const dbh = trx ?? this.dbh | ||
| try { | ||
| const res = await (params ? dbh.raw(sql, params) : dbh.raw(sql)) as T | ||
| return res | ||
| } | ||
| catch (ex) { | ||
| await trx?.rollback() | ||
| throw ex | ||
| } | ||
| } | ||
|
|
||
| protected async startTransaction(): Promise<Transaction> { | ||
| const ret = await this.dbh.transaction() | ||
| assert(ret, 'Transaction is required') | ||
| return ret | ||
| } | ||
|
|
||
| } | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* c8 ignore start */ | ||
|
|
||
| import type { QueueOptionsBase } from '../types.js' | ||
|
|
||
|
|
||
| export interface ShowPartitions { | ||
| partitionSchemaname: string | ||
| partitionTablename: string | ||
| } | ||
|
|
||
| /** | ||
| * @link https://github.com/pgpartman/pg_partman/blob/development/doc/pg_partman.md#show_partitions | ||
| */ | ||
| export interface ShowPartitionsOptions extends QueueOptionsBase { | ||
| /** | ||
| * @default 'ASC' | ||
| */ | ||
| order?: 'ASC' | 'DESC' | ||
| /** | ||
| * @default false | ||
| */ | ||
| includeDefault?: boolean | ||
| } | ||
|
|
||
| export enum PartMsg { | ||
| queueNotManaged = 'Given parent table not managed by pg_partman', | ||
| } | ||
|
|
||
|
|
||
| /** | ||
| * @link https://tembo-io.github.io/pgmq/api/sql/functions/#create_partitioned | ||
| * @link https://tembo-io.github.io/pgmq/#partitioned-queues | ||
| */ | ||
| export interface CreatePartitionedQueueMetaOptions extends QueueOptionsBase { | ||
| /** | ||
| * Pgsql Duration (eg. '1 days') or number string (eg. '100000') | ||
| * @default '1 month' | ||
| */ | ||
| partitionInterval?: string | ||
| /** | ||
| * Pgsql Duration (eg. '1 year') or number string (eg. '10000000') | ||
| * @default '1 year' | ||
| */ | ||
| retentionInterval?: string | ||
| /** | ||
| * Maximum 512 characters | ||
| */ | ||
| queueKey?: string | null | ||
| json?: object | null | ||
| } | ||
|
|
||
|
|
||
| /** | ||
| * @link https://github.com/pgpartman/pg_partman/blob/development/doc/pg_partman.md#run_maintenance | ||
| */ | ||
| export interface RunMaintenanceOptions { | ||
| /** | ||
| * Must prefix with schema name, like 'public.my_queue' | ||
| */ | ||
| queue?: string | ||
| /** | ||
| * @default false | ||
| */ | ||
| analyze?: boolean | ||
| /** | ||
| * @default true | ||
| */ | ||
| jobmon?: boolean | ||
| } | ||
|
|
||
|
|
||
| /* c8 ignore stop */ |
Oops, something went wrong.