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

Operations.migrateReadRequiredOperations fails due to concurrent access when TransactionStatement is prepared 

patch by David Capwell; reviewed by Ariel Weisberg, Caleb Rackliffe for CASSANDRA-18337
This commit is contained in:
David Capwell 2023-03-17 12:03:07 -07:00
parent 68135aaf29
commit 4e95e3a440
8 changed files with 96 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/antlr/Parser.g
View File

@ -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.<ColumnCondition.Raw>emptyList() : conditions,
ifExists,
stmtSrc());
stmtSrc(),
isParsingTxn);
}
;

57
src/java/org/apache/cassandra/cql3/Operations.java
View File

@ -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<Operation>
* 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<Operation>
private final List<ReferenceOperation> regularSubstitutions = new ArrayList<>();
private final List<ReferenceOperation> 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<Operation> src, List<ReferenceOperation> dest)
public Operations forTxn()
{
Iterator<Operation> 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<Operation>
*/
public boolean appliesToStaticColumns()
{
return !staticOperations.isEmpty();
return !staticIsEmpty();
}
/**
@ -94,10 +96,10 @@ public final class Operations implements Iterable<Operation>
*/
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<Operation>
*/
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<Operation>
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<Operation>
*/
public boolean isEmpty()
{
return staticOperations.isEmpty() && regularOperations.isEmpty();
return staticIsEmpty() && regularIsEmpty();
}
/**
@ -175,6 +183,7 @@ public final class Operations implements Iterable<Operation>
{
regularOperations.forEach(p -> p.addFunctionsTo(functions));
staticOperations.forEach(p -> p.addFunctionsTo(functions));
//TODO substitutions as well?
}
public List<ReferenceOperation> allSubstitutions()
@ -201,4 +210,14 @@ public final class Operations implements Iterable<Operation>
{
return staticSubstitutions;
}
private boolean regularIsEmpty()
{
return regularOperations.isEmpty() && regularSubstitutions.isEmpty();
}
private boolean staticIsEmpty()
{
return staticOperations.isEmpty() && staticSubstitutions.isEmpty();
}
}

5
src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
View File

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

13
src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
View File

@ -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<Operation.RawDeletion> 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<ColumnCondition.Raw> 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)
{

31
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

@ -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<Function> 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<ReferenceOperation> 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);

21
src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
View File

@ -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<ColumnIdentifier> columnNames;
private final List<Term.Raw> columnValues;
private final boolean isForTxn;
/**
* A parsed <code>INSERT</code> statement.
@ -158,11 +165,13 @@ public class UpdateStatement extends ModificationStatement
List<ColumnIdentifier> columnNames,
List<Term.Raw> 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<ColumnIdentifier, Operation.RawUpdate> entry : updates.operations)
{

2
src/java/org/apache/cassandra/cql3/transactions/SelectReferenceSource.java
View File

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

1
test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
View File

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