Merge pull request #72 from supabase/feat/replica-identity

Add replica identity fields to /tables

Author: kiwicopple

package.json

@@ -16,7 +16,7 @@
     "dist:pkg": "rimraf bin && pkg --out-path ./bin package.json",
     "dev": "NODE_ENV=development npm-run-all build server",
     "dev:watch": "nodemon",
-    "pretty": "prettier --write \"./src/**/*.{js,json,yml,md,vue,css,scss}\"",
+    "pretty": "prettier --write \"{src,test}/**/*.ts\"",
     "server": "node ./dist/start.js",
     "test": "node -r esm ./node_modules/.bin/mocha 'test/**/*.js' --recursive "
   },

src/api/tables.ts

@@ -1,5 +1,5 @@
 import { Router } from 'express'
-import format from 'pg-format'
+import format, { ident } from 'pg-format'
 import { coalesceRowsToArray, toTransaction } from '../lib/helpers'
 import { RunQuery } from '../lib/connectionPool'
 import { DEFAULT_SYSTEM_SCHEMAS } from '../lib/constants'
@@ -233,16 +233,35 @@ const selectSingleByName = (
 const createTableSqlize = ({
   name,
   schema = 'public',
+  replica_identity,
+  replica_identity_index,
   comment,
 }: {
   name: string
   schema?: string
+  replica_identity?: 'DEFAULT' | 'INDEX' | 'FULL' | 'NOTHING'
+  replica_identity_index?: string
   comment?: string
 }) => {
   const tableSql = format('CREATE TABLE IF NOT EXISTS %I.%I ();', schema, name)
+  let replicaSql: string
+  if (replica_identity === undefined) {
+    replicaSql = ''
+  } else if (replica_identity === 'INDEX') {
+    replicaSql = `ALTER TABLE ${ident(schema)}.${ident(
+      name
+    )} REPLICA IDENTITY USING INDEX ${replica_identity_index};`
+  } else {
+    replicaSql = `ALTER TABLE ${ident(schema)}.${ident(name)} REPLICA IDENTITY ${replica_identity};`
+  }
   const commentSql =
     comment === undefined ? '' : format('COMMENT ON TABLE %I.%I IS %L;', schema, name, comment)
-  return `${tableSql} ${commentSql}`
+  return `
+BEGIN;
+  ${tableSql}
+  ${replicaSql}
+  ${commentSql}
+COMMIT;`
 }
 const alterTableName = (previousName: string, newName: string, schema: string) => {
   return format('ALTER TABLE %I.%I RENAME TO %I;', schema, previousName, newName)
@@ -252,15 +271,19 @@ const alterTableSql = ({
   name,
   rls_enabled,
   rls_forced,
+  replica_identity,
+  replica_identity_index,
   comment,
 }: {
   schema?: string
   name: string
   rls_enabled?: boolean
   rls_forced?: boolean
+  replica_identity?: 'DEFAULT' | 'INDEX' | 'FULL' | 'NOTHING'
+  replica_identity_index?: string
   comment?: string
 }) => {
-  let alter = format('ALTER table %I.%I', schema, name)
+  let alter = format('ALTER TABLE %I.%I', schema, name)
   let enableRls = ''
   if (rls_enabled !== undefined) {
     let enable = `${alter} ENABLE ROW LEVEL SECURITY;`
@@ -273,13 +296,23 @@ const alterTableSql = ({
     let disable = `${alter} NO FORCE ROW LEVEL SECURITY;`
     forceRls = rls_forced ? enable : disable
   }
+  let replicaSql: string
+  if (replica_identity === undefined) {
+    replicaSql = ''
+  } else if (replica_identity === 'INDEX') {
+    replicaSql = `${alter} REPLICA IDENTITY USING INDEX ${replica_identity_index};`
+  } else {
+    replicaSql = `${alter} REPLICA IDENTITY ${replica_identity};`
+  }
   const commentSql =
     comment === undefined ? '' : format('COMMENT ON TABLE %I.%I IS %L;', schema, name, comment)
   return `
+BEGIN;
     ${enableRls}
     ${forceRls}
+    ${replicaSql}
     ${commentSql}
-  `.trim()
+COMMIT;`
 }
 const dropTableSql = (schema: string, name: string, cascade: boolean) => {
   return format(`DROP TABLE %I.%I ${cascade ? 'CASCADE' : 'RESTRICT'};`, schema, name)

src/lib/connectionPool.ts

@@ -6,7 +6,7 @@ import { SQLStatement } from 'sql-template-strings'
 pg.types.setTypeParser(20, 'text', parseInt)
 const { Pool } = pg
 
-export const RunQuery = async (connectionString: any, sql: string|SQLStatement) => {
+export const RunQuery = async (connectionString: any, sql: string | SQLStatement) => {
   const pool = new Pool({ connectionString })
   try {
     const results = await pool.query(sql)

src/lib/interfaces.ts

@@ -65,8 +65,8 @@ export namespace Tables {
     is_typed: boolean
     bytes: number
     size: string
-    rls_enabled: boolean          // determines where RLS has been turned on 
-    rls_forced: boolean           // determines whether RLS should be forced on the table owner
+    rls_enabled: boolean // determines where RLS has been turned on
+    rls_forced: boolean // determines whether RLS should be forced on the table owner
 
     // pg_stat_all_tables columns
     sequential_scans: number

src/lib/sql/tables.sql

@@ -6,6 +6,11 @@ SELECT
   is_insertable_into,
   relrowsecurity::bool as rls_enabled, 
   relforcerowsecurity as rls_forced,
+  CASE WHEN relreplident = 'd' THEN 'DEFAULT'
+       WHEN relreplident = 'i' THEN 'INDEX'
+       WHEN relreplident = 'f' THEN 'FULL'
+       ELSE 'NOTHING'
+  END AS replica_identity,
   is_typed,
   pg_total_relation_size(format('%I.%I', table_schema, table_name))::bigint AS bytes,
   pg_size_pretty(

test/integration/index.spec.js

@@ -250,8 +250,9 @@ describe('/tables', async () => {
     assert.equal(true, !!included)
   })
   it('POST /tables should create a table', async () => {
-    const { data: newTable } = await axios.post(`${URL}/tables`, { name: 'test', comment: 'foo' })
+    const { data: newTable } = await axios.post(`${URL}/tables`, { name: 'test', replica_identity: 'FULL', comment: 'foo' })
     assert.equal(`${newTable.schema}.${newTable.name}`, 'public.test')
+    assert.equal(newTable.replica_identity, 'FULL')
     assert.equal(newTable.comment, 'foo')
 
     const { data: tables } = await axios.get(`${URL}/tables`)
@@ -268,11 +269,13 @@ describe('/tables', async () => {
       name: 'test a',
       rls_enabled: true,
       rls_forced: true,
+      replica_identity: 'NOTHING',
       comment: 'foo',
     })
     assert.equal(updatedTable.name, `test a`)
     assert.equal(updatedTable.rls_enabled, true)
     assert.equal(updatedTable.rls_forced, true)
+    assert.equal(updatedTable.replica_identity, 'NOTHING')
     assert.equal(updatedTable.comment, 'foo')
     await axios.delete(`${URL}/tables/${newTable.id}`)
   })