Skip to content

Commit 124f44f

Browse files
beltranbeltran
committed
Merged PYTHON-848
2 parents fa374a8 + 947b521 commit 124f44f

5 files changed

Lines changed: 167 additions & 6 deletions

File tree

‎CHANGELOG.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ Bug Fixes
2525
* __del__ method in Session is throwing an exception (PYTHON-813)
2626
* LZ4 import issue with recent versions (PYTHON-897)
2727
* ResponseFuture._connection can be None when returning request_id (PYTHON-853)
28+
* ResultSet.was_applied doesn't support batch with LWT statements (PYTHON-848)
2829

2930
Other
3031
-----

‎cassandra/cluster.py‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4321,13 +4321,20 @@ def was_applied(self):
43214321
"""
43224322
For LWT results, returns whether the transaction was applied.
43234323
4324-
Result is indeterminate if called on a result that was not an LWT request.
4324+
Result is indeterminate if called on a result that was not an LWT request or on
4325+
a :class:`.query.BatchStatement` containing LWT. In the latter case either all the batch
4326+
succeeds or fails.
43254327
4326-
Only valid when one of tne of the internal row factories is in use.
4328+
Only valid when one of the of the internal row factories is in use.
43274329
"""
43284330
if self.response_future.row_factory not in (named_tuple_factory, dict_factory, tuple_factory):
4329-
raise RuntimeError("Cannot determine LWT result with row factory %s" % (self.response_future.row_factsory,))
4330-
if len(self.current_rows) != 1:
4331+
raise RuntimeError("Cannot determine LWT result with row factory %s" % (self.response_future.row_factory,))
4332+
4333+
is_batch_statement = isinstance(self.response_future.query, BatchStatement)
4334+
if is_batch_statement and (not self.column_names or self.column_names[0] != "[applied]"):
4335+
raise RuntimeError("No LWT were present in the BatchStatement")
4336+
4337+
if not is_batch_statement and len(self.current_rows) != 1:
43314338
raise RuntimeError("LWT result should have exactly one row. This has %d." % (len(self.current_rows)))
43324339

43334340
row = self.current_rows[0]

‎cassandra/cqlengine/query.py‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
import six
2020
from warnings import warn
2121

22-
from cassandra.query import SimpleStatement, BatchType as CBatchType
22+
from cassandra.query import SimpleStatement, BatchType as CBatchType, BatchStatement
2323
from cassandra.cqlengine import columns, CQLEngineException, ValidationError, UnicodeMixin
2424
from cassandra.cqlengine import connection as conn
2525
from cassandra.cqlengine.functions import Token, BaseQueryFunction, QueryValue
@@ -67,7 +67,9 @@ class MultipleObjectsReturned(QueryException):
6767

6868
def check_applied(result):
6969
"""
70-
Raises LWTException if it looks like a failed LWT request.
70+
Raises LWTException if it looks like a failed LWT request. A LWTException
71+
won't be raised in the special case in which there are several failed LWT
72+
in a :class:`~cqlengine.query.BatchQuery`.
7173
"""
7274
try:
7375
applied = result.was_applied

‎tests/integration/cqlengine/test_lwt_conditional.py‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,13 @@ class TestConditionalModel(Model):
3636
text = columns.Text(required=False)
3737

3838

39+
class TestUpdateModel(Model):
40+
partition = columns.Integer(primary_key=True)
41+
cluster = columns.Integer(primary_key=True)
42+
value = columns.Integer(required=False)
43+
text = columns.Text(required=False, index=True)
44+
45+
3946
@greaterthancass20
4047
class TestConditional(BaseCassEngTestCase):
4148

@@ -135,6 +142,28 @@ def test_batch_update_conditional(self):
135142
updated = TestConditionalModel.objects(id=id).first()
136143
self.assertEqual(updated.text, 'something else')
137144

145+
@unittest.skip("Skipping until PYTHON-943 is resolved")
146+
def test_batch_update_conditional_several_rows(self):
147+
sync_table(TestUpdateModel)
148+
self.addCleanup(drop_table, TestUpdateModel)
149+
150+
first_row = TestUpdateModel.create(partition=1, cluster=1, value=5, text="something")
151+
second_row = TestUpdateModel.create(partition=1, cluster=2, value=5, text="something")
152+
153+
b = BatchQuery()
154+
TestUpdateModel.batch(b).if_not_exists().create(partition=1, cluster=1, value=5, text='something else')
155+
TestUpdateModel.batch(b).if_not_exists().create(partition=1, cluster=2, value=5, text='something else')
156+
TestUpdateModel.batch(b).if_not_exists().create(partition=1, cluster=3, value=5, text='something else')
157+
158+
# The response will be more than two rows because two of the inserts will fail
159+
with self.assertRaises(LWTException):
160+
b.execute()
161+
162+
first_row.delete()
163+
second_row.delete()
164+
b.execute()
165+
166+
138167
def test_delete_conditional(self):
139168
# DML path
140169
t = TestConditionalModel.create(text='something', count=5)

