diff --git a/cinder/tests/unit/volume/drivers/dell_emc/vmax/test_vmax.py b/cinder/tests/unit/volume/drivers/dell_emc/vmax/test_vmax.py
index 2ecd67e4772..1e2094bb229 100644
--- a/cinder/tests/unit/volume/drivers/dell_emc/vmax/test_vmax.py
+++ b/cinder/tests/unit/volume/drivers/dell_emc/vmax/test_vmax.py
@@ -60,6 +60,7 @@ class VMAXCommonData(object):
     storagegroup_name_i = 'OS-HostX-SRP_1-Diamond-DSS-OS-iscsi-PG'
     defaultstoragegroup_name = 'OS-SRP_1-Diamond-DSS-SG'
     default_sg_no_slo = 'OS-no_SLO-SG'
+    default_sg_compr_disabled = 'OS-SRP_1-Diamond-DSS-CD-SG'
     failed_resource = 'OS-failed-resource'
     fake_host = 'HostX@Backend#Diamond+DSS+SRP_1+000197800123'
     new_host = 'HostX@Backend#Silver+OLTP+SRP_1+000197800123'
@@ -155,6 +156,9 @@ class VMAXCommonData(object):
 
     # extra-specs
     vol_type_extra_specs = {'pool_name': u'Diamond+DSS+SRP_1+000197800123'}
+    vol_type_extra_specs_compr_disabled = {
+        'pool_name': u'Diamond+DSS+SRP_1+000197800123',
+        'storagetype:disablecompression': "true"}
     extra_specs = {'pool_name': u'Diamond+DSS+SRP_1+000197800123',
                    'slo': slo,
                    'workload': workload,
@@ -162,6 +166,8 @@ class VMAXCommonData(object):
                    'array': array,
                    'interval': 3,
                    'retries': 120}
+    extra_specs_disable_compression = deepcopy(extra_specs)
+    extra_specs_disable_compression[utils.DISABLECOMPRESSION] = "true"
     extra_specs_intervals_set = deepcopy(extra_specs)
     extra_specs_intervals_set['interval'] = 1
     extra_specs_intervals_set['retries'] = 1
@@ -176,6 +182,7 @@ class VMAXCommonData(object):
         'maskingview_name': masking_view_name_f,
         'parent_sg_name': parent_sg_f,
         'srp': srp,
+        'storagetype:disablecompression': False,
         'port_group_name': port_group_name_f,
         'slo': slo,
         'storagegroup_name': storagegroup_name_f,
@@ -190,6 +197,7 @@ class VMAXCommonData(object):
         'initiator_check': False,
         'maskingview_name': masking_view_name_f,
         'srp': srp,
+        'storagetype:disablecompression': False,
         'port_group_name': port_group_name_f,
         'slo': None,
         'parent_sg_name': parent_sg_f,
@@ -197,8 +205,25 @@ class VMAXCommonData(object):
         'volume_name': test_volume.name,
         'workload': None}
 
+    masking_view_dict_compression_disabled = {
+        'array': array,
+        'connector': connector,
+        'device_id': '00001',
+        'init_group_name': initiatorgroup_name_f,
+        'initiator_check': False,
+        'maskingview_name': masking_view_name_f,
+        'srp': srp,
+        'storagetype:disablecompression': True,
+        'port_group_name': port_group_name_f,
+        'slo': slo,
+        'parent_sg_name': parent_sg_f,
+        'storagegroup_name': 'OS-HostX-SRP_1-DiamondDSS-OS-fibre-PG-CD',
+        'volume_name': test_volume['name'],
+        'workload': workload}
+
     # vmax data
     # sloprovisioning
+    compression_info = {"symmetrixId": ["000197800128"]}
     inititiatorgroup = [{"initiator": [wwpn1],
                          "hostId": initiatorgroup_name_f,
                          "maskingview": [masking_view_name_f]},
@@ -452,6 +477,8 @@ class FakeRequestsSession(object):
                 return_object = self.data.srp_details
             elif 'workloadtype' in url:
                 return_object = self.data.workloadtype
+            elif 'compressionCapable' in url:
+                return_object = self.data.compression_info
             else:
                 return_object = self.data.slo_details
 
@@ -857,6 +884,14 @@ class VMAXUtilsTest(test.TestCase):
             srp_name, slo, workload)
         self.assertEqual(self.data.default_sg_no_slo, sg_name)
 
+    def test_get_default_storage_group_name_compr_disabled(self):
+        srp_name = self.data.srp
+        slo = self.data.slo
+        workload = self.data.workload
+        sg_name = self.utils.get_default_storage_group_name(
+            srp_name, slo, workload, True)
+        self.assertEqual(self.data.default_sg_compr_disabled, sg_name)
+
     def test_get_time_delta(self):
         start_time = 1487781721.09
         end_time = 1487781758.16
@@ -910,6 +945,41 @@ class VMAXUtilsTest(test.TestCase):
         pg3 = self.utils.get_pg_short_name(pg_over_12_chars)
         self.assertEqual(pg2, pg3)
 
