Add Indexes to Schema API (#4240)

* Add Indexes to Schema API

* error handling

* ci errors

* postgres support

* full text compound indexes

* pg clean up

* get indexes on startup

* test compound index on startup

* add default _id to index, full Text index on startup

* lint

* fix test

src/Adapters/Storage/Postgres/PostgresStorageAdapter.js

@@ -98,10 +98,15 @@ const toParseSchema = (schema) => {
   if (schema.classLevelPermissions) {
     clps = {...emptyCLPS, ...schema.classLevelPermissions};
   }
+  let indexes = {};
+  if (schema.indexes) {
+    indexes = {...schema.indexes};
+  }
   return {
     className: schema.className,
     fields: schema.fields,
     classLevelPermissions: clps,
+    indexes,
   };
 }
 
@@ -608,12 +613,64 @@ export class PostgresStorageAdapter {
     });
   }
 
+  setIndexesWithSchemaFormat(className, submittedIndexes, existingIndexes = {}, fields, conn) {
+    conn = conn || this._client;
+    if (submittedIndexes === undefined) {
+      return Promise.resolve();
+    }
+    if (Object.keys(existingIndexes).length === 0) {
+      existingIndexes = { _id_: { _id: 1} };
+    }
+    const deletedIndexes = [];
+    const insertedIndexes = [];
+    Object.keys(submittedIndexes).forEach(name => {
+      const field = submittedIndexes[name];
+      if (existingIndexes[name] && field.__op !== 'Delete') {
+        throw new Parse.Error(Parse.Error.INVALID_QUERY, `Index ${name} exists, cannot update.`);
+      }
+      if (!existingIndexes[name] && field.__op === 'Delete') {
+        throw new Parse.Error(Parse.Error.INVALID_QUERY, `Index ${name} does not exist, cannot delete.`);
+      }
+      if (field.__op === 'Delete') {
+        deletedIndexes.push(name);
+        delete existingIndexes[name];
+      } else {
+        Object.keys(field).forEach(key => {
+          if (!fields.hasOwnProperty(key)) {
+            throw new Parse.Error(Parse.Error.INVALID_QUERY, `Field ${key} does not exist, cannot add index.`);
+          }
+        });
+        existingIndexes[name] = field;
+        insertedIndexes.push({
+          key: field,
+          name,
+        });
+      }
+    });
+    let insertPromise = Promise.resolve();
+    if (insertedIndexes.length > 0) {
+      insertPromise = this.createIndexes(className, insertedIndexes, conn);
+    }
+    let deletePromise = Promise.resolve();
+    if (deletedIndexes.length > 0) {
+      deletePromise = this.dropIndexes(className, deletedIndexes, conn);
+    }
+    return deletePromise
+      .then(() => insertPromise)
+      .then(() => this._ensureSchemaCollectionExists())
+      .then(() => {
+        const values = [className, 'schema', 'indexes', JSON.stringify(existingIndexes)]
+        return conn.none(`UPDATE "_SCHEMA" SET $2:name = json_object_set_key($2:name, $3::text, $4::jsonb) WHERE "className"=$1 `, values);
+      });
+  }
+
   createClass(className, schema) {
     return this._client.tx(t => {
       const q1 = this.createTable(className, schema, t);
       const q2 = t.none('INSERT INTO "_SCHEMA" ("className", "schema", "isParseClass") VALUES ($<className>, $<schema>, true)', { className, schema });
+      const q3 = this.setIndexesWithSchemaFormat(className, schema.indexes, {}, schema.fields, t);
 
-      return t.batch([q1, q2]);
+      return t.batch([q1, q2, q3]);
     })
       .then(() => {
         return toParseSchema(schema)
@@ -1548,6 +1605,25 @@ export class PostgresStorageAdapter {
         console.error(error);
       });
   }
+
+  createIndexes(className, indexes, conn) {
+    return (conn || this._client).tx(t => t.batch(indexes.map(i => {
+      return t.none('CREATE INDEX $1:name ON $2:name ($3:name)', [i.name, className, i.key]);
+    })));
+  }
+
+  dropIndexes(className, indexes, conn) {
+    return (conn || this._client).tx(t => t.batch(indexes.map(i => t.none('DROP INDEX $1:name', i))));
+  }
+
+  getIndexes(className) {
+    const qs = 'SELECT * FROM pg_indexes WHERE tablename = ${className}';
+    return this._client.any(qs, {className});
+  }
+
+  updateSchemaWithIndexes() {
+    return Promise.resolve();
+  }
 }
 
 function convertPolygonToSQL(polygon) {