Citus Dialect

import { Prorm } from 'ts-prorm-orm';

$ npm install ts-prorm-orm pg

$ pnpm add ts-prorm-orm pg

$ yarn add ts-prorm-orm pg

$ bun add ts-prorm-orm pg

import { Prorm } from 'ts-prorm-orm';

export const db = new Prorm({
  dialect: 'citus',
  host: 'coordinator.internal',
  database: 'app',
  username: 'postgres',
  password: process.env.DB_PASSWORD,
});

import { Prorm } from 'ts-prorm-orm';

export const db = new Prorm({
  dialect: 'citus',
  url: 'citus://postgres:secret@coordinator.internal:5432/app',
});

import { DataTypes } from 'ts-prorm-orm';

const User = db.define('User', {
  id: { type: DataTypes.INTEGER(), primaryKey: true, autoIncrement: true },
  email: { type: DataTypes.STRING(255), allowNull: false, unique: true },
  name: { type: DataTypes.STRING(100) },
  meta: { type: DataTypes.JSONB() },
});

await db.sync(); // CREATE TABLE — see Generated SQL below

import { Op } from 'ts-prorm-orm';

await User.create({ email: 'ada@example.com', name: 'Ada' });

const page = await User.findAll({
  where: { name: { [Op.like]: 'A%' } },
  order: [['id', 'ASC']],
  limit: 20,
  offset: 40,
});

await User.create({ email: 'ada@example.com', name: 'Ada' });
await User.update({ name: 'Grace' }, { where: { email: 'ada@example.com' } });
await User.destroy({ where: { id: 1 } });

await User.upsert({ id: 1, email: 'ada@example.com', name: 'Ada Lovelace' });

await User.findAll({
  where: { name: { [Op.like]: 'A%' } },
  order: [['id', 'ASC']],
  limit: 20,
  offset: 40,
});

import { Op } from 'ts-prorm-orm';

await User.findAll({ where: { meta: { [Op.contains]: { plan: 'pro' } } } });

await db.transaction(async (t) => {
  await User.create({ email: 'grace@example.com', name: 'Grace' }, { transaction: t });
  // A nested transaction is a SAVEPOINT; its failure undoes only its own work.
  await db
    .transaction(async (sp) => {
      await User.destroy({ where: { id: 1 }, transaction: sp });
      throw new Error('keep the row');
    }, { transaction: t })
    .catch(() => {});
});

await db.transaction(async (t) => {
  await User.findAll({ where: { name: 'Ada' }, lock: { level: 'UPDATE', skipLocked: true }, transaction: t });
});

const dialect = new Prorm({
  dialect: 'citus',
  host: 'coordinator.internal',
  // port defaults to 5432 (CITUS_DEFAULT_PORT); pass port to override
  username: 'postgres',
  password: process.env.PG_PASSWORD,
  database: 'app',
});
await dialect.query(dialect.createExtensionSQL());
// CREATE EXTENSION IF NOT EXISTS "citus"

dialect.createReferenceTable('app.Countries');
// SELECT create_reference_table('"app"."Countries"')

dialect.createDistributedTable('events', 'tenant_id', { colocateWith: 'tenants', shardCount: 64 });
// SELECT create_distributed_table('"events"', 'tenant_id', colocate_with => '"tenants"', shard_count => 64)

dialect.createDistributedTable('e', 'k', { colocateWith: 'none' });
// SELECT create_distributed_table('"e"', 'k', colocate_with => 'none')  -- 'default'/'none' are keywords, not tables

dialect.createDistributedTable('e', 'k', { concurrently: true });
// SELECT create_distributed_table_concurrently('"e"', 'k')  -- Citus 11.1+, cannot run inside a transaction block

dialect.createReferenceTable('countries');
// SELECT create_reference_table('"countries"')

dialect.undistributeTable('e', { cascadeViaForeignKeys: true });
// SELECT undistribute_table('"e"', cascade_via_foreign_keys => true)

dialect.alterDistributedTable('e', { distributionColumn: 'k2', shardCount: 16, colocateWith: 'none', cascadeToColocated: false });
// SELECT alter_distributed_table('"e"', distribution_column => 'k2', shard_count => 16, colocate_with => 'none', cascade_to_colocated => false)

dialect.truncateLocalDataAfterDistributingTable('e');
// SELECT truncate_local_data_after_distributing_table('"e"')

dialect.getShardIdForDistributionColumn('e', 42);
// SELECT get_shard_id_for_distribution_column('"e"', 42)

dialect.isolateTenantToNewShard('e', 135, { cascade: true });
// SELECT isolate_tenant_to_new_shard('"e"', 135, cascade_option => 'CASCADE')