+    def test_is_compression_disabled_true(self):
+        extra_specs = self.data.extra_specs_disable_compression
+        do_disable_compression = self.utils.is_compression_disabled(
+            extra_specs)
+        self.assertTrue(do_disable_compression)
+
+    def test_is_compression_disabled_false(self):
+        # Path 1: no compressiion extra spec set
+        extra_specs = self.data.extra_specs
+        do_disable_compression = self.utils.is_compression_disabled(
+            extra_specs)
+        self.assertFalse(do_disable_compression)
+        # Path 2: compression extra spec set to false
+        extra_specs2 = deepcopy(extra_specs)
+        extra_specs2.update({utils.DISABLECOMPRESSION: 'false'})
+        do_disable_compression2 = self.utils.is_compression_disabled(
+            extra_specs)
+        self.assertFalse(do_disable_compression2)
+
+    def test_change_compression_type_true(self):
+        source_compr_disabled_true = 'true'
+        new_type_compr_disabled = {
+            'extra_specs': {utils.DISABLECOMPRESSION: 'no'}}
+        ans = self.utils.change_compression_type(
+            source_compr_disabled_true, new_type_compr_disabled)
+        self.assertTrue(ans)
+
+    def test_change_compression_type_false(self):
+        source_compr_disabled_true = True
+        new_type_compr_disabled = {
+            'extra_specs': {utils.DISABLECOMPRESSION: 'true'}}
+        ans = self.utils.change_compression_type(
+            source_compr_disabled_true, new_type_compr_disabled)
+        self.assertFalse(ans)
+
 
 class VMAXRestTest(test.TestCase):
     def setUp(self):
@@ -1115,6 +1185,17 @@ class VMAXRestTest(test.TestCase):
             self.data.slo, self.data.workload)
         self.assertIsNone(headroom_cap)
 
+    def test_is_compression_capable_true(self):
+        compr_capable = self.rest.is_compression_capable('000197800128')
+        self.assertTrue(compr_capable)
+
+    def test_is_compression_capable_false(self):
+        compr_capable = self.rest.is_compression_capable(self.data.array)
+        self.assertFalse(compr_capable)
+        with mock.patch.object(self.rest, 'request', return_value=(200, {})):
+            compr_capable = self.rest.is_compression_capable(self.data.array)
+            self.assertFalse(compr_capable)
+
     def test_get_storage_group(self):
         ref_details = self.data.sg_details[0]
         sg_details = self.rest.get_storage_group(
@@ -1159,6 +1240,28 @@ class VMAXRestTest(test.TestCase):
             None, None, self.data.extra_specs)
         self.assertEqual(self.data.default_sg_no_slo, sg_name)
 
+    def test_create_storage_group_compression_disabled(self):
+        with mock.patch.object(self.rest, '_create_storagegroup',
+                               return_value=(200, self.data.job_list[0])):
+            self.rest.create_storage_group(
+                self.data.array, self.data.default_sg_compr_disabled,
+                self.data.srp, self.data.slo, self.data.workload,
+                self.data.extra_specs, True)
+            payload = {"srpId": self.data.srp,
+                       "storageGroupId": self.data.default_sg_compr_disabled,
+                       "emulation": "FBA",
+                       "create_empty_storage_group": "true",
+                       "sloBasedStorageGroupParam": [
+                           {"num_of_vols": 0,
+                            "sloId": self.data.slo,
+                            "workloadSelection": self.data.workload,
+                            "volumeAttribute": {
+                                "volume_size": "0",
+                                "capacityUnit": "GB"},
+                            "noCompression": "true"}]}
+            self.rest._create_storagegroup.assert_called_once_with(
+                self.data.array, payload)
+
     def test_modify_storage_group(self):
         array = self.data.array
         storagegroup = self.data.defaultstoragegroup_name
@@ -2651,6 +2754,17 @@ class VMAXCommonTest(test.TestCase):
             volume, connector, extra_specs)
         self.assertEqual(ref_mv_dict, masking_view_dict)
 
+    def test_populate_masking_dict_compr_disabled(self):
+        volume = self.data.test_volume
+        connector = self.data.connector
+        extra_specs = deepcopy(self.data.extra_specs)
+        extra_specs['port_group_name'] = self.data.port_group_name_f
+        extra_specs[utils.DISABLECOMPRESSION] = "true"
+        ref_mv_dict = self.data.masking_view_dict_compression_disabled
+        masking_view_dict = self.common._populate_masking_dict(
+            volume, connector, extra_specs)
+        self.assertEqual(ref_mv_dict, masking_view_dict)
+
     def test_create_cloned_volume(self):
         volume = self.data.test_clone_volume
         source_volume = self.data.test_volume
@@ -2823,6 +2937,27 @@ class VMAXCommonTest(test.TestCase):
         extra_specs = self.common._set_vmax_extra_specs({}, srp_record)
         self.assertEqual('Optimized', extra_specs['slo'])
 
+    def test_set_vmax_extra_specs_compr_disabled(self):
+        with mock.patch.object(self.rest, 'is_compression_capable',
+                               return_value=True):
+            srp_record = self.utils.parse_file_to_get_array_map(
+                self.fake_xml)
+            extra_specs = self.common._set_vmax_extra_specs(
+                self.data.vol_type_extra_specs_compr_disabled, srp_record)
+            ref_extra_specs = deepcopy(self.data.extra_specs_intervals_set)
+            ref_extra_specs['port_group_name'] = self.data.port_group_name_f
+            ref_extra_specs[utils.DISABLECOMPRESSION] = "true"
+            self.assertEqual(ref_extra_specs, extra_specs)
+
+    def test_set_vmax_extra_specs_compr_disabled_not_compr_capable(self):
+        srp_record = self.utils.parse_file_to_get_array_map(
+            self.fake_xml)
+        extra_specs = self.common._set_vmax_extra_specs(
+            self.data.vol_type_extra_specs_compr_disabled, srp_record)
+        ref_extra_specs = deepcopy(self.data.extra_specs_intervals_set)
+        ref_extra_specs['port_group_name'] = self.data.port_group_name_f
+        self.assertEqual(ref_extra_specs, extra_specs)
+
     def test_set_vmax_extra_specs_portgroup_as_spec(self):
         srp_record = self.utils.parse_file_to_get_array_map(
             self.fake_xml)
