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

Add guardrail for SELECT IN terms and their cartesian product 

patch by Andrés de la Peña; reviewed by Ekaterina Dimitrova for CASSANDRA-17187

Co-authored-by: Aleksandr Sorokoumov <aleksandr.sorokoumov@gmail.com>
Co-authored-by: Andrés de la Peña <a.penya.garcia@gmail.com>
This commit is contained in:
Andrés de la Peña 2022-03-08 11:08:29 +00:00
parent ac34f28e31
commit 3233c82311
30 changed files with 501 additions and 124 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 SELECT IN terms and their cartesian product (CASSANDRA-17187)
* remove unused imports in cqlsh.py and cqlshlib (CASSANDRA-17413)
* deprecate property windows_timer_interval (CASSANDRA-17404)
* Expose streaming as a vtable (CASSANDRA-17390)

5
conf/cassandra.yaml
View File

@ -1620,6 +1620,11 @@ drop_compact_storage_enabled: false
# The two thresholds default to -1 to disable.
# partition_keys_in_select_warn_threshold: -1
# partition_keys_in_select_fail_threshold: -1
# Guardrail to warn or fail when an IN query creates a cartesian product with a size exceeding threshold,
# eg. "a in (1,2,...10) and b in (1,2...10)" results in cartesian product of 100.
# The two thresholds default to -1 to disable.
# in_select_cartesian_product_warn_threshold: -1
# in_select_cartesian_product_fail_threshold: -1
# Startup Checks are executed as part of Cassandra startup process, not all of them
# are configurable (so you can disable them) but these which are enumerated bellow.

3
src/java/org/apache/cassandra/config/Config.java
View File

@ -767,10 +767,13 @@ public class Config
public volatile int page_size_fail_threshold = DISABLED_GUARDRAIL;
public volatile int partition_keys_in_select_warn_threshold = DISABLED_GUARDRAIL;
public volatile int partition_keys_in_select_fail_threshold = DISABLED_GUARDRAIL;
public volatile int in_select_cartesian_product_warn_threshold = DISABLED_GUARDRAIL;
public volatile int in_select_cartesian_product_fail_threshold = DISABLED_GUARDRAIL;
public volatile Set<String> table_properties_ignored = Collections.emptySet();
public volatile Set<String> table_properties_disallowed = Collections.emptySet();
public volatile boolean user_timestamps_enabled = true;
public volatile boolean read_before_write_list_operations_enabled = true;
public volatile DurationSpec streaming_state_expires = DurationSpec.inDays(3);
public volatile DataStorageSpec streaming_state_size = DataStorageSpec.inMebibytes(40);

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

