diff --git a/CHANGES.txt b/CHANGES.txt index 99c38b89cd..771302e4d9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 541aa6e2ad..b0ab525851 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -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 diff --git a/src/java/org/apache/cassandra/config/GuardrailsOptions.java b/src/java/org/apache/cassandra/config/GuardrailsOptions.java index e019b8450b..ff42f59ba1 100644 --- a/src/java/org/apache/cassandra/config/GuardrailsOptions.java +++ b/src/java/org/apache/cassandra/config/GuardrailsOptions.java @@ -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 void updatePropertyWithLogging(String propertyName, T newValue, Supplier getter, Consumer setter) { T oldValue = getter.get(); diff --git a/src/java/org/apache/cassandra/cql3/Lists.java b/src/java/org/apache/cassandra/cql3/Lists.java index a2c26084f2..cc14e76d14 100644 --- a/src/java/org/apache/cassandra/cql3/Lists.java +++ b/src/java/org/apache/cassandra/cql3/Lists.java @@ -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"); diff --git a/src/java/org/apache/cassandra/cql3/UpdateParameters.java b/src/java/org/apache/cassandra/cql3/UpdateParameters.java index 4272307349..cbf9d51457 100644 --- a/src/java/org/apache/cassandra/cql3/UpdateParameters.java +++ b/src/java/org/apache/cassandra/cql3/UpdateParameters.java @@ -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; diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java index 24877ef23f..dabe05efce 100644 --- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java @@ -265,7 +265,8 @@ public class BatchStatement implements CQLStatement } @VisibleForTesting - public List getMutations(BatchQueryOptions options, + public List 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, diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java index 563a63907f..cc02650d06 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java +++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java @@ -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 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 map = stmt.requiresRead() ? Collections.singletonMap(key, current) : null; UpdateParameters params = new UpdateParameters(metadata, updateBuilder.columns(), + state, options, timestamp, nowInSeconds, diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 0a35e1ffc4..eed528fa1c 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -476,7 +476,8 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa cl.validateForWrite(); List 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 getMutations(QueryOptions options, - boolean local, - long timestamp, - int nowInSeconds, - long queryStartNanoTime) + private List getMutations(ClientState state, + QueryOptions options, + boolean local, + long timestamp, + int nowInSeconds, + long queryStartNanoTime) { List keys = buildPartitionKeyNames(options); HashMultiset 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 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 keys, NavigableSet> 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 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, diff --git a/src/java/org/apache/cassandra/db/guardrails/Guardrails.java b/src/java/org/apache/cassandra/db/guardrails/Guardrails.java index d29ab34768..00f3492332 100644 --- a/src/java/org/apache/cassandra/db/guardrails/Guardrails.java +++ b/src/java/org/apache/cassandra/db/guardrails/Guardrails.java @@ -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 values) { return values == null ? "" : String.join(",", values); diff --git a/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java b/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java index c34f224a9a..2abbdb5536 100644 --- a/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java +++ b/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java @@ -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. diff --git a/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java b/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java index c3509ee20d..197adddec2 100644 --- a/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java +++ b/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java @@ -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); } diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java index 48a0e1f3e4..919c19919f 100644 --- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java @@ -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), diff --git a/src/java/org/apache/cassandra/service/CASRequest.java b/src/java/org/apache/cassandra/service/CASRequest.java index 88fb9bd70d..77d1f5d41b 100644 --- a/src/java/org/apache/cassandra/service/CASRequest.java +++ b/src/java/org/apache/cassandra/service/CASRequest.java @@ -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; } diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index c3c0c88e2b..4780e45b87 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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); diff --git a/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java b/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java index d487e2d8cf..4d10107160 100644 --- a/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java +++ b/test/microbench/org/apache/cassandra/test/microbench/BatchStatementBench.java @@ -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); } diff --git a/test/unit/org/apache/cassandra/cql3/ListsTest.java b/test/unit/org/apache/cassandra/cql3/ListsTest.java index 92dcd96f35..27c36b8216 100644 --- a/test/unit/org/apache/cassandra/cql3/ListsTest.java +++ b/test/unit/org/apache/cassandra/cql3/ListsTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/db/guardrails/GuardrailReadBeforeWriteListOperationsTest.java b/test/unit/org/apache/cassandra/db/guardrails/GuardrailReadBeforeWriteListOperationsTest.java new file mode 100644 index 0000000000..a4049c7e07 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/guardrails/GuardrailReadBeforeWriteListOperationsTest.java @@ -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 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)"); + } + + @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); + } +} diff --git a/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java b/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java index 7028b84c3b..84ac5f1900 100644 --- a/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java +++ b/tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java @@ -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),