@@ -3161,41 +3296,44 @@ class VMAXCommonTest(test.TestCase):
         extra_specs = self.data.extra_specs_intervals_set
         extra_specs['port_group_name'] = self.data.port_group_name_f
         volume = self.data.test_volume
+        new_type = {'extra_specs': {}}
         host = {'host': self.data.new_host}
-        self.common.retype(volume, host)
+        self.common.retype(volume, new_type, host)
         mock_migrate.assert_called_once_with(
-            device_id, volume, host, volume_name, extra_specs)
+            device_id, volume, host, volume_name, new_type, extra_specs)
         mock_migrate.reset_mock()
         with mock.patch.object(
                 self.common, '_find_device_on_array', return_value=None):
-            self.common.retype(volume, host)
+            self.common.retype(volume, new_type, host)
             mock_migrate.assert_not_called()
 
     def test_slo_workload_migration_valid(self):
         device_id = self.data.volume_details[0]['volumeId']
         volume_name = self.data.test_volume['name']
         extra_specs = self.data.extra_specs
+        new_type = {'extra_specs': {}}
         volume = self.data.test_volume
         host = {'host': self.data.new_host}
         with mock.patch.object(self.common, '_migrate_volume'):
             self.common._slo_workload_migration(
-                device_id, volume, host, volume_name, extra_specs)
+                device_id, volume, host, volume_name, new_type, extra_specs)
             self.common._migrate_volume.assert_called_once_with(
                 extra_specs[utils.ARRAY], device_id,
                 extra_specs[utils.SRP], 'Silver',
-                'OLTP', volume_name, extra_specs)
+                'OLTP', volume_name, new_type, extra_specs)
 
     def test_slo_workload_migration_not_valid(self):
         device_id = self.data.volume_details[0]['volumeId']
         volume_name = self.data.test_volume['name']
         extra_specs = self.data.extra_specs
         volume = self.data.test_volume
+        new_type = {'extra_specs': {}}
         host = {'host': self.data.new_host}
         with mock.patch.object(self.common,
                                '_is_valid_for_storage_assisted_migration',
                                return_value=(False, 'Silver', 'OLTP')):
             migrate_status = self.common._slo_workload_migration(
-                device_id, volume, host, volume_name, extra_specs)
+                device_id, volume, host, volume_name, new_type, extra_specs)
             self.assertFalse(migrate_status)
 
     def test_slo_workload_migration_same_hosts(self):
@@ -3204,10 +3342,31 @@ class VMAXCommonTest(test.TestCase):
         extra_specs = self.data.extra_specs
         volume = self.data.test_volume
         host = {'host': self.data.fake_host}
+        new_type = {'extra_specs': {}}
         migrate_status = self.common._slo_workload_migration(
-            device_id, volume, host, volume_name, extra_specs)
+            device_id, volume, host, volume_name, new_type, extra_specs)
         self.assertFalse(migrate_status)
 
+    def test_slo_workload_migration_same_host_change_compression(self):
+        device_id = self.data.volume_details[0]['volumeId']
+        volume_name = self.data.test_volume['name']
+        extra_specs = self.data.extra_specs
+        volume = self.data.test_volume
+        host = {'host': self.data.fake_host}
+        new_type = {'extra_specs': {utils.DISABLECOMPRESSION: "true"}}
+        with mock.patch.object(
+                self.common, '_is_valid_for_storage_assisted_migration',
+                return_value=(True, self.data.slo, self.data.workload)):
+            with mock.patch.object(self.common, '_migrate_volume'):
+                migrate_status = self.common._slo_workload_migration(
+                    device_id, volume, host, volume_name, new_type,
+                    extra_specs)
+                self.assertTrue(migrate_status)
+                self.common._migrate_volume.assert_called_once_with(
+                    extra_specs[utils.ARRAY], device_id,
+                    extra_specs[utils.SRP], self.data.slo,
+                    self.data.workload, volume_name, new_type, extra_specs)
+
     @mock.patch.object(masking.VMAXMasking, 'remove_and_reset_members')
     def test_migrate_volume_success(self, mock_remove):
         with mock.patch.object(self.rest, 'is_volume_in_storagegroup',
@@ -3215,9 +3374,10 @@ class VMAXCommonTest(test.TestCase):
             device_id = self.data.volume_details[0]['volumeId']
             volume_name = self.data.test_volume['name']
             extra_specs = self.data.extra_specs
+            new_type = {'extra_specs': {}}
             migrate_status = self.common._migrate_volume(
                 self.data.array, device_id, self.data.srp, self.data.slo,
-                self.data.workload, volume_name, extra_specs)
+                self.data.workload, volume_name, new_type, extra_specs)
             self.assertTrue(migrate_status)
             mock_remove.assert_called_once_with(
                 self.data.array, device_id, None, extra_specs, False)
