Fix SimpleStatement.is_lwt(): detect LWT from CQL query string

SimpleStatement.is_lwt() always returned False, which meant LWT-aware
routing (TokenAwarePolicy Paxos leader routing) and retry policies
never activated for:
- Raw CQL queries via session.execute(SimpleStatement(...))
- All cqlengine queries (which wrap everything in SimpleStatement)

This adds regex-based LWT detection to SimpleStatement that matches:
- INSERT ... IF NOT EXISTS
- UPDATE/DELETE ... IF EXISTS
- UPDATE/DELETE ... IF <column> <op> <value> (conditional)

The result is cached after the first call. This is a best-effort
heuristic; PreparedStatement continues to get the authoritative
is_lwt flag from the server during PREPARE.

Also updates the existing BatchStatement LWT test to use the new
SimpleStatement.is_lwt() directly instead of a workaround subclass.

Author: mykaul

cassandra/query.py

@@ -23,6 +23,18 @@
 import re
 import struct
 import time
+
+# Regex to detect LWT (Lightweight Transaction) queries in CQL strings.
+# Matches: INSERT ... IF NOT EXISTS, UPDATE/DELETE ... IF EXISTS,
+# and conditional updates/deletes (e.g. UPDATE ... IF col = ...).
+# Uses word boundaries and case-insensitive matching.
+# This is a best-effort heuristic for SimpleStatement; PreparedStatement
+# gets the authoritative is_lwt flag from the server.
+_LWT_PATTERN = re.compile(
+    r'\bIF\s+(?:NOT\s+)?EXISTS\b'   # IF [NOT] EXISTS
+    r'|\bIF\s+[a-zA-Z_]',             # IF <column_name> (conditional)
+    re.IGNORECASE
+)
 import warnings
 
 from cassandra import ConsistencyLevel, OperationTimedOut
@@ -416,6 +428,24 @@ def __str__(self):
                 (self.query_string, consistency))
     __repr__ = __str__
 
