diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 4b104e0a79..45f2499af7 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -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, diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index 4f567cb95f..3b5bf66976 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -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 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); diff --git a/test/unit/org/apache/cassandra/cql3/statements/TransactionStatementTest.java b/test/unit/org/apache/cassandra/cql3/statements/TransactionStatementTest.java index da7d9213e8..2e3371e542 100644 --- a/test/unit/org/apache/cassandra/cql3/statements/TransactionStatementTest.java +++ b/test/unit/org/apache/cassandra/cql3/statements/TransactionStatementTest.java @@ -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) 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) 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() {