diff --git a/spec/RedisCacheAdapter.spec.js b/spec/RedisCacheAdapter.spec.js
index 4f90022870..3b1c78af04 100644
--- a/spec/RedisCacheAdapter.spec.js
+++ b/spec/RedisCacheAdapter.spec.js
@@ -188,7 +188,7 @@ describe_only(() => {
     const object = new TestObject();
     object.set('foo', 'bar');
     await object.save();
-    expect(getSpy.calls.count()).toBe(2);
+    expect(getSpy.calls.count()).toBe(3);
     expect(putSpy.calls.count()).toBe(2);
   });
 
@@ -201,7 +201,7 @@ describe_only(() => {
       booleanField: true,
     });
     await container.save();
-    expect(getSpy.calls.count()).toBe(2);
+    expect(getSpy.calls.count()).toBe(3);
     expect(putSpy.calls.count()).toBe(2);
   });
 
@@ -215,7 +215,7 @@ describe_only(() => {
 
     object.set('foo', 'barz');
     await object.save();
-    expect(getSpy.calls.count()).toBe(2);
+    expect(getSpy.calls.count()).toBe(3);
     expect(putSpy.calls.count()).toBe(0);
   });
 
@@ -233,7 +233,7 @@ describe_only(() => {
       objects.push(object);
     }
     await Parse.Object.saveAll(objects);
-    expect(getSpy.calls.count()).toBe(11);
+    expect(getSpy.calls.count()).toBe(21);
     expect(putSpy.calls.count()).toBe(10);
 
     getSpy.calls.reset();
@@ -258,7 +258,7 @@ describe_only(() => {
       objects.push(object);
     }
     await Parse.Object.saveAll(objects, { batchSize: 5 });
-    expect(getSpy.calls.count()).toBe(12);
+    expect(getSpy.calls.count()).toBe(22);
     expect(putSpy.calls.count()).toBe(5);
 
     getSpy.calls.reset();
@@ -279,7 +279,7 @@ describe_only(() => {
 
     object.set('new', 'barz');
     await object.save();
-    expect(getSpy.calls.count()).toBe(2);
+    expect(getSpy.calls.count()).toBe(3);
     expect(putSpy.calls.count()).toBe(1);
   });
 
@@ -299,7 +299,7 @@ describe_only(() => {
       booleanField: true,
     });
     await object.save();
-    expect(getSpy.calls.count()).toBe(2);
+    expect(getSpy.calls.count()).toBe(3);
     expect(putSpy.calls.count()).toBe(1);
   });
 
@@ -309,7 +309,7 @@ describe_only(() => {
     user.setPassword('testing');
     await user.signUp();
 
-    expect(getSpy.calls.count()).toBe(6);
+    expect(getSpy.calls.count()).toBe(8);
     expect(putSpy.calls.count()).toBe(1);
   });
 
@@ -326,7 +326,7 @@ describe_only(() => {
 
     object.set('foo', 'bar');
     await object.save();
-    expect(getSpy.calls.count()).toBe(3);
+    expect(getSpy.calls.count()).toBe(4);
     expect(putSpy.calls.count()).toBe(1);
 
     getSpy.calls.reset();
diff --git a/spec/Schema.spec.js b/spec/Schema.spec.js
index aaede67e17..cf73927274 100644
--- a/spec/Schema.spec.js
+++ b/spec/Schema.spec.js
@@ -504,7 +504,7 @@ describe('SchemaController', () => {
       schema
         .addClassIfNotExists('_InvalidName', { foo: { type: 'String' } })
         .catch(error => {
-          expect(error.error).toEqual(
+          expect(error.message).toEqual(
             'Invalid classname: _InvalidName, classnames can only have alphanumeric characters and _, and must start with an alpha character '
           );
           done();
@@ -522,7 +522,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INVALID_KEY_NAME);
-        expect(error.error).toEqual('invalid field name: 0InvalidName');
+        expect(error.message).toEqual('invalid field name: 0InvalidName');
         done();
       });
   });
@@ -535,7 +535,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(136);
-        expect(error.error).toEqual('field objectId cannot be added');
+        expect(error.message).toEqual('field objectId cannot be added');
         done();
       });
   });