@ -69,6 +69,7 @@ public class GuardrailsOptions implements GuardrailsConfig
validateIntThreshold(config.page_size_warn_threshold, config.page_size_fail_threshold, "page_size");
validateIntThreshold(config.partition_keys_in_select_warn_threshold,
config.partition_keys_in_select_fail_threshold, "partition_keys_in_select");
validateIntThreshold(config.in_select_cartesian_product_warn_threshold, config.in_select_cartesian_product_fail_threshold, "in_select_cartesian_product");
}
@Override
@ -321,6 +322,31 @@ public class GuardrailsOptions implements GuardrailsConfig
x -> config.read_before_write_list_operations_enabled = x);
}
@Override
public int getInSelectCartesianProductWarnThreshold()
{
return config.in_select_cartesian_product_warn_threshold;
}
@Override
public int getInSelectCartesianProductFailThreshold()
{
return config.in_select_cartesian_product_fail_threshold;
}
public void setInSelectCartesianProductThreshold(int warn, int fail)
{
validateIntThreshold(warn, fail, "in_select_cartesian_product");
updatePropertyWithLogging("in_select_cartesian_product_warn_threshold",
warn,
() -> config.in_select_cartesian_product_warn_threshold,
x -> config.in_select_cartesian_product_warn_threshold = x);
updatePropertyWithLogging("in_select_cartesian_product_fail_threshold",
fail,
() -> config.in_select_cartesian_product_fail_threshold,
x -> config.in_select_cartesian_product_fail_threshold = x);
}
private static <T> void updatePropertyWithLogging(String propertyName, T newValue, Supplier<T> getter, Consumer<T> setter)
{
T oldValue = getter.get();

8
src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.cql3.restrictions;
import java.util.*;
import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.cql3.QueryOptions;
@ -27,6 +28,7 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.btree.BTreeSet;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
@ -101,12 +103,16 @@ final class ClusteringColumnRestrictions extends RestrictionSetWrapper
return false;
}
public NavigableSet<Clustering<?>> valuesAsClustering(QueryOptions options) throws InvalidRequestException
public NavigableSet<Clustering<?>> valuesAsClustering(QueryOptions options, ClientState state) throws InvalidRequestException
{
MultiCBuilder builder = MultiCBuilder.create(comparator, hasIN());
for (SingleRestriction r : restrictions)
{
r.appendTo(builder, options);
if (hasIN() && Guardrails.inSelectCartesianProduct.enabled(state))
Guardrails.inSelectCartesianProduct.guard(builder.buildSize(), "clustering key", state);
if (builder.hasMissingElements())
break;
}

3
src/java/org/apache/cassandra/cql3/restrictions/PartitionKeyRestrictions.java
View File

@ -23,6 +23,7 @@ import java.util.List;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.statements.Bound;
import org.apache.cassandra.service.ClientState;
/**
* A set of restrictions on the partition key.
@ -32,7 +33,7 @@ interface PartitionKeyRestrictions extends Restrictions
{
public PartitionKeyRestrictions mergeWith(Restriction restriction);
public List<ByteBuffer> values(QueryOptions options);
public List<ByteBuffer> values(QueryOptions options, ClientState state);
public List<ByteBuffer> bounds(Bound b, QueryOptions options);

8
src/java/org/apache/cassandra/cql3/restrictions/PartitionKeySingleRestrictionSet.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.cql3.restrictions;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.statements.Bound;
@ -28,6 +29,7 @@ import org.apache.cassandra.db.ClusteringPrefix;
import org.apache.cassandra.db.MultiCBuilder;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.service.ClientState;
/**
* A set of single restrictions on the partition key.
@ -78,12 +80,16 @@ final class PartitionKeySingleRestrictionSet extends RestrictionSetWrapper imple
}
@Override
public List<ByteBuffer> values(QueryOptions options)
public List<ByteBuffer> values(QueryOptions options, ClientState state)
{
MultiCBuilder builder = MultiCBuilder.create(comparator, hasIN());
for (SingleRestriction r : restrictions)
{
r.appendTo(builder, options);
if (hasIN() && Guardrails.inSelectCartesianProduct.enabled(state))
Guardrails.inSelectCartesianProduct.guard(builder.buildSize(), "partition key", state);
if (builder.hasMissingElements())
break;
}

11
src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
View File

@ -35,6 +35,7 @@ import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.btree.BTreeSet;
import org.apache.commons.lang3.builder.ToStringBuilder;
@ -619,11 +620,12 @@ public final class StatementRestrictions
* Returns the partition keys for which the data is requested.
*
* @param options the query options
* @param state the client state
* @return the partition keys for which the data is requested.
*/
public List<ByteBuffer> getPartitionKeys(final QueryOptions options)
public List<ByteBuffer> getPartitionKeys(final QueryOptions options, ClientState state)
{
return partitionKeyRestrictions.values(options);
return partitionKeyRestrictions.values(options, state);
}
/**
@ -741,9 +743,10 @@ public final class StatementRestrictions
* Returns the requested clustering columns.
*
* @param options the query options
* @param state the client state
* @return the requested clustering columns
*/
public NavigableSet<Clustering<?>> getClusteringColumns(QueryOptions options)
public NavigableSet<Clustering<?>> getClusteringColumns(QueryOptions options, ClientState state)
{
// If this is a names command and the table is a static compact one, then as far as CQL is concerned we have
// only a single row which internally correspond to the static parts. In which case we want to return an empty
@ -751,7 +754,7 @@ public final class StatementRestrictions
if (table.isStaticCompactTable())
return BTreeSet.empty(table.comparator);
return clusteringColumnsRestrictions.valuesAsClustering(options);
return clusteringColumnsRestrictions.valuesAsClustering(options, state);
}
/**

10
src/java/org/apache/cassandra/cql3/restrictions/TokenFilter.java
View File

@ -35,6 +35,7 @@ import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.service.ClientState;
import static org.apache.cassandra.cql3.statements.Bound.END;
import static org.apache.cassandra.cql3.statements.Bound.START;
@ -102,9 +103,9 @@ final class TokenFilter implements PartitionKeyRestrictions
}
@Override
public List<ByteBuffer> values(QueryOptions options) throws InvalidRequestException
public List<ByteBuffer> values(QueryOptions options, ClientState state) throws InvalidRequestException
{
return filter(restrictions.values(options), options);
return filter(restrictions.values(options, state), options, state);
}
@Override
@ -139,13 +140,14 @@ final class TokenFilter implements PartitionKeyRestrictions
*
* @param values the values returned by the decorated restriction
* @param options the query options
* @param state the client state
* @return the values matching the token restriction
* @throws InvalidRequestException if the request is invalid
*/
private List<ByteBuffer> filter(List<ByteBuffer> values, QueryOptions options) throws InvalidRequestException
private List<ByteBuffer> filter(List<ByteBuffer> values, QueryOptions options, ClientState state) throws InvalidRequestException
{
RangeSet<Token> rangeSet = tokenRestriction.hasSlice() ? toRangeSet(tokenRestriction, options)
: toRangeSet(tokenRestriction.values(options));
: toRangeSet(tokenRestriction.values(options, state));
return filterWithRangeSet(rangeSet, values);
}

10
src/java/org/apache/cassandra/cql3/restrictions/TokenRestriction.java
View File

@ -31,6 +31,7 @@ import org.apache.cassandra.cql3.statements.Bound;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.service.ClientState;
import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
@ -205,7 +206,10 @@ public abstract class TokenRestriction implements PartitionKeyRestrictions
@Override
public List<ByteBuffer> bounds(Bound b, QueryOptions options) throws InvalidRequestException
{
return values(options);
// ClientState is used by inSelectCartesianProduct guardrail to skip non-ordinary users.
// Passing null here to avoid polluting too many methods, because in case of EQ token restriction,
// it won't generate high cartesian product.
return values(options, null);
}
@Override
@ -221,7 +225,7 @@ public abstract class TokenRestriction implements PartitionKeyRestrictions
}
@Override
public List<ByteBuffer> values(QueryOptions options) throws InvalidRequestException
public List<ByteBuffer> values(QueryOptions options, ClientState state) throws InvalidRequestException
{
return Collections.singletonList(value.bindAndGet(options));
}
@ -254,7 +258,7 @@ public abstract class TokenRestriction implements PartitionKeyRestrictions
}
@Override
public List<ByteBuffer> values(QueryOptions options) throws InvalidRequestException
public List<ByteBuffer> values(QueryOptions options, ClientState state) throws InvalidRequestException
{
throw new UnsupportedOperationException();
}

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

