cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jji...@apache.org
Subject [4/7] cassandra git commit: Range deletes in a CAS batch are ignored
Date Mon, 11 Sep 2017 16:38:14 GMT
Range deletes in a CAS batch are ignored

Patch by Jeff Jirsa; Reviewed by Jay Zhuang, Sylvain Lebresne for CASSANDRA-13655


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/433f24cb
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/433f24cb
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/433f24cb

Branch: refs/heads/cassandra-3.11
Commit: 433f24cb04dbcf74029a918ee73155f78d5f8111
Parents: ae88fd6
Author: Jeff Jirsa <jjirsa@apple.com>
Authored: Mon Sep 11 09:35:01 2017 -0700
Committer: Jeff Jirsa <jjirsa@apple.com>
Committed: Mon Sep 11 09:35:01 2017 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../cql3/statements/BatchStatement.java         |  36 ++++--
 .../cql3/statements/CQL3CasRequest.java         |  31 ++++++
 .../cql3/statements/ModificationStatement.java  |  15 ++-
 .../org/apache/cassandra/cql3/BatchTests.java   |  51 +++++++--
 .../cql3/validation/operations/BatchTest.java   | 111 +++++++++++++++++++
 6 files changed, 219 insertions(+), 26 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/433f24cb/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index f4360be..76d155e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0.15
+ * Range deletes in a CAS batch are ignored (CASSANDRA-13655)
  * Change repair midpoint logging for tiny ranges (CASSANDRA-13603)
  * Better handle corrupt final commitlog segment (CASSANDRA-11995)
  * StreamingHistogram is not thread safe (CASSANDRA-13756)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/433f24cb/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
index 76a6460..cd9358c 100644
--- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
@@ -417,18 +417,36 @@ public class BatchStatement implements CQLStatement
                        "IN on the clustering key columns is not supported with conditional
%s",
                        statement.type.isUpdate()? "updates" : "deletions");
 
-            Clustering clustering = Iterables.getOnlyElement(statement.createClustering(statementOptions));
+            if (statement.hasSlices())
+            {
+                // All of the conditions require meaningful Clustering, not Slices
+                assert !statement.hasConditions();
+
+                Slices slices = statement.createSlices(statementOptions);
+                // If all the ranges were invalid we do not need to do anything.
+                if (slices.isEmpty())
+                    continue;
+
+                for (Slice slice : slices)
+                {
+                    casRequest.addRangeDeletion(slice, statement, statementOptions, timestamp);
+                }
 
-            if (statement.hasConditions())
+            }
+            else
             {
-                statement.addConditions(clustering, casRequest, statementOptions);
-                // As soon as we have a ifNotExists, we set columnsWithConditions to null
so that everything is in the resultSet
-                if (statement.hasIfNotExistCondition() || statement.hasIfExistCondition())
-                    columnsWithConditions = null;
-                else if (columnsWithConditions != null)
-                    Iterables.addAll(columnsWithConditions, statement.getColumnsWithConditions());
+                Clustering clustering = Iterables.getOnlyElement(statement.createClustering(statementOptions));
+                if (statement.hasConditions())
+                {
+                    statement.addConditions(clustering, casRequest, statementOptions);
+                    // As soon as we have a ifNotExists, we set columnsWithConditions to
null so that everything is in the resultSet
+                    if (statement.hasIfNotExistCondition() || statement.hasIfExistCondition())
+                        columnsWithConditions = null;
+                    else if (columnsWithConditions != null)
+                        Iterables.addAll(columnsWithConditions, statement.getColumnsWithConditions());
+                }
+                casRequest.addRowUpdate(clustering, statement, statementOptions, timestamp);
             }
-            casRequest.addRowUpdate(clustering, statement, statementOptions, timestamp);
         }
 
         return Pair.create(casRequest, columnsWithConditions);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/433f24cb/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
index e226a2a..e14ae6c 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
@@ -56,6 +56,7 @@ public class CQL3CasRequest implements CASRequest
     private final TreeMap<Clustering, RowCondition> conditions;
 
     private final List<RowUpdate> updates = new ArrayList<>();
+    private final List<RangeDeletion> rangeDeletions = new ArrayList<>();
 
     public CQL3CasRequest(CFMetaData cfm,
                           DecoratedKey key,
@@ -78,6 +79,11 @@ public class CQL3CasRequest implements CASRequest
         updates.add(new RowUpdate(clustering, stmt, options, timestamp));
     }
 