@@ -3227,7 +3387,7 @@ class VMAXCommonTest(test.TestCase):
                     return_value=[]):
                 migrate_status = self.common._migrate_volume(
                     self.data.array, device_id, self.data.srp, self.data.slo,
-                    self.data.workload, volume_name, extra_specs)
+                    self.data.workload, volume_name, new_type, extra_specs)
                 self.assertTrue(migrate_status)
                 mock_remove.assert_not_called()
 
@@ -3236,24 +3396,26 @@ class VMAXCommonTest(test.TestCase):
         device_id = self.data.volume_details[0]['volumeId']
         volume_name = self.data.test_volume['name']
         extra_specs = self.data.extra_specs
+        new_type = {'extra_specs': {}}
         with mock.patch.object(
                 self.masking, 'get_or_create_default_storage_group',
                 side_effect=exception.VolumeBackendAPIException):
             migrate_status = self.common._migrate_volume(
                 self.data.array, device_id, self.data.srp, self.data.slo,
-                self.data.workload, volume_name, extra_specs)
+                self.data.workload, volume_name, new_type, extra_specs)
             self.assertFalse(migrate_status)
 
     def test_migrate_volume_failed_vol_not_added(self):
         device_id = self.data.volume_details[0]['volumeId']
         volume_name = self.data.test_volume['name']
         extra_specs = self.data.extra_specs
+        new_type = {'extra_specs': {}}
         with mock.patch.object(
                 self.rest, 'is_volume_in_storagegroup',
                 return_value=False):
             migrate_status = self.common._migrate_volume(
                 self.data.array, device_id, self.data.srp, self.data.slo,
-                self.data.workload, volume_name, extra_specs)
+                self.data.workload, volume_name, new_type, extra_specs)
             self.assertFalse(migrate_status)
 
     def test_is_valid_for_storage_assisted_migration_true(self):
@@ -3262,13 +3424,15 @@ class VMAXCommonTest(test.TestCase):
         volume_name = self.data.test_volume['name']
         ref_return = (True, 'Silver', 'OLTP')
         return_val = self.common._is_valid_for_storage_assisted_migration(
-            device_id, host, self.data.array, self.data.srp, volume_name)
+            device_id, host, self.data.array,
+            self.data.srp, volume_name, False)
         self.assertEqual(ref_return, return_val)
         # No current sgs found
         with mock.patch.object(self.rest, 'get_storage_groups_from_volume',
                                return_value=None):
             return_val = self.common._is_valid_for_storage_assisted_migration(
-                device_id, host, self.data.array, self.data.srp, volume_name)
+                device_id, host, self.data.array, self.data.srp,
+                volume_name, False)
             self.assertEqual(ref_return, return_val)
 
     def test_is_valid_for_storage_assisted_migration_false(self):
@@ -3278,22 +3442,26 @@ class VMAXCommonTest(test.TestCase):
         # IndexError
         host = {'host': 'HostX@Backend#Silver+SRP_1+000197800123'}
         return_val = self.common._is_valid_for_storage_assisted_migration(
-            device_id, host, self.data.array, self.data.srp, volume_name)
+            device_id, host, self.data.array,
+            self.data.srp, volume_name, False)
         self.assertEqual(ref_return, return_val)
         # Wrong array
         host2 = {'host': 'HostX@Backend#Silver+OLTP+SRP_1+00012345678'}
         return_val = self.common._is_valid_for_storage_assisted_migration(
-            device_id, host2, self.data.array, self.data.srp, volume_name)
+            device_id, host2, self.data.array,
+            self.data.srp, volume_name, False)
         self.assertEqual(ref_return, return_val)
         # Wrong srp
         host3 = {'host': 'HostX@Backend#Silver+OLTP+SRP_2+000197800123'}
         return_val = self.common._is_valid_for_storage_assisted_migration(
-            device_id, host3, self.data.array, self.data.srp, volume_name)
+            device_id, host3, self.data.array,
+            self.data.srp, volume_name, False)
         self.assertEqual(ref_return, return_val)
         # Already in correct sg
         host4 = {'host': self.data.fake_host}
         return_val = self.common._is_valid_for_storage_assisted_migration(
-            device_id, host4, self.data.array, self.data.srp, volume_name)
+            device_id, host4, self.data.array,
+            self.data.srp, volume_name, False)
         self.assertEqual(ref_return, return_val)
 
 
@@ -3509,11 +3677,12 @@ class VMAXFCTest(test.TestCase):
 
     def test_retype(self):
         host = {'host': self.data.new_host}
+        new_type = {'extra_specs': {}}
         with mock.patch.object(self.common, 'retype',
                                return_value=True):
-            self.driver.retype({}, self.data.test_volume, '', '', host)
+            self.driver.retype({}, self.data.test_volume, new_type, '', host)
             self.common.retype.assert_called_once_with(
-                self.data.test_volume, host)
+                self.data.test_volume, new_type, host)
 
 
 class VMAXISCSITest(test.TestCase):
@@ -3739,11 +3908,12 @@ class VMAXISCSITest(test.TestCase):
 
     def test_retype(self):
         host = {'host': self.data.new_host}
+        new_type = {'extra_specs': {}}
         with mock.patch.object(self.common, 'retype',
                                return_value=True):
-            self.driver.retype({}, self.data.test_volume, '', '', host)
+            self.driver.retype({}, self.data.test_volume, new_type, '', host)
             self.common.retype.assert_called_once_with(
-                self.data.test_volume, host)
+                self.data.test_volume, new_type, host)
 
 
 class VMAXMaskingTest(test.TestCase):
