mirror of https://github.com/apache/cassandra
Accord should block currently unsafe operations
patch by David Capwell; reviewed by Ariel Weisberg for CASSANDRA-20020
This commit is contained in:
David Capwell
parent
5afe99b7ca
commit
1f80f99b71
|
|
@ -422,6 +422,11 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement,
|
|||
return aggregationSpecFactory == null ? null : aggregationSpecFactory.newInstance(options);
|
||||
}
|
||||
|
||||
public boolean hasAggregation()
|
||||
{
|
||||
return aggregationSpecFactory != null;
|
||||
}
|
||||
|
||||
public ReadQuery getQuery(QueryOptions options, long nowInSec) throws RequestValidationException
|
||||
{
|
||||
Selectors selectors = selection.newSelectors(options);
|
||||
|
|
@ -1229,7 +1234,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement,
|
|||
public final Term.Raw limit;
|
||||
public final Term.Raw perPartitionLimit;
|
||||
private ClientState state;
|
||||
private final StatementSource source;
|
||||
public final StatementSource source;
|
||||
|
||||
public RawStatement(QualifiedName cfName,
|
||||
Parameters parameters,
|
||||
|
|
|
|||
|
|
@ -58,6 +58,7 @@ import org.apache.cassandra.cql3.transactions.RowDataReference;
|
|||
import org.apache.cassandra.cql3.transactions.SelectReferenceSource;
|
||||
import org.apache.cassandra.db.SinglePartitionReadCommand;
|
||||
import org.apache.cassandra.db.SinglePartitionReadQuery;
|
||||
import org.apache.cassandra.db.filter.DataLimits;
|
||||
import org.apache.cassandra.db.marshal.AbstractType;
|
||||
import org.apache.cassandra.db.partitions.FilteredPartition;
|
||||
import org.apache.cassandra.exceptions.InvalidRequestException;
|
||||
|
|
@ -99,14 +100,19 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
|
|||
public static final String INCOMPLETE_PRIMARY_KEY_SELECT_MESSAGE = "SELECT must specify either all primary key elements or all partition key elements and LIMIT 1. In both cases partition key elements must be always specified with equality operators; %s %s";
|
||||
public static final String NO_CONDITIONS_IN_UPDATES_MESSAGE = "Updates within transactions may not specify their own conditions; %s statement %s";
|
||||
public static final String NO_TIMESTAMPS_IN_UPDATES_MESSAGE = "Updates within transactions may not specify custom timestamps; %s statement %s";
|
||||
public static final String NO_TTLS_IN_UPDATES_MESSAGE = "Updates within transactions may not specify custom ttls; %s statement %s";
|
||||
public static final String TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE = "Accord transactions are disabled on table (See transactional_mode in table options); %s statement %s";
|
||||
public static final String TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE = "Accord transactions are disabled on table (table is being dropped); %s statement %s";
|
||||
public static final String NO_COUNTERS_IN_TXNS_MESSAGE = "Counter columns cannot be accessed within a transaction; %s statement %s";
|
||||
public static final String NO_AGGREGATION_IN_TXNS_MESSAGE = "No aggregation functions allowed within a transaction; %s statement %s";
|
||||
public static final String NO_ORDER_BY_IN_TXNS_MESSAGE = "No ORDER BY clause allowed within a transaction; %s statement %s";
|
||||
public static final String NO_GROUP_BY_IN_TXNS_MESSAGE = "No GROUP BY clause allowed within a transaction; %s statement %s";
|
||||
public static final String EMPTY_TRANSACTION_MESSAGE = "Transaction contains no reads or writes";
|
||||
public static final String SELECT_REFS_NEED_COLUMN_MESSAGE = "SELECT references must specify a column.";
|
||||
public static final String TRANSACTIONS_DISABLED_MESSAGE = "Accord transactions are disabled. (See accord.enabled in cassandra.yaml)";
|
||||
public static final String ILLEGAL_RANGE_QUERY_MESSAGE = "Range queries are not allowed for reads within a transaction; %s %s";
|
||||
public static final String UNSUPPORTED_MIGRATION = "Transaction Statement is unsupported when migrating away from Accord or before migration to Accord is complete for a range";
|
||||
public static final String NO_PARTITION_IN_CLAUSE_WITH_LIMIT = "Partition key is present in IN clause and there is a LIMIT... this is currently not supported; %s statement %s";
|
||||
|
||||
static class NamedSelect
|
||||
{
|
||||
|
|
@ -465,6 +471,29 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
|
|||
return false;
|
||||
}
|
||||
|
||||
private static void validate(SelectStatement.RawStatement select)
|
||||
{
|
||||
if (select.parameters.orderings != null && !select.parameters.orderings.isEmpty())
|
||||
throw invalidRequest(NO_ORDER_BY_IN_TXNS_MESSAGE, "SELECT", select.source);
|
||||
if (select.parameters.groups != null && !select.parameters.groups.isEmpty())
|
||||
throw invalidRequest(NO_GROUP_BY_IN_TXNS_MESSAGE, "SELECT", select.source);
|
||||
}
|
||||
|
||||
private static void validate(SelectStatement prepared)
|
||||
{
|
||||
if (!prepared.table.isAccordEnabled())
|
||||
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, "SELECT", prepared.source);
|
||||
if (prepared.table.params.pendingDrop)
|
||||
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, "SELECT", prepared.source);
|
||||
if (prepared.table.isCounter())
|
||||
throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, "SELECT", prepared.source);
|
||||
if (prepared.hasAggregation())
|
||||
throw invalidRequest(NO_AGGREGATION_IN_TXNS_MESSAGE, "SELECT", prepared.source);
|
||||
|
||||
if (prepared.getRestrictions().keyIsInRelation())
|
||||
checkTrue(prepared.getLimit(null) == DataLimits.NO_LIMIT, NO_PARTITION_IN_CLAUSE_WITH_LIMIT, "SELECT", prepared.source);
|
||||
}
|
||||
|
||||
public static class Parsed extends QualifiedStatement.Composite
|
||||
{
|
||||
private final List<SelectStatement.RawStatement> assignments;
|
||||
|
|
@ -515,15 +544,10 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
|
|||
checkNotNull(select.parameters.refName, "Assignments must be named");
|
||||
TxnDataName name = TxnDataName.user(select.parameters.refName);
|
||||
checkTrue(selectNames.add(name), DUPLICATE_TUPLE_NAME_MESSAGE, name.name());
|
||||
validate(select);
|
||||
|
||||
SelectStatement prepared = select.prepare(bindVariables);
|
||||
|
||||
if (!prepared.table.isAccordEnabled())
|
||||
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, "SELECT", prepared.source);
|
||||
if (prepared.table.params.pendingDrop)
|
||||
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, "SELECT", prepared.source);
|
||||
if (prepared.table.isCounter())
|
||||
throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, "SELECT", prepared.source);
|
||||
validate(prepared);
|
||||
|
||||
NamedSelect namedSelect = new NamedSelect(name, prepared);
|
||||
checkAtMostOneRowSpecified(namedSelect.select, "LET assignment " + name.name());
|
||||
|
|
@ -538,15 +562,9 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
|
|||
NamedSelect returningSelect = null;
|
||||
if (select != null)
|
||||
{
|
||||
validate(select);
|
||||
SelectStatement prepared = select.prepare(bindVariables);
|
||||
|
||||
if (!prepared.table.isAccordEnabled())
|
||||
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, "SELECT", prepared.source);
|
||||
if (prepared.table.params.pendingDrop)
|
||||
throw invalidRequest(TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, "SELECT", prepared.source);
|
||||
if (prepared.table.isCounter())
|
||||
throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, "SELECT", prepared.source);
|
||||
|
||||
validate(prepared);
|
||||
returningSelect = new NamedSelect(TxnDataName.returning(), prepared);
|
||||
checkAtMostOnePartitionSpecified(returningSelect.select, "returning select");
|
||||
}
|
||||
|
|
@ -573,6 +591,7 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement,
|
|||
checkFalse(prepared.metadata().params.pendingDrop, TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, prepared.type, prepared.source);
|
||||
checkFalse(prepared.hasConditions(), NO_CONDITIONS_IN_UPDATES_MESSAGE, prepared.type, prepared.source);
|
||||
checkFalse(prepared.isTimestampSet(), NO_TIMESTAMPS_IN_UPDATES_MESSAGE, prepared.type, prepared.source);
|
||||
checkFalse(prepared.attrs.isTimeToLiveSet(), NO_TTLS_IN_UPDATES_MESSAGE, prepared.type, prepared.source);
|
||||
|
||||
if (prepared.metadata().isCounter())
|
||||
throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, prepared.type, prepared.source);
|
||||
|
|
|
|||
|
|
@ -18,16 +18,20 @@
|
|||
|
||||
package org.apache.cassandra.cql3.statements;
|
||||
|
||||
import org.apache.cassandra.cql3.ast.*;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.SchemaLoader;
|
||||
import org.apache.cassandra.cql3.CQLStatement;
|
||||
import org.apache.cassandra.cql3.QueryProcessor;
|
||||
import org.apache.cassandra.cql3.ast.Conditional.Is;
|
||||
import org.apache.cassandra.db.Keyspace;
|
||||
import org.apache.cassandra.exceptions.InvalidRequestException;
|
||||
import org.apache.cassandra.exceptions.SyntaxException;
|
||||
import org.apache.cassandra.schema.KeyspaceParams;
|
||||
import org.apache.cassandra.schema.TableId;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.service.ClientState;
|
||||
import org.apache.cassandra.service.QueryState;
|
||||
import org.apache.cassandra.transport.Dispatcher;
|
||||
|
|
@ -39,8 +43,13 @@ import static org.apache.cassandra.cql3.statements.TransactionStatement.EMPTY_TR
|
|||
import static org.apache.cassandra.cql3.statements.TransactionStatement.ILLEGAL_RANGE_QUERY_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.INCOMPLETE_PARTITION_KEY_SELECT_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.INCOMPLETE_PRIMARY_KEY_SELECT_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.NO_AGGREGATION_IN_TXNS_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.NO_CONDITIONS_IN_UPDATES_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.NO_GROUP_BY_IN_TXNS_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.NO_ORDER_BY_IN_TXNS_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.NO_PARTITION_IN_CLAUSE_WITH_LIMIT;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.NO_TIMESTAMPS_IN_UPDATES_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.NO_TTLS_IN_UPDATES_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.SELECT_REFS_NEED_COLUMN_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.TransactionStatement.TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE;
|
||||
import static org.apache.cassandra.cql3.statements.UpdateStatement.CANNOT_SET_KEY_WITH_REFERENCE_MESSAGE;
|
||||
|
|
@ -58,6 +67,7 @@ public class TransactionStatementTest
|
|||
private static final TableId TABLE4_ID = TableId.fromString("00000000-0000-0000-0000-000000000004");
|
||||
private static final TableId TABLE5_ID = TableId.fromString("00000000-0000-0000-0000-000000000005");
|
||||
private static final TableId TABLE6_ID = TableId.fromString("00000000-0000-0000-0000-000000000006");
|
||||
private static final TableId TABLE7_ID = TableId.fromString("00000000-0000-0000-0000-000000000007");
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeClass() throws Exception
|
||||
|
|
@ -69,7 +79,18 @@ public class TransactionStatementTest
|
|||
parse("CREATE TABLE tbl3 (k int PRIMARY KEY, \"with spaces\" int, \"with\"\"quote\" int, \"MiXeD_CaSe\" int) WITH transactional_mode = 'full'", "ks").id(TABLE3_ID),
|
||||
parse("CREATE TABLE tbl4 (k int PRIMARY KEY, int_list list<int>) WITH transactional_mode = 'full'", "ks").id(TABLE4_ID),
|
||||
parse("CREATE TABLE tbl5 (k int PRIMARY KEY, v int) WITH transactional_mode = 'full'", "ks").id(TABLE5_ID),
|
||||
parse("CREATE TABLE tbl6 (k int PRIMARY KEY, v int) WITH transactional_mode = 'off'", "ks").id(TABLE6_ID));
|
||||
parse("CREATE TABLE tbl6 (k int PRIMARY KEY, v int) WITH transactional_mode = 'off'", "ks").id(TABLE6_ID),
|
||||
parse("CREATE TABLE tbl7 (k int PRIMARY KEY, v vector<float, 1>) WITH transactional_mode = 'full'", "ks").id(TABLE7_ID));
|
||||
}
|
||||
|
||||
private static TableMetadata tbl(int num)
|
||||
{
|
||||
return Keyspace.open("ks").getColumnFamilyStore("tbl" + num).metadata();
|
||||
}
|
||||
|
||||
private static TableMetadata tbl5()
|
||||
{
|
||||
return tbl(5);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -342,6 +363,136 @@ public class TransactionStatementTest
|
|||
.hasMessageContaining(String.format(ILLEGAL_RANGE_QUERY_MESSAGE, "LET assignment row1", "at [2:15]"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldRejectTTL()
|
||||
{
|
||||
for (Mutation.Kind kind : Mutation.Kind.values())
|
||||
{
|
||||
if (kind == Mutation.Kind.DELETE) continue; // deletes don't support TTL
|
||||
Mutation mutation;
|
||||
switch (kind)
|
||||
{
|
||||
case INSERT:
|
||||
mutation = Mutation.insert(tbl5())
|
||||
.value("k", 1)
|
||||
.value("v", 2)
|
||||
.ttl(42)
|
||||
.build();
|
||||
break;
|
||||
case UPDATE:
|
||||
mutation = Mutation.update(tbl5())
|
||||
.value("k", 1)
|
||||
.set("v", 2)
|
||||
.ttl(42)
|
||||
.build();
|
||||
break;
|
||||
default:
|
||||
throw new UnsupportedOperationException(kind.name());
|
||||
}
|
||||
String query = Txn.wrap(mutation).toCQL();
|
||||
Assertions.assertThatThrownBy(() -> prepare(query))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_TTLS_IN_UPDATES_MESSAGE, kind.name(), "at"));
|
||||
|
||||
var txn = Txn.builder()
|
||||
.addLet("a", Select.builder()
|
||||
.table(tbl5())
|
||||
.value("k", 1)
|
||||
.build())
|
||||
.addIf(new Is("a", Is.Kind.Null), mutation)
|
||||
.build();
|
||||
Assertions.assertThatThrownBy(() -> prepare(txn.toCQL()))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_TTLS_IN_UPDATES_MESSAGE, kind.name(), "at"));
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldRejectAggFunctions()
|
||||
{
|
||||
var select = Select.builder()
|
||||
.selection(FunctionCall.count("v"))
|
||||
.table(tbl5())
|
||||
.value("k",0)
|
||||
.build();
|
||||
|
||||
Assertions.assertThatThrownBy(() -> prepare(Txn.wrap(select).toCQL()))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_AGGREGATION_IN_TXNS_MESSAGE, "SELECT", "at"));
|
||||
|
||||
var txn = Txn.builder()
|
||||
.addLet("a", select)
|
||||
.addReturnReferences("a.count")
|
||||
.build();
|
||||
|
||||
Assertions.assertThatThrownBy(() -> prepare(txn.toCQL()))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_AGGREGATION_IN_TXNS_MESSAGE, "SELECT", "at"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldRejectOrderBy()
|
||||
{
|
||||
String query = "BEGIN TRANSACTION\n" +
|
||||
" SELECT * FROM ks.tbl7 WHERE k=0 ORDER BY v ANN OF [42] LIMIT 1;" +
|
||||
"COMMIT TRANSACTION;";
|
||||
Assertions.assertThatThrownBy(() -> prepare(query))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_ORDER_BY_IN_TXNS_MESSAGE, "SELECT", "at"));
|
||||
|
||||
// The below code is left commented out as a reminder to think about this case... As of this writing ORDER BY does not parse in a LET clause... if that is ever fixed we should block it right away!
|
||||
// String query2 = "BEGIN TRANSACTION\n" +
|
||||
// " LET a = (SELECT * FROM ks.tbl7 WHERE k=0 ORDER BY v ANN OF [42] LIMIT 1;)" +
|
||||
// " SELECT a.v" +
|
||||
// "COMMIT TRANSACTION;";
|
||||
// Assertions.assertThatThrownBy(() -> prepare(query2))
|
||||
// .isInstanceOf(InvalidRequestException.class)
|
||||
// .hasMessageContaining(String.format(NO_ORDER_BY_IN_TXNS_MESSAGE, "SELECT", "at"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldRejectGroupBy()
|
||||
{
|
||||
String query = "BEGIN TRANSACTION\n" +
|
||||
" SELECT * FROM ks.tbl1 WHERE k=0 GROUP BY c LIMIT 1;" +
|
||||
"COMMIT TRANSACTION;";
|
||||
Assertions.assertThatThrownBy(() -> prepare(query))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_GROUP_BY_IN_TXNS_MESSAGE, "SELECT", "at"));
|
||||
|
||||
// The below code is left commented out as a reminder to think about this case... As of this writing GROUP BY does not parse in a LET clause... if that is ever fixed we should block it right away!
|
||||
// String query2 = "BEGIN TRANSACTION\n" +
|
||||
// " LET a = (SELECT * FROM ks.tbl1 WHERE k=0 GROUP BY c LIMIT 1;)" +
|
||||
// " SELECT a.v" +
|
||||
// "COMMIT TRANSACTION;";
|
||||
// Assertions.assertThatThrownBy(() -> prepare(query2))
|
||||
// .isInstanceOf(InvalidRequestException.class)
|
||||
// .hasMessageContaining(String.format(NO_GROUP_BY_IN_TXNS_MESSAGE, "SELECT", "at"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldRejectInClauseInLet()
|
||||
{
|
||||
// this is blocked not because this isn't safe, but that the logic to handle this is currently in the read coordinator, which Accord doesn't call.
|
||||
// So rather than return bad results to users, IN w/ LIMIT is blocked... until we can fix
|
||||
Select select = Select.builder()
|
||||
.table(tbl(1))
|
||||
.in("k", 0, 1)
|
||||
.limit(Literal.of(1))
|
||||
.build();
|
||||
|
||||
Assertions.assertThatThrownBy(() -> prepare(Txn.wrap(select).toCQL()))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_PARTITION_IN_CLAUSE_WITH_LIMIT, "SELECT", "at"));
|
||||
|
||||
Assertions.assertThatThrownBy(() -> prepare(Txn.builder()
|
||||
.addLet("a", select)
|
||||
.addReturnReferences("a.k")
|
||||
.build().toCQL()))
|
||||
.isInstanceOf(InvalidRequestException.class)
|
||||
.hasMessageContaining(String.format(NO_PARTITION_IN_CLAUSE_WITH_LIMIT, "SELECT", "at"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldRejectLetSelectOnNonTransactionalTable()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in New Issue