+    public void addRangeDeletion(Slice slice, ModificationStatement stmt, QueryOptions options,
long timestamp)
+    {
+        rangeDeletions.add(new RangeDeletion(slice, stmt, options, timestamp));
+    }
+
     public void addNotExist(Clustering clustering) throws InvalidRequestException
     {
         addExistsCondition(clustering, new NotExistCondition(clustering), true);
@@ -226,6 +232,8 @@ public class CQL3CasRequest implements CASRequest
         PartitionUpdate update = new PartitionUpdate(cfm, key, updatedColumns(), conditions.size());
         for (RowUpdate upd : updates)
             upd.applyUpdates(current, update);
+        for (RangeDeletion upd : rangeDeletions)
+            upd.applyUpdates(current, update);
 
         Keyspace.openAndGetStore(cfm).indexManager.validate(update);
 
@@ -264,6 +272,29 @@ public class CQL3CasRequest implements CASRequest
         }
     }
 
+    private class RangeDeletion
+    {
+        private final Slice slice;
+        private final ModificationStatement stmt;
+        private final QueryOptions options;
+        private final long timestamp;
+
+        private RangeDeletion(Slice slice, ModificationStatement stmt, QueryOptions options,
long timestamp)
+        {
+            this.slice = slice;
+            this.stmt = stmt;
+            this.options = options;
+            this.timestamp = timestamp;
+        }
+
+        public void applyUpdates(FilteredPartition current, PartitionUpdate updates) throws
InvalidRequestException
+        {
+            Map<DecoratedKey, Partition> map = stmt.requiresRead() ? Collections.<DecoratedKey,
Partition>singletonMap(key, current) : null;
+            UpdateParameters params = new UpdateParameters(cfm, updates.columns(), options,
timestamp, stmt.getTimeToLive(options), map);
+            stmt.addUpdateForKey(updates, slice, params);
+        }
+    }
+
     private static abstract class RowCondition
     {
         public final Clustering clustering;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/433f24cb/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index 1722f02..0afd34d 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -402,6 +402,13 @@ public abstract class ModificationStatement implements CQLStatement
         return !conditions.isEmpty();
     }
 
+    public boolean hasSlices()
+    {
+        return type.allowClusteringColumnSlices()
+               && getRestrictions().hasClusteringColumnsRestriction()
+               && getRestrictions().isColumnRange();
+    }
+
     public ResultMessage execute(QueryState queryState, QueryOptions options)
     throws RequestExecutionException, RequestValidationException
     {
@@ -626,11 +633,9 @@ public abstract class ModificationStatement implements CQLStatement
     {
         List<ByteBuffer> keys = buildPartitionKeyNames(options);
 
-        if (type.allowClusteringColumnSlices()
-                && restrictions.hasClusteringColumnsRestriction()
-                && restrictions.isColumnRange())
+        if (hasSlices())
         {
-            Slices slices = createSlice(options);
+            Slices slices = createSlices(options);
 
             // If all the ranges were invalid we do not need to do anything.
             if (slices.isEmpty())
@@ -693,7 +698,7 @@ public abstract class ModificationStatement implements CQLStatement
         }
     }
 
-    private Slices createSlice(QueryOptions options)
+    Slices createSlices(QueryOptions options)
     {
         SortedSet<Slice.Bound> startBounds = restrictions.getClusteringColumnsBounds(Bound.START,
options);
         SortedSet<Slice.Bound> endBounds = restrictions.getClusteringColumnsBounds(Bound.END,
options);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/433f24cb/test/unit/org/apache/cassandra/cql3/BatchTests.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/cql3/BatchTests.java b/test/unit/org/apache/cassandra/cql3/BatchTests.java
index 73923fb..260db4e 100644
--- a/test/unit/org/apache/cassandra/cql3/BatchTests.java
+++ b/test/unit/org/apache/cassandra/cql3/BatchTests.java
@@ -30,7 +30,7 @@ import org.junit.Test;
 
 import java.io.IOException;
 
-public class BatchTests
+public class BatchTests extends  CQLTester
 {
     private static EmbeddedCassandraService cassandra;
 
@@ -39,6 +39,7 @@ public class BatchTests
 
     private static PreparedStatement counter;
     private static PreparedStatement noncounter;
+    private static PreparedStatement clustering;
 
     @BeforeClass()
     public static void setup() throws ConfigurationException, IOException
@@ -59,58 +60,79 @@ public class BatchTests
                 "  id int PRIMARY KEY,\n" +
                 "  val counter,\n" +
                 ");");
+        session.execute("CREATE TABLE junit.clustering (\n" +
+                "  id int,\n" +
+                "  clustering1 int,\n" +
+                "  clustering2 int,\n" +
+                "  clustering3 int,\n" +
+                "  val text, \n" +
+                " PRIMARY KEY(id, clustering1, clustering2, clustering3)" +
+                ");");
 
 
         noncounter = session.prepare("insert into junit.noncounter(id, val)values(?,?)");
         counter = session.prepare("update junit.counter set val = val + ? where id = ?");
+        clustering = session.prepare("insert into junit.clustering(id, clustering1, clustering2,
clustering3, val) values(?,?,?,?,?)");
     }
 
     @Test(expected = InvalidQueryException.class)
     public void testMixedInCounterBatch()
     {
-       sendBatch(BatchStatement.Type.COUNTER, true, true);
+       sendBatch(BatchStatement.Type.COUNTER, true, true, false);
     }
 
     @Test(expected = InvalidQueryException.class)
     public void testMixedInLoggedBatch()
     {
-        sendBatch(BatchStatement.Type.LOGGED, true, true);
+        sendBatch(BatchStatement.Type.LOGGED, true, true, false);
     }
 
     @Test(expected = InvalidQueryException.class)
     public void testMixedInUnLoggedBatch()
     {
-        sendBatch(BatchStatement.Type.UNLOGGED, true, true);
+        sendBatch(BatchStatement.Type.UNLOGGED, true, true, false);
     }
 
     @Test(expected = InvalidQueryException.class)
     public void testNonCounterInCounterBatch()
     {
-        sendBatch(BatchStatement.Type.COUNTER, false, true);
+        sendBatch(BatchStatement.Type.COUNTER, false, true, false);
     }
 
     @Test
     public void testNonCounterInLoggedBatch()
     {
-        sendBatch(BatchStatement.Type.LOGGED, false, true);
+        sendBatch(BatchStatement.Type.LOGGED, false, true, false);
     }
 
     @Test
     public void testNonCounterInUnLoggedBatch()
     {
-        sendBatch(BatchStatement.Type.UNLOGGED, false, true);
+        sendBatch(BatchStatement.Type.UNLOGGED, false, true, false);
     }
 
     @Test
     public void testCounterInCounterBatch()
     {
-        sendBatch(BatchStatement.Type.COUNTER, true, false);
+        sendBatch(BatchStatement.Type.COUNTER, true, false, false);
     }
 
     @Test
     public void testCounterInUnLoggedBatch()
     {
-        sendBatch(BatchStatement.Type.UNLOGGED, true, false);
+        sendBatch(BatchStatement.Type.UNLOGGED, true, false, false);
+    }
+
+    @Test
+    public void testTableWithClusteringInLoggedBatch()
+    {
+        sendBatch(BatchStatement.Type.LOGGED, false, false, true);
+    }
+
+    @Test
+    public void testTableWithClusteringInUnLoggedBatch()
+    {
+        sendBatch(BatchStatement.Type.UNLOGGED, false, false, true);
     }
 
     @Test
@@ -123,7 +145,7 @@ public class BatchTests
     @Test(expected = InvalidQueryException.class)
     public void testCounterInLoggedBatch()
     {
-        sendBatch(BatchStatement.Type.LOGGED, true, false);
+        sendBatch(BatchStatement.Type.LOGGED, true, false, false);
     }
 
     @Test(expected = InvalidQueryException.class)
@@ -138,10 +160,10 @@ public class BatchTests
         session.execute(b);
     }
 