+    def is_lwt(self):
+        """
+        Detect whether this query is a Lightweight Transaction (LWT) by
+        inspecting the query string for ``IF [NOT] EXISTS`` or ``IF <condition>``
+        clauses.
+
+        This is a best-effort heuristic. For authoritative LWT detection,
+        use :class:`.PreparedStatement` which gets the ``is_lwt`` flag from
+        the server during PREPARE.
+
+        The result is cached after the first call.
+        """
+        try:
+            return self._cached_is_lwt
+        except AttributeError:
+            self._cached_is_lwt = bool(_LWT_PATTERN.search(self._query_string))
+            return self._cached_is_lwt
+
 
 class PreparedStatement(object):
     """

tests/unit/test_query.py

@@ -103,15 +103,121 @@ def test_is_lwt_propagates_from_statements(self):
         batch_with_bound.add(bound_lwt)
         assert batch_with_bound.is_lwt() is True
 
-        class LwtSimpleStatement(SimpleStatement):
-            def __init__(self):
-                super(LwtSimpleStatement, self).__init__(
-                    "INSERT INTO test.table (id) VALUES (2) IF NOT EXISTS"
-                )
-
-            def is_lwt(self):
-                return True
+        # SimpleStatement now detects LWT from query string (no subclass needed)
+        lwt_simple = SimpleStatement(
+            "INSERT INTO test.table (id) VALUES (2) IF NOT EXISTS"
+        )
+        assert lwt_simple.is_lwt() is True
 
         batch_with_simple = BatchStatement()
-        batch_with_simple.add(LwtSimpleStatement())
+        batch_with_simple.add(lwt_simple)
         assert batch_with_simple.is_lwt() is True
+
+class SimpleStatementIsLwtTest(unittest.TestCase):
+    """Tests for SimpleStatement.is_lwt() CQL-based LWT detection."""
+
+    # --- INSERT IF NOT EXISTS ---
+
+    def test_insert_if_not_exists(self):
+        s = SimpleStatement("INSERT INTO ks.t (a) VALUES (1) IF NOT EXISTS")
+        assert s.is_lwt() is True
+
+    def test_insert_if_not_exists_lowercase(self):
+        s = SimpleStatement("insert into ks.t (a) values (1) if not exists")
+        assert s.is_lwt() is True
+
+    def test_insert_if_not_exists_mixed_case(self):
+        s = SimpleStatement("INSERT INTO ks.t (a) VALUES (1) If Not Exists")
+        assert s.is_lwt() is True
+
+    # --- UPDATE IF EXISTS ---
+
+    def test_update_if_exists(self):
+        s = SimpleStatement("UPDATE ks.t SET a=1 WHERE k=1 IF EXISTS")
+        assert s.is_lwt() is True
+
+    # --- DELETE IF EXISTS ---
+
+    def test_delete_if_exists(self):
+        s = SimpleStatement("DELETE FROM ks.t WHERE k=1 IF EXISTS")
+        assert s.is_lwt() is True
+
+    # --- Conditional UPDATE (IF <column> = <value>) ---
+
+    def test_conditional_update_equals(self):
+        s = SimpleStatement("UPDATE ks.t SET a=1 WHERE k=1 IF a = 2")
+        assert s.is_lwt() is True
+
+    def test_conditional_update_not_equals(self):
+        s = SimpleStatement("UPDATE ks.t SET a=1 WHERE k=1 IF a != 2")
+        assert s.is_lwt() is True
+
+    def test_conditional_update_greater_than(self):
+        s = SimpleStatement("UPDATE ks.t SET a=1 WHERE k=1 IF a > 2")
+        assert s.is_lwt() is True
+
+    def test_conditional_update_multiple_conditions(self):
+        s = SimpleStatement(
+            "UPDATE ks.t SET a=1 WHERE k=1 IF a = 2 AND b = 3")
+        assert s.is_lwt() is True
+
+    # --- Conditional DELETE ---
+
+    def test_conditional_delete(self):
+        s = SimpleStatement("DELETE FROM ks.t WHERE k=1 IF a = 2")
+        assert s.is_lwt() is True
+
+    # --- Non-LWT queries (should return False) ---
+
+    def test_select_not_lwt(self):
+        s = SimpleStatement("SELECT * FROM ks.t WHERE k=1")
+        assert s.is_lwt() is False
+
+    def test_insert_without_if(self):
+        s = SimpleStatement("INSERT INTO ks.t (a) VALUES (1)")
+        assert s.is_lwt() is False
+
+    def test_update_without_if(self):
+        s = SimpleStatement("UPDATE ks.t SET a=1 WHERE k=1")
+        assert s.is_lwt() is False
+
+    def test_delete_without_if(self):
+        s = SimpleStatement("DELETE FROM ks.t WHERE k=1")
+        assert s.is_lwt() is False
+
+    def test_create_table_with_if_not_exists(self):
+        """DDL IF NOT EXISTS should also match — this is harmless since
+        DDL queries don't use token-aware routing anyway."""
+        s = SimpleStatement("CREATE TABLE IF NOT EXISTS ks.t (a int PRIMARY KEY)")
+        assert s.is_lwt() is True  # False positive but harmless
+
+    # --- Caching ---
+
+    def test_result_is_cached(self):
+        s = SimpleStatement("INSERT INTO ks.t (a) VALUES (1) IF NOT EXISTS")
+        assert s.is_lwt() is True
+        assert s.is_lwt() is True  # should use cache
+        assert s._cached_is_lwt is True
+
+    def test_non_lwt_result_is_cached(self):
+        s = SimpleStatement("SELECT * FROM ks.t")
+        assert s.is_lwt() is False
+        assert s._cached_is_lwt is False
+
+    # --- Edge cases ---
+
+    def test_multiline_query(self):
+        s = SimpleStatement("""
+            INSERT INTO ks.t (a, b)
+            VALUES (1, 2)
+            IF NOT EXISTS
+        """)
+        assert s.is_lwt() is True
+
+    def test_extra_whitespace(self):
+        s = SimpleStatement("UPDATE ks.t SET a=1 WHERE k=1  IF   EXISTS")
+        assert s.is_lwt() is True
+
+    def test_tab_separated(self):
+        s = SimpleStatement("DELETE FROM ks.t WHERE k=1\tIF\tEXISTS")
+        assert s.is_lwt() is True