@ -282,7 +282,7 @@ public class BatchStatement implements CQLStatement
ModificationStatement stmt = statements.get(i);
if (metadata != null && !stmt.metadata.id.equals(metadata.id))
metadata = null;
List<ByteBuffer> stmtPartitionKeys = stmt.buildPartitionKeyNames(options.forStatement(i));
List<ByteBuffer> stmtPartitionKeys = stmt.buildPartitionKeyNames(options.forStatement(i), state);
partitionKeys.add(stmtPartitionKeys);
HashMultiset<ByteBuffer> perKeyCountsForTable = partitionCounts.computeIfAbsent(stmt.metadata.id, k -> HashMultiset.create());
for (int stmtIdx = 0, stmtSize = stmtPartitionKeys.size(); stmtIdx < stmtSize; stmtIdx++)
@ -489,7 +489,7 @@ public class BatchStatement implements CQLStatement
ModificationStatement statement = statements.get(i);
QueryOptions statementOptions = options.forStatement(i);
long timestamp = attrs.getTimestamp(batchTimestamp, statementOptions);
List<ByteBuffer> pks = statement.buildPartitionKeyNames(statementOptions);
List<ByteBuffer> pks = statement.buildPartitionKeyNames(statementOptions, state.getClientState());
if (statement.getRestrictions().keyIsInRelation())
throw new IllegalArgumentException("Batch with conditions cannot span multiple partitions (you cannot use IN on the partition key)");
if (key == null)
@ -524,7 +524,7 @@ public class BatchStatement implements CQLStatement
}
else
{
Clustering<?> clustering = Iterables.getOnlyElement(statement.createClustering(statementOptions));
Clustering<?> clustering = Iterables.getOnlyElement(statement.createClustering(statementOptions, state.getClientState()));
if (statement.hasConditions())
{
statement.addConditions(clustering, casRequest, statementOptions);

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

@ -330,23 +330,23 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
return conditions.isIfExists();
}
public List<ByteBuffer> buildPartitionKeyNames(QueryOptions options)
public List<ByteBuffer> buildPartitionKeyNames(QueryOptions options, ClientState state)
throws InvalidRequestException
{
List<ByteBuffer> partitionKeys = restrictions.getPartitionKeys(options);
List<ByteBuffer> partitionKeys = restrictions.getPartitionKeys(options, state);
for (ByteBuffer key : partitionKeys)
QueryProcessor.validateKey(key);
return partitionKeys;
}
public NavigableSet<Clustering<?>> createClustering(QueryOptions options)
public NavigableSet<Clustering<?>> createClustering(QueryOptions options, ClientState state)
throws InvalidRequestException
{
if (appliesOnlyToStaticColumns() && !restrictions.hasClusteringColumnsRestrictions())
return FBUtilities.singleton(CBuilder.STATIC_BUILDER.build(), metadata().comparator);
return restrictions.getClusteringColumns(options);
return restrictions.getClusteringColumns(options, state);
}
/**
@ -508,7 +508,8 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
private CQL3CasRequest makeCasRequest(QueryState queryState, QueryOptions options)
{
List<ByteBuffer> keys = buildPartitionKeyNames(options);
ClientState clientState = queryState.getClientState();
List<ByteBuffer> keys = buildPartitionKeyNames(options, clientState);
// We don't support IN for CAS operation so far
checkFalse(restrictions.keyIsInRelation(),
"IN on the partition key is not supported with conditional %s",
@ -522,7 +523,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
"IN on the clustering key columns is not supported with conditional %s",
type.isUpdate()? "updates" : "deletions");
Clustering<?> clustering = Iterables.getOnlyElement(createClustering(options));
Clustering<?> clustering = Iterables.getOnlyElement(createClustering(options, clientState));
CQL3CasRequest request = new CQL3CasRequest(metadata(), key, conditionColumns(), updatesRegularRows(), updatesStaticRow());
addConditions(clustering, request, options);
@ -695,7 +696,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
int nowInSeconds,
long queryStartNanoTime)
{
List<ByteBuffer> keys = buildPartitionKeyNames(options);
List<ByteBuffer> keys = buildPartitionKeyNames(options, state);
HashMultiset<ByteBuffer> perPartitionKeyCounts = HashMultiset.create(keys);
SingleTableUpdatesCollector collector = new SingleTableUpdatesCollector(metadata, updatedColumns, perPartitionKeyCounts);
addUpdates(collector, keys, state, options, local, timestamp, nowInSeconds, queryStartNanoTime);
@ -741,7 +742,7 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
}
else
{
NavigableSet<Clustering<?>> clusterings = createClustering(options);
NavigableSet<Clustering<?>> clusterings = createClustering(options, state);
// If some of the restrictions were unspecified (e.g. empty IN restrictions) we do not need to do anything.
if (restrictions.hasClusteringColumnsRestrictions() && clusterings.isEmpty())

46
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -293,7 +293,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
DataLimits limit = getDataLimits(userLimit, perPartitionLimit, pageSize);
if (isPartitionRangeQuery)
return getRangeCommand(options, columnFilter, limit, nowInSec);
return getRangeCommand(options, state, columnFilter, limit, nowInSec);
return getSliceCommands(options, state, columnFilter, limit, nowInSec);
}
@ -528,7 +528,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
private ReadQuery getSliceCommands(QueryOptions options, ClientState state, ColumnFilter columnFilter,
DataLimits limit, int nowInSec)
{
Collection<ByteBuffer> keys = restrictions.getPartitionKeys(options);
Collection<ByteBuffer> keys = restrictions.getPartitionKeys(options, state);
if (keys.isEmpty())
return ReadQuery.empty(table);
@ -537,7 +537,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
Guardrails.partitionKeysInSelect.guard(keys.size(), table.name, state);
}
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, columnFilter);
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, state, columnFilter);
if (filter == null || filter.isEmpty(table.comparator))
return ReadQuery.empty(table);
@ -564,8 +564,9 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
public Slices clusteringIndexFilterAsSlices()
{
QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList());
ClientState state = ClientState.forInternalCalls();
ColumnFilter columnFilter = selection.newSelectors(options).getColumnFilter();
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, columnFilter);
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, state, columnFilter);
if (filter instanceof ClusteringIndexSliceFilter)
return ((ClusteringIndexSliceFilter)filter).requestedSlices();
@ -582,8 +583,9 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
public SinglePartitionReadCommand internalReadForView(DecoratedKey key, int nowInSec)
{
QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList());
ClientState state = ClientState.forInternalCalls();
ColumnFilter columnFilter = selection.newSelectors(options).getColumnFilter();
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, columnFilter);
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, state, columnFilter);
RowFilter rowFilter = getRowFilter(options);
return SinglePartitionReadCommand.create(table, nowInSec, columnFilter, rowFilter, DataLimits.NONE, key, filter);
}
@ -596,9 +598,9 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return getRowFilter(QueryOptions.forInternalCalls(Collections.emptyList()));
}
private ReadQuery getRangeCommand(QueryOptions options, ColumnFilter columnFilter, DataLimits limit, int nowInSec)
private ReadQuery getRangeCommand(QueryOptions options, ClientState state, ColumnFilter columnFilter, DataLimits limit, int nowInSec)
{
ClusteringIndexFilter clusteringIndexFilter = makeClusteringIndexFilter(options, columnFilter);
ClusteringIndexFilter clusteringIndexFilter = makeClusteringIndexFilter(options, state, columnFilter);
if (clusteringIndexFilter == null)
return ReadQuery.empty(table);
@ -619,7 +621,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return command;
}
private ClusteringIndexFilter makeClusteringIndexFilter(QueryOptions options, ColumnFilter columnFilter)
private ClusteringIndexFilter makeClusteringIndexFilter(QueryOptions options, ClientState state, ColumnFilter columnFilter)
{
if (parameters.isDistinct)
{
@ -642,7 +644,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return new ClusteringIndexSliceFilter(slices, isReversed);
}
NavigableSet<Clustering<?>> clusterings = getRequestedRows(options);
NavigableSet<Clustering<?>> clusterings = getRequestedRows(options, state);
// We can have no clusterings if either we're only selecting the static columns, or if we have
// a 'IN ()' for clusterings. In that case, we still want to query if some static columns are
// queried. But we're fine otherwise.
@ -774,12 +776,12 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return userLimit;
}
private NavigableSet<Clustering<?>> getRequestedRows(QueryOptions options) throws InvalidRequestException
private NavigableSet<Clustering<?>> getRequestedRows(QueryOptions options, ClientState state) throws InvalidRequestException
{
// Note: getRequestedColumns don't handle static columns, but due to CASSANDRA-5762
// we always do a slice for CQL3 tables, so it's ok to ignore them here
assert !restrictions.isColumnRange();
return restrictions.getClusteringColumns(options);
return restrictions.getClusteringColumns(options, state);
}
/**
@ -841,8 +843,9 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
if (result.shouldWarn(options.getCoordinatorReadSizeWarnThresholdKB()))
{
String msg = String.format("Read on table %s has exceeded the size warning threshold of %,d kb", table, options.getCoordinatorReadSizeWarnThresholdKB());
ClientWarn.instance.warn(msg + " with " + loggableTokens(options));
logger.warn("{} with query {}", msg, asCQL(options));
ClientState state = ClientState.forInternalCalls();
ClientWarn.instance.warn(msg + " with " + loggableTokens(options, state));
logger.warn("{} with query {}", msg, asCQL(options, state));
if (store != null)
store.metric.coordinatorReadSizeWarnings.mark();
}
@ -855,9 +858,10 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
if (result.shouldReject(options.getCoordinatorReadSizeAbortThresholdKB()))
{
String msg = String.format("Read on table %s has exceeded the size failure threshold of %,d kb", table, options.getCoordinatorReadSizeAbortThresholdKB());
String clientMsg = msg + " with " + loggableTokens(options);
ClientState state = ClientState.forInternalCalls();
String clientMsg = msg + " with " + loggableTokens(options, state);
ClientWarn.instance.warn(clientMsg);
logger.warn("{} with query {}", msg, asCQL(options));
logger.warn("{} with query {}", msg, asCQL(options, state));
ColumnFamilyStore store = cfs();
if (store != null)
{
@ -1442,7 +1446,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
}
private String loggableTokens(QueryOptions options)
private String loggableTokens(QueryOptions options, ClientState state)
{
if (restrictions.isKeyRange() || restrictions.usesSecondaryIndexing())
{
@ -1454,7 +1458,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
}
else
{
Collection<ByteBuffer> keys = restrictions.getPartitionKeys(options);
Collection<ByteBuffer> keys = restrictions.getPartitionKeys(options, state);
if (keys.size() == 1)
{
return "token: " + table.partitioner.getToken(Iterables.getOnlyElement(keys)).toString();
@ -1474,7 +1478,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
}
}
private String asCQL(QueryOptions options)
private String asCQL(QueryOptions options, ClientState state)
{
ColumnFilter columnFilter = selection.newSelectors(options).getColumnFilter();
StringBuilder sb = new StringBuilder();
@ -1484,7 +1488,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
if (restrictions.isKeyRange() || restrictions.usesSecondaryIndexing())
{
// partition range
ClusteringIndexFilter clusteringIndexFilter = makeClusteringIndexFilter(options, columnFilter);
ClusteringIndexFilter clusteringIndexFilter = makeClusteringIndexFilter(options, state, columnFilter);
if (clusteringIndexFilter == null)
return "EMPTY";
@ -1515,10 +1519,10 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
else
{
// single partition
Collection<ByteBuffer> keys = restrictions.getPartitionKeys(options);
Collection<ByteBuffer> keys = restrictions.getPartitionKeys(options, state);
if (keys.isEmpty())
return "EMPTY";
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, columnFilter);
ClusteringIndexFilter filter = makeClusteringIndexFilter(options, state, columnFilter);
if (filter == null)
return "EMPTY";

25
src/java/org/apache/cassandra/db/MultiCBuilder.java
View File

@ -129,6 +129,13 @@ public abstract class MultiCBuilder
return comparator.size() - size;
}
/**
* Returns the current number of results when {@link #build()} is called
*
* @return the current number of build results
*/
public abstract int buildSize();
/**
* Checks if the clusterings contains null elements.
*
@ -252,6 +259,12 @@ public abstract class MultiCBuilder
return addEachElementToAll(values.get(0));
}
@Override
public int buildSize()
{
return hasMissingElements ? 0 : 1;
}
public NavigableSet<Clustering<?>> build()
{
built = true;
@ -309,7 +322,7 @@ public abstract class MultiCBuilder
checkUpdateable();
if (elementsList.isEmpty())
elementsList.add(new ArrayList<ByteBuffer>());
elementsList.add(new ArrayList<>());
if (value == null)
containsNull = true;
@ -328,7 +341,7 @@ public abstract class MultiCBuilder
checkUpdateable();
if (elementsList.isEmpty())
elementsList.add(new ArrayList<ByteBuffer>());
elementsList.add(new ArrayList<>());
if (values.isEmpty())
{
@ -365,7 +378,7 @@ public abstract class MultiCBuilder
checkUpdateable();
if (elementsList.isEmpty())
elementsList.add(new ArrayList<ByteBuffer>());
elementsList.add(new ArrayList<>());
if (values.isEmpty())
{
@ -397,6 +410,12 @@ public abstract class MultiCBuilder
return this;
}
@Override
public int buildSize()
{
return hasMissingElements ? 0 : elementsList.size();
}
public NavigableSet<Clustering<?>> build()
{
built = true;

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

@ -148,6 +148,20 @@ public final class Guardrails implements GuardrailsMBean
new DisableFlag(state -> !CONFIG_PROVIDER.getOrCreate(state).getReadBeforeWriteListOperationsEnabled(),
"List operation requiring read before write");
/**
* Guardrail on the number of restrictions created by a cartesian product of a CQL's {@code IN} query.
*/
public static final Threshold inSelectCartesianProduct =
new Threshold(state -> CONFIG_PROVIDER.getOrCreate(state).getInSelectCartesianProductWarnThreshold(),
state -> CONFIG_PROVIDER.getOrCreate(state).getInSelectCartesianProductFailThreshold(),
(isWarning, what, value, threshold) ->
isWarning ? format("The cartesian product of the IN restrictions on %s produces %d values, " +
"this exceeds warning threshold of %s.",
what, value, threshold)
: format("Aborting query because the cartesian product of the IN restrictions on %s " +
"produces %d values, this exceeds fail threshold of %s.",
what, value, threshold));
private Guardrails()
{
MBeanWrapper.instance.registerMBean(this, MBEAN_NAME);
@ -383,6 +397,24 @@ public final class Guardrails implements GuardrailsMBean
return DEFAULT_CONFIG.getPartitionKeysInSelectFailThreshold();
}
@Override
public int getInSelectCartesianProductWarnThreshold()
{
return DEFAULT_CONFIG.getInSelectCartesianProductWarnThreshold();
}
@Override
public int getInSelectCartesianProductFailThreshold()
{
return DEFAULT_CONFIG.getInSelectCartesianProductFailThreshold();
}
@Override
public void setInSelectCartesianProductThreshold(int warn, int fail)
{
DEFAULT_CONFIG.setInSelectCartesianProductThreshold(warn, fail);
}
private static String toCSV(Set<String> values)
{
return values == null ? "" : String.join(",", values);

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

@ -144,4 +144,16 @@ public interface GuardrailsConfig
* @return {@code true} if list operations that require read before write are allowed, {@code false} otherwise.
*/
boolean getReadBeforeWriteListOperationsEnabled();
/**
* @return The threshold to warn when an IN query creates a cartesian product with a size exceeding threshold.
* -1 means disabled.
*/
public int getInSelectCartesianProductWarnThreshold();
/**
* @return The threshold to prevent IN queries creating a cartesian product with a size exceeding threshold.
* -1 means disabled.
*/
public int getInSelectCartesianProductFailThreshold();
}

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

@ -221,15 +221,6 @@ public interface GuardrailsMBean
*/
void setReadBeforeWriteListOperationsEnabled(boolean enabled);
/**
* @param warn The threshold to warn when the number of partition keys in a select statement is greater than
* threshold -1 means disabled.
* @param fail The threshold to prevent when the number of partition keys in a select statement is more than
* threshold -1 means disabled.
*/
void setPartitionKeysInSelectThreshold(int warn, int fail);
/**
* @return The threshold to warn when the number of partition keys in a select statement greater than threshold.
* -1 means disabled.
@ -241,4 +232,32 @@ public interface GuardrailsMBean
* -1 means disabled.
*/
int getPartitionKeysInSelectFailThreshold();
/**
* @param warn The threshold to warn when the number of partition keys in a select statement is greater than
* threshold -1 means disabled.
* @param fail The threshold to prevent when the number of partition keys in a select statement is more than
* threshold -1 means disabled.
*/
void setPartitionKeysInSelectThreshold(int warn, int fail);
/**
* @return The threshold to warn when an IN query creates a cartesian product with a size exceeding threshold.
* -1 means disabled.
*/
public int getInSelectCartesianProductWarnThreshold();
/**
* @return The threshold to prevent IN queries creating a cartesian product with a size exceeding threshold.
* -1 means disabled.
*/
public int getInSelectCartesianProductFailThreshold();
/**
* @param warn The threshold to warn when an IN query creates a cartesian product with a size exceeding threshold.
* -1 means disabled.
* @param fail The threshold to prevent IN queries creating a cartesian product with a size exceeding threshold.
* -1 means disabled.
*/
public void setInSelectCartesianProductThreshold(int warn, int fail);
}

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