diff --git a/cinder/volume/drivers/dell_emc/vmax/common.py b/cinder/volume/drivers/dell_emc/vmax/common.py
index db890238e9d..cda50999a84 100644
--- a/cinder/volume/drivers/dell_emc/vmax/common.py
+++ b/cinder/volume/drivers/dell_emc/vmax/common.py
@@ -19,6 +19,7 @@ import sys
 
 from oslo_config import cfg
 from oslo_log import log as logging
+from oslo_utils import strutils
 import six
 
 from cinder import exception
@@ -854,6 +855,7 @@ class VMAXCommon(object):
         unique_name = self.utils.truncate_string(extra_specs[utils.SRP], 12)
         protocol = self.utils.get_short_protocol_type(self.protocol)
         short_host_name = self.utils.get_host_short_name(host_name)
+        masking_view_dict[utils.DISABLECOMPRESSION] = False
         slo = extra_specs[utils.SLO]
         workload = extra_specs[utils.WORKLOAD]
         short_pg_name = self.utils.get_pg_short_name(
@@ -877,6 +879,12 @@ class VMAXCommon(object):
                    'srpName': unique_name,
                    'combo': slo_wl_combo,
                    'pg': short_pg_name})
+            do_disable_compression = self.utils.is_compression_disabled(
+                extra_specs)
+            if do_disable_compression:
+                child_sg_name = ("%(child_sg_name)s-CD"
+                                 % {'child_sg_name': child_sg_name})
+                masking_view_dict[utils.DISABLECOMPRESSION] = True
         else:
             child_sg_name = (
                 "OS-%(shortHostName)s-No_SLO-%(pg)s"
@@ -1092,9 +1100,13 @@ class VMAXCommon(object):
                    'array': array,
                    'size': volume_size})
 