-    public void sendBatch(BatchStatement.Type type, boolean addCounter, boolean addNonCounter)
+    public void sendBatch(BatchStatement.Type type, boolean addCounter, boolean addNonCounter,
boolean addClustering)
     {
 
-        assert addCounter || addNonCounter;
+        assert addCounter || addNonCounter || addClustering;
         BatchStatement b = new BatchStatement(type);
 
         for (int i = 0; i < 10; i++)
@@ -151,6 +173,11 @@ public class BatchTests
 
             if (addCounter)
                 b.add(counter.bind((long)i, i));
+
+            if (addClustering)
+            {
+                b.add(clustering.bind(i, i, i, i, "foo"));
+            }
         }
 
         session.execute(b);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/433f24cb/test/unit/org/apache/cassandra/cql3/validation/operations/BatchTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/BatchTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/BatchTest.java
index e8f169d..87d0cde 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/operations/BatchTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/operations/BatchTest.java
@@ -239,4 +239,115 @@ public class BatchTest extends CQLTester
                    row(1,2,2),
                    row(1,3,3));
     }
+
+    @Test
+    public void testBatchAndConditionalInteraction() throws Throwable
+    {
+
+        createTable(String.format("CREATE TABLE %s.clustering (\n" +
+                "  id int,\n" +
+                "  clustering1 int,\n" +
+                "  clustering2 int,\n" +
+                "  clustering3 int,\n" +
+                "  val int, \n" +
+                " PRIMARY KEY(id, clustering1, clustering2, clustering3)" +
+                ")", KEYSPACE));
+
+        execute("DELETE FROM " + KEYSPACE +".clustering WHERE id=1");
+
+        String clusteringInsert = "INSERT INTO " + KEYSPACE + ".clustering(id, clustering1,
clustering2, clustering3, val) VALUES(%s, %s, %s, %s, %s); ";
+        String clusteringUpdate = "UPDATE " + KEYSPACE + ".clustering SET val=%s WHERE id=%s
AND clustering1=%s AND clustering2=%s AND clustering3=%s ;";
+        String clusteringConditionalUpdate = "UPDATE " + KEYSPACE + ".clustering SET val=%s
WHERE id=%s AND clustering1=%s AND clustering2=%s AND clustering3=%s IF val=%s ;";
+        String clusteringDelete = "DELETE FROM " + KEYSPACE + ".clustering WHERE id=%s AND
clustering1=%s AND clustering2=%s AND clustering3=%s ;";
+        String clusteringRangeDelete = "DELETE FROM " + KEYSPACE + ".clustering WHERE id=%s
AND clustering1=%s ;";
+
+
+        execute("BEGIN BATCH " + String.format(clusteringInsert, 1, 1, 1, 1, 1) + " APPLY
BATCH");
+
+        assertRows(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"), row(1,
1, 1, 1, 1));
+
+        StringBuilder cmd2 = new StringBuilder();
+        cmd2.append("BEGIN BATCH ");
+        cmd2.append(String.format(clusteringInsert, 1, 1, 1, 2, 2));
+        cmd2.append(String.format(clusteringConditionalUpdate, 11, 1, 1, 1, 1, 1));
+        cmd2.append("APPLY BATCH ");
+        execute(cmd2.toString());
+
+
+        assertRows(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"),
+                row(1, 1, 1, 1, 11),
+                row(1, 1, 1, 2, 2)
+        );
+
+
+        StringBuilder cmd3 = new StringBuilder();
+        cmd3.append("BEGIN BATCH ");
+        cmd3.append(String.format(clusteringInsert, 1, 1, 2, 3, 23));
+        cmd3.append(String.format(clusteringConditionalUpdate, 22, 1, 1, 1, 2, 2));
+        cmd3.append(String.format(clusteringDelete, 1, 1, 1, 1));
+        cmd3.append("APPLY BATCH ");
+        execute(cmd3.toString());
+
+        assertRows(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"),
+                row(1, 1, 1, 2, 22),
+                row(1, 1, 2, 3, 23)
+        );
+
+        StringBuilder cmd4 = new StringBuilder();
+        cmd4.append("BEGIN BATCH ");
+        cmd4.append(String.format(clusteringInsert, 1, 2, 3, 4, 1234));
+        cmd4.append(String.format(clusteringConditionalUpdate, 234, 1, 1, 1, 2, 22));
+        cmd4.append("APPLY BATCH ");
+        execute(cmd4.toString());
+
+        System.out.println(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"));
+        assertRows(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"),
+                row(1, 1, 1, 2, 234),
+                row(1, 1, 2, 3, 23),
+                row(1, 2, 3, 4, 1234)
+        );
+
+        StringBuilder cmd5 = new StringBuilder();
+        cmd5.append("BEGIN BATCH ");
+        cmd5.append(String.format(clusteringRangeDelete, 1, 2));
+        cmd5.append(String.format(clusteringConditionalUpdate, 1234, 1, 1, 1, 2, 234));
+        cmd5.append("APPLY BATCH ");
+        execute(cmd5.toString());
+
+        System.out.println(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"));
+        assertRows(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"),
+                row(1, 1, 1, 2, 1234),
+                row(1, 1, 2, 3, 23)
+        );
+
+        StringBuilder cmd6 = new StringBuilder();
+        cmd6.append("BEGIN BATCH ");
+        cmd6.append(String.format(clusteringUpdate, 345, 1, 3, 4, 5));
+        cmd6.append(String.format(clusteringConditionalUpdate, 1, 1, 1, 1, 2, 1234));
+        cmd6.append("APPLY BATCH ");
+        execute(cmd6.toString());
+
+        System.out.println(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"));
+        assertRows(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"),
+                row(1, 1, 1, 2, 1),
+                row(1, 1, 2, 3, 23),
+                row(1, 3, 4, 5, 345)
+        );
+
+
+        StringBuilder cmd7 = new StringBuilder();
+        cmd7.append("BEGIN BATCH ");
+        cmd7.append(String.format(clusteringDelete, 1, 3, 4, 5));
+        cmd7.append(String.format(clusteringConditionalUpdate, 2300, 1, 1, 2, 3, 1));  //
SHOULD NOT MATCH
+        cmd7.append("APPLY BATCH ");
+        execute(cmd7.toString());
+
+        System.out.println(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"));
+        assertRows(execute("SELECT * FROM " + KEYSPACE+".clustering WHERE id=1"),
+                row(1, 1, 1, 2, 1),
+                row(1, 1, 2, 3, 23),
+                row(1, 3, 4, 5, 345)
+        );
+    }
+
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message