@ -239,8 +239,9 @@ public class CQLSSTableWriter implements Closeable
throw new InvalidRequestException(String.format("Invalid number of arguments, expecting %d values but got %d", boundNames.size(), values.size()));
QueryOptions options = QueryOptions.forInternalCalls(null, values);
List<ByteBuffer> keys = insert.buildPartitionKeyNames(options);
SortedSet<Clustering<?>> clusterings = insert.createClustering(options);
ClientState state = ClientState.forInternalCalls();
List<ByteBuffer> keys = insert.buildPartitionKeyNames(options, state);
SortedSet<Clustering<?>> clusterings = insert.createClustering(options, state);
long now = currentTimeMillis();
// Note that we asks indexes to not validate values (the last 'false' arg below) because that triggers a 'Keyspace.open'

14
test/unit/org/apache/cassandra/db/guardrails/GuardrailColumnsPerTableTest.java
View File

@ -160,11 +160,12 @@ public class GuardrailColumnsPerTableTest extends ThresholdTester
private void assertWarns(long numColumns, String query, String tableName) throws Throwable
{
assertThresholdWarns(format("The table %s has %s columns, this exceeds the warning threshold of %s.",
assertThresholdWarns(format(query, keyspace() + '.' + tableName),
format("The table %s has %s columns, this exceeds the warning threshold of %s.",
tableName,
numColumns,
guardrails().getColumnsPerTableWarnThreshold()),
format(query, keyspace() + '.' + tableName));
guardrails().getColumnsPerTableWarnThreshold())
);
}
private void assertAddColumnFails(String query) throws Throwable
@ -179,10 +180,11 @@ public class GuardrailColumnsPerTableTest extends ThresholdTester
private void assertFails(long numColumns, String query, String tableName) throws Throwable
{
assertThresholdFails(format("Tables cannot have more than %s columns, but %s provided for table %s",
assertThresholdFails(format(query, keyspace() + '.' + tableName),
format("Tables cannot have more than %s columns, but %s provided for table %s",
guardrails().getColumnsPerTableFailThreshold(),
numColumns,
tableName),
format(query, keyspace() + '.' + tableName));
tableName)
);
}
}