+        do_disable_compression = self.utils.is_compression_disabled(
+            extra_specs)
+
         storagegroup_name = self.masking.get_or_create_default_storage_group(
             array, extra_specs[utils.SRP], extra_specs[utils.SLO],
-            extra_specs[utils.WORKLOAD], extra_specs)
+            extra_specs[utils.WORKLOAD], extra_specs,
+            do_disable_compression)
         try:
             volume_dict = self.provision.create_volume_from_sg(
                 array, volume_name, storagegroup_name,
@@ -1187,6 +1199,14 @@ class VMAXCommon(object):
             slo_from_extra_spec = None
         extra_specs[utils.SLO] = slo_from_extra_spec
         extra_specs[utils.WORKLOAD] = workload_from_extra_spec
+        if self.rest.is_compression_capable(extra_specs[utils.ARRAY]):
+            if extra_specs.get(utils.DISABLECOMPRESSION):
+                # If not True remove it.
+                if not strutils.bool_from_string(
+                        extra_specs[utils.DISABLECOMPRESSION]):
+                    extra_specs.pop(utils.DISABLECOMPRESSION, None)
+        else:
+            extra_specs.pop(utils.DISABLECOMPRESSION, None)
 
         LOG.debug("SRP is: %(srp)s "
                   "Array is: %(array)s "
@@ -1536,10 +1556,11 @@ class VMAXCommon(object):
             self.rest.rename_volume(
                 extra_specs[utils.ARRAY], device_id, volume_id)
 
-    def retype(self, volume, host):
+    def retype(self, volume, new_type, host):
         """Migrate volume to another host using retype.
 
         :param volume: the volume object including the volume_type_id
+        :param new_type: the new volume type.
         :param host: The host dict holding the relevant target(destination)
             information
         :returns: boolean -- True if retype succeeded, False if error
@@ -1558,23 +1579,30 @@ class VMAXCommon(object):
             return False
 
         return self._slo_workload_migration(device_id, volume, host,
-                                            volume_name, extra_specs)
+                                            volume_name, new_type, extra_specs)
 
     def _slo_workload_migration(self, device_id, volume, host,
-                                volume_name, extra_specs):
+                                volume_name, new_type, extra_specs):
         """Migrate from SLO/Workload combination to another.
 
         :param device_id: the volume device id
         :param volume: the volume object
         :param host: the host dict
         :param volume_name: the name of the volume
+        :param new_type: the type to migrate to
         :param extra_specs: extra specifications
         :returns: boolean -- True if migration succeeded, False if error.
         """
+        is_compression_disabled = self.utils.is_compression_disabled(
+            extra_specs)
+        # Check if old type and new type have different compression types
+        do_change_compression = (self.utils.change_compression_type(
+            is_compression_disabled, new_type))
         is_valid, target_slo, target_workload = (
             self._is_valid_for_storage_assisted_migration(
                 device_id, host, extra_specs[utils.ARRAY],
-                extra_specs[utils.SRP], volume_name))
+                extra_specs[utils.SRP], volume_name,
+                do_change_compression))
 
         if not is_valid:
             LOG.error(
@@ -1582,23 +1610,24 @@ class VMAXCommon(object):
                 "assisted migration using retype.",
                 {'name': volume_name})
             return False
-        if volume.host != host['host']:
+        if volume.host != host['host'] or do_change_compression:
             LOG.debug(
                 "Retype Volume %(name)s from source host %(sourceHost)s "
-                "to target host %(targetHost)s. ",
+                "to target host %(targetHost)s. Compression change is %(cc)r.",
                 {'name': volume_name,
                  'sourceHost': volume.host,
-                 'targetHost': host['host']})
+                 'targetHost': host['host'],
+                 'cc': do_change_compression})
             return self._migrate_volume(
                 extra_specs[utils.ARRAY], device_id,
                 extra_specs[utils.SRP], target_slo,
-                target_workload, volume_name, extra_specs)
+                target_workload, volume_name, new_type, extra_specs)
 
         return False
 
     def _migrate_volume(
             self, array, device_id, srp, target_slo,
-            target_workload, volume_name, extra_specs):
+            target_workload, volume_name, new_type, extra_specs):
         """Migrate from one slo/workload combination to another.
 
         This requires moving the volume from its current SG to a
@@ -1609,23 +1638,28 @@ class VMAXCommon(object):
         :param target_slo: the target service level
         :param target_workload: the target workload
         :param volume_name: the volume name
+        :param new_type: the volume type to migrate to
         :param extra_specs: the extra specifications
         :return: bool
         """
         storagegroups = self.rest.get_storage_groups_from_volume(
             array, device_id)
         if not storagegroups:
-            LOG.warning(
-                "Volume : %(volume_name)s does not currently "
-                "belong to any storage groups.",
-                {'volume_name': volume_name})
+            LOG.warning("Volume : %(volume_name)s does not currently "
+                        "belong to any storage groups.",
+                        {'volume_name': volume_name})
         else:
             self.masking.remove_and_reset_members(
                 array, device_id, None, extra_specs, False)
 
+        target_extra_specs = new_type['extra_specs']
+        is_compression_disabled = self.utils.is_compression_disabled(
+            target_extra_specs)
+
         try:
             target_sg_name = self.masking.get_or_create_default_storage_group(
-                array, srp, target_slo, target_workload, extra_specs)
+                array, srp, target_slo, target_workload, extra_specs,
+                is_compression_disabled)
         except Exception as e:
             LOG.error("Failed to get or create storage group. "
                       "Exception received was %(e)s.", {'e': e})
@@ -1648,7 +1682,7 @@ class VMAXCommon(object):
 
     def _is_valid_for_storage_assisted_migration(
             self, device_id, host, source_array,
-            source_srp, volume_name):
+            source_srp, volume_name, do_change_compression):
         """Check if volume is suitable for storage assisted (pool) migration.
 
         :param device_id: the volume device id
@@ -1656,6 +1690,7 @@ class VMAXCommon(object):
         :param source_array: the volume's current array serial number
         :param source_srp: the volume's current pool name
         :param volume_name: the name of the volume to be migrated
+        :param do_change_compression: do change compression
         :returns: boolean -- True/False
         :returns: string -- targetSlo
         :returns: string -- targetWorkload
@@ -1711,12 +1746,15 @@ class VMAXCommon(object):
                                       % {'targetSlo': target_slo,
                                          'targetWorkload': target_workload})
                 if target_combination in emc_fast_setting:
-                    LOG.warning(
-                        "No action required. Volume: %(volume_name)s is "
-                        "already part of slo/workload combination: "
-                        "%(targetCombination)s.",
-                        {'volume_name': volume_name,
-                         'targetCombination': target_combination})
+                    # Check if migration is from compression to non compression
+                    # or vice versa
+                    if not do_change_compression:
+                        LOG.warning(
+                            "No action required. Volume: %(volume_name)s is "
+                            "already part of slo/workload combination: "
+                            "%(targetCombination)s.",
+                            {'volume_name': volume_name,
+                             'targetCombination': target_combination})
                     return false_ret
 
         return True, target_slo, target_workload
diff --git a/cinder/volume/drivers/dell_emc/vmax/fc.py b/cinder/volume/drivers/dell_emc/vmax/fc.py
index b5966014c9b..b3d7217129c 100644
--- a/cinder/volume/drivers/dell_emc/vmax/fc.py
+++ b/cinder/volume/drivers/dell_emc/vmax/fc.py
@@ -78,6 +78,7 @@ class VMAXFCDriver(driver.FibreChannelDriver):
         3.0.0 - REST based driver
               - Retype (storage-assisted migration)
               - QoS support
+              - Support for compression on All Flash
     """
 
     VERSION = "3.0.0"
@@ -438,4 +439,4 @@ class VMAXFCDriver(driver.FibreChannelDriver):
             target(destination) information
         :returns: boolean -- True if retype succeeded, False if error
         """
-        return self.common.retype(volume, host)
+        return self.common.retype(volume, new_type, host)
diff --git a/cinder/volume/drivers/dell_emc/vmax/iscsi.py b/cinder/volume/drivers/dell_emc/vmax/iscsi.py
index 82e44d46e1a..555167efa9c 100644
--- a/cinder/volume/drivers/dell_emc/vmax/iscsi.py
+++ b/cinder/volume/drivers/dell_emc/vmax/iscsi.py
@@ -83,6 +83,7 @@ class VMAXISCSIDriver(driver.ISCSIDriver):
         3.0.0 - REST based driver
               - Retype (storage-assisted migration)
               - QoS support
+              - Support for compression on All Flash
     """
 
     VERSION = "3.0.0"
@@ -382,4 +383,4 @@ class VMAXISCSIDriver(driver.ISCSIDriver):
             target(destination) information
         :returns: boolean -- True if retype succeeded, False if error
         """
-        return self.common.retype(volume, host)
+        return self.common.retype(volume, new_type, host)
diff --git a/cinder/volume/drivers/dell_emc/vmax/masking.py b/cinder/volume/drivers/dell_emc/vmax/masking.py
index 8e0488f80ac..a0fc9a7bd0a 100644
--- a/cinder/volume/drivers/dell_emc/vmax/masking.py
+++ b/cinder/volume/drivers/dell_emc/vmax/masking.py
@@ -141,7 +141,8 @@ class VMAXMasking(object):
         default_sg_name = self.utils.get_default_storage_group_name(
             masking_view_dict[utils.SRP],
             masking_view_dict[utils.SLO],
-            masking_view_dict[utils.WORKLOAD])
+            masking_view_dict[utils.WORKLOAD],
+            masking_view_dict[utils.DISABLECOMPRESSION])
 
         check_vol = self.rest.is_volume_in_storagegroup(
             serial_number, device_id, default_sg_name)