@@ -550,7 +550,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(136);
-        expect(error.error).toEqual('field localeIdentifier cannot be added');
+        expect(error.message).toEqual('field localeIdentifier cannot be added');
         done();
       });
   });
@@ -565,7 +565,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INVALID_JSON);
-        expect(error.error).toEqual('invalid JSON');
+        expect(error.message).toEqual('invalid JSON');
         done();
       });
   });
@@ -580,7 +580,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(135);
-        expect(error.error).toEqual('type Pointer needs a class name');
+        expect(error.message).toEqual('type Pointer needs a class name');
         done();
       });
   });
@@ -595,7 +595,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INVALID_JSON);
-        expect(error.error).toEqual('invalid JSON');
+        expect(error.message).toEqual('invalid JSON');
         done();
       });
   });
@@ -610,7 +610,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(135);
-        expect(error.error).toEqual('type Relation needs a class name');
+        expect(error.message).toEqual('type Relation needs a class name');
         done();
       });
   });
@@ -625,7 +625,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INVALID_JSON);
-        expect(error.error).toEqual('invalid JSON');
+        expect(error.message).toEqual('invalid JSON');
         done();
       });
   });
@@ -640,7 +640,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INVALID_CLASS_NAME);
-        expect(error.error).toEqual(
+        expect(error.message).toEqual(
           'Invalid classname: not a valid class name, classnames can only have alphanumeric characters and _, and must start with an alpha character '
         );
         done();
@@ -657,7 +657,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INVALID_CLASS_NAME);
-        expect(error.error).toEqual(
+        expect(error.message).toEqual(
           'Invalid classname: not a valid class name, classnames can only have alphanumeric characters and _, and must start with an alpha character '
         );
         done();
@@ -674,7 +674,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INCORRECT_TYPE);
-        expect(error.error).toEqual('invalid field type: Unknown');
+        expect(error.message).toEqual('invalid field type: Unknown');
         done();
       });
   });