209
test/unit/org/apache/cassandra/db/guardrails/GuardrailInSelectCartesianProductTest.java Normal file
View File

@ -0,0 +1,209 @@
/*
* 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.nio.ByteBuffer;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.junit.Before;
import org.junit.Test;
import org.apache.cassandra.db.marshal.Int32Type;
/**
* Tests the guardrail for the max number of restrictions produced by the cartesian product of the {@code IN}
* restrictions of a query, {@link Guardrails#inSelectCartesianProduct}.
*/
public class GuardrailInSelectCartesianProductTest extends ThresholdTester
{
private static final int WARN_THRESHOLD = 16;
private static final int FAIL_THRESHOLD = 25;
private static final String WARN_MESSAGE = "The cartesian product of the IN restrictions on %s produces %d " +
"values, this exceeds warning threshold of " + WARN_THRESHOLD;
private static final String FAIL_MESSAGE = "Aborting query because the cartesian product of the IN restrictions " +
"on %s produces %d values, this exceeds fail threshold of " + FAIL_THRESHOLD;
public GuardrailInSelectCartesianProductTest()
{
super(WARN_THRESHOLD,
FAIL_THRESHOLD,
"in_select_cartesian_product",
Guardrails::setInSelectCartesianProductThreshold,
Guardrails::getInSelectCartesianProductWarnThreshold,
Guardrails::getInSelectCartesianProductFailThreshold);
}
@Override
protected long currentValue()
{
throw new UnsupportedOperationException();
}
@Before
public void initSchema()
{
createTable("CREATE TABLE %s (pk1 int, pk2 int, ck1 int, ck2 int, PRIMARY KEY((pk1, pk2), ck1, ck2))");
}
@Test
public void testPkCartesianProduct() throws Throwable
{
// below both thresholds
testPkCartesianProduct(1, 1);
testPkCartesianProduct(1, 4);
testPkCartesianProduct(4, 4);
// above warn threshold
testPkCartesianProduct(5, 5);
testPkCartesianProduct(2, 12);
testPkCartesianProduct(8, 3);
// above cartesian product limit
testPkCartesianProduct(1, 26);
testPkCartesianProduct(5, 6);
testPkCartesianProduct(26, 1);
}
@Test
public void testCkCartesianProduct() throws Throwable
{
// below both thresholds
testCkCartesianProduct(3, 8);
testCkCartesianProduct(5, 5);
// above cartesian product limit
testCkCartesianProduct(1, 26);
testCkCartesianProduct(5, 6);
testCkCartesianProduct(6, 5);
testCkCartesianProduct(26, 1);
}
@Test
public void testPkCkCartesianProduct() throws Throwable
{
// below both thresholds
testCartesianProduct(1, 10, 1, 10);
testCartesianProduct(10, 1, 10, 1);
testCartesianProduct(5, 5, 5, 5);
// above cartesian product limit
testCartesianProduct(5, 6, 5, 5);
testCartesianProduct(6, 5, 5, 5);
testCartesianProduct(5, 5, 6, 5);
testCartesianProduct(5, 5, 5, 6);
}
@Test
public void testExcludedUsers() throws Throwable
{
testExcludedUsers(() -> String.format("SELECT * FROM %%s WHERE pk1 in (%s) AND pk2 in (%s)",
terms(5), terms(5)),
() -> String.format("SELECT * FROM %%s WHERE pk1 in (%s) AND pk2 in (%s) AND ck1 in (%s) AND ck2 in (%s)",
terms(5), terms(5), terms(5), terms(6)));
}
@Test
public void testPkCartesianProductMultiColumnBelowThreshold() throws Throwable
{
String inTerms = IntStream.range(0, 5).mapToObj(i -> String.format("(%d, %d)", i, i + 1)).collect(Collectors.joining(", "));
String query = String.format("SELECT * FROM %%s WHERE (pk1, pk2) in (%s)", inTerms);
assertInvalidMessage("Multi-column relations can only be applied to clustering columns but was applied to: pk1", query);
}
private void testPkCartesianProduct(int pk1Terms, int pk2Terms) throws Throwable
{
testCartesianProduct(pk1Terms, pk2Terms, 1, 1);
}
private void testCkCartesianProduct(int ck1Terms, int ck2Terms) throws Throwable
{
testCartesianProduct(1, 1, ck1Terms, ck2Terms);
}
private void testCartesianProduct(int pk1, int pk2, int ck1, int ck2) throws Throwable
{
int keys = pk1 * pk2;
int clusterings = ck1 * ck2;
String query = String.format("SELECT * FROM %%s WHERE pk1 in (%s) AND pk2 in (%s) AND ck1 in (%s) AND ck2 in (%s)",
terms(pk1), terms(pk2), terms(ck1), terms(ck2));
testCartesianProduct(() -> execute(userClientState, query), keys, clusterings);
String queryWithBindVariables = String.format("SELECT * FROM %%s WHERE pk1 in (%s) AND pk2 in (%s) AND ck1 in (%s) AND ck2 in (%s)",
markers(pk1), markers(pk2), markers(ck1), markers(ck2));
testCartesianProduct(() -> execute(userClientState, queryWithBindVariables, bindValues(pk1, pk2, ck1, ck2)), keys, clusterings);
}
private void testCartesianProduct(CheckedFunction function, int keys, int clusterings) throws Throwable
{
String keysFailMessage = String.format(FAIL_MESSAGE, "partition key", keys);
String keysWarnMessage = String.format(WARN_MESSAGE, "partition key", keys);
String clusteringsFailMessage = String.format(FAIL_MESSAGE, "clustering key", clusterings);
String clusteringsWarnMessage = String.format(WARN_MESSAGE, "clustering key", clusterings);
if (keys > FAIL_THRESHOLD)
{
assertFails(function, keysFailMessage);
}
else if (keys > WARN_THRESHOLD)
{
if (clusterings > FAIL_THRESHOLD)
assertFails(function, keysWarnMessage, clusteringsFailMessage);
else if (clusterings > WARN_THRESHOLD)
assertWarns(function, keysWarnMessage, clusteringsWarnMessage);
else
assertWarns(function, keysWarnMessage);
}
else if (clusterings > FAIL_THRESHOLD)
{
assertFails(function, clusteringsFailMessage);
}
else if (clusterings > WARN_THRESHOLD)
{
assertWarns(function, clusteringsWarnMessage);
}
else
{
assertValid(function);
}
}
private static String terms(int terms)
{
assert terms > 0;
return IntStream.range(0, terms).mapToObj(String::valueOf).collect(Collectors.joining(", "));
}
private static String markers(int terms)
{
assert terms > 0;
return IntStream.range(0, terms).mapToObj(i -> "?").collect(Collectors.joining(", "));
}
private static List<ByteBuffer> bindValues(int... termCounts)
{
return IntStream.of(termCounts)
.boxed()
.flatMap(terms -> IntStream.range(0, terms).boxed().map(Int32Type.instance::decompose))
.collect(Collectors.toList());
}
}