@@ -352,12 +353,14 @@ class VMAXMasking(object):
             slo = None
         else:
             slo = extra_specs[utils.SLO]
+        do_disable_compression = (
+            masking_view_dict[utils.DISABLECOMPRESSION])
         storagegroup = self.rest.get_storage_group(
             serial_number, storagegroup_name)
         if storagegroup is None:
             storagegroup = self.provision.create_storage_group(
                 serial_number, storagegroup_name, srp, slo, workload,
-                extra_specs)
+                extra_specs, do_disable_compression)
 
         if storagegroup is None:
             msg = ("Cannot get or create a storage group: "
@@ -1210,16 +1213,19 @@ class VMAXMasking(object):
         :param volume_name: the volume name
         :param extra_specs: the extra specifications
         """
+        do_disable_compression = self.utils.is_compression_disabled(
+            extra_specs)
         storagegroup_name = self.get_or_create_default_storage_group(
             serial_number, extra_specs[utils.SRP], extra_specs[utils.SLO],
-            extra_specs[utils.WORKLOAD], extra_specs)
+            extra_specs[utils.WORKLOAD], extra_specs, do_disable_compression)
 
         self._check_adding_volume_to_storage_group(
             serial_number, device_id, storagegroup_name, volume_name,
             extra_specs)
 
     def get_or_create_default_storage_group(
-            self, serial_number, srp, slo, workload, extra_specs):
+            self, serial_number, srp, slo, workload, extra_specs,
+            do_disable_compression=False):
         """Get or create a default storage group.
 
         :param serial_number: the array serial number
@@ -1227,12 +1233,13 @@ class VMAXMasking(object):
         :param slo: the SLO
         :param workload: the workload
         :param extra_specs: extra specifications
+        :param do_disable_compression: flag for compression
         :returns: storagegroup_name
         :raises: VolumeBackendAPIException
         """
         storagegroup, storagegroup_name = (
             self.rest.get_vmax_default_storage_group(
-                serial_number, srp, slo, workload))
+                serial_number, srp, slo, workload, do_disable_compression))
         if storagegroup is None:
             self.provision.create_storage_group(
                 serial_number, storagegroup_name, srp, slo, workload,
diff --git a/cinder/volume/drivers/dell_emc/vmax/provision.py b/cinder/volume/drivers/dell_emc/vmax/provision.py
index e0104ee1290..25e04cad40e 100644
--- a/cinder/volume/drivers/dell_emc/vmax/provision.py
+++ b/cinder/volume/drivers/dell_emc/vmax/provision.py
@@ -35,7 +35,8 @@ class VMAXProvision(object):
         self.rest = rest
 
     def create_storage_group(
-            self, array, storagegroup_name, srp, slo, workload, extra_specs):
+            self, array, storagegroup_name, srp, slo, workload,
+            extra_specs, do_disable_compression=False):
         """Create a new storage group.
 
         :param array: the array serial number
@@ -44,6 +45,7 @@ class VMAXProvision(object):
         :param slo: the SLO (String)
         :param workload: the workload (String)
         :param extra_specs: additional info
+        :param do_disable_compression: disable compression flag
         :returns: storagegroup - storage group object
         """
         start_time = time.time()
@@ -51,7 +53,8 @@ class VMAXProvision(object):
         @coordination.synchronized("emc-sg-{storage_group}")
         def do_create_storage_group(storage_group):
             storagegroup = self.rest.create_storage_group(
-                array, storage_group, srp, slo, workload, extra_specs)
+                array, storage_group, srp, slo, workload, extra_specs,
+                do_disable_compression)
 
             LOG.debug("Create storage group took: %(delta)s H:MM:SS.",
                       {'delta': self.utils.get_time_delta(start_time,
diff --git a/cinder/volume/drivers/dell_emc/vmax/rest.py b/cinder/volume/drivers/dell_emc/vmax/rest.py
index 20477724ec2..f604d9dfce5 100644
--- a/cinder/volume/drivers/dell_emc/vmax/rest.py
+++ b/cinder/volume/drivers/dell_emc/vmax/rest.py
@@ -463,6 +463,22 @@ class VMAXRest(object):
             remaining_capacity = None
         return remaining_capacity
 
+    def is_compression_capable(self, array):
+        """Check if array is compression capable.
+
+        :param array: array serial number
+        :returns: bool
+        """
+        is_compression_capable = False
+        target_uri = "/84/sloprovisioning/symmetrix?compressionCapable=true"
+        status_code, message = self.request(target_uri, GET)
+        self.check_status_code_success(
+            "Check if compression enabled", status_code, message)
+        if message.get('symmetrixId'):
+            if array in message['symmetrixId']:
+                is_compression_capable = True
+        return is_compression_capable
+
     def get_storage_group(self, array, storage_group_name):
         """Given a name, return storage group details.
 
@@ -565,7 +581,8 @@ class VMAXRest(object):
             array, SLOPROVISIONING, 'storagegroup', payload)
 
     def create_storage_group(self, array, storagegroup_name,
-                             srp, slo, workload, extra_specs):
+                             srp, slo, workload, extra_specs,
+                             do_disable_compression=False):
         """Create the volume in the specified storage group.
 
         :param array: the array serial number
@@ -573,6 +590,7 @@ class VMAXRest(object):
         :param srp: the SRP (String)
         :param slo: the SLO (String)
         :param workload: the workload (String)
+        :param do_disable_compression: flag for disabling compression
         :param extra_specs: additional info
         :returns: storagegroup_name - string
         """
