diff --git a/src/antlr/Parser.g b/src/antlr/Parser.g index f9fc2842df..b3fb490dca 100644 --- a/src/antlr/Parser.g +++ b/src/antlr/Parser.g @@ -573,7 +573,7 @@ normalInsertStatement [QualifiedName qn] returns [UpdateStatement.ParsedInsert e ( K_IF K_NOT K_EXISTS { ifNotExists = true; } )? ( usingClause[attrs] )? { - $expr = new UpdateStatement.ParsedInsert(qn, attrs, columnNames, values, ifNotExists, stmtSrc()); + $expr = new UpdateStatement.ParsedInsert(qn, attrs, columnNames, values, ifNotExists, stmtSrc(), isParsingTxn); } ; @@ -593,7 +593,7 @@ jsonInsertStatement [QualifiedName qn] returns [UpdateStatement.ParsedInsertJson ( K_IF K_NOT K_EXISTS { ifNotExists = true; } )? ( usingClause[attrs] )? { - $expr = new UpdateStatement.ParsedInsertJson(qn, attrs, val, defaultUnset, ifNotExists, stmtSrc()); + $expr = new UpdateStatement.ParsedInsertJson(qn, attrs, val, defaultUnset, ifNotExists, stmtSrc(), isParsingTxn); } ; @@ -674,7 +674,8 @@ deleteStatement returns [DeleteStatement.Parsed expr] wclause.build(), conditions == null ? Collections.emptyList() : conditions, ifExists, - stmtSrc()); + stmtSrc(), + isParsingTxn); } ; diff --git a/src/java/org/apache/cassandra/cql3/Operations.java b/src/java/org/apache/cassandra/cql3/Operations.java index 952305ec0d..8de94015a7 100644 --- a/src/java/org/apache/cassandra/cql3/Operations.java +++ b/src/java/org/apache/cassandra/cql3/Operations.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import com.google.common.base.Preconditions; import com.google.common.collect.Iterators; import org.apache.cassandra.cql3.functions.Function; @@ -38,6 +39,10 @@ public final class Operations implements Iterable * The type of statement. */ private final StatementType type; + /** + * If this operation is for a Transaction; this causes Operations to "migrate" when they require-read + */ + private final boolean isForTxn; /** * The operations on regular columns. @@ -52,29 +57,26 @@ public final class Operations implements Iterable private final List regularSubstitutions = new ArrayList<>(); private final List staticSubstitutions = new ArrayList<>(); - public Operations(StatementType type) + public Operations(StatementType type, boolean isForTxn) { this.type = type; + this.isForTxn = isForTxn; } - public void migrateReadRequiredOperations() + private Operations(Operations other) { - migrateReadRequiredOperations(staticOperations, staticSubstitutions); - migrateReadRequiredOperations(regularOperations, regularSubstitutions); + Preconditions.checkState(!other.isForTxn, "Unable to migrate from txn to txn"); + Preconditions.checkState(other.regularSubstitutions.isEmpty() && other.staticSubstitutions.isEmpty(), "Transaction substitutions are defined for a non-transaction operations! regular=%s, static=%s", other.regularSubstitutions, other.staticSubstitutions); + + type = other.type; + isForTxn = true; + for (Operation opt : other) + add(opt); } - private static void migrateReadRequiredOperations(List src, List dest) + public Operations forTxn() { - Iterator it = src.iterator(); - while (it.hasNext()) - { - Operation next = it.next(); - if (next.requiresRead()) - { - it.remove(); - dest.add(ReferenceOperation.create(next)); - } - } + return new Operations(this); } /** @@ -84,7 +86,7 @@ public final class Operations implements Iterable */ public boolean appliesToStaticColumns() { - return !staticOperations.isEmpty(); + return !staticIsEmpty(); } /** @@ -94,10 +96,10 @@ public final class Operations implements Iterable */ public boolean appliesToRegularColumns() { - // If we have regular operations, this applies to regular columns. + // If we have regular operations, this applies to regular columns. // Otherwise, if the statement is a DELETE and staticOperations is also empty, this means we have no operations, // which for a DELETE means a full row deletion. Which means the operation applies to all columns and regular ones in particular. - return !regularOperations.isEmpty() || (type.isDelete() && staticOperations.isEmpty()); + return !regularIsEmpty() || (type.isDelete() && staticIsEmpty()); } /** @@ -124,6 +126,11 @@ public final class Operations implements Iterable */ public void add(Operation operation) { + if (isForTxn && operation.requiresRead()) + { + add(operation.column, ReferenceOperation.create(operation)); + return; + } if (operation.column.isStatic()) staticOperations.add(operation); else @@ -132,6 +139,7 @@ public final class Operations implements Iterable public void add(ColumnMetadata column, ReferenceOperation operation) { + Preconditions.checkState(isForTxn, "Unable to add a transaction reference to a non-transaction operation"); if (column.isStatic()) staticSubstitutions.add(operation); else @@ -159,7 +167,7 @@ public final class Operations implements Iterable */ public boolean isEmpty() { - return staticOperations.isEmpty() && regularOperations.isEmpty(); + return staticIsEmpty() && regularIsEmpty(); } /** @@ -175,6 +183,7 @@ public final class Operations implements Iterable { regularOperations.forEach(p -> p.addFunctionsTo(functions)); staticOperations.forEach(p -> p.addFunctionsTo(functions)); + //TODO substitutions as well? } public List allSubstitutions() @@ -201,4 +210,14 @@ public final class Operations implements Iterable { return staticSubstitutions; } + + private boolean regularIsEmpty() + { + return regularOperations.isEmpty() && regularSubstitutions.isEmpty(); + } + + private boolean staticIsEmpty() + { + return staticOperations.isEmpty() && staticSubstitutions.isEmpty(); + } } diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java index 96d75d2651..e511af3190 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java +++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java @@ -520,7 +520,10 @@ public class CQL3CasRequest implements CASRequest int idx = 0; for (RowUpdate update : updates) { - ModificationStatement modification = update.stmt; + // Some operations may need to migrate to run in the transaction, so need to call forTxn to make sure this + // happens. + // see CASSANDRA-18337 + ModificationStatement modification = update.stmt.forTxn(); QueryOptions options = update.options; TxnWrite.Fragment fragment = modification.getTxnWriteFragment(idx++, state, options); fragments.add(fragment); diff --git a/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java b/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java index 4ce68fd75d..f2bbec1458 100644 --- a/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java @@ -63,6 +63,12 @@ public class DeleteStatement extends ModificationStatement super(StatementType.DELETE, bindVariables, cfm, operations, restrictions, conditions, attrs, source); } + @Override + protected ModificationStatement withOperations(Operations operations) + { + return new DeleteStatement(bindVariables, metadata, operations, restrictions, conditions, attrs, source); + } + @Override public void addUpdateForKey(PartitionUpdate.Builder updateBuilder, Clustering clustering, UpdateParameters params) throws InvalidRequestException @@ -135,6 +141,7 @@ public class DeleteStatement extends ModificationStatement { private final List deletions; private final WhereClause whereClause; + private final boolean isForTxn; public Parsed(QualifiedName name, Attributes.Raw attrs, @@ -142,11 +149,13 @@ public class DeleteStatement extends ModificationStatement WhereClause whereClause, List conditions, boolean ifExists, - StatementSource source) + StatementSource source, + boolean isForTxn) { super(name, StatementType.DELETE, attrs, conditions, false, ifExists, source); this.deletions = deletions; this.whereClause = whereClause; + this.isForTxn = isForTxn; } @@ -157,7 +166,7 @@ public class DeleteStatement extends ModificationStatement Conditions conditions, Attributes attrs) { - Operations operations = new Operations(type); + Operations operations = new Operations(type, isForTxn); for (Operation.RawDeletion deletion : deletions) { diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index d34b73a8a7..51f51e6407 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -147,19 +147,23 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa protected final VariableSpecifications bindVariables; public final TableMetadata metadata; - private final Attributes attrs; + protected final Attributes attrs; - private final StatementRestrictions restrictions; + protected final StatementRestrictions restrictions; private final Operations operations; private final RegularAndStaticColumns updatedColumns; - private final Conditions conditions; + protected final Conditions conditions; private final RegularAndStaticColumns conditionColumns; private final RegularAndStaticColumns requiresRead; + /** + * Used by {@link #forTxn()} to only compute a migrated copy of this statement for transactions + */ + private ModificationStatement txnStmt; private final List functions; @@ -900,12 +904,24 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa return new TxnReferenceOperations(metadata, clustering, regularOps, staticOps); } - @VisibleForTesting - public void migrateReadRequiredOperations() + public ModificationStatement forTxn() { - operations.migrateReadRequiredOperations(); + if (requiresRead.isEmpty()) return this; + ModificationStatement migrated = txnStmt; + if (migrated == null) + { + synchronized (requiresRead) + { + migrated = txnStmt; + if (migrated == null) + txnStmt = migrated = withOperations(operations.forTxn()); + } + } + return migrated; } + protected abstract ModificationStatement withOperations(Operations operations); + @VisibleForTesting public List getSubstitutions() { @@ -914,9 +930,6 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa public TxnWrite.Fragment getTxnWriteFragment(int index, ClientState state, QueryOptions options) { - // When an Operation requires a read, this cannot be done right away and must be done by the transaction itself, - // so migrate those Operations to a ReferenceOperation (which works properly in this case). - operations.migrateReadRequiredOperations(); PartitionUpdate baseUpdate = getTxnUpdate(state, options); TxnReferenceOperations referenceOps = getTxnReferenceOps(options, state); return new TxnWrite.Fragment(index, baseUpdate, referenceOps); diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java index d5002731f5..0eaf58a9bb 100644 --- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java @@ -87,6 +87,12 @@ public class UpdateStatement extends ModificationStatement super(type, bindVariables, metadata, operations, restrictions, conditions, attrs, source); } + @Override + protected ModificationStatement withOperations(Operations operations) + { + return new UpdateStatement(type, bindVariables, metadata, operations, restrictions, conditions, attrs, source); + } + @Override public void addUpdateForKey(PartitionUpdate.Builder updateBuilder, Clustering clustering, UpdateParameters params) { @@ -143,6 +149,7 @@ public class UpdateStatement extends ModificationStatement { private final List columnNames; private final List columnValues; + private final boolean isForTxn; /** * A parsed INSERT statement. @@ -158,11 +165,13 @@ public class UpdateStatement extends ModificationStatement List columnNames, List columnValues, boolean ifNotExists, - StatementSource source) + StatementSource source, + boolean isForTxn) { super(name, StatementType.INSERT, attrs, null, ifNotExists, false, source); this.columnNames = columnNames; this.columnValues = columnValues; + this.isForTxn = isForTxn; } @Override @@ -182,7 +191,7 @@ public class UpdateStatement extends ModificationStatement checkContainsNoDuplicates(columnNames, "The column names contains duplicates"); WhereClause.Builder whereClause = new WhereClause.Builder(); - Operations operations = new Operations(type); + Operations operations = new Operations(type, isForTxn); boolean hasClusteringColumnsSet = false; for (int i = 0; i < columnNames.size(); i++) @@ -244,12 +253,14 @@ public class UpdateStatement extends ModificationStatement { private final Json.Raw jsonValue; private final boolean defaultUnset; + private final boolean isForTxn; - public ParsedInsertJson(QualifiedName name, Attributes.Raw attrs, Json.Raw jsonValue, boolean defaultUnset, boolean ifNotExists, StatementSource source) + public ParsedInsertJson(QualifiedName name, Attributes.Raw attrs, Json.Raw jsonValue, boolean defaultUnset, boolean ifNotExists, StatementSource source, boolean isForTxn) { super(name, StatementType.INSERT, attrs, null, ifNotExists, false, source); this.jsonValue = jsonValue; this.defaultUnset = defaultUnset; + this.isForTxn = isForTxn; } @Override @@ -265,7 +276,7 @@ public class UpdateStatement extends ModificationStatement Json.Prepared prepared = jsonValue.prepareAndCollectMarkers(metadata, defs, bindVariables); WhereClause.Builder whereClause = new WhereClause.Builder(); - Operations operations = new Operations(type); + Operations operations = new Operations(type, isForTxn); boolean hasClusteringColumnsSet = false; for (ColumnMetadata def : defs) @@ -395,7 +406,7 @@ public class UpdateStatement extends ModificationStatement Conditions conditions, Attributes attrs) { - Operations operations = new Operations(type); + Operations operations = new Operations(type, isForTxn); for (Pair entry : updates.operations) { diff --git a/src/java/org/apache/cassandra/cql3/transactions/SelectReferenceSource.java b/src/java/org/apache/cassandra/cql3/transactions/SelectReferenceSource.java index ae5099aa3b..231539c888 100644 --- a/src/java/org/apache/cassandra/cql3/transactions/SelectReferenceSource.java +++ b/src/java/org/apache/cassandra/cql3/transactions/SelectReferenceSource.java @@ -29,7 +29,7 @@ import static org.apache.cassandra.cql3.statements.RequestValidations.checkTrue; public class SelectReferenceSource implements RowDataReference.ReferenceSource { - public static final String COLUMN_NOT_IN_SELECT_MESSAGE = "%s refererences a column not included in the select"; + public static final String COLUMN_NOT_IN_SELECT_MESSAGE = "%s references a column not included in the select"; private final SelectStatement statement; diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java index 750b003d4c..f6ae047fab 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java @@ -248,7 +248,6 @@ public abstract class AccordTestBase extends TestBaseImpl private static boolean isIdempotent(ModificationStatement update) { - update.migrateReadRequiredOperations(); // ReferenceValue.Constant is used during migration, which means a case like "a += 1" // ReferenceValue.Substitution uses a LET reference, so rerunning would always just see the new state long numConstants = update.getSubstitutions().stream()