cqlengine: Remove two-phase query when deleting values

aboudreault · aboudreault · commit 1e4cd1a60ff3 · 2018-01-22T15:57:40.000-05:00
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -20,6 +20,7 @@ Other
 * cqlengine: disallow Counter create, save operations (PYTHON-497)
 * cqlengine: remove the negative indices slicing support in ModelQuerySet (PYTHON-875)
 * cqlengine: Remove Model.__default_ttl__ (PYTHON-889)
+* cqlengine: Remove two-phase query when deleting values (PYTHON-506)
 
 3.12.0
 ======
diff --git a/cassandra/cqlengine/query.py b/cassandra/cqlengine/query.py
@@ -1280,7 +1280,6 @@ def update_async(self, **values):
         if not values:
             return
 
-        nulled_columns = set()
         updated_columns = set()
         us = UpdateStatement(self.column_family_name, where=self._where, ttl=self._ttl,
                              timestamp=self._timestamp, conditionals=self._conditional, if_exists=self._if_exists)
@@ -1303,25 +1302,13 @@ def update_async(self, **values):
                 # we should not provide default values in this use case.
                 val = col.validate(val)
 
-            if val is None:
-                nulled_columns.add(col_name)
-                continue
-
             us.add_update(col, val, operation=col_op)
             updated_columns.add(col_name)
 
-        futures = []
-        if us.assignments:
-            futures.append(self._execute_async(us))
-
-        if nulled_columns:
-            delete_conditional = [condition for condition in self._conditional
-                                  if condition.field not in updated_columns] if self._conditional else None
-            ds = DeleteStatement(self.column_family_name, fields=nulled_columns,
-                                 where=self._where, conditionals=delete_conditional, if_exists=self._if_exists)
-            futures.append(self._execute_async(ds))
+        if not us.assignments:
+            return CQLEngineFuture()
 
-        return CQLEngineFutureWaiter(futures)
+        return self._execute_async(us)
 
     def update(self, **values):
         """
@@ -1470,20 +1457,20 @@ def batch(self, batch_obj):
         self._batch = batch_obj
         return self
 
-    def _delete_null_columns_async(self, conditionals=None):
+    def _delete_map_keys_async(self, conditionals=None):
         """
-        executes a delete query to remove columns that have changed to null
+        executes a delete query to remove map keys that have changed to null.
+
+        Doing map keys deletion in two phases allow us to update a map collection
+        without creating a tombstone.
         """
         ds = DeleteStatement(self.column_family_name, conditionals=conditionals, if_exists=self._if_exists)
         deleted_fields = False
         static_only = True
+
         for _, v in self.instance._values.items():
             col = v.column
-            if v.deleted:
-                ds.add_field(col.db_field_name)
-                deleted_fields = True
-                static_only &= col.static
-            elif isinstance(col, columns.Map):
+            if isinstance(col, columns.Map):
                 uc = MapDeleteClause(col.db_field_name, v.value, v.previous_value)
                 if uc.get_context_size() > 0:
                     ds.add_field(uc)
@@ -1498,11 +1485,11 @@ def _delete_null_columns_async(self, conditionals=None):
         else:
             return CQLEngineFuture()
 
-    def _delete_null_columns(self, conditionals=None):
+    def _delete_map_keys(self, conditionals=None):
         """
         executes a delete query to remove columns that have changed to null
         """
-        return self._delete_null_columns_async(conditionals=conditionals).result()
+        return self._delete_map_keys_async(conditionals=conditionals).result()
 
     def update_async(self):
         """
@@ -1530,18 +1517,14 @@ def update_async(self):
                 val = getattr(self.instance, name, None)
                 val_mgr = self.instance._values[name]
 
-                if val is None:
-                    continue
-
-                if not val_mgr.changed and not isinstance(col, columns.Counter):
+                if not (val_mgr.changed or val_mgr.deleted) and not isinstance(col, columns.Counter):
                     continue
 
                 static_changed_only = static_changed_only and col.static
                 statement.add_update(col, val, previous=val_mgr.previous_value)
                 updated_columns.add(col.db_field_name)
 
         futures = []
-
         if statement.assignments:
             for name, col in self.model._primary_keys.items():
                 # only include clustering key if clustering key is not null, and non static columns are changed to avoid cql error
@@ -1554,7 +1537,7 @@ def update_async(self):
             # remove conditions on fields that have been updated
             delete_conditionals = [condition for condition in self._conditional
                                    if condition.field not in updated_columns] if self._conditional else None
-            futures.append(self._delete_null_columns_async(delete_conditionals))
+            futures.append(self._delete_map_keys_async(delete_conditionals))
 
         return CQLEngineFutureWaiter(futures)
 
@@ -1580,7 +1563,6 @@ def save_async(self):
         if self.instance._has_counter:
             raise CQLEngineException("Counters can only be updated. Use the 'update' mechanism instead.")
 