dialect.citusSchemaDistribute('tenant_1');   // SELECT citus_schema_distribute('"tenant_1"')
dialect.citusSchemaUndistribute('tenant_1'); // SELECT citus_schema_undistribute('"tenant_1"')

dialect.createDistributedFunction('app.deliver', ['int', 'character varying(10)', 'text[]'], {
  distributionArgName: '$1',
  colocateWith: 'orders',
});
// SELECT create_distributed_function('"app"."deliver"(int, character varying(10), text[])', distribution_arg_name => '$1', colocate_with => '"orders"')

await dialect.createAndDistributeTable(
  'events',
  { tenant_id: { type: T('BIGINT'), primaryKey: true }, id: { type: T('BIGINT'), primaryKey: true } },
  'tenant_id',
  { shardCount: 8 },
);
// 1. assertDistributable(...) — throws first if the PK/UNIQUE keys don't include 'tenant_id'
// 2. CREATE TABLE "events" (...)
// 3. SELECT create_distributed_table('"events"', 'tenant_id', shard_count => 8)

dialect.citusAddNode('worker-1', 5432, { groupId: 3, nodeRole: 'secondary', nodeCluster: 'c1' });
// SELECT citus_add_node('worker-1', 5432, groupid => 3, noderole => 'secondary', nodecluster => 'c1')

dialect.citusRemoveNode('worker-1', 5432);
// SELECT citus_remove_node('worker-1', 5432)

dialect.citusDrainNode('worker-1', 5432, { shardTransferMode: 'block_writes' });
// SELECT citus_drain_node('worker-1', 5432, shard_transfer_mode => 'block_writes'::citus.shard_transfer_mode)

dialect.citusSetCoordinatorHost('coord', 5433);
// SELECT citus_set_coordinator_host('coord', port => 5433)

dialect.rebalanceTableShards({
  table: 'events',
  threshold: 0.1,
  maxShardMoves: 10,
  excludedShardList: [102008, BigInt(102009)],
  shardTransferMode: 'force_logical',
  drainOnly: false,
  rebalanceStrategy: 'by_disk_size',
});
// SELECT rebalance_table_shards(relation => '"events"', threshold => 0.1, max_shard_moves => 10,
//   excluded_shard_list => ARRAY[102008, 102009]::bigint[],
//   shard_transfer_mode => 'force_logical'::citus.shard_transfer_mode, drain_only => false,
//   rebalance_strategy => 'by_disk_size')

dialect.citusRebalanceStart({ drainOnly: true, rebalanceStrategy: 'by_shard_count' });
// SELECT citus_rebalance_start(drain_only => true, rebalance_strategy => 'by_shard_count')  -- background, Citus 11.1+

dialect.citusRebalanceStatus(); // SELECT * FROM citus_rebalance_status()
dialect.citusRebalanceStop();   // SELECT citus_rebalance_stop()
dialect.citusRebalanceWait();   // SELECT citus_rebalance_wait()

dialect.buildCreateColumnarTableSql(
  'metrics',
  {
    id: { type: T('BIGINT'), autoIncrement: true, primaryKey: true },
    at: { type: T('DATE'), allowNull: false, defaultValue: 'CURRENT_TIMESTAMP' },
    host: { type: T('STRING', { length: 64 }) },
  },
  { ifNotExists: true, compression: 'zstd', compressionLevel: 3, stripeRowLimit: 150000 },
);

dialect.alterTableSetAccessMethod('events', 'columnar');
// SELECT alter_table_set_access_method('"events"', 'columnar')

dialect.setColumnarOptions('events', { compression: 'lz4', chunkGroupRowLimit: 5000 });
// ALTER TABLE "events" SET (columnar.compression = lz4, columnar.chunk_group_row_limit = 5000)
// (applies only to data written afterwards; existing stripes keep their encoding)

dialect.buildColumnarIndexSql('m', 'm_id', ['id'], { unique: true });
// CREATE UNIQUE INDEX "m_id" ON "m" USING btree ("id")
dialect.buildColumnarIndexSql('m', 'm_h', ['host'], { using: 'hash' });
// CREATE INDEX "m_h" ON "m" USING hash ("host")

await dialect.addIndex('t', 'i', ['a'], { unique: true, using: 'btree', where: { b: null } });
// CREATE UNIQUE INDEX "i" ON "t" USING btree ("a") WHERE "b" IS NULL

dialect.buildPartialIndexSql('orders', 'open_orders', ['tenant_id', 'id'], {
  where: { status: 'shipped', total: { [Op.gt]: 0 } },
  include: ['total'],
});
// CREATE INDEX "open_orders" ON "orders" ("tenant_id", "id") INCLUDE ("total")
//   WHERE "status" = 'shipped' AND "total" > 0