cqlengine: Remove two-phase query when deleting values

aboudreault · mambocab · commit e606593949be · 2018-04-27T13:16:04.000-04:00
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -3,81 +3,24 @@
 
 Features
 --------
-* Make protocol messages pluggable (PYTHON-956)
-* cqlengine: Remove default limit on QuerySets (PYTHON-517)
 * cqlengine: asynchronous execution support (PYTHON-605)
-* cqlengine: disallow Counter create, save operations (PYTHON-497)
 * cqlengine: Makes Model._table_name_ case sensitive (PYTHON-855)
-
-
-Other
------
-* Deprecate ResultSet indexing (PYTHON-945)
-* Cassandra 2.0 support removal (PYTHON-716)
-* cqlengine: remove the negative indices slicing support in ModelQuerySet (PYTHON-875)
-* Remove legacy execution parameters (PYTHON-876)
-* PreparedStatement.column_metadata should be renamed to bind_metadata (PYTHON-884)
-* Support cassandra.query.BatchType with cqlengine BatchQuery (PYTHON-888)
-* cqlengine: Remove Model.__default_ttl__ (PYTHON-889)
-* Remove Cluster.set_meta_refresh_enabled (PYTHON-890)
-
-
-3.14.0
-======
-
-Features
---------
-* Add one() function to the ResultSet API (PYTHON-947)
-* Create an utility function to fetch concurrently many keys from the same replica (PYTHON-647)
-* Allow filter queries with fields that have an index managed outside of cqlengine (PYTHON-966)
-* Twisted SSL Support (PYTHON-343)
-* Support IS NOT NULL operator in cqlengine (PYTHON-968)
-
-Other
------
-* Fix Broken Links in Docs (PYTHON-916)
-* Reevaluate MONKEY_PATCH_LOOP in test codebase (PYTHON-903)
-* Remove CASS_SERVER_VERSION and replace it for CASSANDRA_VERSION in tests (PYTHON-910)
-* Refactor CASSANDRA_VERSION to a some kind of version object (PYTHON-915)
-* Log warning when driver configures an authenticator, but server does not request authentication (PYTHON-940)
-* Warn users when using the deprecated Session.default_consistency_level (PYTHON-953)
-* Add DSE smoke test to OSS driver tests (PYTHON-894)
-* Document long compilation times and workarounds (PYTHON-868)
-* Improve error for batch WriteTimeouts (PYTHON-941)
-
-3.13.0
-======
-January 30, 2018
-
-Features
---------
 * cqlengine: LIKE filter operator (PYTHON-512)
 
 Bug Fixes
 ---------
-* Calling next() on a ResultSet throws an exception (PYTHON-463)
 * Support retry_policy in PreparedStatement (PYTHON-861)
-* cqlengine: Remove cassandra.cqlengine.query.BatchType in favor of cassandra.query.BatchType (PYTHON-888)
-* AttributeError: 'NoneType' object has no attribute 'add_timer' (PYTHON-862)
-* Support retry_policy in PreparedStatement (PYTHON-861)
-* __del__ method in Session is throwing an exception (PYTHON-813)
-* LZ4 import issue with recent versions (PYTHON-897)
-* ResponseFuture._connection can be None when returning request_id (PYTHON-853)
-* ResultSet.was_applied doesn't support batch with LWT statements (PYTHON-848)
 
 Other
 -----
-* Remove basic equality and indexing support of ResultSet (PYTHON-945)
-* cqlengine: avoid warning when unregistering connection on shutdown (PYTHON-865)
-* Fix DeprecationWarning of log.warn (PYTHON-846)
-* Fix example_mapper.py for python3 (PYTHON-860)
-* Possible deadlock on cassandra.concurrent.execute_concurrent (PYTHON-768)
-* Add some known deprecated warnings for 4.x (PYTHON-877)
-* Remove copyright dates from copyright notices (PYTHON-863)
-* Remove "Experimental" tag from execution profiles documentation (PYTHON-840)
-* request_timer metrics descriptions are slightly incorrect (PYTHON-885)
-* Remove "Experimental" tag from cqlengine connections documentation (PYTHON-892)
-* Set in documentation default consistency for operations is LOCAL_ONE (PYTHON-901)
+* Cassandra 2.0 support removal (PYTHON-716)
+* PreparedStatement.column_metadata should be renamed to bind_metadata (PYTHON-884)
+* Remove Cluster.set_meta_refresh_enabled (PYTHON-890)
+* Remove legacy execution parameters (PYTHON-876)
+* 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
 ======
@@ -93,8 +36,6 @@ Features
 * Include hash of result set metadata in prepared stmt id (PYTHON-808)
 * Add NO_COMPACT startup option (PYTHON-839)
 * Add new exception type for CDC (PYTHON-837)
-* Allow 0ms in ConstantSpeculativeExecutionPolicy (PYTHON-836)
-* Add asyncio reactor (PYTHON-507)
 
 Bug Fixes
 ---------
diff --git a/cassandra/cqlengine/query.py b/cassandra/cqlengine/query.py
@@ -1291,7 +1291,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)
@@ -1314,25 +1313,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):
         """
@@ -1481,20 +1468,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)
@@ -1509,11 +1496,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):
         """
@@ -1541,18 +1528,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
@@ -1565,7 +1548,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)
 
@@ -1591,7 +1574,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:
@@ -1603,9 +1585,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
@@ -1614,17 +1595,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
@@ -214,6 +214,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
@@ -222,7 +227,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:
@@ -253,11 +258,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:
@@ -268,7 +273,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:
@@ -286,12 +291,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
 
@@ -307,13 +317,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:
@@ -388,7 +398,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):
@@ -399,7 +411,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:
@@ -413,7 +425,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 = []
@@ -831,7 +843,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
diff --git a/tests/integration/cqlengine/statements/test_update_statement.py b/tests/integration/cqlengine/statements/test_update_statement.py
