From 299b7be0258dbd5ba55a105e70ddfb919f12ea09 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Tue, 26 Apr 2011 14:53:53 +0000 Subject: [PATCH] add support for insert, delete in cql BATCH patch by Pavel Yaskevich; reviewed by jbellis for CASSANDRA-2537 git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1096771 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + doc/cql/CQL.textile | 37 +++++--- .../cassandra/cql/AbstractModification.java | 74 ++++++++++++++++ .../apache/cassandra/cql/BatchStatement.java | 84 +++++++++++++++++++ src/java/org/apache/cassandra/cql/Cql.g | 54 ++++++++---- .../apache/cassandra/cql/DeleteStatement.java | 66 +++++++++++---- .../apache/cassandra/cql/QueryProcessor.java | 60 ++++++------- .../apache/cassandra/cql/StatementType.java | 2 +- .../apache/cassandra/cql/UpdateStatement.java | 58 ++++++++++--- test/system/test_cql.py | 57 +++++++++++++ 10 files changed, 400 insertions(+), 93 deletions(-) create mode 100644 src/java/org/apache/cassandra/cql/AbstractModification.java create mode 100644 src/java/org/apache/cassandra/cql/BatchStatement.java diff --git a/CHANGES.txt b/CHANGES.txt index 1a6aa6ec44..b9d6870400 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,6 +3,7 @@ * Remove checking all column families on startup for compaction candidates (CASSANDRA-2444) * validate CQL create keyspace options (CASSANDRA-2525) * fix nodetool setcompactionthroughput (CASSANDRA-2550) + * add support for insert, delete in cql BATCH (CASSANDRA-2537) 0.8.0-beta1 diff --git a/doc/cql/CQL.textile b/doc/cql/CQL.textile index 5ab12bd75a..bd0e870fa9 100644 --- a/doc/cql/CQL.textile +++ b/doc/cql/CQL.textile @@ -99,19 +99,6 @@ UPDATE ... SET name1 = value1, name2 = value2 WHERE KEY = keyname; Rows are created or updated by supplying column names and values in term assignment format. Multiple columns can be set by separating the name/value pairs using commas. Each update statement requires exactly one key to be specified using a WHERE clause and the @KEY@ keyword. -Additionally, it is also possible to send multiple UPDATES to a node at once using a batch syntax: - -bc. -BEGIN BATCH [USING ] -UPDATE CF1 SET name1 = value1, name2 = value2 WHERE KEY = keyname1; -UPDATE CF1 SET name3 = value3 WHERE KEY = keyname2; -UPDATE CF2 SET name4 = value4, name5 = value5 WHERE KEY = keyname3; -APPLY BATCH - -When batching UPDATEs, a single consistency level is used for the entire batch, it appears after the @BEGIN BATCH@ statement, and uses the standard "consistency level specification":#consistency. Batch UPDATEs default to @CONSISTENCY.ONE@ when left unspecified. - -_NOTE: While there are no isolation guarantees, @UPDATE@ queries are atomic within a give record._ - h2. DELETE _Synopsis:_ @@ -150,6 +137,30 @@ UPDATE ... WHERE KEY = keyname1 UPDATE ... WHERE KEY IN (keyname1, keyname2) The @WHERE@ clause is used to determine which row(s) a @DELETE@ applies to. The first form allows the specification of a single keyname using the @KEY@ keyword and the @=@ operator. The second form allows a list of keyname terms to be specified using the @IN@ notation and a parenthesized list of comma-delimited keyname terms. + +h2. BATCH + +_Synopsis:_ + +bc. +BATCH BEGIN BATCH [USING CONSISTENCY ] + INSERT or UPDATE or DELETE statements separated by semicolon or "end of line" +APPLY BATCH + +A single consistency level is used for the entire batch, it appears after the @BEGIN BATCH@ statement, and uses the standard "consistency level specification":#consistency. Batch default to @CONSISTENCY.ONE@ when left unspecified. + +_NOTE: While there are no isolation guarantees, @UPDATE@ queries are atomic within a give record._ + +_Example:_ + +bc. +BEGIN BATCH USING CONSISTENCY QUORUM + INSERT INTO users (KEY, password, name) VALUES ('user2', 'ch@ngem3b', 'second user') + UPDATE users SET password = 'ps22dhds' WHERE KEY = 'user2' + INSERT INTO users (KEY, password) VALUES ('user3', 'ch@ngem3c') + DELETE name FROM users WHERE key = 'user2' + INSERT INTO users (KEY, password, name) VALUES ('user4', 'ch@ngem3c', 'Andrew') +APPLY BATCH h2. TRUNCATE diff --git a/src/java/org/apache/cassandra/cql/AbstractModification.java b/src/java/org/apache/cassandra/cql/AbstractModification.java new file mode 100644 index 0000000000..74e8e6d64f --- /dev/null +++ b/src/java/org/apache/cassandra/cql/AbstractModification.java @@ -0,0 +1,74 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.cassandra.cql; + +import org.apache.cassandra.db.RowMutation; +import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.thrift.ConsistencyLevel; + +import java.util.List; + +public abstract class AbstractModification +{ + public static final ConsistencyLevel defaultConsistency = ConsistencyLevel.ONE; + + protected final String columnFamily; + protected final ConsistencyLevel cLevel; + + public AbstractModification(String columnFamily, ConsistencyLevel cLevel) + { + this.columnFamily = columnFamily; + this.cLevel = cLevel; + } + + public String getColumnFamily() + { + return columnFamily; + } + + public ConsistencyLevel getConsistencyLevel() + { + return (cLevel != null) ? cLevel : defaultConsistency; + } + + /** + * True if an explicit consistency level was parsed from the statement. + * + * @return true if a consistency was parsed, false otherwise. + */ + public boolean isSetConsistencyLevel() + { + return cLevel != null; + } + + /** + * Convert statement into a list of mutations to apply on the server + * + * @param keyspace The working keyspace + * @param clientState current client status + * + * @return list of the mutations + * + * @throws org.apache.cassandra.thrift.InvalidRequestException on the wrong request + */ + public abstract List prepareRowMutations(String keyspace, ClientState clientState) + throws org.apache.cassandra.thrift.InvalidRequestException; +} diff --git a/src/java/org/apache/cassandra/cql/BatchStatement.java b/src/java/org/apache/cassandra/cql/BatchStatement.java new file mode 100644 index 0000000000..e609374fe5 --- /dev/null +++ b/src/java/org/apache/cassandra/cql/BatchStatement.java @@ -0,0 +1,84 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.cassandra.cql; + +import java.util.LinkedList; +import java.util.List; + +import org.apache.cassandra.db.RowMutation; +import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.thrift.ConsistencyLevel; +import org.apache.cassandra.thrift.InvalidRequestException; + +/** + * A BATCH statement parsed from a CQL query. + * + */ +public class BatchStatement +{ + // statements to execute + protected final List statements; + + // global consistency level + protected final ConsistencyLevel consistency; + + + /** + * Creates a new BatchStatement from a list of statements and a + * Thrift consistency level. + * + * @param statements a list of UpdateStatements + * @param level Thrift consistency level enum + */ + public BatchStatement(List statements, ConsistencyLevel level) + { + this.statements = statements; + consistency = level; + } + + public List getStatements() + { + return statements; + } + + public ConsistencyLevel getConsistencyLevel() + { + return consistency; + } + + public List getMutations(String keyspace, ClientState clientState) throws InvalidRequestException + { + List batch = new LinkedList(); + + for (AbstractModification statement : statements) + { + batch.addAll(statement.prepareRowMutations(keyspace, clientState)); + } + + return batch; + } + + + public String toString() + { + return String.format("BatchStatement(statements=%s, consistency=%s)", statements, consistency); + } +} diff --git a/src/java/org/apache/cassandra/cql/Cql.g b/src/java/org/apache/cassandra/cql/Cql.g index 6d4c707f07..dc76fe0881 100644 --- a/src/java/org/apache/cassandra/cql/Cql.g +++ b/src/java/org/apache/cassandra/cql/Cql.g @@ -102,12 +102,12 @@ options { query returns [CQLStatement stmnt] : selectStatement { $stmnt = new CQLStatement(StatementType.SELECT, $selectStatement.expr); } - | insertStatement { $stmnt = new CQLStatement(StatementType.INSERT, $insertStatement.expr); } + | insertStatement endStmnt { $stmnt = new CQLStatement(StatementType.INSERT, $insertStatement.expr); } | updateStatement endStmnt { $stmnt = new CQLStatement(StatementType.UPDATE, $updateStatement.expr); } - | batchUpdateStatement { $stmnt = new CQLStatement(StatementType.BATCH_UPDATE, $batchUpdateStatement.expr); } + | batchStatement { $stmnt = new CQLStatement(StatementType.BATCH, $batchStatement.expr); } | useStatement { $stmnt = new CQLStatement(StatementType.USE, $useStatement.keyspace); } | truncateStatement { $stmnt = new CQLStatement(StatementType.TRUNCATE, $truncateStatement.cfam); } - | deleteStatement { $stmnt = new CQLStatement(StatementType.DELETE, $deleteStatement.expr); } + | deleteStatement endStmnt { $stmnt = new CQLStatement(StatementType.DELETE, $deleteStatement.expr); } | createKeyspaceStatement { $stmnt = new CQLStatement(StatementType.CREATE_KEYSPACE, $createKeyspaceStatement.expr); } | createColumnFamilyStatement { $stmnt = new CQLStatement(StatementType.CREATE_COLUMNFAMILY, $createColumnFamilyStatement.expr); } | createIndexStatement { $stmnt = new CQLStatement(StatementType.CREATE_INDEX, $createIndexStatement.expr); } @@ -191,7 +191,7 @@ whereClause returns [WhereClause clause] */ insertStatement returns [UpdateStatement expr] : { - ConsistencyLevel cLevel = ConsistencyLevel.ONE; + ConsistencyLevel cLevel = null; Map columns = new HashMap(); List columnNames = new ArrayList(); @@ -202,32 +202,54 @@ insertStatement returns [UpdateStatement expr] K_VALUES '(' key=term ( ',' column_value=term { columnValues.add($column_value.item); })+ ')' ( K_USING K_CONSISTENCY K_LEVEL { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); } )? - endStmnt { return new UpdateStatement($columnFamily.text, cLevel, columnNames, columnValues, key); } ; /** - * BEGIN BATCH [USING CONSISTENCY.] - * UPDATE SET name1 = value1 WHERE KEY = keyname1; - * UPDATE SET name2 = value2 WHERE KEY = keyname2; - * UPDATE SET name3 = value3 WHERE KEY = keyname3; + * BEGIN BATCH [USING CONSISTENCY ] + * UPDATE SET name1 = value1 WHERE KEY = keyname1; + * UPDATE SET name2 = value2 WHERE KEY = keyname2; + * UPDATE SET name3 = value3 WHERE KEY = keyname3; + * ... + * APPLY BATCH + * + * OR + * + * BEGIN BATCH [USING CONSISTENCY ] + * INSERT INTO (KEY, ) VALUES ('', ''); + * INSERT INTO (KEY, ) VALUES ('', ''); + * ... + * APPLY BATCH + * + * OR + * + * BEGIN BATCH [USING CONSISTENCY ] + * DELETE name1, name2 FROM WHERE key = + * DELETE name3, name4 FROM WHERE key = + * ... * APPLY BATCH */ -batchUpdateStatement returns [BatchUpdateStatement expr] +batchStatement returns [BatchStatement expr] : { ConsistencyLevel cLevel = ConsistencyLevel.ONE; - List updates = new ArrayList(); + List statements = new ArrayList(); } K_BEGIN K_BATCH ( K_USING K_CONSISTENCY K_LEVEL { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); } )? - u1=updateStatement ';'? { updates.add(u1); } ( uN=updateStatement ';'? { updates.add(uN); } )* - K_APPLY K_BATCH EOF + s1=batchStatementObjective ';'? { statements.add(s1); } ( sN=batchStatementObjective ';'? { statements.add(sN); } )* + K_APPLY K_BATCH endStmnt { - return new BatchUpdateStatement(updates, cLevel); + return new BatchStatement(statements, cLevel); } ; +batchStatementObjective returns [AbstractModification statement] + : i=insertStatement { $statement = i; } + | u=updateStatement { $statement = u; } + | d=deleteStatement { $statement = d; } + ; + /** * UPDATE * @@ -265,7 +287,7 @@ updateStatement returns [UpdateStatement expr] */ deleteStatement returns [DeleteStatement expr] : { - ConsistencyLevel cLevel = ConsistencyLevel.ONE; + ConsistencyLevel cLevel = null; List keyList = null; List columnsList = Collections.emptyList(); } @@ -274,7 +296,7 @@ deleteStatement returns [DeleteStatement expr] K_FROM columnFamily=( IDENT | STRING_LITERAL | INTEGER ) ( K_USING K_CONSISTENCY K_LEVEL )? K_WHERE ( K_KEY '=' key=term { keyList = Collections.singletonList(key); } | K_KEY K_IN '(' keys=termList { keyList = $keys.items; } ')' - )? endStmnt + )? { return new DeleteStatement(columnsList, $columnFamily.text, cLevel, keyList); } diff --git a/src/java/org/apache/cassandra/cql/DeleteStatement.java b/src/java/org/apache/cassandra/cql/DeleteStatement.java index 557cb2a8e3..cc6437fe72 100644 --- a/src/java/org/apache/cassandra/cql/DeleteStatement.java +++ b/src/java/org/apache/cassandra/cql/DeleteStatement.java @@ -20,26 +20,37 @@ */ package org.apache.cassandra.cql; +import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.List; +import org.apache.cassandra.auth.Permission; +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.RowMutation; +import org.apache.cassandra.db.filter.QueryPath; +import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.service.ClientState; import org.apache.cassandra.thrift.ConsistencyLevel; +import org.apache.cassandra.thrift.InvalidRequestException; + +import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily; +import static org.apache.cassandra.cql.QueryProcessor.validateColumnName; /** * A DELETE parsed from a CQL query statement. * */ -public class DeleteStatement +public class DeleteStatement extends AbstractModification { private List columns; - private String columnFamily; - private ConsistencyLevel cLevel; private List keys; public DeleteStatement(List columns, String columnFamily, ConsistencyLevel cLevel, List keys) { + super(columnFamily, cLevel); + this.columns = columns; - this.columnFamily = columnFamily; - this.cLevel = cLevel; this.keys = keys; } @@ -48,21 +59,44 @@ public class DeleteStatement return columns; } - public String getColumnFamily() - { - return columnFamily; - } - - public ConsistencyLevel getConsistencyLevel() - { - return cLevel; - } - public List getKeys() { return keys; } - + + /** {@inheritDoc} */ + public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException + { + clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE); + CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false); + + AbstractType comparator = metadata.getComparatorFor(null); + AbstractType keyType = DatabaseDescriptor.getCFMetaData(keyspace, columnFamily).getKeyValidator(); + + List rowMutations = new ArrayList(); + + for (Term key : keys) + { + RowMutation rm = new RowMutation(keyspace, key.getByteBuffer(keyType)); + + if (columns.size() < 1) // No columns, delete the row + rm.delete(new QueryPath(columnFamily), System.currentTimeMillis()); + else // Delete specific columns + { + for (Term column : columns) + { + ByteBuffer columnName = column.getByteBuffer(comparator); + validateColumnName(columnName); + rm.delete(new QueryPath(columnFamily, null, columnName), System.currentTimeMillis()); + } + } + + rowMutations.add(rm); + } + + return rowMutations; + } + public String toString() { return String.format("DeleteStatement(columns=%s, columnFamily=%s, consistency=%s keys=%s)", diff --git a/src/java/org/apache/cassandra/cql/QueryProcessor.java b/src/java/org/apache/cassandra/cql/QueryProcessor.java index 338b0a9d52..ed2e64c8a7 100644 --- a/src/java/org/apache/cassandra/cql/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql/QueryProcessor.java @@ -367,7 +367,7 @@ public class QueryProcessor } } - private static void validateKey(ByteBuffer key) throws InvalidRequestException + public static void validateKey(ByteBuffer key) throws InvalidRequestException { if (key == null || key.remaining() == 0) { @@ -396,13 +396,13 @@ public class QueryProcessor } } - private static void validateColumnName(ByteBuffer column) + public static void validateColumnName(ByteBuffer column) throws InvalidRequestException { validateColumnNames(Arrays.asList(column)); } - private static void validateColumn(CFMetaData metadata, ByteBuffer name, ByteBuffer value) + public static void validateColumn(CFMetaData metadata, ByteBuffer name, ByteBuffer value) throws InvalidRequestException { validateColumnName(name); @@ -543,15 +543,27 @@ public class QueryProcessor result.type = CqlResultType.VOID; return result; - case BATCH_UPDATE: - BatchUpdateStatement batch = (BatchUpdateStatement)statement.statement; - - for (UpdateStatement up : batch.getUpdates()) + case BATCH: + BatchStatement batch = (BatchStatement) statement.statement; + + for (AbstractModification up : batch.getStatements()) if (up.isSetConsistencyLevel()) throw new InvalidRequestException( - "Consistency level must be set on the BATCH, not individual UPDATE statements"); - - batchUpdate(clientState, batch.getUpdates(), batch.getConsistencyLevel()); + "Consistency level must be set on the BATCH, not individual statements"); + + try + { + StorageProxy.mutate(batch.getMutations(keyspace, clientState), batch.getConsistencyLevel()); + } + catch (org.apache.cassandra.thrift.UnavailableException e) + { + throw new UnavailableException(); + } + catch (TimeoutException e) + { + throw new TimedOutException(); + } + result.type = CqlResultType.VOID; return result; @@ -583,34 +595,10 @@ public class QueryProcessor case DELETE: DeleteStatement delete = (DeleteStatement)statement.statement; - clientState.hasColumnFamilyAccess(delete.getColumnFamily(), Permission.WRITE); - metadata = validateColumnFamily(keyspace, delete.getColumnFamily(), false); - comparator = metadata.getComparatorFor(null); - AbstractType keyType = DatabaseDescriptor.getCFMetaData(keyspace, - delete.getColumnFamily()).getKeyValidator(); - - List rowMutations = new ArrayList(); - for (Term key : delete.getKeys()) - { - RowMutation rm = new RowMutation(keyspace, key.getByteBuffer(keyType)); - if (delete.getColumns().size() < 1) // No columns, delete the row - rm.delete(new QueryPath(delete.getColumnFamily()), System.currentTimeMillis()); - else // Delete specific columns - { - for (Term column : delete.getColumns()) - { - ByteBuffer columnName = column.getByteBuffer(comparator); - validateColumnName(columnName); - rm.delete(new QueryPath(delete.getColumnFamily(), null, columnName), - System.currentTimeMillis()); - } - } - rowMutations.add(rm); - } - + try { - StorageProxy.mutate(rowMutations, delete.getConsistencyLevel()); + StorageProxy.mutate(delete.prepareRowMutations(keyspace, clientState), delete.getConsistencyLevel()); } catch (TimeoutException e) { diff --git a/src/java/org/apache/cassandra/cql/StatementType.java b/src/java/org/apache/cassandra/cql/StatementType.java index ec2400d71d..330f89a302 100644 --- a/src/java/org/apache/cassandra/cql/StatementType.java +++ b/src/java/org/apache/cassandra/cql/StatementType.java @@ -24,7 +24,7 @@ import java.util.EnumSet; public enum StatementType { - SELECT, INSERT, UPDATE, BATCH_UPDATE, USE, TRUNCATE, DELETE, CREATE_KEYSPACE, CREATE_COLUMNFAMILY, CREATE_INDEX, + SELECT, INSERT, UPDATE, BATCH, USE, TRUNCATE, DELETE, CREATE_KEYSPACE, CREATE_COLUMNFAMILY, CREATE_INDEX, DROP_KEYSPACE, DROP_COLUMNFAMILY; // Statement types that don't require a keyspace to be set. diff --git a/src/java/org/apache/cassandra/cql/UpdateStatement.java b/src/java/org/apache/cassandra/cql/UpdateStatement.java index c604e38d98..c017aa2259 100644 --- a/src/java/org/apache/cassandra/cql/UpdateStatement.java +++ b/src/java/org/apache/cassandra/cql/UpdateStatement.java @@ -21,24 +21,29 @@ package org.apache.cassandra.cql; import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; +import org.apache.cassandra.auth.Permission; +import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.RowMutation; +import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.service.ClientState; import org.apache.cassandra.thrift.ConsistencyLevel; import org.apache.cassandra.thrift.InvalidRequestException; +import static org.apache.cassandra.cql.QueryProcessor.validateKey; +import static org.apache.cassandra.cql.QueryProcessor.validateColumn; + +import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily; + /** * An UPDATE statement parsed from a CQL query statement. * */ -public class UpdateStatement +public class UpdateStatement extends AbstractModification { - public static final ConsistencyLevel defaultConsistency = ConsistencyLevel.ONE; - private String columnFamily; - private ConsistencyLevel cLevel = null; private Map columns; private List columnNames, columnValues; private Term key; @@ -54,8 +59,8 @@ public class UpdateStatement */ public UpdateStatement(String columnFamily, ConsistencyLevel cLevel, Map columns, Term key) { - this.columnFamily = columnFamily; - this.cLevel = cLevel; + super(columnFamily, cLevel); + this.columns = columns; this.key = key; } @@ -86,8 +91,8 @@ public class UpdateStatement */ public UpdateStatement(String columnFamily, ConsistencyLevel cLevel, List columnNames, List columnValues, Term key) { - this.columnFamily = columnFamily; - this.cLevel = cLevel; + super(columnFamily, cLevel); + this.columnNames = columnNames; this.columnValues = columnValues; this.key = key; @@ -114,6 +119,37 @@ public class UpdateStatement return (cLevel != null); } + /** {@inheritDoc} */ + public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException + { + List cfamsSeen = new ArrayList(); + + CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false); + + // Avoid unnecessary authorizations. + if (!(cfamsSeen.contains(columnFamily))) + { + clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE); + cfamsSeen.add(columnFamily); + } + + ByteBuffer key = this.key.getByteBuffer(getKeyType(keyspace)); + validateKey(key); + AbstractType comparator = getComparator(keyspace); + + RowMutation rm = new RowMutation(keyspace, key); + for (Map.Entry column : getColumns().entrySet()) + { + ByteBuffer colName = column.getKey().getByteBuffer(comparator); + ByteBuffer colValue = column.getValue().getByteBuffer(getValueValidator(keyspace, colName)); + + validateColumn(metadata, colName, colValue); + rm.add(new QueryPath(columnFamily, null, colName), colValue, System.currentTimeMillis()); + } + + return Arrays.asList(rm); + } + public String getColumnFamily() { return columnFamily; diff --git a/test/system/test_cql.py b/test/system/test_cql.py index b716776b0f..6b587fad86 100644 --- a/test/system/test_cql.py +++ b/test/system/test_cql.py @@ -699,3 +699,60 @@ class TestCql(ThriftTester): r = cursor.fetchone() assert r[1] == "some_value", \ "unrecognized value '%s'" % r[1] + + def test_batch_with_mixed_statements(self): + "handle BATCH with INSERT/UPDATE/DELETE statements mixed in it" + cursor = init() + cursor.compression = 'NONE' + cursor.execute(""" + BEGIN BATCH USING CONSISTENCY ONE + UPDATE StandardString1 SET :name = :val WHERE KEY = :key1 + INSERT INTO StandardString1 (KEY, :col1) VALUES (:key2, :val) + INSERT INTO StandardString1 (KEY, :col2) VALUES (:key3, :val) + DELETE :col2 FROM StandardString1 WHERE key = :key3 + APPLY BATCH + """, + dict(key1="bKey1", key2="bKey2", key3="bKey3", name="bName", col1="bCol1", col2="bCol2", val="bVal")) + + cursor.execute("SELECT :name FROM StandardString1 WHERE KEY = :key", + dict(name="bName", key="bKey1")) + + assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + assert colnames[1] == "bName", \ + "unrecognized name '%s'" % colnames[1] + r = cursor.fetchone() + assert r[1] == "bVal", \ + "unrecognized value '%s'" % r[1] + + cursor.execute("SELECT :name FROM StandardString1 WHERE KEY = :key", + dict(name="bCol2", key="bKey3")) + + assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + assert colnames[1] == "bCol2", \ + "unrecognized name '%s'" % colnames[1] + # was deleted by DELETE statement + r = cursor.fetchone() + assert r[1] == None, \ + "unrecognized value '%s'" % r[1] + + cursor.execute("SELECT :name FROM StandardString1 WHERE KEY = :key", + dict(name="bCol1", key="bKey2")) + + assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + assert colnames[1] == "bCol1", \ + "unrecognized name '%s'" % colnames[1] + r = cursor.fetchone() + assert r[1] == "bVal", \ + "unrecognized value '%s'" % r[1] + + assert_raises(cql.ProgrammingError, + cursor.execute, + """ + BEGIN BATCH USING CONSISTENCY ONE + UPDATE USING CONSISTENCY QUORUM StandardString1 SET 'name' = 'value' WHERE KEY = 'bKey4' + DELETE 'name' FROM StandardString1 WHERE KEY = 'bKey4' + APPLY BATCH + """)