CQLSSTableWriter does not allow Update statement Patch by Alex Petrov; reviewed by Stefania Alborghetti for CASSANDRA-12450
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/47c473ae Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/47c473ae Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/47c473ae Branch: refs/heads/trunk Commit: 47c473ae3c0455e02b7a8529746988afdfdd9c54 Parents: 6216ce8 Author: Alex Petrov <oleksandr.pet...@gmail.com> Authored: Fri Oct 7 12:09:12 2016 +0800 Committer: Stefania Alborghetti <stefania.alborghe...@datastax.com> Committed: Fri Oct 7 12:11:32 2016 +0800 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../apache/cassandra/cql3/QueryProcessor.java | 18 ++++- .../cassandra/io/sstable/CQLSSTableWriter.java | 36 +++------- .../io/sstable/CQLSSTableWriterTest.java | 71 ++++++++++++++++++-- .../apache/cassandra/stress/StressProfile.java | 5 +- 5 files changed, 93 insertions(+), 38 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/47c473ae/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 08de041..f566b1b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.10 + * CQLSSTableWriter does not allow Update statement (CASSANDRA-12450) * Config class uses boxed types but DD exposes primitive types (CASSANDRA-12199) * Add pre- and post-shutdown hooks to Storage Service (CASSANDRA-12461) * Add hint delivery metrics (CASSANDRA-12693) http://git-wip-us.apache.org/repos/asf/cassandra/blob/47c473ae/src/java/org/apache/cassandra/cql3/QueryProcessor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index c91105e..5313a1a 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -546,6 +546,22 @@ public class QueryProcessor implements QueryHandler return statement.prepare(); } + public static <T extends ParsedStatement> T parseStatement(String queryStr, Class<T> klass, String type) throws SyntaxException + { + try + { + ParsedStatement stmt = parseStatement(queryStr); + + if (!klass.isAssignableFrom(stmt.getClass())) + throw new IllegalArgumentException("Invalid query, must be a " + type + " statement but was: " + stmt.getClass()); + + return klass.cast(stmt); + } + catch (RequestValidationException e) + { + throw new IllegalArgumentException(e.getMessage(), e); + } + } public static ParsedStatement parseStatement(String queryStr) throws SyntaxException { try @@ -622,7 +638,7 @@ public class QueryProcessor implements QueryHandler while (iterator.hasNext()) { Map.Entry<MD5Digest, ParsedStatement.Prepared> entry = iterator.next(); - if (shouldInvalidate(ksName, cfName, entry.getValue().statement)) + if (shouldInvalidate(ksName, cfName, entry.getValue().statement)) { SystemKeyspace.removePreparedStatement(entry.getKey()); iterator.remove(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/47c473ae/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java index 8a9d01d..a195235 100644 --- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java @@ -41,6 +41,7 @@ import org.apache.cassandra.cql3.UpdateParameters; import org.apache.cassandra.cql3.functions.UDHelper; import org.apache.cassandra.cql3.statements.CreateTableStatement; import org.apache.cassandra.cql3.statements.CreateTypeStatement; +import org.apache.cassandra.cql3.statements.ModificationStatement; import org.apache.cassandra.cql3.statements.ParsedStatement; import org.apache.cassandra.cql3.statements.UpdateStatement; import org.apache.cassandra.db.Clustering; @@ -50,7 +51,6 @@ import org.apache.cassandra.db.partitions.Partition; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.exceptions.InvalidRequestException; -import org.apache.cassandra.exceptions.RequestValidationException; import org.apache.cassandra.exceptions.SyntaxException; import org.apache.cassandra.io.sstable.format.SSTableFormat; import org.apache.cassandra.schema.KeyspaceMetadata; @@ -343,7 +343,7 @@ public class CQLSSTableWriter implements Closeable private CreateTableStatement.RawStatement schemaStatement; private final List<CreateTypeStatement> typeStatements; - private UpdateStatement.ParsedInsert insertStatement; + private ModificationStatement.Parsed insertStatement; private IPartitioner partitioner; private boolean sorted = false; @@ -391,7 +391,7 @@ public class CQLSSTableWriter implements Closeable public Builder withType(String typeDefinition) throws SyntaxException { - typeStatements.add(parseStatement(typeDefinition, CreateTypeStatement.class, "CREATE TYPE")); + typeStatements.add(QueryProcessor.parseStatement(typeDefinition, CreateTypeStatement.class, "CREATE TYPE")); return this; } @@ -411,7 +411,7 @@ public class CQLSSTableWriter implements Closeable */ public Builder forTable(String schema) { - this.schemaStatement = parseStatement(schema, CreateTableStatement.RawStatement.class, "CREATE TABLE"); + this.schemaStatement = QueryProcessor.parseStatement(schema, CreateTableStatement.RawStatement.class, "CREATE TABLE"); return this; } @@ -432,14 +432,13 @@ public class CQLSSTableWriter implements Closeable } /** - * The INSERT statement defining the order of the values to add for a given CQL row. + * The INSERT or UPDATE statement defining the order of the values to add for a given CQL row. * <p> * Please note that the provided INSERT statement <b>must</b> use a fully-qualified - * table name, one that include the keyspace name. Morewover, said statement must use - * bind variables since it is those bind variables that will be bound to values by the - * resulting writer. + * table name, one that include the keyspace name. Moreover, said statement must use + * bind variables since these variables will be bound to values by the resulting writer. * <p> - * This is a mandatory option, and this needs to be called after foTable(). + * This is a mandatory option. * * @param insert an insertion statement that defines the order * of column values to use. @@ -450,7 +449,7 @@ public class CQLSSTableWriter implements Closeable */ public Builder using(String insert) { - this.insertStatement = parseStatement(insert, UpdateStatement.ParsedInsert.class, "INSERT"); + this.insertStatement = QueryProcessor.parseStatement(insert, ModificationStatement.Parsed.class, "INSERT/UPDATE"); return this; } @@ -586,21 +585,4 @@ public class CQLSSTableWriter implements Closeable return Pair.create(insert, cqlStatement.boundNames); } } - - private static <T extends ParsedStatement> T parseStatement(String query, Class<T> klass, String type) - { - try - { - ParsedStatement stmt = QueryProcessor.parseStatement(query); - - if (!stmt.getClass().equals(klass)) - throw new IllegalArgumentException("Invalid query, must be a " + type + " statement but was: " + stmt.getClass()); - - return klass.cast(stmt); - } - catch (RequestValidationException e) - { - throw new IllegalArgumentException(e.getMessage(), e); - } - } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/47c473ae/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java index 3c80b9e..ac7f4ad 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java @@ -27,7 +27,6 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.io.Files; -import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -48,6 +47,7 @@ import com.datastax.driver.core.UDTValue; import com.datastax.driver.core.UserType; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.fail; public class CQLSSTableWriterTest @@ -126,7 +126,7 @@ public class CQLSSTableWriterTest } } - @Test(expected = IllegalArgumentException.class) + @Test public void testForbidCounterUpdates() throws Exception { String KS = "cql_keyspace"; @@ -142,10 +142,18 @@ public class CQLSSTableWriterTest " PRIMARY KEY (my_id)" + ")"; String insert = String.format("UPDATE cql_keyspace.counter1 SET my_counter = my_counter - ? WHERE my_id = ?"); - CQLSSTableWriter.builder().inDirectory(dataDir) - .forTable(schema) - .withPartitioner(Murmur3Partitioner.instance) - .using(insert).build(); + try + { + CQLSSTableWriter.builder().inDirectory(dataDir) + .forTable(schema) + .withPartitioner(Murmur3Partitioner.instance) + .using(insert).build(); + fail("Counter update statements should not be supported"); + } + catch (IllegalArgumentException e) + { + assertEquals(e.getMessage(), "Counter update statements are not supported"); + } } @Test @@ -167,8 +175,8 @@ public class CQLSSTableWriterTest String insert = "INSERT INTO ks.test (k, v) VALUES (?, ?)"; CQLSSTableWriter writer = CQLSSTableWriter.builder() .inDirectory(dataDir) - .forTable(schema) .using(insert) + .forTable(schema) .withBufferSizeInMB(1) .build(); @@ -532,6 +540,55 @@ public class CQLSSTableWriterTest assertEquals("5", r5.getString("v")); } + @Test + public void testUpdateSatement() throws Exception + { + final String KS = "cql_keyspace6"; + final String TABLE = "table6"; + + final String schema = "CREATE TABLE " + KS + "." + TABLE + " (" + + " k int," + + " c1 int," + + " c2 int," + + " v text," + + " PRIMARY KEY (k, c1, c2)" + + ")"; + + File tempdir = Files.createTempDir(); + File dataDir = new File(tempdir.getAbsolutePath() + File.separator + KS + File.separator + TABLE); + assert dataDir.mkdirs(); + + CQLSSTableWriter writer = CQLSSTableWriter.builder() + .inDirectory(dataDir) + .forTable(schema) + .using("UPDATE " + KS + "." + TABLE + " SET v = ? " + + "WHERE k = ? AND c1 = ? AND c2 = ?") + .build(); + + writer.addRow("a", 1, 2, 3); + writer.addRow("b", 4, 5, 6); + writer.addRow(null, 7, 8, 9); + writer.addRow(CQLSSTableWriter.UNSET_VALUE, 10, 11, 12); + writer.close(); + loadSSTables(dataDir, KS); + + UntypedResultSet resultSet = QueryProcessor.executeInternal("SELECT * FROM " + KS + "." + TABLE); + assertEquals(2, resultSet.size()); + + Iterator<UntypedResultSet.Row> iter = resultSet.iterator(); + UntypedResultSet.Row r1 = iter.next(); + assertEquals(1, r1.getInt("k")); + assertEquals(2, r1.getInt("c1")); + assertEquals(3, r1.getInt("c2")); + assertEquals("a", r1.getString("v")); + UntypedResultSet.Row r2 = iter.next(); + assertEquals(4, r2.getInt("k")); + assertEquals(5, r2.getInt("c1")); + assertEquals(6, r2.getInt("c2")); + assertEquals("b", r2.getString("v")); + assertFalse(iter.hasNext()); + } + private static void loadSSTables(File dataDir, String ks) throws ExecutionException, InterruptedException { SSTableLoader loader = new SSTableLoader(dataDir, new SSTableLoader.Client() http://git-wip-us.apache.org/repos/asf/cassandra/blob/47c473ae/tools/stress/src/org/apache/cassandra/stress/StressProfile.java ---------------------------------------------------------------------- diff --git a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java index 84d4abd..9c0be4e 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressProfile.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressProfile.java @@ -41,6 +41,7 @@ import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ColumnDefinition; import org.apache.cassandra.cql3.CQLFragmentParser; import org.apache.cassandra.cql3.CqlParser; +import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.statements.CreateTableStatement; import org.apache.cassandra.exceptions.RequestValidationException; import org.apache.cassandra.exceptions.SyntaxException; @@ -62,8 +63,6 @@ import org.yaml.snakeyaml.Yaml; import org.yaml.snakeyaml.constructor.Constructor; import org.yaml.snakeyaml.error.YAMLException; -import static org.apache.cassandra.io.sstable.StressCQLSSTableWriter.parseStatement; - public class StressProfile implements Serializable { private String keyspaceCql; @@ -449,7 +448,7 @@ public class StressProfile implements Serializable public CreateTableStatement.RawStatement getCreateStatement() { - CreateTableStatement.RawStatement createStatement = parseStatement(tableCql, CreateTableStatement.RawStatement.class, "CREATE TABLE"); + CreateTableStatement.RawStatement createStatement = QueryProcessor.parseStatement(tableCql, CreateTableStatement.RawStatement.class, "CREATE TABLE"); createStatement.prepareKeyspace(keyspaceName); return createStatement;
