RisingWave Dialect

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

const db = new Prorm({
  dialect: 'risingwave',
  host: 'localhost',
  // port defaults to 4566, not 5432 — inheriting Postgres's default would
  // make every connection fail against a healthy deployment
  username: 'root', // a local RisingWave's default user
  database: 'dev', // a local RisingWave's default database
});

const d = new Prorm({
  dialect: 'risingwave',
  host: 'localhost',
  username: 'root',
  database: 'dev',
});
// or: const d = createRisingWaveDialect({ ... });

$ 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: 'risingwave',
  host: 'localhost',
  database: 'dev',
  username: 'root',
});

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

export const db = new Prorm({
  dialect: 'risingwave',
  url: 'risingwave://root@localhost:4566/dev',
});

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

const User = db.define('User', {
  id: { type: DataTypes.INTEGER(), primaryKey: true, allowNull: false },
  email: { type: DataTypes.STRING(255), allowNull: false },
  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({ id: 1, 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({ id: 1, email: 'ada@example.com', name: 'Ada' });
await User.update({ name: 'Grace' }, { where: { email: 'ada@example.com' } });
await User.destroy({ where: { id: 1 } });

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' } } } });

d.buildCreateTableSql(
  'events',
  { id: { type: DataTypes.BIGINT, primaryKey: true }, ts: { type: DataTypes.DATE } },
  { watermark: { column: 'ts', delay: '5 seconds' }, onConflict: 'do_update_if_not_null' },
);

d.buildCreateTableSql('t', { a: { type: DataTypes.BIGINT } }, {
  appendOnly: true,
  connector: { connector: 'kafka', topic: 't', 'properties.bootstrap.server': 'k:9092' },
  format: { format: 'PLAIN', encode: 'JSON' },
});

d.buildAddColumnSql('t', 'c', { type: DataTypes.STRING, defaultValue: 'x' });
// ALTER TABLE "t" ADD COLUMN "c" VARCHAR DEFAULT 'x'

d.buildCreateSourceSql({
  name: 'clicks',
  ifNotExists: true,
  columns: { user_id: { type: DataTypes.BIGINT }, ts: { type: DataTypes.DATE } },
  watermark: { column: 'ts', delay: '5 seconds' },
  with: { connector: 'kafka', topic: 'clicks', 'properties.bootstrap.server': 'k:9092' },
  format: { format: 'PLAIN', encode: 'AVRO', options: { 'schema.registry': 'http://r:8081' } },
});

d.buildCreateSinkSql({
  name: 'out', ifNotExists: true,
  as: {
    from: 'clicks', timeColumn: 'ts',
    window: { type: 'tumble', size: '1 minute' },
    aggregates: [{ fn: 'count', as: 'n' }],
    emitOnWindowClose: true,
  },
  with: { connector: 'blackhole' },
});

d.buildWindowedAggregateQuery({
  from: 'clicks', timeColumn: 'ts',
  window: { type: 'tumble', size: '1 minute' },
  groupBy: ['page'],
  aggregates: [
    { fn: 'count', as: 'views' },
    { fn: 'count', column: 'user_id', distinct: true, as: 'users' },
  ],
  where: { country: 'NZ' },
});

d.buildWindowedAggregateQuery({
  from: 'clicks', timeColumn: 'ts',
  window: { type: 'hop', hop: '1 minute', size: '5 minutes', offset: '30 seconds' },
  aggregates: [{ fn: 'sum', column: 'amount', as: 'total' }],
});

d.buildCreateMaterializedViewSql({ name: 'mv', query: 'SELECT 1', ifNotExists: true });
// CREATE MATERIALIZED VIEW IF NOT EXISTS "mv" AS SELECT 1

d.buildCreateSubscriptionSql({ name: 'sub', from: 'mv', retention: '1D' });
// CREATE SUBSCRIPTION "sub" FROM "mv" WITH (retention = '1D')

d.buildDropSubscriptionSql('sub', { ifExists: true });
// DROP SUBSCRIPTION IF EXISTS "sub"

d.buildDeclareSubscriptionCursorSql('c', 'sub', 'full');
// DECLARE "c" SUBSCRIPTION CURSOR FOR "sub" FULL

d.buildDeclareSubscriptionCursorSql('c', 'sub', 1717000000000);
// DECLARE "c" SUBSCRIPTION CURSOR FOR "sub" SINCE 1717000000000

d.buildFetchSubscriptionCursorSql('c', { count: 100, timeout: '1s' });
// FETCH 100 FROM "c" WITH (timeout = '1s')

d.buildCloseCursorSql('c');
// CLOSE "c"

d.buildCreateIndexSql('orders', {
  name: 'idx', ifNotExists: true,
  columns: ['customer', { name: 'ts', order: 'DESC' }],
  include: ['total'],
  distributedBy: ['customer'],
});

d.buildCreateVectorIndexSql('docs', {
  name: 'vi', column: 'emb', method: 'hnsw', distanceType: 'cosine',
  include: ['id'], params: { m: 16, ef_construction: 64 },
});

await d.startTransaction();
// START TRANSACTION READ ONLY