natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

CQLSSTableWriter does not allow Update statement 

Patch by Alex Petrov; reviewed by Stefania Alborghetti for CASSANDRA-12450
This commit is contained in:
Alex Petrov 2016-10-07 12:09:12 +08:00 committed by Stefania Alborghetti
parent 6216ce8794
commit 47c473ae3c
5 changed files with 93 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

18
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -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();

36
src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
View File

@ -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);
}
}
}

71
test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
View File

@ -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()

5
tools/stress/src/org/apache/cassandra/stress/StressProfile.java
View File

@ -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;