fixes 707: use sql.unnest instead of sql.join for bulk insert (getodk#771)

We use a function called insertMany to do bulk inserting of certain things including Form Fields, Submission Attachments, Comments, Audits, and Client Audits.

insertMany was using sql.join internally, but broke down in extreme cases such as a very large form with 25K fields. In such a form, we store ~10 columns about each form field, so sql.join was forming a query with 250,000 parameters that was too much. Slonik throws 'Maximum call stack size exceeded' due to a bug gajus/slonik#344. Even after solving that bug, it is not possible to execute sql query with 250K parameters in Postgresql because database supports only 65535 parameters.

Right way to insert bulk rows in postgresql is to use unnest function, for e.g:

INSERT INTO table (col1, col2)
    SELECT * FROM
      UNNEST($1, $2) as t

where $1 is an array of values for col1 and $2 is an array of values for col2.

Slonik has sql.unnest to implement similar query, e.g., when inserting 25K of 10-column form fields, it will have 10 parameters where each is a 25K length array of every value for that one column. The one caveat of using sql.unnest is that it takes a second required argument to list out the types of each column.

Most of the code in this PR is about modifying our frames to also define field types on any frame that needs to use insertMany. If field types are missing and insertMany is called anyway, there is a 500 Problem explaining that.

To summarize
- insertMany changed to use sql.unnest instead of sql.join.
- Frames that use insertMany must now also specify a fieldTypes array for the properties in that frame.

Author: sadiqkhoja

lib/model/frame.js

@@ -8,7 +8,7 @@
 // except according to the terms contained in the LICENSE file.
 
 const { raw } = require('slonik-sql-tag-raw');
-const { pick, without } = require('ramda');
+const { pick, without, remove, indexOf } = require('ramda');
 const uuid = require('uuid/v4');
 const { pickAll, noargs } = require('../util/util');
 const Option = require('../util/option');
@@ -26,6 +26,10 @@ const table = (name, to) => (def) => {
   def.from = name;
   def.to = to || name.substring(0, name.length - 1); // take off the s
 };
+const fieldTypes = (types) => (def) => {
+  def.fieldTypes = types;
+};
+
 const from = (fro) => (def) => { def.from = fro; };
 const into = (to) => (def) => { def.to = to; };
 
@@ -66,6 +70,8 @@ class Frame {
       const Frame = class extends this { static get def() { return def; } };
       Frame.fieldlist = raw(def.fields.map((s) => `"${s}"`).join(','));
       Frame.insertfields = without([ 'id' ], def.fields);
+      const indexOfId = indexOf('id', def.fields);
+      Frame.insertFieldTypes = indexOfId > -1 ? remove(indexOf('id', def.fields), 1, def.fieldTypes ?? []) : def.fieldTypes;
       Frame.insertlist = raw(Frame.insertfields.map((s) => `"${s}"`).join(','));
       Frame.hasCreatedAt = def.fields.includes('createdAt');
       Frame.hasUpdatedAt = def.fields.includes('updatedAt');
@@ -156,5 +162,5 @@ Frame.def = { fields: [], readable: [], hasCreatedAt: false, hasUpdatedAt: false
 // util func to ensure the frame we fetched successfully joined to the intended def.
 const ensureDef = rejectIf(((f) => f.def.id == null), noargs(Problem.user.notFound));
 
-module.exports = { Frame, table, from, into, aux, species, embedded, readable, writable, ensureDef };
+module.exports = { Frame, table, fieldTypes, from, into, aux, species, embedded, readable, writable, ensureDef };
 

lib/model/frames.js

@@ -8,7 +8,7 @@
 // except according to the terms contained in the LICENSE file.
 
 const { sql } = require('slonik');
-const { Frame, table, readable, writable, aux, species, embedded } = require('./frame');
+const { Frame, table, readable, writable, aux, species, embedded, fieldTypes } = require('./frame');
 const { isBlank } = require('../util/util');
 const Option = require('../util/option');
 
@@ -45,6 +45,7 @@ class Audit extends Frame.define(
   'details',    readable,               'loggedAt',     readable,
   'notes',      readable,
   'claimed', 'processed', 'lastFailure', 'failures',
+  fieldTypes(['int4', 'int4', 'text', 'varchar', 'jsonb', 'timestamptz', 'text', 'timestamptz', 'timestamptz', 'timestamptz', 'int4']),
   embedded('actor'), embedded('actee')
 ) {
   // TODO: sort of duplicative of Audits.log
@@ -71,7 +72,12 @@ class Audit extends Frame.define(
 const { Blob } = require('./frames/blob');
 
 const { headers } = require('../data/client-audits');
-const ClientAudit = Frame.define(table('client_audits'), 'blobId', 'remainder', ...headers);
+const ClientAudit = Frame.define(
+  table('client_audits'),
+  'blobId',       'remainder',
+  ...headers,
+  fieldTypes(['int4', 'jsonb', ...headers.map(() => 'text')])
+);
 
 const Comment = Frame.define(
   table('comments'),
@@ -103,7 +109,8 @@ Form.Attachment = class extends Frame.define(
   'formId',                             'formDefId',
   'blobId',                             'datasetId',
   'name',         readable,             'type',       readable,
-  'updatedAt',    readable
+  'updatedAt',    readable,
+  fieldTypes(['int4', 'int4', 'int4', 'int4', 'text', 'text', 'timestamptz'])
 ) {
   forApi() {
     const data = { name: this.name, type: this.type, exists: (this.blobId != null || this.datasetId != null),  blobExists: this.blobId != null, datasetExists: this.datasetId != null };
@@ -171,7 +178,8 @@ class Session extends Frame.define(
 const { Submission } = require('./frames/submission');
 Submission.Attachment = class extends Frame.define(
   table('submission_attachments'),
-  'submissionDefId', 'blobId', 'name', 'index', 'isClientAudit'
+  'submissionDefId', 'blobId', 'name', 'index', 'isClientAudit',
+  fieldTypes(['int4', 'int4', 'text', 'int4', 'bool'])
 ) {
   forApi() {
     return { name: this.name, exists: (this.blobId != null) };

lib/model/frames/form.js

@@ -25,7 +25,7 @@
 // As with Users/Actors, this information is merged together upon return via the
 // API. This is partly for legacy reasons: Forms and FormsDef did not used to
 
-const { Frame, table, readable, writable, into, embedded } = require('../frame');
+const { Frame, table, readable, writable, into, embedded, fieldTypes } = require('../frame');
 const { injectPublicKey, addVersionSuffix } = require('../../data/schema');
 const { Key } = require('./key');
 const { md5sum, shasum, sha256sum, digestWith, generateVersionSuffix } = require('../../util/crypto');
@@ -184,15 +184,17 @@ Form.Field = class extends Frame.define(
   'formId',                             'schemaId',
   'path',       readable,               'name',         readable,
   'type',       readable,               'binary',       readable,
-  'order',                              'selectMultiple', readable
+  'order',                              'selectMultiple', readable,
+  fieldTypes(['int4', 'int4', 'text', 'text', 'varchar', 'bool', 'int4', 'bool' ])
 ) {
   isStructural() { return (this.type === 'repeat') || (this.type === 'structure'); }
 };
 
 Form.FieldValue = Frame.define(
   table('form_field_values'),
   'formId',                             'submissionDefId',
-  'path',                               'value'
+  'path',                               'value',
+  fieldTypes(['int4', 'int4', 'text', 'text'])
 );
 
 

lib/util/db.js

@@ -8,7 +8,7 @@
 // except according to the terms contained in the LICENSE file.
 
 const { inspect } = require('util');
-const { mergeRight, pick, always } = require('ramda');
+const { mergeRight, pick, always, without, remove, indexOf } = require('ramda');
 const { sql } = require('slonik');
 const { raw } = require('slonik-sql-tag-raw');
 const { reject } = require('./promise');
@@ -151,15 +151,35 @@ values (${sql.join(keys.map(_assign(obj)), sql`,`)})
 returning *`;
 };
 
+// Arguments:
+// objs: An array of Frames to be inserted in the database table.
+//       Frame should defined fieldTypes for this function to work
+//       because Slonik's sql.unnest function has a mandatory columnType argument
 const insertMany = (objs) => {
   if (objs.length === 0) return sql`select true`;
   const Type = objs[0].constructor;
+  if (!Type.def.fieldTypes) throw Problem.internal.fieldTypesNotDefined(Type.from);
+
+  let columns; let rows; let columnTypes; let
+    selectExp;
+
+  // we need to set clock_timestamp if there's createdAt column
+  // Slonik doesn't support setting sql identitfier for sql.unnest yet
+  if (Type.hasCreatedAt) {
+    columns = sql`"createdAt", ${raw(without(['createdAt'], Type.insertfields).map((s) => `"${s}"`).join(','))}`;
+    rows = objs.map(obj => without(['createdAt'], Type.insertfields).map(_assign(obj)));
+    columnTypes = remove(indexOf(['createdAt'], Type.insertfields), 1, Type.insertFieldTypes);
+    selectExp = sql`clock_timestamp(), *`;
+  } else {
+    columns = Type.insertlist;
+    rows = objs.map(obj => Type.insertfields.map(_assign(obj)));
+    columnTypes = Type.insertFieldTypes;
+    selectExp = sql`*`;
+  }
+
   return sql`
-insert into ${raw(Type.table)} (${Type.insertlist})
-values ${sql.join(
-    objs.map((obj) => sql`(${sql.join(Type.insertfields.map(_assign(obj)), sql`,`)})`),
-    sql`,`
-  )}`;
+  INSERT INTO ${raw(Type.table)} (${columns})
+  SELECT ${selectExp} FROM ${sql.unnest(rows, columnTypes)} AS t`;
 };
 
 // generic update utility

lib/util/problem.js

@@ -226,7 +226,9 @@ const problems = {
     enketoUnexpectedResponse: problem(500.4, () => 'The Enketo service returned an unexpected response.'),
 
     // returned if ODK Analytics returns an unexpected response.
-    analyticsUnexpectedResponse: problem(500.5, () => 'The Analytics service returned an unexpected response.')
+    analyticsUnexpectedResponse: problem(500.5, () => 'The Analytics service returned an unexpected response.'),
+
+    fieldTypesNotDefined: problem(500.6, (frame) => `fieldTypes are not defined on the ${frame} Frame, please define them to use insertMany.`)
   }
 };
 

test/integration/other/encryption.js

@@ -143,11 +143,13 @@ describe('managed encryption', () => {
       return Submission.fromXml(xml)
         .then((partial) => hijacked.SubmissionAttachments.create(partial, {}, []))
         .then(() => {
+          // values to sql query of insertMany are passed as array of arrays
           results[0].values.should.eql([
-            null, null, 'zulu.file', 0, false,
-            null, null, 'alpha.file', 1, false,
-            null, null, 'bravo.file', 2, false,
-            null, null, 'submission.xml.enc', 3, null
+            [null, null, null, null],
+            [null, null, null, null],
+            ['zulu.file', 'alpha.file', 'bravo.file', 'submission.xml.enc'],
+            [0, 1, 2, 3],
+            [false, false, false, null]
           ]);
         });
     }));

test/unit/util/db.js

@@ -1,5 +1,6 @@
 const appRoot = require('app-root-path');
 const { sql } = require('slonik');
+const { fieldTypes } = require('../../../lib/model/frame');
 // eslint-disable-next-line import/no-dynamic-require
 const { Frame, table, into } = require(appRoot + '/lib/model/frame');
 // eslint-disable-next-line import/no-dynamic-require
@@ -312,23 +313,38 @@ returning *`);
 
   describe('insertMany', () => {
     const { insertMany } = util;
-    const T = Frame.define(table('dogs'), 'x', 'y');
+    const T = Frame.define(table('dogs'), 'x', 'y', fieldTypes(['text', 'text']));
 
     it('should do nothing if given no data', () => {
       insertMany([]).should.eql(sql`select true`);
     });
 
     it('should insert all data', () => {
-      insertMany([ new T({ x: 2 }), new T({ y: 3 }) ]).should.eql(sql`
-insert into dogs ("x","y")
-values (${2},${null}),(${null},${3})`);
+      const query = insertMany([ new T({ x: 2 }), new T({ y: 3 }) ]);
+      query.sql.should.be.eql(`
+  INSERT INTO dogs ("x","y")
+  SELECT * FROM unnest($1::"text"[], $2::"text"[]) AS t`);
+      query.values.should.be.eql([
+        [2, null],
+        [null, 3]
+      ]);
     });
 
     it('should insert createdAt and strange values', () => {
+      const U = Frame.define(table('dogs'), 'x', 'createdAt', fieldTypes(['timestamptz', 'timestamptz']));
+      const query = insertMany([ new U({ x: new Date('2000-01-01') }), new U() ]);
+      query.sql.should.be.eql(`
+  INSERT INTO dogs ("createdAt", "x")
+  SELECT clock_timestamp(), * FROM unnest($1::"timestamptz"[]) AS t`);
+      query.values.should.be.eql([
+        ['2000-01-01T00:00:00.000Z', null]
+      ]);
+    });
+
+    it('should throw fieldTypesNotDefined', () => {
       const U = Frame.define(table('dogs'), 'x', 'createdAt');
-      insertMany([ new U({ x: new Date('2000-01-01') }), new U() ]).should.eql(sql`
-insert into dogs ("x","createdAt")
-values (${'2000-01-01T00:00:00.000Z'},${sql`clock_timestamp()`}),(${null},${sql`clock_timestamp()`})`);
+      (() => insertMany([ new U({ x: new Date('2000-01-01') }), new U() ]))
+        .should.throw('fieldTypes are not defined on the dogs Frame, please define them to use insertMany.');
     });
   });