14
test/unit/org/apache/cassandra/db/guardrails/GuardrailKeyspacesTest.java
View File

@ -89,18 +89,20 @@ public class GuardrailKeyspacesTest extends ThresholdTester
private String assertCreateKeyspaceWarns() throws Throwable
{
String keyspaceName = createKeyspaceName();
assertThresholdWarns(format("Creating keyspace %s, current number of keyspaces %d exceeds warning threshold of %d",
keyspaceName, currentValue() + 1, WARN_THRESHOLD),
createKeyspaceQuery(keyspaceName));
assertThresholdWarns(createKeyspaceQuery(keyspaceName),
format("Creating keyspace %s, current number of keyspaces %d exceeds warning threshold of %d",
keyspaceName, currentValue() + 1, WARN_THRESHOLD)
);
return keyspaceName;
}
private void assertCreateKeyspaceFails() throws Throwable
{
String keyspaceName = createKeyspaceName();
assertThresholdFails(format("Cannot have more than %d keyspaces, aborting the creation of keyspace %s",
FAIL_THRESHOLD, keyspaceName),
createKeyspaceQuery(keyspaceName));
assertThresholdFails(createKeyspaceQuery(keyspaceName),
format("Cannot have more than %d keyspaces, aborting the creation of keyspace %s",
FAIL_THRESHOLD, keyspaceName)
);
}
private String createKeyspaceQuery()