-        nulled_fields = set()
         if self.instance._can_update():
             return self.update_async()
         else:
@@ -1592,9 +1574,8 @@ def save_async(self):
                 if static_save_only and not col.static and not col.partition_key:
                     continue
                 val = getattr(self.instance, name, None)
-                if col._val_is_null(val):
-                    if self.instance._values[name].changed:
-                        nulled_fields.add(col.db_field_name)
+                if (col._val_is_null(val) and
+                    not (self.instance._values[name].changed or self.instance._values[name].deleted)):
                     continue
                 if col.has_default and not self.instance._values[name].changed:
                     # Ensure default columns included in a save() are marked as explicit, to get them *persisted* properly
@@ -1603,17 +1584,10 @@ def save_async(self):
 
         # skip query execution if it's empty
         # caused by pointless update queries
-        if insert.is_empty and static_save_only:
+        if insert.is_empty:
             return CQLEngineFuture()
 
-        futures = []
-        if not insert.is_empty:
-            futures.append(self._execute_async(insert))
-        # delete any nulled columns
-        if not static_save_only:
-            futures.append(self._delete_null_columns_async())
-
-        return CQLEngineFutureWaiter(futures)
+        return self._execute_async(insert)
 
     def save(self):
         """
diff --git a/cassandra/cqlengine/statements.py b/cassandra/cqlengine/statements.py
@@ -196,6 +196,11 @@ class SetUpdateClause(ContainerUpdateClause):
     _additions = None
     _removals = None
 
+    @property
+    def _is_assignment(self):
+        return not (self._additions or self._removals or
+                self._operation or self.value == self.previous)
+
     def __unicode__(self):
         qs = []
         ctx_id = self.context_id
@@ -204,7 +209,7 @@ def __unicode__(self):
                 self._additions is None and
                 self._removals is None):
             qs += ['"{0}" = %({1})s'.format(self.field, ctx_id)]
-        if self._assignments is not None:
+        if self._is_assignment:
             qs += ['"{0}" = %({1})s'.format(self.field, ctx_id)]
             ctx_id += 1
         if self._additions is not None:
@@ -235,11 +240,11 @@ def get_context_size(self):
         if not self._analyzed:
             self._analyze()
         if (self.previous is None and
-                not self._assignments and
+                not self._is_assignment and
                 self._additions is None and
                 self._removals is None):
             return 1
-        return int(bool(self._assignments)) + int(bool(self._additions)) + int(bool(self._removals))
+        return int(bool(self._is_assignment)) + int(bool(self._additions)) + int(bool(self._removals))
 
     def update_context(self, ctx):
         if not self._analyzed:
@@ -250,7 +255,7 @@ def update_context(self, ctx):
                 self._additions is None and
                 self._removals is None):
             ctx[str(ctx_id)] = set()
-        if self._assignments is not None:
+        if self._is_assignment:
             ctx[str(ctx_id)] = self._assignments
             ctx_id += 1
         if self._additions is not None:
@@ -268,12 +273,17 @@ class ListUpdateClause(ContainerUpdateClause):
     _append = None
     _prepend = None
 
+    @property
+    def _is_assignment(self):
+        return not (self._append or self._prepend or
+                    self._operation or self.value == self.previous)
+
     def __unicode__(self):
         if not self._analyzed:
             self._analyze()
         qs = []
         ctx_id = self.context_id
-        if self._assignments is not None:
+        if self._is_assignment:
             qs += ['"{0}" = %({1})s'.format(self.field, ctx_id)]
             ctx_id += 1
 
@@ -289,13 +299,13 @@ def __unicode__(self):
     def get_context_size(self):
         if not self._analyzed:
             self._analyze()
-        return int(self._assignments is not None) + int(bool(self._append)) + int(bool(self._prepend))
+        return int(self._is_assignment) + int(bool(self._append)) + int(bool(self._prepend))
 
     def update_context(self, ctx):
         if not self._analyzed:
             self._analyze()
         ctx_id = self.context_id
-        if self._assignments is not None:
+        if self._is_assignment:
             ctx[str(ctx_id)] = self._assignments
             ctx_id += 1
         if self._prepend is not None:
@@ -370,7 +380,9 @@ def _analyze(self):
             if self.previous is None:
                 self._updates = sorted([k for k, v in self.value.items()])
             else:
-                self._updates = sorted([k for k, v in self.value.items() if v != self.previous.get(k)]) or None
+                if self.value:
+                    self._updates = sorted([k for k, v in self.value.items() if v != self.previous.get(k)]) or None
+
         self._analyzed = True
 
     def get_context_size(self):
@@ -381,7 +393,7 @@ def get_context_size(self):
     def update_context(self, ctx):
         ctx_id = self.context_id
         if self.is_assignment:
-            ctx[str(ctx_id)] = {}
+            ctx[str(ctx_id)] = {} if self.value is None else self.value
         elif self._removals is not None:
             ctx[str(ctx_id)] = self._removals
         else:
@@ -395,7 +407,7 @@ def update_context(self, ctx):
     def is_assignment(self):
         if not self._analyzed:
             self._analyze()
-        return self.previous is None and not self._updates and not self._removals
+        return not self._updates and not self._removals
 
     def __unicode__(self):
         qs = []
@@ -813,7 +825,8 @@ def add_update(self, column, value, operation=None, previous=None):
             clause = CounterUpdateClause(column.db_field_name, value, previous)
         else:
             clause = AssignmentClause(column.db_field_name, value)
-        if clause.get_context_size():  # this is to exclude map removals from updates. Can go away if we drop support for C* < 1.2.4 and remove two-phase updates
+
+        if clause.get_context_size():
             self._add_assignment_clause(clause)
 
 
diff --git a/tests/integration/cqlengine/query/test_updates.py b/tests/integration/cqlengine/query/test_updates.py
@@ -106,7 +106,7 @@ def test_null_update_deletes_column(self):
             self.assertEqual(row.count, i)
             self.assertEqual(row.text, None if i == 3 else str(i))
 
-    @execute_count(9)
+    @execute_count(8)
     def test_mixed_value_and_null_update(self):
         """ tests that updating a columns value, and removing another works properly """
         partition = uuid4()
diff --git a/tests/integration/cqlengine/statements/test_assignment_clauses.py b/tests/integration/cqlengine/statements/test_assignment_clauses.py
@@ -49,21 +49,22 @@ def test_update_from_none(self):
         self.assertEqual(ctx, {'0': set((1, 2))})
 
     def test_null_update(self):
-        """ tests setting a set to None creates an empty update statement """
+        """ tests setting a set to None creates an update statement """
         c = SetUpdateClause('s', None, previous=set((1, 2)))
         c._analyze()
         c.set_context_id(0)
 
+        self.assertTrue(c._is_assignment)
         self.assertIsNone(c._assignments)
         self.assertIsNone(c._additions)
         self.assertIsNone(c._removals)
 
-        self.assertEqual(c.get_context_size(), 0)
-        self.assertEqual(str(c), '')
+        self.assertEqual(c.get_context_size(), 1)
+        self.assertEqual(str(c), '"s" = %(0)s')
 
         ctx = {}
         c.update_context(ctx)
-        self.assertEqual(ctx, {})
+        self.assertEqual(ctx, {'0': None})
 
     def test_no_update(self):
         """ tests an unchanged value creates an empty update statement """
@@ -167,6 +168,23 @@ def test_update_from_none(self):
         c.update_context(ctx)
         self.assertEqual(ctx, {'0': [1, 2, 3]})
 
+    def test_update_to_none(self):
+        c = ListUpdateClause('s', None, previous=[1, 2, 3])
+        c._analyze()
+        c.set_context_id(0)
+
+        self.assertTrue(c._is_assignment)
+        self.assertIsNone(c._assignments)
+        self.assertIsNone(c._append)
+        self.assertIsNone(c._prepend)
+
+        self.assertEqual(c.get_context_size(), 1)
+        self.assertEqual(str(c), '"s" = %(0)s')
+
+        ctx = {}
+        c.update_context(ctx)
+        self.assertEqual(ctx, {'0': None})
+
     def test_update_from_empty(self):
         c = ListUpdateClause('s', [1, 2, 3], previous=[])
         c._analyze()
diff --git a/tests/integration/cqlengine/statements/test_update_statement.py b/tests/integration/cqlengine/statements/test_update_statement.py
@@ -69,6 +69,11 @@ def test_update_set_add(self):
         us.add_update(Set(Text, db_field='a'), set((1,)), 'add')
         self.assertEqual(six.text_type(us), 'UPDATE table SET "a" = "a" + %(0)s')
 
+    def test_update_set_with_none(self):
+        us = UpdateStatement('table')
+        us.add_update(Set(Text, db_field='a'), None)
+        self.assertTrue(us.assignments)
+
     def test_update_empty_set_add_does_not_assign(self):
         us = UpdateStatement('table')
         us.add_update(Set(Text, db_field='a'), set(), 'add')
@@ -79,6 +84,11 @@ def test_update_empty_set_removal_does_not_assign(self):
         us.add_update(Set(Text, db_field='a'), set(), 'remove')
         self.assertFalse(us.assignments)
 
+    def test_update_list_with_none(self):
+        us = UpdateStatement('table')
+        us.add_update(List(Text, db_field='a'), None, previous=[1])
+        self.assertTrue(us.assignments)
+
     def test_update_list_prepend_with_empty_list(self):
         us = UpdateStatement('table')
         us.add_update(List(Text, db_field='a'), [], 'prepend')
