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

Add guardrail for list operations that require read before write 

patch by Andrés de la Peña; reviewed by Ekaterina Dimitrova for CASSANDRA-17154
This commit is contained in:
Andrés de la Peña 2021-12-15 16:31:24 +00:00
parent bae92ee139
commit 95522f85d5
18 changed files with 303 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.1
* Add guardrail for list operations that require read before write (CASSANDRA-17154)
* Migrate thresholds for number of keyspaces and tables to guardrails (CASSANDRA-17195)
* Remove self-reference in SSTableTidier (CASSANDRA-17205)
* Add guardrail for query page size (CASSANDRA-17189)

3
conf/cassandra.yaml
View File

@ -1618,3 +1618,6 @@ enable_drop_compact_storage: false
# page_size:
# warn_threshold: -1
# abort_threshold: -1
# Guardrail to allow/disallow list operations that require read before write, i.e. setting list element by index and
# removing list elements by either index or value. Defaults to true.
# read_before_write_list_operations_enabled: true

15
src/java/org/apache/cassandra/config/GuardrailsOptions.java
View File

@ -66,6 +66,7 @@ public class GuardrailsOptions implements GuardrailsConfig
public final IntThreshold page_size = new IntThreshold();
public volatile boolean user_timestamps_enabled = true;
public volatile boolean read_before_write_list_operations_enabled = true;
public void validate()
{
@ -150,6 +151,20 @@ public class GuardrailsOptions implements GuardrailsConfig
x -> user_timestamps_enabled = x);
}
@Override
public boolean getReadBeforeWriteListOperationsEnabled()
{
return read_before_write_list_operations_enabled;
}
public void setReadBeforeWriteListOperationsEnabled(boolean enabled)
{
updatePropertyWithLogging(NAME_PREFIX + "read_before_write_list_operations_enabled",
enabled,
() -> read_before_write_list_operations_enabled,
x -> read_before_write_list_operations_enabled = x);
}
private static <T> void updatePropertyWithLogging(String propertyName, T newValue, Supplier<T> getter, Consumer<T> setter)
{
T oldValue = getter.get();

11
src/java/org/apache/cassandra/cql3/Lists.java
View File

@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.schema.ColumnMetadata;
import com.google.common.annotations.VisibleForTesting;
@ -448,6 +449,9 @@ public abstract class Lists
// we should not get here for frozen lists
assert column.type.isMultiCell() : "Attempted to set an individual element on a frozen list";
Guardrails.readBeforeWriteListOperationsEnabled
.ensureEnabled("Setting of list items by index requiring read before write", params.clientState);
ByteBuffer index = idx.bindAndGet(params.options);
ByteBuffer value = t.bindAndGet(params.options);
@ -565,6 +569,9 @@ public abstract class Lists
{
assert column.type.isMultiCell() : "Attempted to delete from a frozen list";
Guardrails.readBeforeWriteListOperationsEnabled
.ensureEnabled("Removal of list items requiring read before write", params.clientState);
// We want to call bind before possibly returning to reject queries where the value provided is not a list.
Term.Terminal value = t.bind(params.options);
@ -602,6 +609,10 @@ public abstract class Lists
public void execute(DecoratedKey partitionKey, UpdateParameters params) throws InvalidRequestException
{
assert column.type.isMultiCell() : "Attempted to delete an item by index from a frozen list";
Guardrails.readBeforeWriteListOperationsEnabled
.ensureEnabled("Removal of list items by index requiring read before write", params.clientState);
Term.Terminal index = t.bind(params.options);
if (index == null)
throw new InvalidRequestException("Invalid null value for list index");

4
src/java/org/apache/cassandra/cql3/UpdateParameters.java
View File

@ -27,6 +27,7 @@ import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.service.ClientState;
/**
* Groups the parameters of an update query, and make building updates easier.
@ -35,6 +36,7 @@ public class UpdateParameters
{
public final TableMetadata metadata;
public final RegularAndStaticColumns updatedColumns;
public final ClientState clientState;
public final QueryOptions options;
private final int nowInSec;
@ -54,6 +56,7 @@ public class UpdateParameters
public UpdateParameters(TableMetadata metadata,
RegularAndStaticColumns updatedColumns,
ClientState clientState,
QueryOptions options,
long timestamp,
int nowInSec,
@ -63,6 +66,7 @@ public class UpdateParameters
{
this.metadata = metadata;
this.updatedColumns = updatedColumns;
this.clientState = clientState;
this.options = options;
this.nowInSec = nowInSec;

12
src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
View File

@ -265,7 +265,8 @@ public class BatchStatement implements CQLStatement
}
@VisibleForTesting
public List<? extends IMutation> getMutations(BatchQueryOptions options,
public List<? extends IMutation> getMutations(ClientState state,
BatchQueryOptions options,
boolean local,
long batchTimestamp,
int nowInSeconds,
@ -306,7 +307,7 @@ public class BatchStatement implements CQLStatement
}
QueryOptions statementOptions = options.forStatement(i);
long timestamp = attrs.getTimestamp(batchTimestamp, statementOptions);
statement.addUpdates(collector, partitionKeys.get(i), statementOptions, local, timestamp, nowInSeconds, queryStartNanoTime);
statement.addUpdates(collector, partitionKeys.get(i), state, statementOptions, local, timestamp, nowInSeconds, queryStartNanoTime);
}
if (tablesWithZeroGcGs != null)
@ -415,7 +416,8 @@ public class BatchStatement implements CQLStatement
if (updatesVirtualTables)
executeInternalWithoutCondition(queryState, options, queryStartNanoTime);
else
executeWithoutConditions(getMutations(options, false, timestamp, nowInSeconds, queryStartNanoTime), options.getConsistency(), queryStartNanoTime);
executeWithoutConditions(getMutations(queryState.getClientState(), options, false, timestamp, nowInSeconds, queryStartNanoTime),
options.getConsistency(), queryStartNanoTime);
return new ResultMessage.Void();
}
@ -560,7 +562,7 @@ public class BatchStatement implements CQLStatement
long timestamp = batchOptions.getTimestamp(queryState);
int nowInSeconds = batchOptions.getNowInSeconds(queryState);
for (IMutation mutation : getMutations(batchOptions, true, timestamp, nowInSeconds, queryStartNanoTime))
for (IMutation mutation : getMutations(queryState.getClientState(), batchOptions, true, timestamp, nowInSeconds, queryStartNanoTime))
mutation.apply();
return null;
}
@ -577,7 +579,7 @@ public class BatchStatement implements CQLStatement
long timestamp = options.getTimestamp(state);
int nowInSeconds = options.getNowInSeconds(state);
try (RowIterator result = ModificationStatement.casInternal(request, timestamp, nowInSeconds))
try (RowIterator result = ModificationStatement.casInternal(state.getClientState(), request, timestamp, nowInSeconds))
{
ResultSet resultSet =
ModificationStatement.buildCasResultSet(ksName,

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

@ -34,6 +34,7 @@ import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.service.CASRequest;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.Pair;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
@ -228,13 +229,13 @@ public class CQL3CasRequest implements CASRequest
return builder.build();
}
public PartitionUpdate makeUpdates(FilteredPartition current) throws InvalidRequestException
public PartitionUpdate makeUpdates(FilteredPartition current, ClientState state) throws InvalidRequestException
{
PartitionUpdate.Builder updateBuilder = new PartitionUpdate.Builder(metadata, key, updatedColumns(), conditions.size());
for (RowUpdate upd : updates)
upd.applyUpdates(current, updateBuilder);
upd.applyUpdates(current, updateBuilder, state);
for (RangeDeletion upd : rangeDeletions)
upd.applyUpdates(current, updateBuilder);
upd.applyUpdates(current, updateBuilder, state);
PartitionUpdate partitionUpdate = updateBuilder.build();
IndexRegistry.obtain(metadata).validate(partitionUpdate);
@ -265,12 +266,13 @@ public class CQL3CasRequest implements CASRequest
this.nowInSeconds = nowInSeconds;
}
void applyUpdates(FilteredPartition current, PartitionUpdate.Builder updateBuilder)
void applyUpdates(FilteredPartition current, PartitionUpdate.Builder updateBuilder, ClientState state)
{
Map<DecoratedKey, Partition> map = stmt.requiresRead() ? Collections.singletonMap(key, current) : null;
UpdateParameters params =
new UpdateParameters(metadata,
updateBuilder.columns(),
state,
options,
timestamp,
nowInSeconds,
@ -297,13 +299,14 @@ public class CQL3CasRequest implements CASRequest
this.nowInSeconds = nowInSeconds;
}
void applyUpdates(FilteredPartition current, PartitionUpdate.Builder updateBuilder)
void applyUpdates(FilteredPartition current, PartitionUpdate.Builder updateBuilder, ClientState state)
{
// No slice statements currently require a read, but this maintains consistency with RowUpdate, and future proofs us
Map<DecoratedKey, Partition> map = stmt.requiresRead() ? Collections.singletonMap(key, current) : null;
UpdateParameters params =
new UpdateParameters(metadata,
updateBuilder.columns(),
state,
options,
timestamp,
nowInSeconds,

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

@ -476,7 +476,8 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
cl.validateForWrite();
List<? extends IMutation> mutations =
getMutations(options,
getMutations(queryState.getClientState(),
options,
false,
options.getTimestamp(queryState),
options.getNowInSeconds(queryState),
@ -639,7 +640,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
{
long timestamp = options.getTimestamp(queryState);
int nowInSeconds = options.getNowInSeconds(queryState);
for (IMutation mutation : getMutations(options, true, timestamp, nowInSeconds, queryStartNanoTime))
for (IMutation mutation : getMutations(queryState.getClientState(), options, true, timestamp, nowInSeconds, queryStartNanoTime))
mutation.apply();
return null;
}
@ -648,13 +649,13 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
{
CQL3CasRequest request = makeCasRequest(state, options);
try (RowIterator result = casInternal(request, options.getTimestamp(state), options.getNowInSeconds(state)))
try (RowIterator result = casInternal(state.getClientState(), request, options.getTimestamp(state), options.getNowInSeconds(state)))
{
return new ResultMessage.Rows(buildCasResultSet(result, state, options));
}
}
static RowIterator casInternal(CQL3CasRequest request, long timestamp, int nowInSeconds)
static RowIterator casInternal(ClientState state, CQL3CasRequest request, long timestamp, int nowInSeconds)
{
UUID ballot = UUIDGen.getTimeUUIDFromMicros(timestamp);
@ -669,7 +670,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
if (!request.appliesTo(current))
return current.rowIterator();
PartitionUpdate updates = request.makeUpdates(current);
PartitionUpdate updates = request.makeUpdates(current, state);
updates = TriggerExecutor.instance.execute(updates);
Commit proposal = Commit.newProposal(ballot, updates);
@ -680,27 +681,30 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
/**
* Convert statement into a list of mutations to apply on the server
*
* @param state the client state
* @param options value for prepared statement markers
* @param local if true, any requests (for collections) performed by getMutation should be done locally only.
* @param timestamp the current timestamp in microseconds to use if no timestamp is user provided.
*
* @return list of the mutations
*/
private List<? extends IMutation> getMutations(QueryOptions options,
boolean local,
long timestamp,
int nowInSeconds,
long queryStartNanoTime)
private List<? extends IMutation> getMutations(ClientState state,
QueryOptions options,
boolean local,
long timestamp,
int nowInSeconds,
long queryStartNanoTime)
{
List<ByteBuffer> keys = buildPartitionKeyNames(options);
HashMultiset<ByteBuffer> perPartitionKeyCounts = HashMultiset.create(keys);
SingleTableUpdatesCollector collector = new SingleTableUpdatesCollector(metadata, updatedColumns, perPartitionKeyCounts);
addUpdates(collector, keys, options, local, timestamp, nowInSeconds, queryStartNanoTime);
addUpdates(collector, keys, state, options, local, timestamp, nowInSeconds, queryStartNanoTime);
return collector.toMutations();
}
final void addUpdates(UpdatesCollector collector,
List<ByteBuffer> keys,
ClientState state,
QueryOptions options,
boolean local,
long timestamp,
@ -717,6 +721,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
UpdateParameters params = makeUpdateParameters(keys,
new ClusteringIndexSliceFilter(slices, false),
state,
options,
DataLimits.NONE,
local,
@ -742,7 +747,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
if (restrictions.hasClusteringColumnsRestrictions() && clusterings.isEmpty())
return;
UpdateParameters params = makeUpdateParameters(keys, clusterings, options, local, timestamp, nowInSeconds, queryStartNanoTime);
UpdateParameters params = makeUpdateParameters(keys, clusterings, state, options, local, timestamp, nowInSeconds, queryStartNanoTime);
for (ByteBuffer key : keys)
{
@ -789,6 +794,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
private UpdateParameters makeUpdateParameters(Collection<ByteBuffer> keys,
NavigableSet<Clustering<?>> clusterings,
ClientState state,
QueryOptions options,
boolean local,
long timestamp,
@ -798,6 +804,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
if (clusterings.contains(Clustering.STATIC_CLUSTERING))
return makeUpdateParameters(keys,
new ClusteringIndexSliceFilter(Slices.ALL, false),
state,
options,
DataLimits.cqlLimits(1),
local,
@ -807,6 +814,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
return makeUpdateParameters(keys,
new ClusteringIndexNamesFilter(clusterings, false),
state,
options,
DataLimits.NONE,
local,
@ -817,6 +825,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
private UpdateParameters makeUpdateParameters(Collection<ByteBuffer> keys,
ClusteringIndexFilter filter,
ClientState state,
QueryOptions options,
DataLimits limits,
boolean local,
@ -836,6 +845,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
return new UpdateParameters(metadata(),
updatedColumns(),
state,
options,
getTimestamp(timestamp, options),
nowInSeconds,

18
src/java/org/apache/cassandra/db/guardrails/Guardrails.java
View File

@ -127,6 +127,13 @@ public final class Guardrails implements GuardrailsMBean
: format("Aborting query for table %s, page size %s exceeds abort threshold of %s.",
what, value, threshold));
/**
* Guardrail disabling operations on lists that require read before write.
*/
public static final DisableFlag readBeforeWriteListOperationsEnabled =
new DisableFlag(state -> !CONFIG_PROVIDER.getOrCreate(state).getReadBeforeWriteListOperationsEnabled(),
"List operation requiring read before write");
private Guardrails()
{
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
@ -333,6 +340,17 @@ public final class Guardrails implements GuardrailsMBean
DEFAULT_CONFIG.getPageSize().setThresholds(warn, abort);
}
public boolean getReadBeforeWriteListOperationsEnabled()
{
return DEFAULT_CONFIG.getReadBeforeWriteListOperationsEnabled();
}
@Override
public void setReadBeforeWriteListOperationsEnabled(boolean enabled)
{
DEFAULT_CONFIG.setReadBeforeWriteListOperationsEnabled(enabled);
}
private static String toCSV(Set<String> values)
{
return values == null ? "" : String.join(",", values);

7
src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
View File

@ -92,6 +92,13 @@ public interface GuardrailsConfig
*/
IntThreshold getPageSize();
/**
* Returns whether list operations that require read before write are allowed.
*
* @return {@code true} if list operations that require read before write are allowed, {@code false} otherwise.
*/
boolean getReadBeforeWriteListOperationsEnabled();
/**
* Configuration of {@code int}-based thresholds to check if the guarded value should trigger a warning or abort the
* operation.

14
src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
View File

@ -206,4 +206,18 @@ public interface GuardrailsMBean
* @param abort The threshold to prevent requesting pages with more elements than threshold. -1 means disabled.
*/
void setPageSizeThreshold(int warn, int abort);
/**
* Returns whether list operations that require read before write are allowed.
*
* @return {@code true} if list operations that require read before write are allowed, {@code false} otherwise.
*/
boolean getReadBeforeWriteListOperationsEnabled();
/**
* Sets whether list operations that require read before write are allowed.
*
* @param enabled {@code true} if list operations that require read before write are allowed, {@code false} otherwise.
*/
void setReadBeforeWriteListOperationsEnabled(boolean enabled);
}

2
src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
View File

@ -52,6 +52,7 @@ import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.schema.*;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -247,6 +248,7 @@ public class CQLSSTableWriter implements Closeable
// and that forces a lot of initialization that we don't want.
UpdateParameters params = new UpdateParameters(insert.metadata,
insert.updatedColumns(),
ClientState.forInternalCalls(),
options,
insert.getTimestamp(TimeUnit.MILLISECONDS.toMicros(now), options),
(int) TimeUnit.MILLISECONDS.toSeconds(now),

2
src/java/org/apache/cassandra/service/CASRequest.java
View File

@ -42,5 +42,5 @@ public interface CASRequest
* The updates to perform of a CAS success. The values fetched using the readFilter()
* are passed as argument.
*/
public PartitionUpdate makeUpdates(FilteredPartition current) throws InvalidRequestException;
public PartitionUpdate makeUpdates(FilteredPartition current, ClientState state) throws InvalidRequestException;
}

2
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -315,7 +315,7 @@ public class StorageProxy implements StorageProxyMBean
}
// Create the desired updates
PartitionUpdate updates = request.makeUpdates(current);
PartitionUpdate updates = request.makeUpdates(current, state);
long size = updates.dataSize();
casWriteMetrics.mutationSize.update(size);

3
test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java
View File

@ -41,6 +41,7 @@ import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.FBUtilities;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@ -124,7 +125,7 @@ public class BatchStatementBench
@Benchmark
public void bench()
{
bs.getMutations(bqo, false, nowInSec, nowInSec, queryStartTime);
bs.getMutations(ClientState.forInternalCalls(), bqo, false, nowInSec, nowInSec, queryStartTime);
}

11
test/unit/org/apache/cassandra/cql3/ListsTest.java
View File

@ -36,6 +36,7 @@ import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
@ -141,8 +142,14 @@ public class ListsTest extends CQLTester
ByteBuffer keyBuf = ByteBufferUtil.bytes("key");
DecoratedKey key = Murmur3Partitioner.instance.decorateKey(keyBuf);
UpdateParameters parameters =
new UpdateParameters(metaData, null, QueryOptions.DEFAULT, System.currentTimeMillis(), FBUtilities.nowInSeconds(), 1000, Collections.emptyMap());
UpdateParameters parameters = new UpdateParameters(metaData,
null,
ClientState.forInternalCalls(),
QueryOptions.DEFAULT,
System.currentTimeMillis(),
FBUtilities.nowInSeconds(),
1000,
Collections.emptyMap());
Clustering<?> clustering = Clustering.make(ByteBufferUtil.bytes(1));
parameters.newRow(clustering);
prepender.execute(key, parameters);

177
test/unit/org/apache/cassandra/db/guardrails/GuardrailReadBeforeWriteListOperationsTest.java Normal file
View File

@ -0,0 +1,177 @@
/*
* 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.db.guardrails;
import java.util.Arrays;
import java.util.Collection;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
/**
* Tests the guardrail for read-before-write list operations, {@link Guardrails#readBeforeWriteListOperationsEnabled}.
*/
@RunWith(Parameterized.class)
public class GuardrailReadBeforeWriteListOperationsTest extends GuardrailTester
{
@Parameterized.Parameter
public boolean enabled;
@Parameterized.Parameters(name = "read_before_write_list_operations_enabled={0}")
public static Collection<Object> data()
{
return Arrays.asList(false, true);
}
@Before
public void before()
{
guardrails().setReadBeforeWriteListOperationsEnabled(enabled);
Assert.assertEquals(enabled, guardrails().getReadBeforeWriteListOperationsEnabled());
createTable("CREATE TABLE %s (k int PRIMARY KEY, l list<int>)");
}
@Test
public void tesInsertFullValue() throws Throwable
{
// insert from scratch
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2])");
assertRows(row(0, list(1, 2)));
// insert overriding previous value
assertValid("INSERT INTO %s (k, l) VALUES (0, [2, 3])");
assertRows(row(0, list(2, 3)));
}
@Test
public void testUpdateFullValue() throws Throwable
{
// update from scratch
assertValid("UPDATE %s SET l = [1, 2] WHERE k = 0");
assertRows(row(0, list(1, 2)));
// update overriding previous value
assertValid("UPDATE %s SET l = [2, 3] WHERE k = 0");
assertRows(row(0, list(2, 3)));
}
@Test
public void testDeleteFullValue() throws Throwable
{
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2])");
assertValid("DELETE l FROM %s WHERE k = 0");
assertRows(row(0, null));
}
@Test
public void testAppend() throws Throwable
{
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2])");
assertValid("UPDATE %s SET l = l + [3, 4] WHERE k = 0");
assertRows(row(0, list(1, 2, 3, 4)));
}
@Test
public void testPrepend() throws Throwable
{
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2])");
assertValid("UPDATE %s SET l = [3, 4] + l WHERE k = 0");
assertRows(row(0, list(3, 4, 1, 2)));
}
@Test
public void testUpdateByIndex() throws Throwable
{
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2, 3])");
testGuardrail("UPDATE %s SET l[1] = 4 WHERE k = 0",
"Setting of list items by index requiring read before write is not allowed",
row(0, list(1, 4, 3)));
}
@Test
public void testDeleteByIndex() throws Throwable
{
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2, 3])");
testGuardrail("DELETE l[1] FROM %s WHERE k = 0",
"Removal of list items by index requiring read before write is not allowed",
row(0, list(1, 3)));
}
@Test
public void testDeleteByItem() throws Throwable
{
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2, 3])");
testGuardrail("UPDATE %s SET l = l - [2] WHERE k = 0",
"Removal of list items requiring read before write is not allowed",
row(0, list(1, 3)));
}
@Test
public void testBatch() throws Throwable
{
assertValid("INSERT INTO %s (k, l) VALUES (0, [1, 2, 3])");
testGuardrail("BEGIN BATCH UPDATE %s SET l[1] = 0 WHERE k = 0; APPLY BATCH",
"Setting of list items by index requiring read before write is not allowed",
row(0, list(1, 0, 3)));
testGuardrail("BEGIN BATCH DELETE l[1] FROM %s WHERE k = 0; APPLY BATCH",
"Removal of list items by index requiring read before write is not allowed",
row(0, list(1, 3)));
testGuardrail("BEGIN BATCH UPDATE %s SET l = l - [3] WHERE k = 0; APPLY BATCH",
"Removal of list items requiring read before write is not allowed",
row(0, list(1)));
}
@Test
public void testExcludedUsers() throws Throwable
{
testExcludedUsers(() -> "INSERT INTO %s (k, l) VALUES (0, [1, 2, 3, 4, 5])",
() -> "UPDATE %s SET l[1] = 4 WHERE k = 0",
() -> "DELETE l[1] FROM %s WHERE k = 0",
() -> "INSERT INTO %s (k, l) VALUES (0, [1, 2, 3])",
() -> "UPDATE %s SET l = l - [2] WHERE k = 0",
() -> "BEGIN BATCH UPDATE %s SET l[1] = 0 WHERE k = 0; APPLY BATCH",
() -> "BEGIN BATCH DELETE l[1] FROM %s WHERE k = 0; APPLY BATCH",
() -> "BEGIN BATCH UPDATE %s SET l = l - [3] WHERE k = 0; APPLY BATCH");
}
private void testGuardrail(String query, String expectedMessage, Object[]... rows) throws Throwable
{
if (enabled)
{
assertValid(query);
assertRows(rows);
}
else
{
assertAborts(expectedMessage, query);
}
}
private void assertRows(Object[]... rows) throws Throwable
{
assertRowsNet(executeNet("SELECT * FROM %s"), rows);
}
}

1
tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java
View File

@ -252,6 +252,7 @@ public class StressCQLSSTableWriter implements Closeable
// and that forces a lot of initialization that we don't want.
UpdateParameters params = new UpdateParameters(insert.metadata(),
insert.updatedColumns(),
ClientState.forInternalCalls(),
options,
insert.getTimestamp(TimeUnit.MILLISECONDS.toMicros(now), options),
(int) TimeUnit.MILLISECONDS.toSeconds(now),