13
test/unit/org/apache/cassandra/db/guardrails/GuardrailPartitionKeysInSelectTest.java
View File

@ -52,14 +52,13 @@ public class GuardrailPartitionKeysInSelectTest extends ThresholdTester
assertValid("SELECT k, c, v FROM %s WHERE k = 2 and c IN (2, 3, 4, 5, 6, 7)");
assertWarns(String.format("Query with partition keys in IN clause on table %s, with " +
"number of partition keys 4 exceeds warning threshold of 3.", tableName),
"SELECT k, c, v FROM %s WHERE k IN (2, 3, 4, 5)");
assertWarns("SELECT k, c, v FROM %s WHERE k IN (2, 3, 4, 5)",
String.format("Query with partition keys in IN clause on table %s, with " +
"number of partition keys 4 exceeds warning threshold of 3.", tableName));
assertFails(String.format("Aborting query with partition keys in IN clause on table %s, " +
"number of partition keys 6 exceeds fail threshold of 5.", tableName) ,
"SELECT k, c, v FROM %s WHERE k IN (2, 3, 4, 5, 6, 7)"
);
assertFails("SELECT k, c, v FROM %s WHERE k IN (2, 3, 4, 5, 6, 7)",
String.format("Aborting query with partition keys in IN clause on table %s, " +
"number of partition keys 6 exceeds fail threshold of 5.", tableName));
}
@Test

2
test/unit/org/apache/cassandra/db/guardrails/GuardrailReadBeforeWriteListOperationsTest.java
View File

@ -166,7 +166,7 @@ public class GuardrailReadBeforeWriteListOperationsTest extends GuardrailTester
}
else
{
assertFails(expectedMessage, query);
assertFails(query, expectedMessage);
}
}

19
test/unit/org/apache/cassandra/db/guardrails/GuardrailSecondaryIndexesPerTable.java
View File

@ -97,24 +97,27 @@ public class GuardrailSecondaryIndexesPerTable extends ThresholdTester
private void assertCreateIndexWarns(String column, String indexName) throws Throwable
{
assertThresholdWarns(format("Creating secondary index %son table %s, current number of indexes %s exceeds warning threshold of %s.",
assertThresholdWarns(format("CREATE INDEX %s ON %%s(%s)", indexName, column),
format("Creating secondary index %son table %s, current number of indexes %s exceeds warning threshold of %s.",
(Strings.isNullOrEmpty(indexName) ? "" : indexName + " "),
currentTable(),
currentValue() + 1,
guardrails().getSecondaryIndexesPerTableWarnThreshold()),
format("CREATE INDEX %s ON %%s(%s)", indexName, column));
guardrails().getSecondaryIndexesPerTableWarnThreshold())
);
}
private void assertCreateIndexFails(String column, String indexName) throws Throwable
{
assertThresholdFails(format("aborting the creation of secondary index %son table %s",
Strings.isNullOrEmpty(indexName) ? "" : indexName + " ", currentTable()),
format("CREATE INDEX %s ON %%s(%s)", indexName, column));
assertThresholdFails(format("CREATE INDEX %s ON %%s(%s)", indexName, column),
format("aborting the creation of secondary index %son table %s",
Strings.isNullOrEmpty(indexName) ? "" : indexName + " ", currentTable())
);
}
private void assertCreateCustomIndexFails(String column) throws Throwable
{
assertThresholdFails(format("aborting the creation of secondary index on table %s", currentTable()),
format("CREATE CUSTOM INDEX ON %%s (%s) USING 'org.apache.cassandra.index.sasi.SASIIndex'", column));
assertThresholdFails(format("CREATE CUSTOM INDEX ON %%s (%s) USING 'org.apache.cassandra.index.sasi.SASIIndex'", column),
format("aborting the creation of secondary index on table %s", currentTable())
);
}
}

10
test/unit/org/apache/cassandra/db/guardrails/GuardrailTablesTest.java
View File

@ -90,16 +90,18 @@ public class GuardrailTablesTest extends ThresholdTester
private String assertCreateTableWarns() throws Throwable
{
String tableName = createTableName();
assertThresholdWarns(format("Creating table %s, current number of tables 2 exceeds warning threshold of 1", tableName),
createTableQuery(tableName));
assertThresholdWarns(createTableQuery(tableName),
format("Creating table %s, current number of tables 2 exceeds warning threshold of 1", tableName)
);
return tableName;
}
private void assertCreateTableFails() throws Throwable
{
String tableName = createTableName();
assertThresholdFails(format("Cannot have more than 2 tables, aborting the creation of table %s", tableName),
createTableQuery(tableName));
assertThresholdFails(createTableQuery(tableName),
format("Cannot have more than 2 tables, aborting the creation of table %s", tableName)
);
}
private String createTableQuery()