@@ -589,6 +607,11 @@ class VMAXRest(object):
                          "volumeAttribute": {
                              "volume_size": "0",
                              "capacityUnit": "GB"}}
+            if do_disable_compression:
+                slo_param.update({"noCompression": "true"})
+            elif self.is_compression_capable(array):
+                slo_param.update({"noCompression": "false"})
+
             payload.update({"sloBasedStorageGroupParam": [slo_param]})
 
         status_code, job = self._create_storagegroup(array, payload)
@@ -783,13 +806,15 @@ class VMAXRest(object):
                 return_value = False
         return return_value
 
-    def get_vmax_default_storage_group(self, array, srp, slo, workload):
+    def get_vmax_default_storage_group(self, array, srp, slo, workload,
+                                       do_disable_compression=False):
         """Get the default storage group.
 
         :param array: the array serial number
         :param srp: the pool name
         :param slo: the SLO
         :param workload: the workload
+        :param do_disable_compression: flag for disabling compression
         :returns: the storage group dict (or None), the storage group name
         """
         storagegroup_name = self.utils.get_default_storage_group_name(
diff --git a/cinder/volume/drivers/dell_emc/vmax/utils.py b/cinder/volume/drivers/dell_emc/vmax/utils.py
index ebbf78e75d4..653ea2662e6 100644
--- a/cinder/volume/drivers/dell_emc/vmax/utils.py
+++ b/cinder/volume/drivers/dell_emc/vmax/utils.py
@@ -20,6 +20,7 @@ import re
 from xml.dom import minidom
 
 from oslo_log import log as logging
+from oslo_utils import strutils
 import six
 
 from cinder import exception
@@ -52,6 +53,7 @@ PARENT_SG_NAME = 'parent_sg_name'
 CONNECTOR = 'connector'
 VOL_NAME = 'volume_name'
 EXTRA_SPECS = 'extra_specs'
+DISABLECOMPRESSION = 'storagetype:disablecompression'
 
 
 class VMAXUtils(object):
@@ -144,18 +146,24 @@ class VMAXUtils(object):
         return six.text_type(datetime.timedelta(seconds=int(delta)))
 
     @staticmethod
-    def get_default_storage_group_name(srp_name, slo, workload):
+    def get_default_storage_group_name(
+            srp_name, slo, workload, is_compression_disabled=False):
         """Determine default storage group from extra_specs.
 
         :param srp_name: the name of the srp on the array
         :param slo: the service level string e.g Bronze
         :param workload: the workload string e.g DSS
+        :param is_compression_disabled:  flag for disabling compression
         :returns: storage_group_name
         """
         if slo and workload:
             prefix = ("OS-%(srpName)s-%(slo)s-%(workload)s"
                       % {'srpName': srp_name, 'slo': slo,
                          'workload': workload})
+
+            if is_compression_disabled:
+                prefix += "-CD"
+
         else:
             prefix = "OS-no_SLO"
 
@@ -399,3 +407,30 @@ class VMAXUtils(object):
             raise exception.VolumeBackendAPIException(
                 data=exception_message)
         return array, device_id
+
+    @staticmethod
+    def is_compression_disabled(extra_specs):
+        """Check is compression is to be disabled.
+
+        :param extra_specs: extra specifications
+        :returns: boolean
+        """
+        do_disable_compression = False
+        if DISABLECOMPRESSION in extra_specs:
+            if strutils.bool_from_string(extra_specs[DISABLECOMPRESSION]):
+                do_disable_compression = True
+        return do_disable_compression
+
+    def change_compression_type(self, is_source_compr_disabled, new_type):
+        """Check if volume type have different compression types
+
+        :param is_source_compr_disabled: from source
+        :param new_type: from target
+        :returns: boolean
+        """
+        extra_specs = new_type['extra_specs']
+        is_target_compr_disabled = self.is_compression_disabled(extra_specs)
+        if is_target_compr_disabled == is_source_compr_disabled:
+            return False
+        else:
+            return True
diff --git a/releasenotes/notes/vmax-rest-compression-10c2590052a9465e.yaml b/releasenotes/notes/vmax-rest-compression-10c2590052a9465e.yaml
new file mode 100644
index 00000000000..77920f92a1b
--- /dev/null
+++ b/releasenotes/notes/vmax-rest-compression-10c2590052a9465e.yaml
@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Adding compression functionality to VMAX driver version 3.0.