You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jj...@apache.org on 2017/09/11 16:38:14 UTC
[4/7] cassandra git commit: Range deletes in a CAS batch are ignored
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 <jj...@apple.com>
Authored: Mon Sep 11 09:35:01 2017 -0700
Committer: Jeff Jirsa <jj...@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