50
test/unit/org/apache/cassandra/db/guardrails/GuardrailTester.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.db.guardrails;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import java.util.function.BiConsumer;
@ -163,7 +164,7 @@ public abstract class GuardrailTester extends CQLTester
assertValid(() -> execute(userClientState, query));
}
protected void assertWarns(CheckedFunction function, String message) throws Throwable
protected void assertWarns(CheckedFunction function, String... messages) throws Throwable
{
// We use client warnings to check we properly warn as this is the most convenient. Technically,
// this doesn't validate we also log the warning, but that's probably fine ...
@ -171,7 +172,7 @@ public abstract class GuardrailTester extends CQLTester
try
{
function.apply();
assertWarnings(message);
assertWarnings(messages);
}
finally
{
@ -179,17 +180,17 @@ public abstract class GuardrailTester extends CQLTester
}
}
protected void assertWarns(String message, String query) throws Throwable
protected void assertWarns(String query, String... messages) throws Throwable
{
assertWarns(() -> execute(userClientState, query), message);
assertWarns(() -> execute(userClientState, query), messages);
}
protected void assertFails(CheckedFunction function, String message) throws Throwable
protected void assertFails(CheckedFunction function, String... messages) throws Throwable
{
assertFails(function, message, true);
assertFails(function, true, messages);
}
protected void assertFails(CheckedFunction function, String message, boolean thrown) throws Throwable
protected void assertFails(CheckedFunction function, boolean thrown, String... messages) throws Throwable
{
ClientWarn.instance.captureWarnings();
try
@ -203,10 +204,12 @@ public abstract class GuardrailTester extends CQLTester
{
assertTrue("Expect no exception thrown", thrown);
assertTrue(format("Full error message '%s' does not contain expected message '%s'", e.getMessage(), message),
e.getMessage().contains(message));
// the last message is the one raising the guardrail failure, the previous messages are warnings
String failMessage = messages[messages.length - 1];
assertTrue(format("Full error message '%s' does not contain expected message '%s'", e.getMessage(), failMessage),
e.getMessage().contains(failMessage));
assertWarnings(message);
assertWarnings(messages);
}
finally
{
@ -214,23 +217,27 @@ public abstract class GuardrailTester extends CQLTester
}
}
protected void assertFails(String message, String query) throws Throwable
protected void assertFails(String query, String... messages) throws Throwable
{
assertFails(() -> execute(userClientState, query), message);
assertFails(() -> execute(userClientState, query), messages);
}
private void assertWarnings(String message)
private void assertWarnings(String... messages)
{
List<String> warnings = getWarnings();
assertFalse("Expected to warn, but no warning was received", warnings == null || warnings.isEmpty());
assertEquals(format("Got more thant 1 warning (got %d => %s)", warnings.size(), warnings),
1,
assertEquals(format("Expected %d warnings but got %d: %s", messages.length, warnings.size(), warnings),
messages.length,
warnings.size());
String warning = warnings.get(0);
assertTrue(format("Warning log message '%s' does not contain expected message '%s'", warning, message),
warning.contains(message));
for (int i = 0; i < messages.length; i++)
{
String message = messages[i];
String warning = warnings.get(i);
assertTrue(format("Warning log message '%s' does not contain expected message '%s'", warning, message),
warning.contains(message));
}
}
private void assertEmptyWarnings()
@ -270,6 +277,11 @@ public abstract class GuardrailTester extends CQLTester
}
protected ResultMessage execute(ClientState state, String query)
{
return execute(state, query, Collections.emptyList());
}
protected ResultMessage execute(ClientState state, String query, List<ByteBuffer> values)
{
QueryState queryState = new QueryState(state);
@ -277,7 +289,7 @@ public abstract class GuardrailTester extends CQLTester
CQLStatement statement = QueryProcessor.parseStatement(formattedQuery, queryState.getClientState());
statement.validate(state);
QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList());
QueryOptions options = QueryOptions.forInternalCalls(values);
return statement.executeLocally(queryState, options);
}

2
test/unit/org/apache/cassandra/db/guardrails/GuardrailUserTimestampsTest.java
View File

@ -119,6 +119,6 @@ public class GuardrailUserTimestampsTest extends GuardrailTester
private void assertFails(String query) throws Throwable
{
assertFails("User provided timestamps (USING TIMESTAMP) is not allowed", query);
assertFails(query, "User provided timestamps (USING TIMESTAMP) is not allowed");
}
}

14
test/unit/org/apache/cassandra/db/guardrails/GuardrailViewsPerTableTest.java
View File

@ -108,16 +108,18 @@ public class GuardrailViewsPerTableTest extends ThresholdTester
private void assertCreateViewWarns() throws Throwable
{
String viewName = createViewName();
assertThresholdWarns(format("Creating materialized view %s on table %s, current number of views %s exceeds warning threshold of %s.",
viewName, currentTable(), currentValue() + 1, guardrails().getMaterializedViewsPerTableWarnThreshold()),
format(CREATE_VIEW, viewName));
assertThresholdWarns(format(CREATE_VIEW, viewName),
format("Creating materialized view %s on table %s, current number of views %s exceeds warning threshold of %s.",
viewName, currentTable(), currentValue() + 1, guardrails().getMaterializedViewsPerTableWarnThreshold())
);
}
private void assertCreateViewFails() throws Throwable
{
String viewName = createViewName();
assertThresholdFails(format("aborting the creation of materialized view %s on table %s",
viewName, currentTable()),
format(CREATE_VIEW, viewName));
assertThresholdFails(format(CREATE_VIEW, viewName),
format("aborting the creation of materialized view %s on table %s",
viewName, currentTable())
);
}
}

8
test/unit/org/apache/cassandra/db/guardrails/ThresholdTester.java
View File

@ -116,18 +116,18 @@ public abstract class ThresholdTester extends GuardrailTester
.isLessThanOrEqualTo(failGetter.applyAsLong(guardrails()));
}
protected void assertThresholdWarns(String message, String query) throws Throwable
protected void assertThresholdWarns(String query, String... messages) throws Throwable
{
assertWarns(message, query);
assertWarns(query, messages);
Assertions.assertThat(currentValue())
.isGreaterThan(warnGetter.applyAsLong(guardrails()))
.isLessThanOrEqualTo(failGetter.applyAsLong(guardrails()));
}
protected void assertThresholdFails(String message, String query) throws Throwable
protected void assertThresholdFails(String query, String... messages) throws Throwable
{
assertFails(message, query);
assertFails(query, messages);
Assertions.assertThat(currentValue())
.isGreaterThanOrEqualTo(warnGetter.applyAsLong(guardrails()))

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

@ -244,8 +244,9 @@ public class StressCQLSSTableWriter implements Closeable
throw new InvalidRequestException(String.format("Invalid number of arguments, expecting %d values but got %d", boundNames.size(), values.size()));
QueryOptions options = QueryOptions.forInternalCalls(null, values);
List<ByteBuffer> keys = insert.buildPartitionKeyNames(options);
SortedSet<Clustering<?>> clusterings = insert.createClustering(options);
ClientState state = ClientState.forInternalCalls();
List<ByteBuffer> keys = insert.buildPartitionKeyNames(options, state);
SortedSet<Clustering<?>> clusterings = insert.createClustering(options, state);
long now = currentTimeMillis();
// Note that we asks indexes to not validate values (the last 'false' arg below) because that triggers a 'Keyspace.open'