‎tests/integration/standard/test_query.py‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -842,11 +842,20 @@ def setUp(self):
842842
v int )'''
843843
self.session.execute(ddl)
844844

845+
ddl = '''
846+
CREATE TABLE test3rf.lwt_clustering (
847+
k int,
848+
c int,
849+
v int,
850+
PRIMARY KEY (k, c))'''
851+
self.session.execute(ddl)
852+
845853
def tearDown(self):
846854
"""
847855
Shutdown cluster
848856
"""
849857
self.session.execute("DROP TABLE test3rf.lwt")
858+
self.session.execute("DROP TABLE test3rf.lwt_clustering")
850859
self.cluster.shutdown()
851860

852861
def test_no_connection_refused_on_timeout(self):
@@ -892,6 +901,119 @@ def test_no_connection_refused_on_timeout(self):
892901
# Make sure test passed
893902
self.assertTrue(received_timeout)
894903

904+
def test_was_applied_batch_stmt(self):
905+
"""
906+
Test to ensure `:attr:cassandra.cluster.ResultSet.was_applied` works as expected
907+
with Batchstatements.
908+
909+
For both type of batches verify was_applied has the correct result
910+
under different scenarios:
911+
- If on LWT fails the rest of the statements fail including normal UPSERTS
912+
- If on LWT fails the rest of the statements fail
913+
- All the queries succeed
914+
915+
@since 3.14
916+
@jira_ticket PYTHON-848
917+
@expected_result `:attr:cassandra.cluster.ResultSet.was_applied` is updated as
918+
expected
919+
920+
@test_category query
921+
"""
922+
for batch_type in (BatchType.UNLOGGED, BatchType.LOGGED):
923+
batch_statement = BatchStatement(batch_type)
924+
batch_statement.add_all(["INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 0, 10);",
925+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 1, 10);",
926+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 2, 10);"], [None] * 3)
927+
result = self.session.execute(batch_statement)
928+
#self.assertTrue(result.was_applied)
929+
930+
# Should fail since (0, 0, 10) have already been written
931+
# The non conditional insert shouldn't be written as well
932+
batch_statement = BatchStatement(batch_type)
933+
batch_statement.add_all(["INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 0, 10) IF NOT EXISTS;",
934+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 3, 10) IF NOT EXISTS;",
935+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 4, 10);",
936+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 5, 10) IF NOT EXISTS;"], [None] * 4)
937+
result = self.session.execute(batch_statement)
938+
self.assertFalse(result.was_applied)
939+
940+
all_rows = self.session.execute("SELECT * from test3rf.lwt_clustering")
941+
# Verify the non conditional insert hasn't been inserted
942+
self.assertEqual(len(all_rows.current_rows), 3)
943+
944+
# Should fail since (0, 0, 10) have already been written
945+
batch_statement = BatchStatement(batch_type)
946+
batch_statement.add_all(["INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 0, 10) IF NOT EXISTS;",
947+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 3, 10) IF NOT EXISTS;",
948+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 5, 10) IF NOT EXISTS;"], [None] * 3)
949+
result = self.session.execute(batch_statement)
950+
self.assertFalse(result.was_applied)
951+
952+
# Should fail since (0, 0, 10) have already been written
953+
batch_statement.add("INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 0, 10) IF NOT EXISTS;")
954+
result = self.session.execute(batch_statement)
955+
self.assertFalse(result.was_applied)
956+
957+
# Should succeed
958+
batch_statement = BatchStatement(batch_type)
959+
batch_statement.add_all(["INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 3, 10) IF NOT EXISTS;",
960+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 4, 10) IF NOT EXISTS;",
961+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 5, 10) IF NOT EXISTS;"], [None] * 3)
962+
963+
result = self.session.execute(batch_statement)
964+
self.assertTrue(result.was_applied)
965+
966+
all_rows = self.session.execute("SELECT * from test3rf.lwt_clustering")
967+
for i, row in enumerate(all_rows):
968+
self.assertEqual((0, i, 10), (row[0], row[1], row[2]))
969+
970+
self.session.execute("TRUNCATE TABLE test3rf.lwt_clustering")
971+
972+
def test_empty_batch_statement(self):
973+
"""
974+
Test to ensure `:attr:cassandra.cluster.ResultSet.was_applied` works as expected
975+
with empty Batchstatements.
976+
977+
@since 3.14
978+
@jira_ticket PYTHON-848
979+
@expected_result an Exception is raised
980+
expected
981+
982+
@test_category query
983+
"""
984+
batch_statement = BatchStatement()
985+
results = self.session.execute(batch_statement)
986+
with self.assertRaises(RuntimeError):
987+
results.was_applied
988+
989+
@unittest.skip("Skipping until PYTHON-943 is resolved")
990+
def test_was_applied_batch_string(self):
991+
batch_statement = BatchStatement(BatchType.LOGGED)
992+
batch_statement.add_all(["INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 0, 10);",
993+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 1, 10);",
994+
"INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 2, 10);"], [None] * 3)
995+
self.session.execute(batch_statement)
996+
997+
batch_str = """
998+
BEGIN unlogged batch
999+
INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 0, 10) IF NOT EXISTS;
1000+
INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 1, 10) IF NOT EXISTS;
1001+
INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 2, 10) IF NOT EXISTS;
1002+
APPLY batch;
1003+
"""
1004+
result = self.session.execute(batch_str)
1005+
self.assertFalse(result.was_applied)
1006+
1007+
batch_str = """
1008+
BEGIN unlogged batch
1009+
INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 3, 10) IF NOT EXISTS;
1010+
INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 4, 10) IF NOT EXISTS;
1011+
INSERT INTO test3rf.lwt_clustering (k, c, v) VALUES (0, 5, 10) IF NOT EXISTS;
1012+
APPLY batch;
1013+
"""
1014+
result = self.session.execute(batch_str)
1015+
self.assertTrue(result.was_applied)
1016+
8951017

8961018
class BatchStatementDefaultRoutingKeyTests(unittest.TestCase):
8971019
# Test for PYTHON-126: BatchStatement.add() should set the routing key of the first added prepared statement

0 commit comments

Comments
 (0)