@@ -929,7 +929,7 @@ describe('SchemaController', () => {
       )
       .catch(error => {
         expect(error.code).toEqual(Parse.Error.INCORRECT_TYPE);
-        expect(error.error).toEqual(
+        expect(error.message).toEqual(
           'currently, only one GeoPoint field may exist in an object. Adding geo2 when geo1 already exists.'
         );
         done();
diff --git a/spec/schemas.spec.js b/spec/schemas.spec.js
index 20ca8e5b83..e0969aad37 100644
--- a/spec/schemas.spec.js
+++ b/spec/schemas.spec.js
@@ -406,6 +406,124 @@ describe('schemas', () => {
     });
   });
 
+  it('responds with all fields and options when you create a class with field options', done => {
+    request({
+      url: 'http://localhost:8378/1/schemas',
+      method: 'POST',
+      headers: masterKeyHeaders,
+      json: true,
+      body: {
+        className: 'NewClassWithOptions',
+        fields: {
+          foo1: { type: 'Number' },
+          foo2: { type: 'Number', required: true, defaultValue: 10 },
+          foo3: {
+            type: 'String',
+            required: false,
+            defaultValue: 'some string',
+          },
+          foo4: { type: 'Date', required: true },
+          foo5: { type: 'Number', defaultValue: 5 },
+          ptr: { type: 'Pointer', targetClass: 'SomeClass', required: false },
+        },
+      },
+    }).then(async response => {
+      expect(response.data).toEqual({
+        className: 'NewClassWithOptions',
+        fields: {
+          ACL: { type: 'ACL' },
+          createdAt: { type: 'Date' },
+          updatedAt: { type: 'Date' },
+          objectId: { type: 'String' },
+          foo1: { type: 'Number' },
+          foo2: { type: 'Number', required: true, defaultValue: 10 },
+          foo3: {
+            type: 'String',
+            required: false,
+            defaultValue: 'some string',
+          },
+          foo4: { type: 'Date', required: true },
+          foo5: { type: 'Number', defaultValue: 5 },
+          ptr: { type: 'Pointer', targetClass: 'SomeClass', required: false },
+        },
+        classLevelPermissions: defaultClassLevelPermissions,
+      });
+      const obj = new Parse.Object('NewClassWithOptions');
+      try {
+        await obj.save();
+        fail('should fail');
+      } catch (e) {
+        expect(e.code).toEqual(142);
+      }
+      const date = new Date();
+      obj.set('foo4', date);
+      await obj.save();
+      expect(obj.get('foo1')).toBeUndefined();
+      expect(obj.get('foo2')).toEqual(10);
+      expect(obj.get('foo3')).toEqual('some string');
+      expect(obj.get('foo4')).toEqual(date);
+      expect(obj.get('foo5')).toEqual(5);
+      expect(obj.get('ptr')).toBeUndefined();
+      done();
+    });
+  });
+
+  it('validated the data type of default values when creating a new class', async () => {
+    try {
+      await request({
+        url: 'http://localhost:8378/1/schemas',
+        method: 'POST',
+        headers: masterKeyHeaders,
+        json: true,
+        body: {
+          className: 'NewClassWithValidation',
+          fields: {
+            foo: { type: 'String', defaultValue: 10 },
+          },
+        },
+      });
+      fail('should fail');
+    } catch (e) {
+      expect(e.data.error).toEqual(
+        'schema mismatch for NewClassWithValidation.foo default value; expected String but got Number'
+      );
+    }
+  });
+
+  it('validated the data type of default values when adding new fields', async () => {
+    try {
+      await request({
+        url: 'http://localhost:8378/1/schemas',
+        method: 'POST',
+        headers: masterKeyHeaders,
+        json: true,
+        body: {
+          className: 'NewClassWithValidation',
+          fields: {
+            foo: { type: 'String', defaultValue: 'some value' },
+          },
+        },
+      });
+      await request({
+        url: 'http://localhost:8378/1/schemas/NewClassWithValidation',
+        method: 'PUT',
+        headers: masterKeyHeaders,
+        json: true,
+        body: {
+          className: 'NewClassWithValidation',
+          fields: {
+            foo2: { type: 'String', defaultValue: 10 },
+          },
+        },
+      });
+      fail('should fail');
+    } catch (e) {
+      expect(e.data.error).toEqual(
+        'schema mismatch for NewClassWithValidation.foo2 default value; expected String but got Number'
+      );
+    }
+  });
+
   it('responds with all fields when getting incomplete schema', done => {
     config.database
       .loadSchema()
@@ -740,6 +858,319 @@ describe('schemas', () => {
     });
   });
 
+  it('lets you add fields with options', done => {
+    request({
+      url: 'http://localhost:8378/1/schemas/NewClass',
+      method: 'POST',
+      headers: masterKeyHeaders,
+      json: true,
+      body: {},
+    }).then(() => {
+      request({
+        method: 'PUT',
+        url: 'http://localhost:8378/1/schemas/NewClass',
+        headers: masterKeyHeaders,
+        json: true,
+        body: {
+          fields: {
+            newField: {
+              type: 'String',
+              required: true,
+              defaultValue: 'some value',
+            },
+          },
+        },
+      }).then(response => {
+        expect(
+          dd(response.data, {
+            className: 'NewClass',
+            fields: {
+              ACL: { type: 'ACL' },
+              createdAt: { type: 'Date' },
+              objectId: { type: 'String' },
+              updatedAt: { type: 'Date' },
+              newField: {
+                type: 'String',
+                required: true,
+                defaultValue: 'some value',
+              },
+            },
+            classLevelPermissions: defaultClassLevelPermissions,
+          })
+        ).toEqual(undefined);
+        request({
+          url: 'http://localhost:8378/1/schemas/NewClass',
+          headers: masterKeyHeaders,
+          json: true,
+        }).then(response => {
+          expect(response.data).toEqual({
+            className: 'NewClass',
+            fields: {
+              ACL: { type: 'ACL' },
+              createdAt: { type: 'Date' },
+              updatedAt: { type: 'Date' },
+              objectId: { type: 'String' },
+              newField: {
+                type: 'String',
+                required: true,
+                defaultValue: 'some value',
+              },
+            },
+            classLevelPermissions: defaultClassLevelPermissions,
+          });
+          done();
+        });
+      });
+    });
+  });
+
+  it('should validate required fields', done => {
+    request({
+      url: 'http://localhost:8378/1/schemas/NewClass',
+      method: 'POST',
+      headers: masterKeyHeaders,
+      json: true,
+      body: {},
+    }).then(() => {
+      request({
+        method: 'PUT',
+        url: 'http://localhost:8378/1/schemas/NewClass',
+        headers: masterKeyHeaders,
+        json: true,
+        body: {
+          fields: {
+            newRequiredField: {
+              type: 'String',
+              required: true,
+            },
+            newRequiredFieldWithDefaultValue: {
+              type: 'String',
+              required: true,
+              defaultValue: 'some value',
+            },
+            newNotRequiredField: {
+              type: 'String',
+              required: false,
+            },
+            newNotRequiredFieldWithDefaultValue: {
+              type: 'String',
+              required: false,
+              defaultValue: 'some value',
+            },
+            newRegularFieldWithDefaultValue: {
+              type: 'String',
+              defaultValue: 'some value',
+            },
+            newRegularField: {
+              type: 'String',
+            },
+          },
+        },
+      }).then(async () => {
+        let obj = new Parse.Object('NewClass');
+        try {
+          await obj.save();
+          fail('Should fail');
+        } catch (e) {
+          expect(e.code).toEqual(142);
+          expect(e.message).toEqual('newRequiredField is required');
+        }
+        obj.set('newRequiredField', 'some value');
+        await obj.save();
+        expect(obj.get('newRequiredField')).toEqual('some value');
+        expect(obj.get('newRequiredFieldWithDefaultValue')).toEqual(
+          'some value'
+        );
+        expect(obj.get('newNotRequiredField')).toEqual(undefined);
+        expect(obj.get('newNotRequiredFieldWithDefaultValue')).toEqual(
+          'some value'
+        );
+        expect(obj.get('newRegularField')).toEqual(undefined);
+        obj.set('newRequiredField', null);
+        try {
+          await obj.save();
+          fail('Should fail');
+        } catch (e) {
+          expect(e.code).toEqual(142);
+          expect(e.message).toEqual('newRequiredField is required');
+        }
+        obj.unset('newRequiredField');
+        try {
+          await obj.save();
+          fail('Should fail');
+        } catch (e) {
+          expect(e.code).toEqual(142);
+          expect(e.message).toEqual('newRequiredField is required');
+        }
+        obj.set('newRequiredField', 'some value2');
+        await obj.save();
+        expect(obj.get('newRequiredField')).toEqual('some value2');
+        expect(obj.get('newRequiredFieldWithDefaultValue')).toEqual(
+          'some value'
+        );
+        expect(obj.get('newNotRequiredField')).toEqual(undefined);
+        expect(obj.get('newNotRequiredFieldWithDefaultValue')).toEqual(
+          'some value'
+        );
+        expect(obj.get('newRegularField')).toEqual(undefined);
+        obj.unset('newRequiredFieldWithDefaultValue');
+        try {
+          await obj.save();
+          fail('Should fail');
+        } catch (e) {
+          expect(e.code).toEqual(142);
+          expect(e.message).toEqual(
+            'newRequiredFieldWithDefaultValue is required'
+          );
+        }
+        obj.set('newRequiredFieldWithDefaultValue', '');
+        try {
+          await obj.save();
+          fail('Should fail');
+        } catch (e) {
+          expect(e.code).toEqual(142);
+          expect(e.message).toEqual(
+            'newRequiredFieldWithDefaultValue is required'
+          );
+        }
+        obj.set('newRequiredFieldWithDefaultValue', 'some value2');
+        obj.set('newNotRequiredField', '');
+        obj.set('newNotRequiredFieldWithDefaultValue', null);
+        obj.unset('newRegularField');
+        await obj.save();
+        expect(obj.get('newRequiredField')).toEqual('some value2');
+        expect(obj.get('newRequiredFieldWithDefaultValue')).toEqual(
+          'some value2'
+        );
+        expect(obj.get('newNotRequiredField')).toEqual('');
+        expect(obj.get('newNotRequiredFieldWithDefaultValue')).toEqual(null);
+        expect(obj.get('newRegularField')).toEqual(undefined);
+        obj = new Parse.Object('NewClass');
+        obj.set('newRequiredField', 'some value3');
+        obj.set('newRequiredFieldWithDefaultValue', 'some value3');
+        obj.set('newNotRequiredField', 'some value3');
+        obj.set('newNotRequiredFieldWithDefaultValue', 'some value3');
+        obj.set('newRegularField', 'some value3');
+        await obj.save();
+        expect(obj.get('newRequiredField')).toEqual('some value3');
+        expect(obj.get('newRequiredFieldWithDefaultValue')).toEqual(
+          'some value3'
+        );
+        expect(obj.get('newNotRequiredField')).toEqual('some value3');
+        expect(obj.get('newNotRequiredFieldWithDefaultValue')).toEqual(
+          'some value3'
+        );
+        expect(obj.get('newRegularField')).toEqual('some value3');
+        done();
+      });
+    });
+  });
+
+  it('should validate required fields and set default values after before save trigger', async () => {
+    await request({
+      url: 'http://localhost:8378/1/schemas',
+      method: 'POST',
+      headers: masterKeyHeaders,
+      json: true,
+      body: {
+        className: 'NewClassForBeforeSaveTest',
+        fields: {
+          foo1: { type: 'String' },
+          foo2: { type: 'String', required: true },
+          foo3: {
+            type: 'String',
+            required: true,
+            defaultValue: 'some default value 3',
+          },
+          foo4: { type: 'String', defaultValue: 'some default value 4' },
+        },
+      },
+    });
+
+    Parse.Cloud.beforeSave('NewClassForBeforeSaveTest', req => {
+      req.object.set('foo1', 'some value 1');
+      req.object.set('foo2', 'some value 2');
+      req.object.set('foo3', 'some value 3');
+      req.object.set('foo4', 'some value 4');
+    });
+
+    let obj = new Parse.Object('NewClassForBeforeSaveTest');
+    await obj.save();
+
+    expect(obj.get('foo1')).toEqual('some value 1');
+    expect(obj.get('foo2')).toEqual('some value 2');
+    expect(obj.get('foo3')).toEqual('some value 3');
+    expect(obj.get('foo4')).toEqual('some value 4');
+
+    Parse.Cloud.beforeSave('NewClassForBeforeSaveTest', req => {
+      req.object.set('foo1', 'some value 1');
+      req.object.set('foo2', 'some value 2');
+    });
+
+    obj = new Parse.Object('NewClassForBeforeSaveTest');
+    await obj.save();
+
+    expect(obj.get('foo1')).toEqual('some value 1');
+    expect(obj.get('foo2')).toEqual('some value 2');
+    expect(obj.get('foo3')).toEqual('some default value 3');
+    expect(obj.get('foo4')).toEqual('some default value 4');
+
+    Parse.Cloud.beforeSave('NewClassForBeforeSaveTest', req => {
+      req.object.set('foo1', 'some value 1');
+      req.object.set('foo2', 'some value 2');
+      req.object.set('foo3', undefined);
+      req.object.unset('foo4');
+    });
+
+    obj = new Parse.Object('NewClassForBeforeSaveTest');
+    obj.set('foo3', 'some value 3');
+    obj.set('foo4', 'some value 4');
+    await obj.save();
+
+    expect(obj.get('foo1')).toEqual('some value 1');
+    expect(obj.get('foo2')).toEqual('some value 2');
+    expect(obj.get('foo3')).toEqual('some default value 3');
+    expect(obj.get('foo4')).toEqual('some default value 4');
+
+    Parse.Cloud.beforeSave('NewClassForBeforeSaveTest', req => {
+      req.object.set('foo1', 'some value 1');
+      req.object.set('foo2', undefined);
+      req.object.set('foo3', undefined);
+      req.object.unset('foo4');
+    });
+
+    obj = new Parse.Object('NewClassForBeforeSaveTest');
+    obj.set('foo2', 'some value 2');
+    obj.set('foo3', 'some value 3');
+    obj.set('foo4', 'some value 4');
+
+    try {
+      await obj.save();
+      fail('should fail');
+    } catch (e) {
+      expect(e.message).toEqual('foo2 is required');
+    }
+
+    Parse.Cloud.beforeSave('NewClassForBeforeSaveTest', req => {
+      req.object.set('foo1', 'some value 1');
+      req.object.unset('foo2');
+      req.object.set('foo3', undefined);
+      req.object.unset('foo4');
+    });
+
+    obj = new Parse.Object('NewClassForBeforeSaveTest');
+    obj.set('foo2', 'some value 2');
+    obj.set('foo3', 'some value 3');
+    obj.set('foo4', 'some value 4');
+
+    try {
+      await obj.save();
+      fail('should fail');
+    } catch (e) {
+      expect(e.message).toEqual('foo2 is required');
+    }
+  });
+
   it('lets you add fields to system schema', done => {
     request({
       method: 'POST',
diff --git a/src/Adapters/Storage/Mongo/MongoSchemaCollection.js b/src/Adapters/Storage/Mongo/MongoSchemaCollection.js
index 632a3f500e..46ba3e8165 100644
--- a/src/Adapters/Storage/Mongo/MongoSchemaCollection.js
+++ b/src/Adapters/Storage/Mongo/MongoSchemaCollection.js
@@ -46,6 +46,17 @@ function mongoSchemaFieldsToParseSchemaFields(schema) {
   );
   var response = fieldNames.reduce((obj, fieldName) => {
     obj[fieldName] = mongoFieldToParseSchemaField(schema[fieldName]);
+    if (
+      schema._metadata &&
+      schema._metadata.fields_options &&
+      schema._metadata.fields_options[fieldName]
+    ) {
+      obj[fieldName] = Object.assign(
+        {},
+        obj[fieldName],
+        schema._metadata.fields_options[fieldName]
+      );
+    }
     return obj;
   }, {});
   response.ACL = { type: 'ACL' };
@@ -212,7 +223,7 @@ class MongoSchemaCollection {
   // Support additional types that Mongo doesn't, like Money, or something.
 
   // TODO: don't spend an extra query on finding the schema if the type we are trying to add isn't a GeoPoint.
-  addFieldIfNotExists(className: string, fieldName: string, type: string) {
+  addFieldIfNotExists(className: string, fieldName: string, fieldType: string) {
     return this._fetchOneSchemaFrom_SCHEMA(className)
       .then(
         schema => {
@@ -221,7 +232,7 @@ class MongoSchemaCollection {
             return;
           }
           // The schema exists. Check for existing GeoPoints.
-          if (type.type === 'GeoPoint') {
+          if (fieldType.type === 'GeoPoint') {
             // Make sure there are not other geopoint fields
             if (
               Object.keys(schema.fields).some(
@@ -247,13 +258,37 @@ class MongoSchemaCollection {
         }
       )
       .then(() => {
+        const { type, targetClass, ...fieldOptions } = fieldType;
         // We use $exists and $set to avoid overwriting the field type if it
         // already exists. (it could have added inbetween the last query and the update)
-        return this.upsertSchema(
-          className,
-          { [fieldName]: { $exists: false } },
-          { $set: { [fieldName]: parseFieldTypeToMongoFieldType(type) } }
-        );
+        if (fieldOptions && Object.keys(fieldOptions).length > 0) {
+          return this.upsertSchema(
+            className,
+            { [fieldName]: { $exists: false } },
+            {
+              $set: {
+                [fieldName]: parseFieldTypeToMongoFieldType({
+                  type,
+                  targetClass,
+                }),
+                [`_metadata.fields_options.${fieldName}`]: fieldOptions,
+              },
+            }
+          );
+        } else {
+          return this.upsertSchema(
+            className,
+            { [fieldName]: { $exists: false } },
+            {
+              $set: {
+                [fieldName]: parseFieldTypeToMongoFieldType({
+                  type,
+                  targetClass,
+                }),
+              },
+            }
+          );
+        }
       });
   }
 }
diff --git a/src/Adapters/Storage/Mongo/MongoStorageAdapter.js b/src/Adapters/Storage/Mongo/MongoStorageAdapter.js
index 33282ce2b9..14128bbc54 100644
--- a/src/Adapters/Storage/Mongo/MongoStorageAdapter.js
+++ b/src/Adapters/Storage/Mongo/MongoStorageAdapter.js
@@ -84,9 +84,19 @@ const mongoSchemaFromFieldsAndClassNameAndCLP = (
   };
 
   for (const fieldName in fields) {
+    const { type, targetClass, ...fieldOptions } = fields[fieldName];
     mongoObject[
       fieldName
-    ] = MongoSchemaCollection.parseFieldTypeToMongoFieldType(fields[fieldName]);
+    ] = MongoSchemaCollection.parseFieldTypeToMongoFieldType({
+      type,
+      targetClass,
+    });
+    if (fieldOptions && Object.keys(fieldOptions).length > 0) {
+      mongoObject._metadata = mongoObject._metadata || {};
+      mongoObject._metadata.fields_options =
+        mongoObject._metadata.fields_options || {};
+      mongoObject._metadata.fields_options[fieldName] = fieldOptions;
+    }
   }
 
   if (typeof classLevelPermissions !== 'undefined') {
@@ -425,6 +435,7 @@ export class MongoStorageAdapter implements StorageAdapter {
     const schemaUpdate = { $unset: {} };
     fieldNames.forEach(name => {
       schemaUpdate['$unset'][name] = null;
+      schemaUpdate['$unset'][`_metadata.fields_options.${name}`] = null;
     });
 
     return this._adaptiveCollection(className)
diff --git a/src/Controllers/SchemaController.js b/src/Controllers/SchemaController.js
index e3dfc040b9..d6a95a5034 100644
--- a/src/Controllers/SchemaController.js
+++ b/src/Controllers/SchemaController.js
@@ -654,6 +654,13 @@ export default class SchemaController {
       classLevelPermissions
     );
     if (validationError) {
+      if (validationError instanceof Parse.Error) {
+        return Promise.reject(validationError);
+      } else if (validationError.code && validationError.error) {
+        return Promise.reject(
+          new Parse.Error(validationError.code, validationError.error)
+        );
+      }
       return Promise.reject(validationError);
     }
 
@@ -874,8 +881,23 @@ export default class SchemaController {
             error: 'field ' + fieldName + ' cannot be added',
           };
         }
-        const error = fieldTypeIsInvalid(fields[fieldName]);
+        const type = fields[fieldName];
+        const error = fieldTypeIsInvalid(type);
         if (error) return { code: error.code, error: error.message };
+        if (type.defaultValue !== undefined) {
+          let defaultValueType = getType(type.defaultValue);
+          if (typeof defaultValueType === 'string') {
+            defaultValueType = { type: defaultValueType };
+          }
+          if (!dbTypeMatchesObjectType(type, defaultValueType)) {
+            return {
+              code: Parse.Error.INCORRECT_TYPE,
+              error: `schema mismatch for ${className}.${fieldName} default value; expected ${typeToString(
+                type
+              )} but got ${typeToString(defaultValueType)}`,
+            };
+          }
+        }
       }
     }
 
@@ -937,7 +959,22 @@ export default class SchemaController {
 
     const expectedType = this.getExpectedType(className, fieldName);
     if (typeof type === 'string') {
-      type = { type };
+      type = ({ type }: SchemaField);
+    }
+
+    if (type.defaultValue !== undefined) {
+      let defaultValueType = getType(type.defaultValue);
+      if (typeof defaultValueType === 'string') {
+        defaultValueType = { type: defaultValueType };
+      }
+      if (!dbTypeMatchesObjectType(type, defaultValueType)) {
+        throw new Parse.Error(
+          Parse.Error.INCORRECT_TYPE,
+          `schema mismatch for ${className}.${fieldName} default value; expected ${typeToString(
+            type
+          )} but got ${typeToString(defaultValueType)}`
+        );
+      }
     }
 
     if (expectedType) {
diff --git a/src/Controllers/types.js b/src/Controllers/types.js
index 77a67f6c6d..2bd3298935 100644
--- a/src/Controllers/types.js
+++ b/src/Controllers/types.js
@@ -5,6 +5,8 @@ export type LoadSchemaOptions = {
 export type SchemaField = {
   type: string,
   targetClass?: ?string,
+  required?: ?boolean,
+  defaultValue?: ?any,
 };
 
 export type SchemaFields = { [string]: SchemaField };
diff --git a/src/RestWrite.js b/src/RestWrite.js
index 3fea6043de..3ac917cd3a 100644
--- a/src/RestWrite.js
+++ b/src/RestWrite.js
@@ -323,16 +323,66 @@ RestWrite.prototype.runBeforeLoginTrigger = async function(userData) {
 
 RestWrite.prototype.setRequiredFieldsIfNeeded = function() {
   if (this.data) {
-    // Add default fields
-    this.data.updatedAt = this.updatedAt;
-    if (!this.query) {
-      this.data.createdAt = this.updatedAt;
+    return this.validSchemaController.getAllClasses().then(allClasses => {
+      const schema = allClasses.find(
+        oneClass => oneClass.className === this.className
+      );
+      const setRequiredFieldIfNeeded = (fieldName, setDefault) => {
+        if (
+          this.data[fieldName] === undefined ||
+          this.data[fieldName] === null ||
+          this.data[fieldName] === '' ||
+          (typeof this.data[fieldName] === 'object' &&
+            this.data[fieldName].__op === 'Delete')
+        ) {
+          if (
+            setDefault &&
+            schema.fields[fieldName] &&
+            schema.fields[fieldName].defaultValue &&
+            (this.data[fieldName] === undefined ||
+              (typeof this.data[fieldName] === 'object' &&
+                this.data[fieldName].__op === 'Delete'))
+          ) {
+            this.data[fieldName] = schema.fields[fieldName].defaultValue;
+            this.storage.fieldsChangedByTrigger =
+              this.storage.fieldsChangedByTrigger || [];
+            if (this.storage.fieldsChangedByTrigger.indexOf(fieldName) < 0) {
+              this.storage.fieldsChangedByTrigger.push(fieldName);
+            }
+          } else if (
+            schema.fields[fieldName] &&
+            schema.fields[fieldName].required === true
+          ) {
+            throw new Parse.Error(
+              Parse.Error.VALIDATION_ERROR,
+              `${fieldName} is required`
+            );
+          }
+        }
+      };
+
+      // Add default fields
+      this.data.updatedAt = this.updatedAt;
+      if (!this.query) {
+        this.data.createdAt = this.updatedAt;
 
-      // Only assign new objectId if we are creating new object
-      if (!this.data.objectId) {
-        this.data.objectId = cryptoUtils.newObjectId(this.config.objectIdSize);
+        // Only assign new objectId if we are creating new object
+        if (!this.data.objectId) {
+          this.data.objectId = cryptoUtils.newObjectId(
+            this.config.objectIdSize
+          );
+        }
+        if (schema) {
+          Object.keys(schema.fields).forEach(fieldName => {
+            setRequiredFieldIfNeeded(fieldName, true);
+          });
+        }
+      } else if (schema) {
+        Object.keys(this.data).forEach(fieldName => {
+          setRequiredFieldIfNeeded(fieldName, false);
+        });
       }
-    }
+    });
   }
   return Promise.resolve();
 };