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

Avoid possible consistency violations for SAI intersection queries over partially updated rows at consistency levels that require reconciliation 

- Replica filtring protection now correctly accounts for short reads caused by coordinator filtering of merged rows.
- There is a new flag in ReadCommand that indicates whether the user query it belongs to requires reconciliation.
- Local SAI queries now degrade intersections to unions on unrepaired data if reconciliation is required and multiple mutable columns are restricted.

patch by Caleb Rackliffe; reviewed by Andres de la Peña and Alex Petrov for CASSANDRA-19018

Co-authored-by: Caleb Rackliffe <calebrackliffe@gmail.com>
Co-authored-by: Alex Petrov <oleksandr.petrov@gmail.com>
This commit is contained in:
Caleb Rackliffe 2023-11-27 18:35:26 -06:00
parent 54cf65476a
commit f7984627e7
36 changed files with 1409 additions and 623 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0-beta2
* Avoid consistency violations for SAI intersections over unrepaired data at consistency levels requiring reconciliation (CASSANDRA-19018)
* Fix NullPointerException in ANN+WHERE when adding rows in another partition (CASSANDRA-19404)
* Record latencies for SAI post-filtering reads against local storage (CASSANDRA-18940)
* Fix VectorMemoryIndex#update logic to compare vectors. Fix Index view (CASSANDRA-19168)

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

@ -779,7 +779,11 @@ public final class StatementRestrictions
if (filterRestrictions.isEmpty())
return RowFilter.none();
RowFilter filter = RowFilter.create();
// If there is only one replica, we don't need reconciliation at any consistency level.
boolean needsReconciliation = options.getConsistency().needsReconciliation()
&& Keyspace.open(table.keyspace).getReplicationStrategy().getReplicationFactor().allReplicas > 1;
RowFilter filter = RowFilter.create(needsReconciliation);
for (Restrictions restrictions : filterRestrictions.getRestrictions())
restrictions.addToRowFilter(filter, indexRegistry, options);

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

@ -97,12 +97,12 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Encapsulates a completely parsed SELECT query, including the target
* column family, expression, result count, and ordering clause.
*
* <p>
* A number of public methods here are only used internally. However,
* many of these are made accessible for the benefit of custom
* QueryHandler implementations, so before reducing their accessibility
* due consideration should be given.
*
* <p>
* Note that select statements can be accessed by multiple threads, so we cannot rely on mutable attributes.
*/
@ThreadSafe
@ -305,9 +305,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
options.getConsistency() == ConsistencyLevel.SERIAL,
String.format(TOPK_CONSISTENCY_LEVEL_ERROR, options.getConsistency()));
if (options.getConsistency() != ConsistencyLevel.ONE &&
options.getConsistency() != ConsistencyLevel.LOCAL_ONE &&
options.getConsistency() != ConsistencyLevel.NODE_LOCAL)
if (options.getConsistency().needsReconciliation())
{
ConsistencyLevel supplied = options.getConsistency();
ConsistencyLevel downgrade = supplied.isDatacenterLocal() ? ConsistencyLevel.LOCAL_ONE : ConsistencyLevel.ONE;
@ -362,7 +360,6 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
return rows;
}
public AggregationSpecification getAggregationSpec(QueryOptions options)
{
return aggregationSpecFactory == null ? null : aggregationSpecFactory.newInstance(options);

2
src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java
View File

@ -43,7 +43,7 @@ public abstract class AbstractReadCommandBuilder
protected boolean reversed = false;
protected Set<ColumnIdentifier> columns;
protected final RowFilter filter = RowFilter.create();
protected final RowFilter filter = RowFilter.create(true);
private ClusteringBound<?> lowerClusteringBound;
private ClusteringBound<?> upperClusteringBound;

15
src/java/org/apache/cassandra/db/ConsistencyLevel.java
View File

@ -65,12 +65,12 @@ public enum ConsistencyLevel
}
}
private ConsistencyLevel(int code)
ConsistencyLevel(int code)
{
this(code, false);
}
private ConsistencyLevel(int code, boolean isDCLocal)
ConsistencyLevel(int code, boolean isDCLocal)
{
this.code = code;
this.isDCLocal = isDCLocal;
@ -258,6 +258,17 @@ public enum ConsistencyLevel
throw new InvalidRequestException("Counter operations are inherently non-serializable");
}
/**
* With a replication factor greater than one, reads that contact more than one replica will require
* reconciliation of the individual replica results at the coordinator.
*
* @return true if reads at this consistency level require merging at the coordinator
*/
public boolean needsReconciliation()
{
return this != ConsistencyLevel.ONE && this != ConsistencyLevel.LOCAL_ONE && this != ConsistencyLevel.NODE_LOCAL;
}
private void requireNetworkTopologyStrategy(AbstractReplicationStrategy replicationStrategy) throws InvalidRequestException
{
if (!(replicationStrategy instanceof NetworkTopologyStrategy))

38
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -1011,6 +1011,12 @@ public abstract class ReadCommand extends AbstractReadQuery
@VisibleForTesting
public static class Serializer implements IVersionedSerializer<ReadCommand>
{
private static final int IS_DIGEST = 0x01;
private static final int IS_FOR_THRIFT = 0x02;
private static final int HAS_INDEX = 0x04;
private static final int ACCEPTS_TRANSIENT = 0x08;
private static final int NEEDS_RECONCILIATION = 0x10;
private final SchemaProvider schema;
public Serializer()
@ -1026,22 +1032,22 @@ public abstract class ReadCommand extends AbstractReadQuery
private static int digestFlag(boolean isDigest)
{
return isDigest ? 0x01 : 0;
return isDigest ? IS_DIGEST : 0;
}
private static boolean isDigest(int flags)
{
return (flags & 0x01) != 0;
return (flags & IS_DIGEST) != 0;
}
private static boolean acceptsTransient(int flags)
{
return (flags & 0x08) != 0;
return (flags & ACCEPTS_TRANSIENT) != 0;
}
private static int acceptsTransientFlag(boolean acceptsTransient)
{
return acceptsTransient ? 0x08 : 0;
return acceptsTransient ? ACCEPTS_TRANSIENT : 0;
}
// We don't set this flag anymore, but still look if we receive a
@ -1051,17 +1057,27 @@ public abstract class ReadCommand extends AbstractReadQuery
// used by these release for thrift and would thus confuse things)
private static boolean isForThrift(int flags)
{
return (flags & 0x02) != 0;
return (flags & IS_FOR_THRIFT) != 0;
}
private static int indexFlag(boolean hasIndex)
{
return hasIndex ? 0x04 : 0;
return hasIndex ? HAS_INDEX : 0;
}
private static boolean hasIndex(int flags)
{
return (flags & 0x04) != 0;
return (flags & HAS_INDEX) != 0;
}
private static int needsReconciliationFlag(boolean needsReconciliation)
{
return needsReconciliation ? NEEDS_RECONCILIATION : 0;
}
private static boolean needsReconciliation(int flags)
{
return (flags & NEEDS_RECONCILIATION) != 0;
}
public void serialize(ReadCommand command, DataOutputPlus out, int version) throws IOException
@ -1071,6 +1087,7 @@ public abstract class ReadCommand extends AbstractReadQuery
digestFlag(command.isDigestQuery())
| indexFlag(null != command.indexQueryPlan())
| acceptsTransientFlag(command.acceptsTransient())
| needsReconciliationFlag(command.rowFilter().needsReconciliation())
);
if (command.isDigestQuery())
out.writeUnsignedVInt32(command.digestVersion());
@ -1104,10 +1121,12 @@ public abstract class ReadCommand extends AbstractReadQuery
boolean hasIndex = hasIndex(flags);
int digestVersion = isDigest ? in.readUnsignedVInt32() : 0;
boolean needsReconciliation = needsReconciliation(flags);
TableMetadata metadata = schema.getExistingTableMetadata(TableId.deserialize(in));
long nowInSec = version >= MessagingService.VERSION_50 ? CassandraUInt.toLong(in.readInt()) : in.readInt();
ColumnFilter columnFilter = ColumnFilter.serializer.deserialize(in, version, metadata);
RowFilter rowFilter = RowFilter.serializer.deserialize(in, version, metadata);
RowFilter rowFilter = RowFilter.serializer.deserialize(in, version, metadata, needsReconciliation);
DataLimits limits = DataLimits.serializer.deserialize(in, version, metadata);
Index.QueryPlan indexQueryPlan = null;
@ -1119,7 +1138,8 @@ public abstract class ReadCommand extends AbstractReadQuery
indexQueryPlan = indexGroup.queryPlanFor(rowFilter);
}
return kind.selectionDeserializer.deserialize(in, version, isDigest, digestVersion, acceptsTransient, metadata, nowInSec, columnFilter, rowFilter, limits, indexQueryPlan);
return kind.selectionDeserializer.deserialize(in, version, isDigest, digestVersion, acceptsTransient,
metadata, nowInSec, columnFilter, rowFilter, limits, indexQueryPlan);
}
private IndexMetadata deserializeIndexMetadata(DataInputPlus in, int version, TableMetadata metadata) throws IOException

29
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -265,31 +265,24 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
return new UnfilteredRowIterators.MergeListener()
{
public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions)
{
}
@Override
public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions) {}
public Row onMergedRows(Row merged, Row[] versions)
@Override
public void onMergedRows(Row merged, Row[] versions)
{
indexTransaction.start();
indexTransaction.onRowMerge(merged, versions);
indexTransaction.commit();
return merged;
}
public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker mergedMarker, RangeTombstoneMarker[] versions)
{
}
@Override
public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker mergedMarker, RangeTombstoneMarker[] versions) {}
public void close()
{
}
@Override
public void close() {}
};
}
public void close()
{
}
};
}
@ -426,8 +419,8 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
DeletionTime openDeletionTime = DeletionTime.LIVE;
DeletionTime partitionDeletionTime;
DeletionTime activeDeletionTime;
Unfiltered tombNext = null;
Unfiltered dataNext = null;
Unfiltered tombNext;
Unfiltered dataNext;
Unfiltered next = null;
/**
@ -530,7 +523,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
tombOpenDeletionTime);
boolean supersededBefore = openDeletionTime.isLive();
boolean supersededAfter = !dataOpenDeletionTime.supersedes(activeDeletionTime);
// If a range open was not issued because it was superseded and the deletion isn't superseded any more, we need to open it now.
// If a range open was not issued because it was superseded and the deletion isn't superseded anymore, we need to open it now.
if (supersededBefore && !supersededAfter)
next = new RangeTombstoneBoundMarker(((RangeTombstoneMarker) tombNext).closeBound(false).invert(), dataOpenDeletionTime);
// If the deletion begins to be superseded, we don't close the range yet. This can save us a close/open pair if it ends after the superseding range.

244
src/java/org/apache/cassandra/db/filter/RowFilter.java
View File

@ -19,12 +19,13 @@ package org.apache.cassandra.db.filter;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import com.google.common.base.Objects;
import org.slf4j.Logger;
@ -32,14 +33,36 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.context.*;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionPurger;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.CollectionType;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.db.marshal.ListType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.db.rows.ComplexColumnData;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.IndexRegistry;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
@ -60,32 +83,37 @@ import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNu
* be handled by a 2ndary index, and the rest is simply filtered out from the
* result set (the later can only happen if the query was using ALLOW FILTERING).
*/
public abstract class RowFilter implements Iterable<RowFilter.Expression>
public class RowFilter implements Iterable<RowFilter.Expression>
{
private static final Logger logger = LoggerFactory.getLogger(RowFilter.class);
public static final Serializer serializer = new Serializer();
private static final RowFilter NONE = new RowFilter(Collections.emptyList(), false);
protected final List<Expression> expressions;
protected RowFilter(List<Expression> expressions)
private final boolean needsReconciliation;
protected RowFilter(List<Expression> expressions, boolean needsReconciliation)
{
this.expressions = expressions;
this.needsReconciliation = needsReconciliation;
}
public static RowFilter create()
/**
*
* @param needsReconciliation whether or not this filter belongs to a read that requires coordinator reconciliation
*
* @return a new {@link RowFilter} with an empty {@link Expression} list
*/
public static RowFilter create(boolean needsReconciliation)
{
return new CQLFilter(new ArrayList<>());
}
public static RowFilter create(int capacity)
{
return new CQLFilter(new ArrayList<>(capacity));
return new RowFilter(new ArrayList<>(), needsReconciliation);
}
public static RowFilter none()
{
return CQLFilter.NONE;
return NONE;
}
public SimpleExpression add(ColumnMetadata def, Operator op, ByteBuffer value)
@ -111,16 +139,35 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
expressions.add(expression);
}
public void addUserExpression(UserExpression e)
{
expressions.add(e);
}
public List<Expression> getExpressions()
{
return expressions;
}
/**
* @return true if this filter belongs to a read that requires reconciliation at the coordinator
* @see StatementRestrictions#getRowFilter(IndexRegistry, QueryOptions)
*/
public boolean needsReconciliation()
{
return needsReconciliation;
}
/**
* If this filter belongs to a read that requires reconciliation at the coordinator, and it contains an intersection
* on two or more non-key (and therefore mutable) columns, we cannot strictly apply it to local, unrepaired rows.
* When this occurs, we must downgrade the intersection of expressions to a union and leave the coordinator to
* filter strictly before sending results to the client.
*
* @return true if strict filtering is safe
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-19018">CASSANDRA-19018</a>
*/
public boolean isStrict()
{
return !needsReconciliation || expressions.stream().filter(e -> !e.column.isPrimaryKeyColumn()).count() <= 1;
}
/**
* Checks if some of the expressions apply to clustering or regular columns.
* @return {@code true} if some of the expressions apply to clustering or regular columns, {@code false} otherwise.
@ -136,7 +183,74 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
return false;
}
protected abstract Transformation<BaseRowIterator<?>> filter(TableMetadata metadata, long nowInSec);
/**
* Note that the application of this transformation does not yet take {@link #isStrict()} into account. This means
* that even when strict filtering is not safe, expressions will be applied as intersections rather than unions.
* The filter will always be evaluated strictly in conjunction with replica filtering protection at the
* coordinator, however, even after CASSANDRA-19007 is addressed.
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-190007">CASSANDRA-19007</a>
*/
protected Transformation<BaseRowIterator<?>> filter(TableMetadata metadata, long nowInSec)
{
List<Expression> partitionLevelExpressions = new ArrayList<>();
List<Expression> rowLevelExpressions = new ArrayList<>();
for (Expression e: expressions)
{
if (e.column.isStatic() || e.column.isPartitionKey())
partitionLevelExpressions.add(e);
else
rowLevelExpressions.add(e);
}
long numberOfRegularColumnExpressions = rowLevelExpressions.size();
final boolean filterNonStaticColumns = numberOfRegularColumnExpressions > 0;
return new Transformation<>()
{
DecoratedKey pk;
@Override
protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
{
pk = partition.partitionKey();
// Short-circuit all partitions that won't match based on static and partition keys
for (Expression e : partitionLevelExpressions)
if (!e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow()))
{
partition.close();
return null;
}
BaseRowIterator<?> iterator = partition instanceof UnfilteredRowIterator
? Transformation.apply((UnfilteredRowIterator) partition, this)
: Transformation.apply((RowIterator) partition, this);
if (filterNonStaticColumns && !iterator.hasNext())
{
iterator.close();
return null;
}
return iterator;
}
@Override
public Row applyToRow(Row row)
{
Row purged = row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
if (purged == null)
return null;
for (Expression e : rowLevelExpressions)
if (!e.isSatisfiedBy(metadata, pk, purged))
return null;
return row;
}
};
}
/**
* Filters the provided iterator so that only the row satisfying the expression of this filter
@ -267,13 +381,11 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
return withNewExpressions(Collections.emptyList());
}
public RowFilter restrict(Predicate<Expression> filter)
protected RowFilter withNewExpressions(List<Expression> expressions)
{
return withNewExpressions(expressions.stream().filter(filter).collect(Collectors.toList()));
return new RowFilter(expressions, needsReconciliation);
}
protected abstract RowFilter withNewExpressions(List<Expression> expressions);
public boolean isEmpty()
{
return expressions.isEmpty();
@ -312,80 +424,6 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
return sb.toString();
}
private static class CQLFilter extends RowFilter
{
static CQLFilter NONE = new CQLFilter(Collections.emptyList());
private CQLFilter(List<Expression> expressions)
{
super(expressions);
}
protected Transformation<BaseRowIterator<?>> filter(TableMetadata metadata, long nowInSec)
{
List<Expression> partitionLevelExpressions = new ArrayList<>();
List<Expression> rowLevelExpressions = new ArrayList<>();
for (Expression e: expressions)
{
if (e.column.isStatic() || e.column.isPartitionKey())
partitionLevelExpressions.add(e);
else
rowLevelExpressions.add(e);
}
long numberOfRegularColumnExpressions = rowLevelExpressions.size();
final boolean filterNonStaticColumns = numberOfRegularColumnExpressions > 0;
return new Transformation<BaseRowIterator<?>>()
{
DecoratedKey pk;
protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
{
pk = partition.partitionKey();
// Short-circuit all partitions that won't match based on static and partition keys
for (Expression e : partitionLevelExpressions)
if (!e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow()))
{
partition.close();
return null;
}
BaseRowIterator<?> iterator = partition instanceof UnfilteredRowIterator
? Transformation.apply((UnfilteredRowIterator) partition, this)
: Transformation.apply((RowIterator) partition, this);
if (filterNonStaticColumns && !iterator.hasNext())
{
iterator.close();
return null;
}
return iterator;
}
public Row applyToRow(Row row)
{
Row purged = row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
if (purged == null)
return null;
for (Expression e : rowLevelExpressions)
if (!e.isSatisfiedBy(metadata, pk, purged))
return null;
return row;
}
};
}
protected RowFilter withNewExpressions(List<Expression> expressions)
{
return new CQLFilter(expressions);
}
}
public static abstract class Expression
{
private static final Serializer serializer = new Serializer();
@ -1058,7 +1096,7 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
}
public RowFilter deserialize(DataInputPlus in, int version, TableMetadata metadata) throws IOException
public RowFilter deserialize(DataInputPlus in, int version, TableMetadata metadata, boolean needsReconciliation) throws IOException
{
in.readBoolean(); // Unused
int size = in.readUnsignedVInt32();
@ -1066,7 +1104,7 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
for (int i = 0; i < size; i++)
expressions.add(Expression.serializer.deserialize(in, version, metadata));
return new CQLFilter(expressions);
return new RowFilter(expressions, needsReconciliation);
}
public long serializedSize(RowFilter filter, int version)

33
src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java
View File

@ -66,7 +66,7 @@ public abstract class UnfilteredRowIterators
* @param versions the partition level deletion for the sources of the merge. Elements of the array will never
* be null, but be "live".
**/
public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions);
void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions);
/**
* Called once for every row participating in the merge.
@ -76,16 +76,12 @@ public abstract class UnfilteredRowIterators
* that is shadowed by another source range tombstone or partition level deletion).
*
* @param merged the result of the merge. This cannot be {@code null} (so that listener can always access the
* clustering from this safely)but can be empty, in which case this is a placeholder for when at least one
* clustering from this safely) but can be empty, in which case this is a placeholder for when at least one
* source has a row, but that row is shadowed in the merged output.
* @param versions for each source, the row in that source corresponding to {@code merged}. This can be
* {@code null} for some sources if the source has not such row.
* @return the row to use as result of the merge (can be {@code null}). Most implementations should simply
* return {@code merged}, but this allows some implementations to impact the merge result if necessary. If this
* returns either {@code null} or an empty row, then the row is skipped from the merge result. If this returns a
* non {@code null} result, then the returned row <b>must</b> have the same clustering than {@code merged}.
*/
public Row onMergedRows(Row merged, Row[] versions);
void onMergedRows(Row merged, Row[] versions);
/**
* Called once for every range tombstone marker participating in the merge.
@ -108,7 +104,7 @@ public abstract class UnfilteredRowIterators
{
public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions) {}
public Row onMergedRows(Row merged, Row[] versions) {return merged;}
public void onMergedRows(Row merged, Row[] versions) {}
public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker merged, RangeTombstoneMarker[] versions) {}
@ -503,12 +499,9 @@ public abstract class UnfilteredRowIterators
Row merged = merger.merge(partitionDeletion);
if (merged == null)
merged = Rows.EMPTY_STATIC_ROW;
if (listener == null)
return merged;
merged = listener.onMergedRows(merged, merger.mergedRows());
// Note that onMergedRows can have returned null even though his input wasn't null
return merged == null ? Rows.EMPTY_STATIC_ROW : merged;
if (listener != null)
listener.onMergedRows(merged, merger.mergedRows());
return merged;
}
private static RegularAndStaticColumns collectColumns(List<UnfilteredRowIterator> iterators)
@ -584,15 +577,9 @@ public abstract class UnfilteredRowIterators
if (nextKind == Unfiltered.Kind.ROW)
{
Row merged = rowMerger.merge(markerMerger.activeDeletion());
if (listener == null)
return merged;
merged = listener.onMergedRows(merged == null
? BTreeRow.emptyRow(rowMerger.mergedClustering())
: merged,
rowMerger.mergedRows());
return merged == null || merged.isEmpty() ? null : merged;
if (listener != null)
listener.onMergedRows(merged == null ? BTreeRow.emptyRow(rowMerger.mergedClustering()) : merged, rowMerger.mergedRows());
return merged;
}
else
{

14
src/java/org/apache/cassandra/index/sai/QueryContext.java
View File

@ -18,18 +18,21 @@
package org.apache.cassandra.index.sai;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.exceptions.QueryCancelledException;
import org.apache.cassandra.index.sai.plan.FilterTree;
import org.apache.cassandra.index.sai.plan.QueryController;
import org.apache.cassandra.utils.Clock;
import static org.apache.cassandra.config.CassandraRelevantProperties.SAI_TEST_DISABLE_TIMEOUT;
/**
* Tracks state relevant to the execution of a single query, including metrics and timeout monitoring.
*
* <p>
* Fields here are non-volatile, as they are accessed from a single thread.
*/
@NotThreadSafe
@ -58,6 +61,15 @@ public class QueryContext
public boolean queryTimedOut = false;
/**
* {@code true} if the local query for this context has matches from Memtable-attached indexes or indexes on
* unrepaired SSTables, and {@code false} otherwise. When this is {@code false}, {@link FilterTree} can ignore the
* coordinator suggestion to downgrade to non-strict filtering, potentially reducing the number of false positives.
*
* @see QueryController#getIndexQueryResults(Collection)
* */
public boolean hasUnrepairedMatches = false;
private VectorQueryContext vectorContext;
public QueryContext(ReadCommand readCommand, long executionQuotaMs)

17
src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java
View File

@ -33,8 +33,6 @@ import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture; //checkstyle: permit this import
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.BooleanSupplier;
import java.util.stream.Collectors;
@ -44,8 +42,8 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.Futures; //checkstyle: permit this import
import com.google.common.util.concurrent.ListenableFuture; //checkstyle: permit this import
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.FutureCombiner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -109,6 +107,7 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.apache.cassandra.utils.concurrent.OpOrder;
import static org.apache.cassandra.config.CassandraRelevantProperties.SAI_MAX_FROZEN_TERM_SIZE;
@ -826,7 +825,7 @@ public class StorageAttachedIndex implements Index
{
logger.debug(indexIdentifier.logMessage("Skipping validation and building in initialization task, as pre-join has already made the storage-attached index queryable..."));
initBuildStarted = true;
return CompletableFuture.completedFuture(null);
return ImmediateFuture.success(null);
}
// stop in-progress compaction tasks to prevent compacted sstable not being indexed.
@ -852,13 +851,11 @@ public class StorageAttachedIndex implements Index
List<SSTableReader> nonIndexed = findNonIndexedSSTables(baseCfs, indexGroup, validation);
if (nonIndexed.isEmpty())
{
return CompletableFuture.completedFuture(null);
}
return ImmediateFuture.success(null);
// split sorted sstables into groups with similar size and build each group in separate compaction thread
List<List<SSTableReader>> groups = groupBySize(nonIndexed, DatabaseDescriptor.getConcurrentIndexBuilders());
List<ListenableFuture<?>> futures = new ArrayList<>();
List<Future<?>> futures = new ArrayList<>();
for (List<SSTableReader> group : groups)
{
@ -869,7 +866,7 @@ public class StorageAttachedIndex implements Index
}
logger.info(indexIdentifier.logMessage("Submitting {} parallel initial index builds over {} total sstables..."), futures.size(), nonIndexed.size());
return Futures.allAsList(futures);
return FutureCombiner.allOf(futures);
}
private Future<?> startPreJoinTask()

14
src/java/org/apache/cassandra/index/sai/disk/IndexSearchResultIterator.java
View File

@ -61,13 +61,17 @@ public class IndexSearchResultIterator extends KeyRangeIterator
public static IndexSearchResultIterator build(Expression expression,
Collection<SSTableIndex> sstableIndexes,
AbstractBounds<PartitionPosition> keyRange,
QueryContext queryContext)
QueryContext queryContext,
boolean includeMemtables)
{
List<KeyRangeIterator> subIterators = new ArrayList<>(1 + sstableIndexes.size());
List<KeyRangeIterator> subIterators = new ArrayList<>(sstableIndexes.size() + (includeMemtables ? 1 : 0));
KeyRangeIterator memtableIterator = expression.getIndex().memtableIndexManager().searchMemtableIndexes(queryContext, expression, keyRange);
if (memtableIterator != null)
subIterators.add(memtableIterator);
if (includeMemtables)
{
KeyRangeIterator memtableIterator = expression.getIndex().memtableIndexManager().searchMemtableIndexes(queryContext, expression, keyRange);
if (memtableIterator != null)
subIterators.add(memtableIterator);
}
for (SSTableIndex sstableIndex : sstableIndexes)
{

4
src/java/org/apache/cassandra/index/sai/disk/PerColumnIndexWriter.java
View File

@ -31,6 +31,8 @@ public interface PerColumnIndexWriter
{
/**
* Adds a row to this index.
* <p>
* Note: Callers are responsible for ensuring that rows are added in {@link PrimaryKey} order.
*/
void addRow(PrimaryKey key, Row row, long sstableRowId) throws IOException;
@ -41,7 +43,7 @@ public interface PerColumnIndexWriter
/**
* Aborts accumulating data. Allows to clean up resources on error.
*
* <p>
* Note: Implementations should be idempotent, i.e. safe to call multiple times without producing undesirable side-effects.
*/
void abort(Throwable cause);

39
src/java/org/apache/cassandra/index/sai/disk/RowMapping.java
View File

@ -76,17 +76,6 @@ public class RowMapping
private boolean complete = false;
public PrimaryKey minKey;
public PrimaryKey maxKey;
public PrimaryKey minStaticKey;
public PrimaryKey maxStaticKey;
public long maxSSTableRowId = -1;
public long maxStaticSSTableRowId = -1;
public int rowCount;
public int staticRowCount;
private RowMapping()
{}
@ -165,29 +154,8 @@ public class RowMapping
*/
public void add(PrimaryKey key, long sstableRowId) throws InMemoryTrie.SpaceExhaustedException
{
assert !complete : "Cannot modify built RowMapping.";
assert !complete : "Cannot modify and already built RowMapping.";
rowMapping.putSingleton(key, sstableRowId, OVERWRITE_TRANSFORMER);
// Data is written in primary key order. If a schema contains clustering keys, it may also contain static
// columns. We track min, max, and count for static keys separately here so that we can pass them to the segment
// metadata for indexes on static columns.
if (key.kind() == PrimaryKey.Kind.STATIC)
{
if (minStaticKey == null)
minStaticKey = key;
maxStaticKey = key;
staticRowCount++;
maxStaticSSTableRowId = Math.max(maxStaticSSTableRowId, sstableRowId);
}
else
{
if (minKey == null)
minKey = key;
maxKey = key;
rowCount++;
maxSSTableRowId = Math.max(maxSSTableRowId, sstableRowId);
}
}
/**
@ -202,9 +170,4 @@ public class RowMapping
Long sstableRowId = rowMapping.get(key);
return sstableRowId == null ? -1 : Math.toIntExact(sstableRowId);
}
public boolean hasRows()
{
return maxSSTableRowId >= 0 || maxStaticSSTableRowId >= 0;
}
}

38
src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java
View File

@ -53,6 +53,7 @@ import org.apache.cassandra.utils.bytecomparable.ByteComparable;
public class MemtableIndexWriter implements PerColumnIndexWriter
{
private static final Logger logger = LoggerFactory.getLogger(MemtableIndexWriter.class);
private static final int NO_ROWS = -1;
private final IndexDescriptor indexDescriptor;
private final IndexTermType indexTermType;
@ -61,6 +62,11 @@ public class MemtableIndexWriter implements PerColumnIndexWriter
private final MemtableIndex memtable;
private final RowMapping rowMapping;
private PrimaryKey minKey;
private PrimaryKey maxKey;
private long maxSSTableRowId = NO_ROWS;
private int rowCount;
public MemtableIndexWriter(MemtableIndex memtable,
IndexDescriptor indexDescriptor,
IndexTermType indexTermType,
@ -83,13 +89,31 @@ public class MemtableIndexWriter implements PerColumnIndexWriter
{
// Memtable indexes are flushed directly to disk with the aid of a mapping between primary
// keys and row IDs in the flushing SSTable. This writer, therefore, does nothing in
// response to the flushing of individual rows.
// response to the flushing of individual rows except for keeping index-specific statistics.
boolean isStatic = indexTermType.columnMetadata().isStatic();
// Indexes on static columns should only track static rows, and indexes on non-static columns
// should only track non-static rows. (Within a partition, the row ID for a static row will always
// come before any non-static row.)
if (key.kind() == PrimaryKey.Kind.STATIC && isStatic || key.kind() != PrimaryKey.Kind.STATIC && !isStatic)
{
if (minKey == null)
minKey = key;
maxKey = key;
rowCount++;
maxSSTableRowId = Math.max(maxSSTableRowId, sstableRowId);
}
}
@Override
public void abort(Throwable cause)
{
logger.warn(indexIdentifier.logMessage("Aborting index memtable flush for {}..."), indexDescriptor.sstableDescriptor, cause);
if (cause == null)
// This commonly occurs when a Memtable has no rows to flush, and is harmless:
logger.debug(indexIdentifier.logMessage("Aborting index memtable flush for {}..."), indexDescriptor.sstableDescriptor);
else
logger.warn(indexIdentifier.logMessage("Aborting index memtable flush for {}..."), indexDescriptor.sstableDescriptor, cause);
indexDescriptor.deleteColumnIndex(indexTermType, indexIdentifier);
}
@ -102,7 +126,7 @@ public class MemtableIndexWriter implements PerColumnIndexWriter
try
{
if (!rowMapping.hasRows() || memtable == null || memtable.isEmpty())
if (maxSSTableRowId == -1 || memtable == null || memtable.isEmpty())
{
logger.debug(indexIdentifier.logMessage("No indexed rows to flush from SSTable {}."), indexDescriptor.sstableDescriptor);
// Write a completion marker even though we haven't written anything to the index,
@ -155,9 +179,6 @@ public class MemtableIndexWriter implements PerColumnIndexWriter
return 0;
}
PrimaryKey minKey = indexTermType.columnMetadata().isStatic() ? rowMapping.minStaticKey : rowMapping.minKey;
PrimaryKey maxKey = indexTermType.columnMetadata().isStatic() ? rowMapping.maxStaticKey : rowMapping.maxKey;
// During index memtable flush, the data is sorted based on terms.
SegmentMetadata metadata = new SegmentMetadata(0,
numRows,
@ -176,11 +197,6 @@ public class MemtableIndexWriter implements PerColumnIndexWriter
private void flushVectorIndex(long startTime, Stopwatch stopwatch) throws IOException
{
int rowCount = indexTermType.columnMetadata().isStatic() ? rowMapping.staticRowCount : rowMapping.rowCount;
PrimaryKey minKey = indexTermType.columnMetadata().isStatic() ? rowMapping.minStaticKey : rowMapping.minKey;
PrimaryKey maxKey = indexTermType.columnMetadata().isStatic() ? rowMapping.maxStaticKey : rowMapping.maxKey;
long maxSSTableRowId = indexTermType.columnMetadata().isStatic() ? rowMapping.maxStaticSSTableRowId : rowMapping.maxSSTableRowId;
SegmentMetadata.ComponentMetadataMap metadataMap = memtable.writeDirect(indexDescriptor, indexIdentifier, rowMapping::get);
completeIndexFlush(rowCount, startTime, stopwatch);

2
src/java/org/apache/cassandra/index/sai/iterators/KeyRangeIterator.java
View File

@ -54,7 +54,7 @@ public abstract class KeyRangeIterator extends AbstractGuavaIterator<PrimaryKey>
{
boolean isComplete = min != null && max != null && count != 0;
boolean isEmpty = min == null && max == null && (count == 0 || count == -1);
Preconditions.checkArgument(isComplete || isEmpty, "Range: [" + min + ',' + max + "], Count: " + count);
Preconditions.checkArgument(isComplete || isEmpty, "Range: [%s,%s], Count: %d", min, max, count);
this.min = min;
this.current = min;

38
src/java/org/apache/cassandra/index/sai/iterators/KeyRangeUnionIterator.java
View File

@ -44,30 +44,34 @@ public class KeyRangeUnionIterator extends KeyRangeIterator
public PrimaryKey computeNext()
{
candidates.clear();
PrimaryKey candidate = null;
PrimaryKey candidateKey = null;
for (KeyRangeIterator range : ranges)
{
if (range.hasNext())
{
// Avoid repeated values but only if we have read at least one value
while (next != null && range.hasNext() && range.peek().compareTo(getCurrent()) == 0)
range.next();
if (!range.hasNext())
continue;
if (candidate == null)
if (candidateKey == null)
{
candidate = range.peek();
candidateKey = range.peek();
candidates.add(range);
}
else
{
int cmp = candidate.compareTo(range.peek());
PrimaryKey peeked = range.peek();
int cmp = candidateKey.compareTo(peeked);
if (cmp == 0)
{
// Replace any existing candidate key if this one is STATIC:
if (peeked.kind() == PrimaryKey.Kind.STATIC)
candidateKey = peeked;
candidates.add(range);
}
else if (cmp > 0)
{
candidates.clear();
candidate = range.peek();
candidateKey = peeked;
candidates.add(range);
}
}
@ -75,8 +79,18 @@ public class KeyRangeUnionIterator extends KeyRangeIterator
}
if (candidates.isEmpty())
return endOfData();
candidates.forEach(KeyRangeIterator::next);
return candidate;
for (KeyRangeIterator candidate : candidates)
{
do
{
// Consume the remaining values equal to the candidate key:
candidate.next();
}
while (candidate.hasNext() && candidate.peek().compareTo(candidateKey) == 0);
}
return candidateKey;
}
@Override

58
src/java/org/apache/cassandra/index/sai/plan/FilterTree.java
View File

@ -27,7 +27,7 @@ import com.google.common.collect.ListMultimap;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.index.sai.QueryContext;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.ColumnMetadata.Kind;
import org.apache.cassandra.utils.FBUtilities;
@ -43,14 +43,18 @@ import static org.apache.cassandra.index.sai.plan.Operation.BooleanOperator;
*/
public class FilterTree
{
protected final BooleanOperator op;
protected final BooleanOperator baseOperator;
protected final ListMultimap<ColumnMetadata, Expression> expressions;
protected final List<FilterTree> children = new ArrayList<>();
private final boolean isStrict;
private final QueryContext context;
FilterTree(BooleanOperator operation, ListMultimap<ColumnMetadata, Expression> expressions)
FilterTree(BooleanOperator baseOperator, ListMultimap<ColumnMetadata, Expression> expressions, boolean isStrict, QueryContext context)
{
this.op = operation;
this.baseOperator = baseOperator;
this.expressions = expressions;
this.isStrict = isStrict;
this.context = context;
}
void addChild(FilterTree child)
@ -58,29 +62,47 @@ public class FilterTree
children.add(child);
}
public boolean isSatisfiedBy(DecoratedKey key, Unfiltered unfiltered, Row staticRow)
/**
* @return true if this node of the tree or any of its children filter a non-static column
*/
public boolean restrictsNonStaticRow()
{
boolean result = localSatisfiedBy(key, unfiltered, staticRow);
for (ColumnMetadata column : expressions.keySet())
if (!column.isStatic())
return true;
for (FilterTree child : children)
result = op.apply(result, child.isSatisfiedBy(key, unfiltered, staticRow));
if (child.restrictsNonStaticRow())
return true;
return false;
}
public boolean isSatisfiedBy(DecoratedKey key, Row row, Row staticRow)
{
boolean result = localSatisfiedBy(key, row, staticRow);
for (FilterTree child : children)
result = baseOperator.apply(result, child.isSatisfiedBy(key, row, staticRow));
return result;
}
private boolean localSatisfiedBy(DecoratedKey key, Unfiltered unfiltered, Row staticRow)
private boolean localSatisfiedBy(DecoratedKey key, Row row, Row staticRow)
{
if (unfiltered == null || !unfiltered.isRow())
if (row == null)
return false;
final long now = FBUtilities.nowInSeconds();
boolean result = op == BooleanOperator.AND;
// Downgrade AND to OR unless the coordinator indicates strict filtering is safe or all matches are repaired:
BooleanOperator localOperator = (isStrict || !context.hasUnrepairedMatches) ? baseOperator : BooleanOperator.OR;
boolean result = localOperator == BooleanOperator.AND;
Iterator<ColumnMetadata> columnIterator = expressions.keySet().iterator();
while (columnIterator.hasNext())
{
ColumnMetadata column = columnIterator.next();
Row row = column.kind == Kind.STATIC ? staticRow : (Row) unfiltered;
Row localRow = column.kind == Kind.STATIC ? staticRow : row;
// If there is a column with multiple expressions that can mean an OR, or (in the case of map
// collections) it can mean different map indexes.
@ -95,18 +117,22 @@ public class FilterTree
if (filter.getIndexTermType().isNonFrozenCollection())
{
Iterator<ByteBuffer> valueIterator = filter.getIndexTermType().valuesOf(row, now);
result = op.apply(result, collectionMatch(valueIterator, filter));
Iterator<ByteBuffer> valueIterator = filter.getIndexTermType().valuesOf(localRow, now);
result = localOperator.apply(result, collectionMatch(valueIterator, filter));
}
else
{
ByteBuffer value = filter.getIndexTermType().valueOf(key, row, now);
result = op.apply(result, singletonMatch(value, filter));
ByteBuffer value = filter.getIndexTermType().valueOf(key, localRow, now);
result = localOperator.apply(result, singletonMatch(value, filter));
}
// If the operation is an AND then exit early if we get a single false
if (op == BooleanOperator.AND && !result)
if ((localOperator == BooleanOperator.AND) && !result)
return false;
// If the operation is an OR then exit early if we get a single true
if (localOperator == BooleanOperator.OR && result)
return true;
}
}
return result;

32
src/java/org/apache/cassandra/index/sai/plan/Operation.java
View File

@ -36,6 +36,7 @@ import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CollectionType;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.sai.QueryContext;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.sai.analyzer.AbstractAnalyzer;
import org.apache.cassandra.index.sai.iterators.KeyRangeIterator;
@ -46,7 +47,8 @@ public class Operation
{
public enum BooleanOperator
{
AND((a, b) -> a & b);
AND((a, b) -> a & b),
OR((a, b) -> a | b);
private final BiFunction<Boolean, Boolean, Boolean> func;
@ -256,13 +258,13 @@ public class Operation
*/
static KeyRangeIterator buildIterator(QueryController controller)
{
var orderings = controller.filterOperation().getExpressions()
var orderings = controller.indexFilter().getExpressions()
.stream().filter(e -> e.operator() == Operator.ANN).collect(Collectors.toList());
assert orderings.size() <= 1;
if (controller.filterOperation().getExpressions().size() == 1 && orderings.size() == 1)
if (controller.indexFilter().getExpressions().size() == 1 && orderings.size() == 1)
// If we only have one expression, we just use the ANN index to order and limit.
return controller.getTopKRows(orderings.get(0));
var iterator = Node.buildTree(controller.filterOperation()).analyzeTree(controller).rangeIterator(controller);
var iterator = Node.buildTree(controller.indexFilter()).analyzeTree(controller).rangeIterator(controller);
if (orderings.isEmpty())
return iterator;
return controller.getTopKRows(iterator, orderings.get(0));
@ -277,9 +279,9 @@ public class Operation
*
* @return root of the filter tree.
*/
static FilterTree buildFilter(QueryController controller)
static FilterTree buildFilter(QueryController controller, boolean strict)
{
return Node.buildTree(controller.filterOperation()).buildFilter(controller);
return Node.buildTree(controller.indexFilter()).buildFilter(controller, strict);
}
static abstract class Node
@ -308,7 +310,7 @@ public class Operation
abstract void analyze(List<RowFilter.Expression> expressionList, QueryController controller);
abstract FilterTree filterTree();
abstract FilterTree filterTree(boolean strict, QueryContext context);
abstract KeyRangeIterator rangeIterator(QueryController controller);
@ -347,13 +349,13 @@ public class Operation
}
}
FilterTree buildFilter(QueryController controller)
FilterTree buildFilter(QueryController controller, boolean isStrict)
{
analyzeTree(controller);
FilterTree tree = filterTree();
FilterTree tree = filterTree(isStrict, controller.queryContext);
for (Node child : children())
if (child.canFilter())
tree.addChild(child.buildFilter(controller));
tree.addChild(child.buildFilter(controller, isStrict));
return tree;
}
}
@ -384,9 +386,9 @@ public class Operation
}
@Override
FilterTree filterTree()
FilterTree filterTree(boolean isStrict, QueryContext context)
{
return new FilterTree(BooleanOperator.AND, expressionMap);
return new FilterTree(BooleanOperator.AND, expressionMap, isStrict, context);
}
@Override
@ -411,12 +413,14 @@ public class Operation
public void analyze(List<RowFilter.Expression> expressionList, QueryController controller)
{
expressionMap = buildIndexExpressions(controller, expressionList);
assert expressionMap.size() == 1 : "Expression nodes should only have a single expression!";
}
@Override
FilterTree filterTree()
FilterTree filterTree(boolean isStrict, QueryContext context)
{
return new FilterTree(BooleanOperator.AND, expressionMap);
// There should only be one expression, so AND/OR would both work here.
return new FilterTree(BooleanOperator.AND, expressionMap, isStrict, context);
}
public ExpressionNode(RowFilter.Expression expression)

93
src/java/org/apache/cassandra/index/sai/plan/QueryController.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.index.sai.plan;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@ -44,6 +45,7 @@ import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.guardrails.Guardrails;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Range;
@ -69,10 +71,11 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.SAI_VECTOR
public class QueryController
{
final QueryContext queryContext;
private final ColumnFamilyStore cfs;
private final ReadCommand command;
private final QueryContext queryContext;
private final RowFilter filterOperation;
private final RowFilter indexFilter;
private final List<DataRange> ranges;
private final AbstractBounds<PartitionPosition> mergeRange;
private final PrimaryKey.Factory keyFactory;
@ -82,13 +85,13 @@ public class QueryController
public QueryController(ColumnFamilyStore cfs,
ReadCommand command,
RowFilter filterOperation,
RowFilter indexFilter,
QueryContext queryContext)
{
this.cfs = cfs;
this.command = command;
this.queryContext = queryContext;
this.filterOperation = filterOperation;
this.indexFilter = indexFilter;
this.ranges = dataRanges(command);
DataRange first = ranges.get(0);
DataRange last = ranges.get(ranges.size() - 1);
@ -119,9 +122,14 @@ public class QueryController
return command.metadata();
}
public RowFilter filterOperation()
public RowFilter indexFilter()
{
return this.filterOperation;
return this.indexFilter;
}
public boolean usesStrictFiltering()
{
return command.rowFilter().isStrict();
}
/**
@ -167,33 +175,84 @@ public class QueryController
* the {@link SSTableIndex}s that will satisfy the expression.
* <p>
* Each (expression, SSTable indexes) pair is then passed to
* {@link IndexSearchResultIterator#build(Expression, Collection, AbstractBounds, QueryContext)}
* {@link IndexSearchResultIterator#build(Expression, Collection, AbstractBounds, QueryContext, boolean)}
* to search the in-memory index associated with the expression and the SSTable indexes, the results of
* which are unioned and returned.
* <p>
* The results from each call to {@link IndexSearchResultIterator#build(Expression, Collection, AbstractBounds, QueryContext)}
* are added to a {@link KeyRangeIntersectionIterator} and returned.
* The results from each call to {@link IndexSearchResultIterator#build(Expression, Collection, AbstractBounds, QueryContext, boolean)}
* are added to a {@link KeyRangeIntersectionIterator} and returned if strict filtering is allowed.
* <p>
* If strict filtering is not allowed, indexes are split into two groups according to the repaired status of their
* backing SSTables. Results from searches over the repaired group are added to a
* {@link KeyRangeIntersectionIterator}, which is then added, along with results from searches on the unrepaired
* set, to a top-level {@link KeyRangeUnionIterator}, and returned. This is done to ensure that AND queries do not
* prematurely filter out matches on un-repaired partial updates. Post-filtering must also take this into
* account. (see {@link FilterTree#isSatisfiedBy(DecoratedKey, Row, Row)}) Note that Memtable-attached
* indexes are treated as part of the unrepaired set.
*/
public KeyRangeIterator.Builder getIndexQueryResults(Collection<Expression> expressions)
{
// VSTODO move ANN out of expressions and into its own abstraction? That will help get generic ORDER BY support
expressions = expressions.stream().filter(e -> e.getIndexOperator() != Expression.IndexOperator.ANN).collect(Collectors.toList());
KeyRangeIterator.Builder builder = KeyRangeIntersectionIterator.builder(expressions.size());
QueryViewBuilder queryViewBuilder = new QueryViewBuilder(expressions, mergeRange);
QueryViewBuilder.QueryView queryView = queryViewBuilder.build();
KeyRangeIterator.Builder builder = command.rowFilter().isStrict()
? KeyRangeIntersectionIterator.builder(expressions.size())
: KeyRangeUnionIterator.builder(expressions.size());
QueryViewBuilder.QueryView queryView = new QueryViewBuilder(expressions, mergeRange).build();
try
{
maybeTriggerGuardrails(queryView);
for (Pair<Expression, Collection<SSTableIndex>> queryViewPair : queryView.view)
if (command.rowFilter().isStrict())
{
KeyRangeIterator indexIterator = IndexSearchResultIterator.build(queryViewPair.left, queryViewPair.right, mergeRange, queryContext);
// If strict filtering is enabled, evaluate indexes for both repaired and un-repaired SSTables together.
// This usually means we are making this local index query in the context of a user query that reads
// from a single replica and thus can safely perform local intersections.
for (Pair<Expression, Collection<SSTableIndex>> queryViewPair : queryView.view)
builder.add(IndexSearchResultIterator.build(queryViewPair.left, queryViewPair.right, mergeRange, queryContext, true));
}
else
{
KeyRangeIterator.Builder repairedBuilder = KeyRangeIntersectionIterator.builder(expressions.size());
builder.add(indexIterator);
for (Pair<Expression, Collection<SSTableIndex>> queryViewPair : queryView.view)
{
// The initial sizes here reflect little more than an effort to avoid resizing for
// partition-restricted searches w/ LCS:
List<SSTableIndex> repaired = new ArrayList<>(5);
List<SSTableIndex> unrepaired = new ArrayList<>(5);
// Split SSTable indexes into repaired and un-reparired:
for (SSTableIndex index : queryViewPair.right)
if (index.getSSTable().isRepaired())
repaired.add(index);
else
unrepaired.add(index);
// Always build an iterator for the un-repaired set, given this must include Memtable indexes...
IndexSearchResultIterator unrepairedIterator =
IndexSearchResultIterator.build(queryViewPair.left, unrepaired, mergeRange, queryContext, true);
// ...but ignore it if our combined results are empty.
if (unrepairedIterator.getCount() > 0)
{
builder.add(unrepairedIterator);
queryContext.hasUnrepairedMatches = true;
}
else
{
// We're not going to use this, so release the resources it holds.
unrepairedIterator.close();
}
// ...then only add an iterator to the repaired intersection if repaired SSTable indexes exist.
if (!repaired.isEmpty())
repairedBuilder.add(IndexSearchResultIterator.build(queryViewPair.left, repaired, mergeRange, queryContext, false));
}
if (repairedBuilder.rangeCount() > 0)
builder.add(repairedBuilder.build());
}
}
catch (Throwable t)

78
src/java/org/apache/cassandra/index/sai/plan/QueryViewBuilder.java
View File

@ -20,12 +20,9 @@ package org.apache.cassandra.index.sai.plan;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.NavigableSet;
import java.util.Set;
import java.util.TreeSet;
import java.util.stream.Collectors;
import org.apache.cassandra.db.PartitionPosition;
@ -93,9 +90,6 @@ public class QueryViewBuilder
private Collection<Pair<Expression, Collection<SSTableIndex>>> getQueryView(Collection<Expression> expressions)
{
// first let's determine the most selective expression
Pair<Expression, Collection<SSTableIndex>> mostSelective = calculateMostSelective(expressions);
List<Pair<Expression, Collection<SSTableIndex>>> queryView = new ArrayList<>();
for (Expression expression : expressions)
@ -105,82 +99,18 @@ public class QueryViewBuilder
if (expression.isNotIndexed())
continue;
// If we didn't get a most selective expression then none of the
// expressions select anything so, add an empty entry for the
// expression. We need the empty entry because we may have in-memory
// data for the expression
if (mostSelective == null)
{
queryView.add(Pair.create(expression, Collections.emptyList()));
continue;
}
// If this expression is the most selective then just add it to the
// query view
if (expression.equals(mostSelective.left))
{
queryView.add(mostSelective);
continue;
}
// Finally, we select all the sstable indexes for this expression that
// have overlapping keys with the sstable indexes of the most selective
// and have a term range that is satisfied by the expression.
// Select all the sstable indexes that have a term range that is satisfied by this expression and
// overlap with the key range being queried.
View view = expression.getIndex().view();
Set<SSTableIndex> indexes = new TreeSet<>(SSTableIndex.COMPARATOR);
indexes.addAll(view.match(expression)
.stream()
.filter(index -> sstableIndexOverlaps(index, mostSelective.right))
.collect(Collectors.toList()));
queryView.add(Pair.create(expression, indexes));
queryView.add(Pair.create(expression, selectIndexesInRange(view.match(expression))));
}
return queryView;
}
private boolean sstableIndexOverlaps(SSTableIndex sstableIndex, Collection<SSTableIndex> sstableIndexes)
{
return sstableIndexes.stream().anyMatch(index -> index.bounds().contains(sstableIndex.bounds().left) ||
index.bounds().contains(sstableIndex.bounds().right));
}
// The purpose of this method is to calculate the most selective expression. This is the
// expression with the most sstable indexes that match the expression by term and lie
// within the key range being queried.
//
// The result can be null. This indicates that none of the expressions select any
// sstable indexes.
private Pair<Expression, Collection<SSTableIndex>> calculateMostSelective(Collection<Expression> expressions)
{
Expression mostSelectiveExpression = null;
NavigableSet<SSTableIndex> mostSelectiveIndexes = null;
for (Expression expression : expressions)
{
if (expression.isNotIndexed())
continue;
View view = expression.getIndex().view();
NavigableSet<SSTableIndex> indexes = new TreeSet<>(SSTableIndex.COMPARATOR);
indexes.addAll(selectIndexesInRange(view.match(expression)));
if (indexes.isEmpty())
continue;
if (mostSelectiveExpression == null || mostSelectiveIndexes.size() > indexes.size())
{
mostSelectiveIndexes = indexes;
mostSelectiveExpression = expression;
}
}
return mostSelectiveExpression == null ? null : Pair.create(mostSelectiveExpression, mostSelectiveIndexes);
}
private List<SSTableIndex> selectIndexesInRange(Collection<SSTableIndex> indexes)
{
return indexes.stream().filter(this::indexInRange).collect(Collectors.toList());
return indexes.stream().filter(this::indexInRange).sorted(SSTableIndex.COMPARATOR).collect(Collectors.toList());
}
private boolean indexInRange(SSTableIndex index)

30
src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexQueryPlan.java
View File

@ -30,6 +30,7 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.sai.metrics.TableQueryMetrics;
@ -37,23 +38,26 @@ import org.apache.cassandra.schema.TableMetadata;
public class StorageAttachedIndexQueryPlan implements Index.QueryPlan
{
public static final String UNSUPPORTED_NON_STRICT_OPERATOR =
"Operator %s is only supported in intersections for reads that do not require replica reconciliation.";
private final ColumnFamilyStore cfs;
private final TableQueryMetrics queryMetrics;
private final RowFilter postIndexFilter;
private final RowFilter filterOperation;
private final RowFilter indexFilter;
private final Set<Index> indexes;
private final boolean isTopK;
private StorageAttachedIndexQueryPlan(ColumnFamilyStore cfs,
TableQueryMetrics queryMetrics,
RowFilter postIndexFilter,
RowFilter filterOperation,
RowFilter indexFilter,
ImmutableSet<Index> indexes)
{
this.cfs = cfs;
this.queryMetrics = queryMetrics;
this.postIndexFilter = postIndexFilter;
this.filterOperation = filterOperation;
this.indexFilter = indexFilter;
this.indexes = indexes;
this.isTopK = indexes.stream().anyMatch(i -> i instanceof StorageAttachedIndex && ((StorageAttachedIndex) i).termType().isVector());
}
@ -62,27 +66,33 @@ public class StorageAttachedIndexQueryPlan implements Index.QueryPlan
public static StorageAttachedIndexQueryPlan create(ColumnFamilyStore cfs,
TableQueryMetrics queryMetrics,
Set<StorageAttachedIndex> indexes,
RowFilter rowFilter)
RowFilter filter)
{
ImmutableSet.Builder<Index> selectedIndexesBuilder = ImmutableSet.builder();
RowFilter preIndexFilter = rowFilter;
RowFilter postIndexFilter = rowFilter;
RowFilter preIndexFilter = filter;
RowFilter postIndexFilter = filter;
for (RowFilter.Expression expression : rowFilter)
for (RowFilter.Expression expression : filter)
{
// we ignore any expressions here (currently IN and user-defined expressions) where we don't have a way to
// translate their #isSatifiedBy method, they will be included in the filter returned by QueryPlan#postIndexQueryFilter()
// We ignore any expressions here (currently IN and user-defined expressions) where we don't have a way to
// translate their #isSatifiedBy method, they will be included in the filter returned by
// QueryPlan#postIndexQueryFilter(). If strict filtering is not allowed, we must reject the query until the
// expression(s) in question are compatible with #isSatifiedBy.
//
// Note: For both the pre- and post-filters we need to check that the expression exists before removing it
// because the without method assert if the expression doesn't exist. This can be the case if we are given
// a duplicate expression - a = 1 and a = 1. The without method removes all instances of the expression.
if (expression.operator().isIN() || expression.isUserDefined())
{
if (!filter.isStrict())
throw new InvalidRequestException(String.format(UNSUPPORTED_NON_STRICT_OPERATOR, expression.operator()));
if (preIndexFilter.getExpressions().contains(expression))
preIndexFilter = preIndexFilter.without(expression);
continue;
}
if (postIndexFilter.getExpressions().contains(expression))
postIndexFilter = postIndexFilter.without(expression);
@ -129,7 +139,7 @@ public class StorageAttachedIndexQueryPlan implements Index.QueryPlan
return new StorageAttachedIndexSearcher(cfs,
queryMetrics,
command,
filterOperation,
indexFilter,
DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS));
}

68
src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java
View File

@ -27,8 +27,6 @@ import java.util.function.Supplier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import com.google.common.collect.Iterators;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
@ -67,12 +65,12 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
public StorageAttachedIndexSearcher(ColumnFamilyStore cfs,
TableQueryMetrics tableQueryMetrics,
ReadCommand command,
RowFilter filterOperation,
RowFilter indexFilter,
long executionQuotaMs)
{
this.command = command;
this.queryContext = new QueryContext(command, executionQuotaMs);
this.queryController = new QueryController(cfs, command, filterOperation, queryContext);
this.queryController = new QueryController(cfs, command, indexFilter, queryContext);
this.tableQueryMetrics = tableQueryMetrics;
}
@ -85,10 +83,10 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
@Override
public PartitionIterator filterReplicaFilteringProtection(PartitionIterator fullResponse)
{
for (RowFilter.Expression expression : queryController.filterOperation())
for (RowFilter.Expression expression : queryController.indexFilter())
{
if (queryController.hasAnalyzer(expression))
return applyIndexFilter(fullResponse, Operation.buildFilter(queryController), queryContext);
return applyIndexFilter(fullResponse, Operation.buildFilter(queryController, true), queryContext);
}
// if no analyzer does transformation
@ -140,7 +138,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
this.keyRanges = queryController.dataRanges().iterator();
this.currentKeyRange = keyRanges.next().keyRange();
this.resultKeyIterator = Operation.buildIterator(queryController);
this.filterTree = Operation.buildFilter(queryController);
this.filterTree = Operation.buildFilter(queryController, queryController.usesStrictFiltering());
this.executionController = executionController;
this.keyFactory = queryController.primaryKeyFactory();
this.firstPrimaryKey = queryController.firstPrimaryKeyInRange();
@ -375,7 +373,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
queryContext.partitionsRead++;
queryContext.checkpoint();
UnfilteredRowIterator filtered = applyIndexFilter(key, partition, filterTree, queryContext);
UnfilteredRowIterator filtered = applyIndexFilter(key, partition, filterTree);
// Note that we record the duration of the read after post-filtering, which actually
// materializes the rows from disk.
@ -385,59 +383,60 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
}
}
private UnfilteredRowIterator applyIndexFilter(PrimaryKey key, UnfilteredRowIterator partition, FilterTree tree, QueryContext queryContext)
private UnfilteredRowIterator applyIndexFilter(PrimaryKey key, UnfilteredRowIterator partition, FilterTree tree)
{
Row staticRow = partition.staticRow();
List<Unfiltered> clusters = new ArrayList<>();
List<Unfiltered> matchingRows = new ArrayList<>();
boolean hasMatch = false;
// We need to filter the partition rows before filtering on the static row. If this is done in the other
// order then we get incorrect results if we are filtering on a partition key index on a table with a
// composite partition key.
while (partition.hasNext())
{
Unfiltered row = partition.next();
Unfiltered unfiltered = partition.next();
queryContext.rowsFiltered++;
if (tree.isSatisfiedBy(partition.partitionKey(), row, staticRow))
if (unfiltered.isRow())
{
clusters.add(row);
queryContext.rowsFiltered++;
if (tree.isSatisfiedBy(partition.partitionKey(), (Row) unfiltered, staticRow))
{
matchingRows.add(unfiltered);
hasMatch = true;
}
}
}
if (clusters.isEmpty())
if (!hasMatch)
{
queryContext.rowsFiltered++;
if (tree.isSatisfiedBy(key.partitionKey(), staticRow, staticRow))
{
clusters.add(staticRow);
}
hasMatch = true;
}
/*
* If {@code clusters} is empty, which means either all clustering row and static row pairs failed,
* or static row and static row pair failed. In both cases, we should not return any partition.
* If {@code clusters} is not empty, which means either there are some clustering row and static row pairs match the filters,
* or static row and static row pair matches the filters. In both cases, we should return a partition with static row,
* and remove the static row marker from the {@code clusters} for the latter case.
*/
if (clusters.isEmpty())
if (!hasMatch)
{
// shadowed by expired TTL or row tombstone or range tombstone
if (topK)
queryContext.vectorContext().recordShadowedPrimaryKey(key);
// If there are no matches, return an empty partition. If reconciliation is required at the
// coordinator, replica filtering protection may make a second round trip to complete its view
// of the partition.
return null;
}
return new PartitionIterator(partition, staticRow, Iterators.filter(clusters.iterator(), u -> !((Row)u).isStatic()));
// Return all matches found, along with the static row...
return new PartitionIterator(partition, staticRow, matchingRows.iterator());
}
private class PartitionIterator extends AbstractUnfilteredRowIterator
{
private final Iterator<Unfiltered> rows;
public PartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator<Unfiltered> content)
public PartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator<Unfiltered> rows)
{
super(partition.metadata(),
partition.partitionKey(),
@ -447,7 +446,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
partition.isReverseOrder(),
partition.stats());
rows = content;
this.rows = rows;
}
@Override
@ -475,7 +474,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
* Used by {@link StorageAttachedIndexSearcher#filterReplicaFilteringProtection} to filter rows for columns that
* have transformations so won't get handled correctly by the row filter.
*/
private static PartitionIterator applyIndexFilter(PartitionIterator response, FilterTree tree, QueryContext queryContext)
private static PartitionIterator applyIndexFilter(PartitionIterator response, FilterTree tree, QueryContext context)
{
return new PartitionIterator()
{
@ -497,6 +496,11 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
RowIterator delegate = response.next();
Row staticRow = delegate.staticRow();
// If we only restrict static columns, and we pass the filter, simply pass through the delegate, as all
// non-static rows are matches. If we fail on the filter, no rows are matches, so return nothing.
if (!tree.restrictsNonStaticRow())
return tree.isSatisfiedBy(delegate.partitionKey(), staticRow, staticRow) ? delegate : null;
return new RowIterator()
{
Row next;
@ -542,7 +546,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
while (delegate.hasNext())
{
Row row = delegate.next();
queryContext.rowsFiltered++;
context.rowsFiltered++;
if (tree.isSatisfiedBy(delegate.partitionKey(), row, staticRow))
return row;
}

55
src/java/org/apache/cassandra/service/reads/DataResolver.java
View File

@ -121,7 +121,7 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
if (usesReplicaFilteringProtection())
return resolveWithReplicaFilteringProtection(replicas, repairedDataTracker);
ResolveContext context = new ResolveContext(replicas);
ResolveContext context = new ResolveContext(replicas, true);
return resolveWithReadRepair(context,
i -> shortReadProtectedResponse(i, context, runOnShortRead),
UnaryOperator.identity(),
@ -148,7 +148,11 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
private final E replicas;
private final DataLimits.Counter mergedResultCounter;
private ResolveContext(E replicas)
/**
* @param replicas the collection of {@link Endpoints} involved in the query
* @param enforceLimits whether or not to enforce counter limits in this context
*/
private ResolveContext(E replicas, boolean enforceLimits)
{
this.replicas = replicas;
this.mergedResultCounter = command.limits().newCounter(command.nowInSec(),
@ -157,8 +161,8 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
enforceStrictLiveness);
// In case of top-k query, do not trim reconciled rows here because QueryPlan#postProcessor()
// needs to compare all rows
if (command.isTopK())
// needs to compare all rows. Also avoid enforcing the limit if explicitly requested.
if (command.isTopK() || !enforceLimits)
this.mergedResultCounter.onlyCount();
}
@ -180,7 +184,7 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
if (command.isTopK())
return false;
// If we have only one result, there is no read repair to do and we can't get short reads
// If we have only one result, there is no read repair to do, and we can't get short reads
// Also, so-called "short reads" stems from nodes returning only a subset of the results they have for a
// partition due to the limit, but that subset not being enough post-reconciliation. So if we don't have limit,
// don't bother protecting against short reads.
@ -229,36 +233,36 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
// Protecting against inconsistent replica filtering (some replica returning a row that is outdated but that
// wouldn't be removed by normal reconciliation because up-to-date replica have filtered the up-to-date version
// of that row) involves 3 main elements:
// 1) We combine short-read protection and a merge listener that identifies potentially "out-of-date"
// rows to create an iterator that is guaranteed to produce enough valid row results to satisfy the query
// limit if enough actually exist. A row is considered out-of-date if its merged from is non-empty and we
// receive not response from at least one replica. In this case, it is possible that filtering at the
// "silent" replica has produced a more up-to-date result.
// 1) We combine an unlimited, short-read-protected iterator of unfiltered partitions with a merge listener
// that identifies potentially "out-of-date" rows. A row is considered out-of-date if its merged form is
// non-empty, and we receive no response from some replica. It is also potentially out-of-date if there is
// disagreement around the value of a single column in a non-empty row, and some replica provides no value
// for that column. In either case, it is possible that filtering at the "silent" replica has produced a
// more up-to-date result.
// 2) This iterator is passed to the standard resolution process with read-repair, but is first wrapped in a
// response provider that lazily "completes" potentially out-of-date rows by directly querying them on the
// replicas that were previously silent. As this iterator is consumed, it caches valid data for potentially
// out-of-date rows, and this cached data is merged with the fetched data as rows are requested. If there
// is no replica divergence, only rows in the partition being evalutated will be cached (then released
// when the partition is consumed).
// out-of-date rows, and this cached data is merged with the fetched data as rows are requested. Only rows
// in the partition being evalutated will be cached (then released when the partition is consumed).
// 3) After a "complete" row is materialized, it must pass the row filter supplied by the original query
// before it counts against the limit.
// before it counts against the limit. If this "pre-count" filter causes a short read, additional rows
// will be fetched from the first-phase iterator.
// We need separate contexts, as each context has his own counter
ResolveContext firstPhaseContext = new ResolveContext(replicas);
ResolveContext secondPhaseContext = new ResolveContext(replicas);
ReplicaFilteringProtection<E> rfp = new ReplicaFilteringProtection<>(replicaPlan().keyspace(),
command,
replicaPlan().consistencyLevel(),
queryStartNanoTime,
firstPhaseContext.replicas,
replicas,
DatabaseDescriptor.getCachedReplicaRowsWarnThreshold(),
DatabaseDescriptor.getCachedReplicaRowsFailThreshold());
ResolveContext firstPhaseContext = new ResolveContext(replicas, false);
PartitionIterator firstPhasePartitions = resolveInternal(firstPhaseContext,
rfp.mergeController(),
i -> shortReadProtectedResponse(i, firstPhaseContext, null),
UnaryOperator.identity());
null);
ResolveContext secondPhaseContext = new ResolveContext(replicas, true);
PartitionIterator completedPartitions = resolveWithReadRepair(secondPhaseContext,
i -> rfp.queryProtectedPartitions(firstPhasePartitions, i),
preCountFilterForReplicaFilteringProtection(),
@ -282,7 +286,7 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
private PartitionIterator resolveInternal(ResolveContext context,
UnfilteredPartitionIterators.MergeListener mergeListener,
ResponseProvider responseProvider,
UnaryOperator<PartitionIterator> preCountFilter)
@Nullable UnaryOperator<PartitionIterator> preCountFilter)
{
int count = context.replicas.size();
List<UnfilteredPartitionIterator> results = new ArrayList<>(count);
@ -306,7 +310,10 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
UnfilteredPartitionIterator merged = UnfilteredPartitionIterators.merge(results, mergeListener);
Filter filter = new Filter(command.nowInSec(), command.metadata().enforceStrictLiveness());
FilteredPartitions filtered = FilteredPartitions.filter(merged, filter);
PartitionIterator counted = Transformation.apply(preCountFilter.apply(filtered), context.mergedResultCounter);
PartitionIterator counted = preCountFilter == null
? filtered
: Transformation.apply(preCountFilter.apply(filtered), context.mergedResultCounter);
return Transformation.apply(counted, new EmptyPartitionsDiscarder());
}
@ -376,11 +383,11 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
}
}
public Row onMergedRows(Row merged, Row[] versions)
public void onMergedRows(Row merged, Row[] versions)
{
try
{
return rowListener.onMergedRows(merged, versions);
rowListener.onMergedRows(merged, versions);
}
catch (AssertionError e)
{
@ -401,7 +408,7 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
{
try
{
// The code for merging range tombstones is a tad complex and we had the assertions there triggered
// The code for merging range tombstones is a tad complex, and we had the assertions there triggered
// unexpectedly in a few occasions (CASSANDRA-13237, CASSANDRA-13719). It's hard to get insights
// when that happen without more context that what the assertion errors give us however, hence the
// catch here that basically gather as much as context as reasonable.

86
src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java
View File

@ -66,6 +66,7 @@ import org.apache.cassandra.locator.ReplicaPlan;
import org.apache.cassandra.locator.ReplicaPlans;
import org.apache.cassandra.metrics.TableMetrics;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.service.StorageProxy;
@ -78,12 +79,15 @@ import org.apache.cassandra.utils.btree.BTreeSet;
* Helper in charge of collecting additional queries to be done on the coordinator to protect against invalid results
* being included due to replica-side filtering (secondary indexes or {@code ALLOW * FILTERING}).
* <p>
* When using replica-side filtering with CL>ONE, a replica can send a stale result satisfying the filter, while updated
* replicas won't send a corresponding tombstone to discard that result during reconciliation. This helper identifies
* the rows in a replica response that don't have a corresponding row in other replica responses, and requests them by
* primary key to the "silent" replicas in a second fetch round.
* <p>
* See CASSANDRA-8272, CASSANDRA-8273, and CASSANDRA-15907 for further details.
* When using replica-side filtering with CL > ONE, a replica can send a stale result satisfying the filter, while
* updated replicas won't send a corresponding tombstone to discard that result during reconciliation. This helper
* identifies the rows in a replica response that don't have a corresponding row in other replica responses (or don't
* have corresponding cell values), and requests them by primary key on the "silent" replicas in a second fetch round.
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-8272">CASSANDRA-8272</a>
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-8273">CASSANDRA-8273</a>
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-15907">CASSANDRA-15907</a>
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-19018">CASSANDRA-19018</a>
*/
public class ReplicaFilteringProtection<E extends Endpoints<E>>
{
@ -166,13 +170,8 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
}
/**
* Returns a merge listener that skips the merged rows for which any of the replicas doesn't have a version,
* pessimistically assuming that they are outdated. It is intended to be used during a first merge of per-replica
* query results to ensure we fetch enough results from the replicas to ensure we don't miss any potentially
* outdated result.
* <p>
* The listener will track both the accepted data and the primary keys of the rows that are considered as outdated.
* That way, once the query results would have been merged using this listener, further calls to
* This listener tracks both the accepted data and the primary keys of the rows that may be incomplete.
* That way, once the query results are merged using this listener, subsequent calls to
* {@link #queryProtectedPartitions(PartitionIterator, int)} will use the collected data to return a copy of the
* data originally collected from the specified replica, completed with the potentially outdated rows.
*/
@ -197,6 +196,9 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
for (int i = 0; i < sources.size(); i++)
builders.add(i, new PartitionBuilder(partitionKey, sources.get(i), columns, stats));
boolean[] silentRowAt = new boolean[builders.size()];
boolean[] silentColumnAt = new boolean[builders.size()];
return new UnfilteredRowIterators.MergeListener()
{
@Override
@ -208,34 +210,49 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
}
@Override
public Row onMergedRows(Row merged, Row[] versions)
public void onMergedRows(Row merged, Row[] versions)
{
// cache the row versions to be able to regenerate the original row iterator
// Cache the row versions to be able to regenerate the original row iterator:
for (int i = 0; i < versions.length; i++)
builders.get(i).addRow(versions[i]);
// If all versions are empty, there's no divergence to resolve:
if (merged.isEmpty())
return merged;
return;
boolean isPotentiallyOutdated = false;
boolean isStatic = merged.isStatic();
Arrays.fill(silentRowAt, false);
// Mark replicas silent if they provide no data for the row:
for (int i = 0; i < versions.length; i++)
if (versions[i] == null || (merged.isStatic() && versions[i].isEmpty()))
silentRowAt[i] = true;
// Even if there are no completely missing rows, replicas may still be silent about individual
// columns, so we need to check for divergence at the column level:
for (ColumnMetadata column : columns)
{
Row version = versions[i];
if (version == null || (isStatic && version.isEmpty()))
Arrays.fill(silentColumnAt, false);
boolean allSilent = true;
for (int i = 0; i < versions.length; i++)
{
isPotentiallyOutdated = true;
builders.get(i).addToFetch(merged);
// If the version at this replica is null, we've already marked it as silent:
if (versions[i] != null && versions[i].getColumnData(column) == null)
silentColumnAt[i] = true;
else
allSilent = false;
}
for (int i = 0; i < versions.length; i++)
// Mark the replica silent if it is silent about this column and there is actually
// divergence between the replicas. (i.e. If all replicas are silent for this
// column, there is nothing to fetch to complete the row anyway.)
silentRowAt[i] |= silentColumnAt[i] && !allSilent;
}
// If the row is potentially outdated (because some replica didn't send anything and so it _may_ be
// an outdated result that is only present because other replica have filtered the up-to-date result
// out), then we skip the row. In other words, the results of the initial merging of results by this
// protection assume the worst case scenario where every row that might be outdated actually is.
// This ensures that during this first phase (collecting additional row to fetch) we are guaranteed
// to look at enough data to ultimately fulfill the query limit.
return isPotentiallyOutdated ? null : merged;
for (int i = 0; i < silentRowAt.length; i++)
if (silentRowAt[i])
builders.get(i).addToFetch(merged);
}
@Override
@ -334,8 +351,7 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
public boolean hasNext()
{
// If there are no cached partition builders for this source, advance the first phase iterator, which
// will force the RFP merge listener to load at least the next protected partition. Note that this may
// load more than one partition if any divergence between replicas is discovered by the merge listener.
// will force the RFP merge listener to load at least the next protected partition.
if (partitions.isEmpty())
{
PartitionIterators.consumeNext(merged);
@ -367,6 +383,8 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
private BTreeSet.Builder<Clustering<?>> toFetch;
private int partitionRowsCached;
private boolean unresolvedStatic = false;
private PartitionBuilder(DecoratedKey key, Replica source, RegularAndStaticColumns columns, EncodingStats stats)
{
this.key = key;
@ -412,7 +430,9 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
// ClusteringIndexNamesFilter we'll build from this later does not expect it), but the fact
// we created a builder in the first place will act as a marker that the static row must be
// fetched, even if no other rows are added for this partition.
if (!row.isStatic())
if (row.isStatic())
unresolvedStatic = true;
else
toFetch.add(row.clustering());
}
@ -519,7 +539,7 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
// build the read command taking into account that we could be requesting only in the static row
DataLimits limits = clusterings.isEmpty() ? DataLimits.cqlLimits(1) : DataLimits.NONE;
ClusteringIndexFilter filter = new ClusteringIndexNamesFilter(clusterings, command.isReversed());
ClusteringIndexFilter filter = unresolvedStatic ? command.clusteringIndexFilter(key) : new ClusteringIndexNamesFilter(clusterings, command.isReversed());
SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(),
command.nowInSec(),
command.columnFilter(),

10
src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java
View File

@ -80,9 +80,9 @@ public class RowIteratorMergeListener<E extends Endpoints<E>>
// For each source, record if there is an open range to send as repair, and from where.
private final ClusteringBound<?>[] markerToRepair;
private final ReadRepair readRepair;
private final ReadRepair<E, ?> readRepair;
public RowIteratorMergeListener(DecoratedKey partitionKey, RegularAndStaticColumns columns, boolean isReversed, ReplicaPlan.ForRead<E, ?> readPlan, ReadCommand command, ReadRepair readRepair)
public RowIteratorMergeListener(DecoratedKey partitionKey, RegularAndStaticColumns columns, boolean isReversed, ReplicaPlan.ForRead<E, ?> readPlan, ReadCommand command, ReadRepair<E, ?> readRepair)
{
this.partitionKey = partitionKey;
this.columns = columns;
@ -204,13 +204,13 @@ public class RowIteratorMergeListener<E extends Endpoints<E>>
}
}
public Row onMergedRows(Row merged, Row[] versions)
public void onMergedRows(Row merged, Row[] versions)
{
// If a row was shadowed post merged, it must be by a partition level or range tombstone, and we handle
// those case directly in their respective methods (in other words, it would be inefficient to send a row
// deletion as repair when we know we've already send a partition level or range tombstone that covers it).
if (merged.isEmpty())
return merged;
return;
Rows.diff(diffListener, merged, versions);
for (int i = 0; i < currentRows.length; i++)
@ -222,8 +222,6 @@ public class RowIteratorMergeListener<E extends Endpoints<E>>
}
}
Arrays.fill(currentRows, null);
return merged;
}
private DeletionTime currentDeletion()

78
test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java
View File

@ -47,7 +47,8 @@ import static org.junit.Assert.assertEquals;
public class ReplicaFilteringProtectionTest extends TestBaseImpl
{
private static final int REPLICAS = 2;
private static final int ROWS = 3;
private static final int PARTITIONS = 3;
private static final int ROWS_PER_PARTITION = 3;
private static Cluster cluster;
@ -75,39 +76,39 @@ public class ReplicaFilteringProtectionTest extends TestBaseImpl
public void testMissedUpdatesBelowCachingWarnThreshold()
{
String tableName = "missed_updates_no_warning";
cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int, c int, v text, PRIMARY KEY (k, c))"));
// The warning threshold provided is one more than the total number of rows returned
// to the coordinator from all replicas and therefore should not be triggered.
testMissedUpdates(tableName, REPLICAS * ROWS, Integer.MAX_VALUE, false);
testMissedUpdates(tableName, REPLICAS * ROWS_PER_PARTITION, Integer.MAX_VALUE, false);
}
@Test
public void testMissedUpdatesAboveCachingWarnThreshold()
{
String tableName = "missed_updates_cache_warn";
cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int, c int, v text, PRIMARY KEY (k, c))"));
// The warning threshold provided is one less than the total number of rows returned
// to the coordinator from all replicas and therefore should be triggered but not fail the query.
testMissedUpdates(tableName, REPLICAS * ROWS - 1, Integer.MAX_VALUE, true);
testMissedUpdates(tableName, REPLICAS * ROWS_PER_PARTITION - 1, Integer.MAX_VALUE, true);
}
@Test
public void testMissedUpdatesAroundCachingFailThreshold()
{
String tableName = "missed_updates_cache_fail";
cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int, c int, v text, PRIMARY KEY (k, c))"));
// The failure threshold provided is exactly the total number of rows returned
// to the coordinator from all replicas and therefore should just warn.
testMissedUpdates(tableName, 1, REPLICAS * ROWS, true);
testMissedUpdates(tableName, 1, REPLICAS * ROWS_PER_PARTITION, true);
try
{
// The failure threshold provided is one less than the total number of rows returned
// to the coordinator from all replicas and therefore should fail the query.
testMissedUpdates(tableName, 1, REPLICAS * ROWS - 1, true);
testMissedUpdates(tableName, 1, REPLICAS * ROWS_PER_PARTITION - 1, true);
}
catch (RuntimeException e)
{
@ -122,18 +123,20 @@ public class ReplicaFilteringProtectionTest extends TestBaseImpl
String fullTableName = KEYSPACE + '.' + tableName;
// Case 1: Insert and query rows at ALL to verify base line.
for (int i = 0; i < ROWS; i++)
{
cluster.coordinator(1).execute("INSERT INTO " + fullTableName + "(k, v) VALUES (?, 'old')", ALL, i);
}
// Case 1: Insert and query rows at ALL to verify baseline.
for (int i = 0; i < PARTITIONS; i++)
for (int j = 0; j < ROWS_PER_PARTITION; j++)
cluster.coordinator(1).execute("INSERT INTO " + fullTableName + "(k, c, v) VALUES (?, ?, 'old')", ALL, i, j);
long histogramSampleCount = rowsCachedPerQueryCount(cluster.get(1), tableName);
String query = "SELECT * FROM " + fullTableName + " WHERE v = ? LIMIT ? ALLOW FILTERING";
Object[][] initialRows = cluster.coordinator(1).execute(query, ALL, "old", ROWS);
assertRows(initialRows, row(1, "old"), row(0, "old"), row(2, "old"));
Object[][] initialRows = cluster.coordinator(1).execute(query, ALL, "old", PARTITIONS * ROWS_PER_PARTITION);
assertRows(initialRows,
row(1, 0, "old"), row(1, 1, "old"), row(1, 2, "old"),
row(0, 0, "old"), row(0, 1, "old"), row(0, 2, "old"),
row(2, 0, "old"), row(2, 1, "old"), row(2, 2, "old"));
// Make sure only one sample was recorded for the query.
assertEquals(histogramSampleCount + 1, rowsCachedPerQueryCount(cluster.get(1), tableName));
@ -143,16 +146,15 @@ public class ReplicaFilteringProtectionTest extends TestBaseImpl
// The replica that missed the results creates a mismatch at every row, and we therefore cache a version
// of that row for all replicas.
SimpleQueryResult oldResult = cluster.coordinator(1).executeWithResult(query, ALL, "old", ROWS);
SimpleQueryResult oldResult = cluster.coordinator(1).executeWithResult(query, ALL, "old", PARTITIONS * ROWS_PER_PARTITION);
assertRows(oldResult.toObjectArrays());
verifyWarningState(shouldWarn, oldResult);
// We should have made 3 row "completion" requests.
assertEquals(ROWS, protectionQueryCount(cluster.get(1), tableName));
assertEquals(PARTITIONS, protectionQueryCount(cluster.get(1), tableName));
// In all cases above, the queries should be caching 1 row per partition per replica, but
// 6 for the whole query, given every row is potentially stale.
assertEquals(ROWS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName));
// Queries should be caching 3 rows per partition per replica at any given time.
assertEquals(PARTITIONS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName));
// Make sure only one more sample was recorded for the query.
assertEquals(histogramSampleCount + 2, rowsCachedPerQueryCount(cluster.get(1), tableName));
@ -162,17 +164,20 @@ public class ReplicaFilteringProtectionTest extends TestBaseImpl
// The previous query peforms a blocking read-repair, which removes replica divergence. This
// will only warn, therefore, if the warning threshold is actually below the number of replicas.
// (i.e. The row cache counter is decremented/reset as each partition is consumed.)
SimpleQueryResult newResult = cluster.coordinator(1).executeWithResult(query, ALL, "new", ROWS);
SimpleQueryResult newResult = cluster.coordinator(1).executeWithResult(query, ALL, "new", PARTITIONS * ROWS_PER_PARTITION);
Object[][] newRows = newResult.toObjectArrays();
assertRows(newRows, row(1, "new"), row(0, "new"), row(2, "new"));
assertRows(newRows,
row(1, 0, "new"), row(1, 1, "new"), row(1, 2, "new"),
row(0, 0, "new"), row(0, 1, "new"), row(0, 2, "new"),
row(2, 0, "new"), row(2, 1, "new"), row(2, 2, "new"));
verifyWarningState(warnThreshold < REPLICAS, newResult);
verifyWarningState(warnThreshold < REPLICAS * ROWS_PER_PARTITION, newResult);
// We still sould only have made 3 row "completion" requests, with no replica divergence in the last query.
assertEquals(ROWS, protectionQueryCount(cluster.get(1), tableName));
assertEquals(PARTITIONS, protectionQueryCount(cluster.get(1), tableName));
// With no replica divergence, we only cache a single partition at a time across 2 replicas.
assertEquals(REPLICAS, minRowsCachedPerQuery(cluster.get(1), tableName));
// Queries should be caching 3 rows per partition per replica at any given time.
assertEquals(REPLICAS * ROWS_PER_PARTITION, minRowsCachedPerQuery(cluster.get(1), tableName));
// Make sure only one more sample was recorded for the query.
assertEquals(histogramSampleCount + 3, rowsCachedPerQueryCount(cluster.get(1), tableName));
@ -182,18 +187,20 @@ public class ReplicaFilteringProtectionTest extends TestBaseImpl
updateAllRowsOn(1, fullTableName, "future");
// Another mismatch is introduced, and we once again cache a version of each row during resolution.
SimpleQueryResult futureResult = cluster.coordinator(1).executeWithResult(query, ALL, "future", ROWS);
SimpleQueryResult futureResult = cluster.coordinator(1).executeWithResult(query, ALL, "future", PARTITIONS * ROWS_PER_PARTITION);
Object[][] futureRows = futureResult.toObjectArrays();
assertRows(futureRows, row(1, "future"), row(0, "future"), row(2, "future"));
assertRows(futureRows,
row(1, 0, "future"), row(1, 1, "future"), row(1, 2, "future"),
row(0, 0, "future"), row(0, 1, "future"), row(0, 2, "future"),
row(2, 0, "future"), row(2, 1, "future"), row(2, 2, "future"));
verifyWarningState(shouldWarn, futureResult);
// We sould have made 3 more row "completion" requests.
assertEquals(ROWS * 2, protectionQueryCount(cluster.get(1), tableName));
assertEquals(PARTITIONS * 2, protectionQueryCount(cluster.get(1), tableName));
// In all cases above, the queries should be caching 1 row per partition, but 6 for the
// whole query, given every row is potentially stale.
assertEquals(ROWS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName));
// Queries should be caching 3 rows per partition per replica at any given time.
assertEquals(PARTITIONS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName));
// Make sure only one more sample was recorded for the query.
assertEquals(histogramSampleCount + 4, rowsCachedPerQueryCount(cluster.get(1), tableName));
@ -201,10 +208,9 @@ public class ReplicaFilteringProtectionTest extends TestBaseImpl
private void updateAllRowsOn(int node, String table, String value)
{
for (int i = 0; i < ROWS; i++)
{
cluster.get(node).executeInternal("UPDATE " + table + " SET v = ? WHERE k = ?", value, i);
}
for (int i = 0; i < PARTITIONS; i++)
for (int j = 0; j < ROWS_PER_PARTITION; j++)
cluster.get(node).executeInternal("UPDATE " + table + " SET v = ? WHERE k = ? and c = ?", value, i, j);
}
private void verifyWarningState(boolean shouldWarn, SimpleQueryResult futureResult)

9
test/distributed/org/apache/cassandra/distributed/test/sai/ConcurrencyFactorTest.java
View File

@ -37,6 +37,8 @@ import org.apache.cassandra.distributed.impl.TracingUtil;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.utils.TimeUUID;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.awaitility.Awaitility.await;
public class ConcurrencyFactorTest extends TestBaseImpl
@ -49,7 +51,9 @@ public class ConcurrencyFactorTest extends TestBaseImpl
@Before
public void init() throws IOException
{
cluster = init(Cluster.build(NODES).withTokenSupplier(generateTokenSupplier()).withTokenCount(1).start());
cluster = init(Cluster.build(NODES).withTokenSupplier(generateTokenSupplier())
.withTokenCount(1)
.withConfig(config -> config.with(GOSSIP).with(NETWORK)).start());
}
@After
@ -63,7 +67,8 @@ public class ConcurrencyFactorTest extends TestBaseImpl
{
cluster.schemaChange(String.format("CREATE TABLE %s.%s (pk int, state ascii, gdp bigint, PRIMARY KEY (pk)) WITH compaction = " +
" {'class' : 'SizeTieredCompactionStrategy', 'enabled' : false }", KEYSPACE, SAI_TABLE));
cluster.schemaChange(String.format("CREATE CUSTOM INDEX ON %s.%s (gdp) USING 'StorageAttachedIndex'", KEYSPACE, SAI_TABLE));
cluster.schemaChange(String.format("CREATE INDEX ON %s.%s (gdp) USING 'sai'", KEYSPACE, SAI_TABLE));
SAIUtil.waitForIndexQueryable(cluster, KEYSPACE);
String template = "INSERT INTO %s.%s (pk, state, gdp) VALUES (%s, %s)";
Random rnd = new Random();

494
test/distributed/org/apache/cassandra/distributed/test/sai/PartialUpdateHandlingTest.java Normal file
View File

@ -0,0 +1,494 @@
/*
* 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.distributed.test.sai;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import com.google.common.collect.Lists;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.apache.cassandra.cql3.statements.StatementType;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.index.sai.plan.Expression;
import static org.junit.Assert.assertEquals;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.index.sai.plan.Expression.IndexOperator.EQ;
import static org.apache.cassandra.index.sai.plan.Expression.IndexOperator.RANGE;
/**
* SAI queries, like all filtering queries, must correctly resolve divergent views of row data across replicas. In
* particular, cases where writes do not propagate to all replicas can, if not resolved correctly, lead to consistency
* violations, as rows that should not appear in our results do, while rows that should appear do not.
* <p>
* The variables that affect the behavior of these queries are, at minimum, the following:
* <p>
* 1.) The combination of write & read consistency levels used by the client. Reads at ONE/LOCAL_ONE trivially
* avoid having to resolve data from diverging replicas.
* 2.) Interaction of existing data (if there is any), with partial updates and deletes. The same query that
* erroneously returns stale matches with naïve resolution and partial updates on top of existing data might
* fail to return live matches with partial updates and no previous data.
* 3.) The number of query clauses and their targets. Clauses may target partition keys, clustering keys, static
* columns, and regular columns, and some combinations are more problematic than others.
* 4.) The repaired state of SSTables that participate in the query. A fully repaired on-disk set of SSTables cannot
* produce erroneous results due to split rows (i.e. rows which contain partial updates of different columns
* across different partitions).
* 5.) Whether data resides in SSTables or Memtables. The latter is implicitly unrepaired.
* 6.) Interaction w/ existing mechanisms on the distributed read path that deal with short reads, replica filtering
* protection, etc.
* 7.) The relationship between columns selected and columns restricted by queries. (If coordinator filtering is
* involved at the implementation level, retrieving enough information to do that filtering is important.)
* 8.) The timestamps of partial updates and deletes, especially for single-column queries that might produce
* stale matches if not resolved correctly.
*/
@RunWith(Parameterized.class)
public class PartialUpdateHandlingTest extends TestBaseImpl
{
private static final String TEST_TABLE_NAME = "test_partial_updates";
private static final int PARTITIONS_PER_TEST = 20;
private static final int NODES = 2;
private static Cluster CLUSTER;
@BeforeClass
public static void setUpCluster() throws IOException
{
CLUSTER = init(Cluster.build(NODES).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK)).start());
// All parameterized test scenarios share the same table and attached indexes, but write to different partitions
// that are deleted after each scenario completes.
String createTableDDL = String.format("CREATE TABLE %s.%s (pk int, pk2 int, ck int, s int static, y int static, a int, b int, x int, PRIMARY KEY ((pk, pk2), ck)) WITH read_repair = 'NONE'",
KEYSPACE, TEST_TABLE_NAME);
CLUSTER.schemaChange(createTableDDL);
CLUSTER.disableAutoCompaction(KEYSPACE);
CLUSTER.schemaChange(String.format("CREATE INDEX pk2_idx ON %s.%s(pk2) USING 'sai'", KEYSPACE, TEST_TABLE_NAME));
CLUSTER.schemaChange(String.format("CREATE INDEX ck_idx ON %s.%s(ck) USING 'sai'", KEYSPACE, TEST_TABLE_NAME));
CLUSTER.schemaChange(String.format("CREATE INDEX s_idx ON %s.%s(s) USING 'sai'", KEYSPACE, TEST_TABLE_NAME));
CLUSTER.schemaChange(String.format("CREATE INDEX a_idx ON %s.%s(a) USING 'sai'", KEYSPACE, TEST_TABLE_NAME));
CLUSTER.schemaChange(String.format("CREATE INDEX b_idx ON %s.%s(b) USING 'sai'", KEYSPACE, TEST_TABLE_NAME));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
}
static class Specification
{
final boolean restrictPartitionKey;
final String[] columns;
final boolean existing;
final StatementType partialUpdateType;
final int partitionKey;
final boolean flushPartials;
final Expression.IndexOperator validationMode;
Specification(boolean restrictPartitionKey,
String[] columns,
boolean existing,
StatementType partialUpdateType,
int partitionKey,
boolean flushPartials,
Expression.IndexOperator validationMode)
{
this.restrictPartitionKey = restrictPartitionKey;
this.columns = columns;
this.existing = existing;
this.partialUpdateType = partialUpdateType;
this.partitionKey = partitionKey;
this.flushPartials = flushPartials;
this.validationMode = validationMode;
}
public String[] nonKeyColumns()
{
return Arrays.stream(columns).filter(c -> !c.equals("ck") && !c.equals("pk") && !c.equals("pk2")).toArray(String[]::new);
}
public String tableName()
{
return TEST_TABLE_NAME;
}
@Override
public String toString()
{
return "restrictPartitionKey=" + restrictPartitionKey +
", columns=" + Arrays.toString(columns) +
", existing=" + existing +
", partialUpdateType=" + partialUpdateType +
", partitionKey=" + partitionKey +
", flushPartials=" + flushPartials +
", validationMode=" + validationMode;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Specification that = (Specification) o;
return Arrays.equals(columns, that.columns)
&& existing == that.existing && restrictPartitionKey == that.restrictPartitionKey
&& partialUpdateType == that.partialUpdateType && partitionKey == that.partitionKey && flushPartials == that.flushPartials;
}
@Override
public int hashCode()
{
int result = Objects.hash(existing, restrictPartitionKey, partialUpdateType, partitionKey, flushPartials);
result = 31 * result + Arrays.hashCode(columns);
return result < 0 ? -result : result;
}
}
static class Model
{
final Specification specification;
final List<Map<String, Integer>> previousRows = new ArrayList<>(PARTITIONS_PER_TEST);
final List<Map<String, Integer>> currentRows = new ArrayList<>(PARTITIONS_PER_TEST);
private int nextCellValue = 1;
private int nextTimestamp = 1;
Model(Specification specification)
{
this.specification = specification;
}
public void writeRepairedRows()
{
for (int i = 0; i < PARTITIONS_PER_TEST; i++)
{
StringBuilder insert = new StringBuilder("INSERT INTO ").append(KEYSPACE).append('.').append(specification.tableName());
insert.append("(pk, pk2, ck");
for (Object column : specification.nonKeyColumns())
insert.append(", ").append(column);
int partitionKey = specification.partitionKey + i;
insert.append(") VALUES (").append(partitionKey).append(", ").append(partitionKey).append(", 0");
Map<String, Integer> row = new HashMap<>();
row.put("pk", partitionKey);
row.put("pk2", partitionKey);
row.put("ck", 0);
for (String column : specification.nonKeyColumns())
{
int value = nextCellValue++;
row.put(column, value);
insert.append(", ").append(value);
}
insert.append(") USING TIMESTAMP ").append(nextTimestamp++);
currentRows.add(row);
CLUSTER.coordinator(1).execute(insert.toString(), ConsistencyLevel.ALL);
}
CLUSTER.get(1).nodetoolResult("repair", KEYSPACE).asserts().success();
}
public void writeUnrepairedRows()
{
// Bookmark the model state before partial updates are applied:
for (Map<String, Integer> row : currentRows)
previousRows.add(new HashMap<>(row));
int node = 1;
for (int i = 0; i < PARTITIONS_PER_TEST; i++)
{
for (String column : specification.nonKeyColumns())
{
if (specification.partialUpdateType == StatementType.INSERT)
node = updateReplica(node, column, i);
else if (specification.partialUpdateType == StatementType.DELETE)
node = deleteReplica(node, column, i);
else
throw new IllegalStateException("Partial update must be either INSERT or DELETE");
}
}
}
private int updateReplica(int node, String column, int partitionIndex)
{
int value = nextCellValue++;
int partitionKey = specification.partitionKey + partitionIndex;
if (currentRows.size() > partitionIndex)
{
// A row already exists, so just update it:
currentRows.get(partitionIndex).put(column, value);
}
else
{
// Create a new row with the appropriate cells and add it to the model:
Map<String, Integer> row = new HashMap<>();
row.put("pk", partitionKey);
row.put("pk2", partitionKey);
row.put("ck", 0);
row.put(column, value);
currentRows.add(row);
assert currentRows.size() == partitionIndex + 1 : "Partition " + partitionIndex + " added at position " + (currentRows.size() - 1);
}
String dml = String.format("INSERT INTO %s.%s(pk, pk2, ck, %s) VALUES (?, ?, 0, ?) USING TIMESTAMP %d",
KEYSPACE, specification.tableName(), column, nextTimestamp++);
CLUSTER.get(node).executeInternal(dml, partitionKey, partitionKey, value);
node = nextNode(node);
return node;
}
private int deleteReplica(int node, Object column, int partitionIndex)
{
// Deletion should only happen when we've written the initial set of repaired partitions:
assert currentRows.size() == PARTITIONS_PER_TEST : "Delete requested with only " + currentRows.size() + " model rows";
currentRows.get(partitionIndex).remove((String) column);
int partitionKey = specification.partitionKey + partitionIndex;
String dml = String.format("DELETE %s FROM %s.%s USING TIMESTAMP %d WHERE pk = %d AND pk2 = %d AND ck = 0",
column, KEYSPACE, specification.tableName(), nextTimestamp++, partitionKey, partitionKey);
if (isStatic((String) column))
dml = String.format("DELETE %s FROM %s.%s USING TIMESTAMP %d WHERE pk = %d AND pk2 = %d",
column, KEYSPACE, specification.tableName(), nextTimestamp++, partitionKey, partitionKey);
CLUSTER.get(node).executeInternal(dml);
node = nextNode(node);
return node;
}
private static boolean isStatic(String column)
{
return column.equals("s") || column.equals("y");
}
private static int nextNode(int node)
{
return Math.max(1, (node + 1) % (NODES + 1));
}
public void validateCurrent()
{
Object[][] result = queryWithModel(currentRows);
assert specification.validationMode == EQ || specification.validationMode == RANGE : "Validation mode must be EQ or RANGE";
int resultRowCount = specification.validationMode == EQ ? 1 : PARTITIONS_PER_TEST / 2;
Object[][] expectedRows = new Object[resultRowCount][];
for (int i = 0; i < resultRowCount; i++)
{
int partitionKey = specification.partitionKey + i;
List<Object> expectedRow = Lists.newArrayList(partitionKey, partitionKey, 0);
for (String column : specification.nonKeyColumns())
expectedRow.add(currentRows.get(i).get(column));
expectedRows[i] = expectedRow.toArray();
}
// Sort by partition key value to make the result sets comparable:
Arrays.sort(result, Comparator.comparingInt(row -> (Integer) row[0]));
assertRows(result, expectedRows);
}
public void validatePrevious()
{
// Ensure queries against the previous version of the row no longer match.
assertRows(queryWithModel(previousRows));
}
private Object[][] queryWithModel(List<Map<String, Integer>> modelRows)
{
StringBuilder select = new StringBuilder("SELECT pk, pk2, ck");
for (Object column : specification.nonKeyColumns())
select.append(", ").append(column);
select.append(" FROM ").append(KEYSPACE).append('.').append(specification.tableName()).append(" WHERE ");
ArrayList<String> restricted = Lists.newArrayList(specification.columns);
if (specification.restrictPartitionKey)
{
restricted.add("pk");
restricted.add("pk2");
}
List<String> clauses = new ArrayList<>();
boolean needsAllowFiltering = false;
if (specification.validationMode == EQ)
{
Map<String, Integer> primaryRow = modelRows.get(0);
assertEquals(specification.partitionKey, primaryRow.get("pk").intValue());
for (String column : restricted)
{
clauses.add(column + " = " + primaryRow.get(column));
needsAllowFiltering |= isNotIndexed(column);
}
}
else if (specification.validationMode == RANGE)
{
// Attempt to match the first half of the model's rows...
for (String column : restricted)
{
int min = modelRows.get(0).get(column);
clauses.add(column + " >= " + min);
int max = modelRows.get(PARTITIONS_PER_TEST / 2).get(column);
clauses.add(column + " < " + max);
needsAllowFiltering |= isNotIndexed(column);
}
}
else
throw new IllegalStateException("Validation mode must be EQ or RANGE");
select.append(String.join(" AND ", clauses));
if (needsAllowFiltering)
select.append(" ALLOW FILTERING");
Object[][] fullResult = CLUSTER.coordinator(1).execute(select.toString(), ALL);
Iterator<Object[]> pagedResult = CLUSTER.coordinator(1).executeWithPaging(select.toString(), ALL, 1);
assertRows(pagedResult, fullResult);
return fullResult;
}
private static boolean isNotIndexed(String column)
{
return column.equals("x") || column.equals("y");
}
}
@Parameterized.Parameter
public Specification specification;
@Parameterized.Parameters(name = "{0}")
public static List<Object[]> parameters()
{
List<Object[]> parameters = new ArrayList<>();
// Each test scenario operates over a different set of partition keys. This starting key is the one
// used in partition-restricted queries.
int nextPartitionKey = 0;
for (boolean flushPartials : new boolean[] { false, true })
{
for (boolean restrictPartitionKey : new boolean[] { false, true })
{
for (String[] columns : new String[][] { { "ck", "a" }, { "ck", "s" }, { "s", "a" }, { "a", "b" }, { "s", "x" }, { "s", "y" }, { "a", "x" }, { "a", "y" }, { "a" }, { "s" } })
for (boolean existing : new boolean[] { false, true })
{
parameters.add(new Object[] { new Specification(restrictPartitionKey, columns, existing, StatementType.INSERT, nextPartitionKey, flushPartials, EQ) });
nextPartitionKey += PARTITIONS_PER_TEST;
}
// Deletion scenarios assume existing data.
for (String[] columns : new String[][] { { "s", "a" }, { "a", "b" }, { "s", "x" }, { "a", "x" }, { "a", "y" }, { "a" }, { "s" } })
{
parameters.add(new Object[] { new Specification(restrictPartitionKey, columns, true, StatementType.DELETE, nextPartitionKey, flushPartials, EQ) });
nextPartitionKey += PARTITIONS_PER_TEST;
}
}
// Note that scenarios around indexes on a partition key element only appear here where we neither
// delete nor restrict on partition, as both would be nonsensical.
for (String[] columns : new String[][] { { "pk2", "a" }, { "s", "a" }, { "a", "b" }, { "s", "x" }, { "a", "x" }, { "a", "y" }, { "a" }, { "s" } })
for (boolean existing : new boolean[] { false, true })
{
parameters.add(new Object[]{ new Specification(false, columns, existing, StatementType.INSERT, nextPartitionKey, flushPartials, RANGE) });
nextPartitionKey += PARTITIONS_PER_TEST;
}
// Deletion scenarios assume existing data.
for (String[] columns : new String[][] { { "s", "a" }, { "a", "b" }, { "s", "x" }, { "a", "x" }, { "a", "y" }, { "a" }, { "s" } })
{
parameters.add(new Object[]{ new Specification(false, columns, true, StatementType.DELETE, nextPartitionKey, flushPartials, RANGE) });
nextPartitionKey += PARTITIONS_PER_TEST;
}
}
return parameters;
}
@Test
public void testPartialUpdateResolution()
{
Model model = new Model(specification);
// Write and repair rows that contain values for all columns we might query:
if (specification.existing)
{
model.writeRepairedRows();
model.validateCurrent();
}
// Introduce partial writes that span replicas:
model.writeUnrepairedRows();
if (specification.flushPartials)
// Flushg partial rows from Memtable-attached indexes to SSTable indexes:
CLUSTER.stream().forEach(i -> i.flush(KEYSPACE));
// If we wrote an initial (repaired) version of the row, do negative validation.
// (i.e. Ensure queries that would have initially produced matches no longer do.)
if (specification.existing)
model.validatePrevious();
// In DELETE scenarios, which always have existing data, (negative) validation is already complete by now:
if (specification.partialUpdateType == StatementType.INSERT)
model.validateCurrent();
}
@After
public void truncateTable()
{
CLUSTER.coordinator(1).execute(String.format("TRUNCATE TABLE %s.%s ", KEYSPACE, specification.tableName()), ALL);
}
@AfterClass
public static void shutDownCluster()
{
if (CLUSTER != null)
CLUSTER.close();
}
}

67
test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringProtectionTest.java
View File

@ -1,67 +0,0 @@
/*
* 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.distributed.test.sai;
import java.io.IOException;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
public class ReplicaFilteringProtectionTest extends TestBaseImpl
{
private static final int REPLICAS = 2;
@Test
public void testRFPWithIndexTransformations() throws IOException
{
try (Cluster cluster = init(Cluster.build()
.withNodes(REPLICAS)
.withConfig(config -> config.set("hinted_handoff_enabled", false)
.set("commitlog_sync", "batch")).start()))
{
String tableName = "sai_rfp";
String fullTableName = KEYSPACE + '.' + tableName;
cluster.schemaChange("CREATE TABLE " + fullTableName + " (k int PRIMARY KEY, v text)");
cluster.schemaChange("CREATE CUSTOM INDEX ON " + fullTableName + "(v) USING 'StorageAttachedIndex' " +
"WITH OPTIONS = { 'case_sensitive' : false}");
// both nodes have the old value
cluster.coordinator(1).execute("INSERT INTO " + fullTableName + "(k, v) VALUES (0, 'OLD')", ALL);
String select = "SELECT * FROM " + fullTableName + " WHERE v = 'old'";
Object[][] initialRows = cluster.coordinator(1).execute(select, ALL);
assertRows(initialRows, row(0, "OLD"));
// only one node gets the new value
cluster.get(1).executeInternal("UPDATE " + fullTableName + " SET v = 'new' WHERE k = 0");
// querying by the old value shouldn't return the old surviving row
SimpleQueryResult oldResult = cluster.coordinator(1).executeWithResult(select, ALL);
assertRows(oldResult.toObjectArrays());
}
}
}

227
test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java Normal file
View File

@ -0,0 +1,227 @@
/*
* 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.distributed.test.sai;
import java.io.IOException;
import java.util.Iterator;
import org.assertj.core.api.Assertions;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.shared.AssertUtils;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.index.sai.plan.StorageAttachedIndexQueryPlan;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
/**
* This class contains a small set of hand-crafted tests that document corner cases around how SAI handles
* resolving partial updates, unrepaired data, and post-index filtering.
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-19018">CASSANDRA-19018</a>
*/
public class StrictFilteringTest extends TestBaseImpl
{
private static Cluster CLUSTER;
@BeforeClass
public static void setUpCluster() throws IOException
{
CLUSTER = init(Cluster.build(2).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK)).start());
}
@Test
public void shouldRejectNonStrictIN()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.reject_in (k int PRIMARY KEY, a int, b int) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.reject_in(a) USING 'sai'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.reject_in(b) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert an unrepaired row
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.reject_in(k, a) VALUES (0, 1)"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.reject_in(k, b) VALUES (0, 2)"));
String select = withKeyspace("SELECT * FROM %s.reject_in WHERE a = 1 AND b IN (2, 3) ALLOW FILTERING");
// This should fail, as strict filtering is not allowed:
Assertions.assertThatThrownBy(() -> CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL))
.hasMessageContaining(String.format(StorageAttachedIndexQueryPlan.UNSUPPORTED_NON_STRICT_OPERATOR, Operator.IN));
// Repair fixes the split row, although we still only allow the query when reconciliation is not required:
CLUSTER.get(1).nodetoolResult("repair", KEYSPACE).asserts().success();
assertRows(CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ONE), row(0, 1, 2));
}
@Test
public void testPartialUpdates()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.partial_updates (k int PRIMARY KEY, a int, b int) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates(a) USING 'sai'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates(b) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert a split row
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.partial_updates(k, a) VALUES (0, 1) USING TIMESTAMP 1"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.partial_updates(k, b) VALUES (0, 2) USING TIMESTAMP 2"));
String select = withKeyspace("SELECT * FROM %s.partial_updates WHERE a = 1 AND b = 2");
Object[][] initialRows = CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL);
assertRows(initialRows, row(0, 1, 2));
}
@Test
public void testPartialUpdatesWithDeleteBetween()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.partial_updates_delete_between (k int, c int, a int, b int, x int, y int, PRIMARY KEY (k, c)) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates_delete_between(a) USING 'sai'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates_delete_between(b) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert a split row w/ a range tombstone sandwiched in the middle
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_delete_between(k, c, a, x) VALUES (0, 1, 1, 100) USING TIMESTAMP 1"));
CLUSTER.get(2).executeInternal(withKeyspace("DELETE FROM %s.partial_updates_delete_between USING TIMESTAMP 2 WHERE k = 0 AND c > 0"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_delete_between(k, c, b, y) VALUES (0, 1, 2, 200) USING TIMESTAMP 3"));
String select = withKeyspace("SELECT * FROM %s.partial_updates_delete_between WHERE a = 1 AND b = 2");
Object[][] initialRows = CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL);
assertRows(initialRows);
}
@Test
public void testDanglingUnfilteredColumnWithDeleteBetween()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.dangling_unfiltered_delete_between (k int, c int, a int, b int, x int, PRIMARY KEY (k, c)) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.dangling_unfiltered_delete_between(a) USING 'sai'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.dangling_unfiltered_delete_between(b) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert a split row w/ a range tombstone sandwiched in the middle
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.dangling_unfiltered_delete_between(k, c, a, b, x) VALUES (0, 1, 1, 2, 100) USING TIMESTAMP 1"));
CLUSTER.get(2).executeInternal(withKeyspace("DELETE FROM %s.dangling_unfiltered_delete_between USING TIMESTAMP 2 WHERE k = 0 AND c > 0"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.dangling_unfiltered_delete_between(k, c, a, b) VALUES (0, 1, 1, 2) USING TIMESTAMP 3"));
String select = withKeyspace("SELECT * FROM %s.dangling_unfiltered_delete_between WHERE a = 1 AND b = 2");
Object[][] initialRows = CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL);
assertRows(initialRows, row(0, 1, 1, 2, null));
}
@Test
public void testPartialUpdatesStaticOnly()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.partial_updates_statics (k int, c int, s int static, b int, PRIMARY KEY (k, c)) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates_statics(s) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert a split row
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_statics(k, s) VALUES (0, 2) USING TIMESTAMP 100"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_statics(k, c, s, b) VALUES (0, 0, 1, 2) USING TIMESTAMP 10"));
String select = withKeyspace("SELECT * FROM %s.partial_updates_statics WHERE s = 2");
Object[][] initialRows = CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL);
assertRows(initialRows, row(0, 0, 2, 2));
}
@Test
public void testShortReadWithRegularColumns()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.partial_updates_short_read (k int PRIMARY KEY, a int, b int) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates_short_read(a) USING 'sai'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates_short_read(b) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert a split row
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read(k, a) VALUES (0, 1) USING TIMESTAMP 1"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read(k, b) VALUES (0, 2) USING TIMESTAMP 2"));
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read(k, a, b) VALUES (1, 4, 2) USING TIMESTAMP 3"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read(k, a, b) VALUES (1, 1, 4) USING TIMESTAMP 4"));
String select = withKeyspace("SELECT * FROM %s.partial_updates_short_read WHERE a = 1 AND b = 2 LIMIT 1");
Iterator<Object[]> initialRows = CLUSTER.coordinator(1).executeWithPaging(select, ConsistencyLevel.ALL, 2);
assertRows(initialRows, row(0, 1, 2));
}
@Test
public void testShortReadWithStaticColumn()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.partial_updates_short_read_static (k int, c int, a int, b int static, PRIMARY KEY(k, c)) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates_short_read_static(a) USING 'sai'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.partial_updates_short_read_static(b) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert a split row
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read_static(k, c, a) VALUES (0, 0, 1) USING TIMESTAMP 1"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read_static(k, c, b) VALUES (0, 0, 2) USING TIMESTAMP 2"));
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read_static(k, c, a, b) VALUES (1, 1, 4, 2) USING TIMESTAMP 3"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.partial_updates_short_read_static(k, c, a, b) VALUES (1, 1, 1, 4) USING TIMESTAMP 4"));
String select = withKeyspace("SELECT k, a, b FROM %s.partial_updates_short_read_static WHERE a = 1 AND b = 2 LIMIT 1");
Object[][] rows = CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL);
assertRows(rows, row(0, 1, 2));
}
@Test
public void testTimestampCollision()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.timestamp_collision (k int PRIMARY KEY, a int, b int) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.timestamp_collision(a) USING 'sai'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.timestamp_collision(b) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
// insert a split row
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.timestamp_collision(k, a, b) VALUES (0, 1, 2) USING TIMESTAMP 1"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.timestamp_collision(k, a, b) VALUES (0, 2, 1) USING TIMESTAMP 1"));
String select = withKeyspace("SELECT * FROM %s.timestamp_collision WHERE a = 2 AND b = 2");
Object[][] initialRows = CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL);
assertRows(initialRows, AssertUtils.row(0, 2, 2));
}
@Test
public void testPartialUpdateOnOneColumn()
{
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.one_column (k int PRIMARY KEY, a int) WITH read_repair = 'NONE'"));
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.one_column(a) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.one_column(k, a) VALUES (0, 1) USING TIMESTAMP 1"));
CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s.one_column(k, a) VALUES (0, 100) USING TIMESTAMP 1"));
// resolved via replica filtering protection
Object[][] initialRows = CLUSTER.coordinator(1).execute(withKeyspace("SELECT * FROM %s.one_column WHERE a = 1"), ConsistencyLevel.ALL);
assertRows(initialRows);
}
@AfterClass
public static void shutDownCluster()
{
if (CLUSTER != null)
CLUSTER.close();
}
}

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

@ -169,7 +169,7 @@ public class CleanupTest
while (!cfs.getBuiltIndexes().contains(indexName) && nanoTime() - start < TimeUnit.SECONDS.toNanos(10))
Thread.sleep(10);
RowFilter cf = RowFilter.create();
RowFilter cf = RowFilter.create(true);
cf.add(cdef, Operator.EQ, VALUE);
assertEquals(LOOPS, Util.getAll(Util.cmd(cfs).filterOn("birthdate", Operator.EQ, VALUE).build()).size());

6
test/unit/org/apache/cassandra/db/ReadCommandTest.java
View File

@ -359,7 +359,7 @@ public class ReadCommandTest
List<ByteBuffer> buffers = new ArrayList<>(groups.length);
long nowInSeconds = FBUtilities.nowInSeconds();
ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build();
RowFilter rowFilter = RowFilter.create();
RowFilter rowFilter = RowFilter.create(true);
Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP);
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.with(cfs.metadata().comparator, slice), false);
@ -526,7 +526,7 @@ public class ReadCommandTest
List<ByteBuffer> buffers = new ArrayList<>(groups.length);
long nowInSeconds = FBUtilities.nowInSeconds();
ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build();
RowFilter rowFilter = RowFilter.create();
RowFilter rowFilter = RowFilter.create(true);
Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP);
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(
Slices.with(cfs.metadata().comparator, slice), false);
@ -602,7 +602,7 @@ public class ReadCommandTest
List<ByteBuffer> buffers = new ArrayList<>(groups.length);
long nowInSeconds = FBUtilities.nowInSeconds();
ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build();
RowFilter rowFilter = RowFilter.create();
RowFilter rowFilter = RowFilter.create(true);
Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP);
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(
Slices.with(cfs.metadata().comparator, slice), false);

6
test/unit/org/apache/cassandra/index/sai/cql/intersection/RandomIntersectionTester.java
View File

@ -80,7 +80,8 @@ public abstract class RandomIntersectionTester extends SAIRandomizedTester
Map<Integer, List<TestRow>> testRowMap = buildAndLoadTestRows();
beforeAndAfterFlush(() -> {
for (int queryCount = 0; queryCount < nextInt(10, 100); queryCount++)
int queryCount = nextInt(10, 80);
for (int i = 0; i < queryCount; i++)
{
int pk = testRowMap.keySet().stream().skip(nextInt(0, testRowMap.size())).findFirst().orElseThrow();
int v1 = nextV1();
@ -127,7 +128,8 @@ public abstract class RandomIntersectionTester extends SAIRandomizedTester
Map<Integer, List<TestRow>> testRowMap = buildAndLoadTestRows();
beforeAndAfterFlush(() -> {
for (int queryCount = 0; queryCount < nextInt(10, 100); queryCount++)
int queryCount = nextInt(10, 80);
for (int i = 0; i < queryCount; i++)
{
int v1 = nextV1();
int v2 = nextV2();

86
test/unit/org/apache/cassandra/index/sai/plan/OperationTest.java
View File

@ -112,22 +112,20 @@ public class OperationTest
public void beforeTest()
{
ReadCommand command = PartitionRangeReadCommand.allDataRead(BACKEND.metadata(), FBUtilities.nowInSeconds());
controller = new QueryController(BACKEND,
command,
null,
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)));
controller = new QueryController(BACKEND, command, null, contextWithUnrepairedMatches(command));
command = PartitionRangeReadCommand.allDataRead(CLUSTERING_BACKEND.metadata(), FBUtilities.nowInSeconds());
controllerClustering = new QueryController(CLUSTERING_BACKEND,
command,
null,
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)));
controllerClustering = new QueryController(CLUSTERING_BACKEND, command, null, contextWithUnrepairedMatches(command));
command = PartitionRangeReadCommand.allDataRead(STATIC_BACKEND.metadata(), FBUtilities.nowInSeconds());
controllerStatic = new QueryController(STATIC_BACKEND,
command,
null,
new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS)));
controllerStatic = new QueryController(STATIC_BACKEND, command, null, contextWithUnrepairedMatches(command));
}
private static QueryContext contextWithUnrepairedMatches(ReadCommand command)
{
QueryContext context = new QueryContext(command, DatabaseDescriptor.getRangeRpcTimeout(TimeUnit.MILLISECONDS));
context.hasUnrepairedMatches = true;
return context;
}
@Test
@ -154,34 +152,34 @@ public class OperationTest
final ColumnMetadata age = getColumn(UTF8Type.instance.decompose("age"));
Operation.Node node = new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(5)));
FilterTree filterTree = node.buildFilter(controller);
FilterTree filterTree = node.buildFilter(controller, true);
DecoratedKey key = buildKey("0");
Unfiltered row = buildRow(buildCell(age, instance.decompose(6), System.currentTimeMillis()));
Row staticRow = buildRow(Clustering.STATIC_CLUSTERING);
assertFalse(filterTree.isSatisfiedBy(key, row, staticRow));
assertFalse(filterTree.isSatisfiedBy(key, (Row) row, staticRow));
row = buildRow(buildCell(age, instance.decompose(5), System.currentTimeMillis()));
assertTrue(filterTree.isSatisfiedBy(key, row, staticRow));
assertTrue(filterTree.isSatisfiedBy(key, (Row) row, staticRow));
row = buildRow(buildCell(age, instance.decompose(6), System.currentTimeMillis()));
assertFalse(filterTree.isSatisfiedBy(key, row, staticRow));
assertFalse(filterTree.isSatisfiedBy(key, (Row) row, staticRow));
// range with exclusions - age > 1 AND age <= 10
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GT, Int32Type.instance.decompose(1))));
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.LTE, Int32Type.instance.decompose(10))));
filterTree = node.buildFilter(controller);
filterTree = node.buildFilter(controller, true);
Set<Integer> exclusions = Sets.newHashSet(0, 1, 11);
for (int i = 0; i <= 11; i++)
{
row = buildRow(buildCell(age, instance.decompose(i), System.currentTimeMillis()));
boolean result = filterTree.isSatisfiedBy(key, row, staticRow);
boolean result = filterTree.isSatisfiedBy(key, (Row) row, staticRow);
assertTrue(exclusions.contains(i) != result);
}
@ -191,13 +189,13 @@ public class OperationTest
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GTE, Int32Type.instance.decompose(0))));
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.LT, Int32Type.instance.decompose(10))));
filterTree = node.buildFilter(controller);
filterTree = node.buildFilter(controller, true);
for (int i = 0; i < 10; i++)
{
row = buildRow(buildCell(age, instance.decompose(i), System.currentTimeMillis()));
boolean result = filterTree.isSatisfiedBy(key, row, staticRow);
boolean result = filterTree.isSatisfiedBy(key, (Row) row, staticRow);
assertTrue(result);
}
@ -206,23 +204,27 @@ public class OperationTest
node.add(new Operation.ExpressionNode(new SimpleExpression(timestamp, Operator.GTE, LongType.instance.decompose(10L))));
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(5))));
filterTree = node.buildFilter(controller);
FilterTree filterTreeStrict = node.buildFilter(controller, true);
FilterTree filterTreeNonStrict = node.buildFilter(controller, false);
row = buildRow(buildCell(age, instance.decompose(6), System.currentTimeMillis()),
buildCell(timestamp, LongType.instance.decompose(11L), System.currentTimeMillis()));
long startTime = System.currentTimeMillis();
row = buildRow(buildCell(age, instance.decompose(6), startTime),
buildCell(timestamp, LongType.instance.decompose(11L), startTime + 1));
assertFalse(filterTree.isSatisfiedBy(key, row, staticRow));
assertFalse(filterTreeStrict.isSatisfiedBy(key, (Row) row, staticRow));
assertTrue(filterTreeNonStrict.isSatisfiedBy(key, (Row) row, staticRow)); // matches on timestamp >= 10
row = buildRow(buildCell(age, instance.decompose(5), System.currentTimeMillis()),
buildCell(timestamp, LongType.instance.decompose(22L), System.currentTimeMillis()));
row = buildRow(buildCell(age, instance.decompose(5), startTime + 2),
buildCell(timestamp, LongType.instance.decompose(22L), startTime + 3));
assertTrue(filterTree.isSatisfiedBy(key, row, staticRow));
assertTrue(filterTreeStrict.isSatisfiedBy(key, (Row) row, staticRow));
assertTrue(filterTreeNonStrict.isSatisfiedBy(key, (Row) row, staticRow));
row = buildRow(buildCell(age, instance.decompose(5), System.currentTimeMillis()),
buildCell(timestamp, LongType.instance.decompose(9L), System.currentTimeMillis()));
assertFalse(filterTree.isSatisfiedBy(key, row, staticRow));
row = buildRow(buildCell(age, instance.decompose(5), startTime + 4),
buildCell(timestamp, LongType.instance.decompose(9L), startTime + 5));
assertFalse(filterTreeStrict.isSatisfiedBy(key, (Row) row, staticRow));
assertTrue(filterTreeNonStrict.isSatisfiedBy(key, (Row) row, staticRow)); // matches on age = 5
}
@Test
@ -271,7 +273,7 @@ public class OperationTest
ColumnMetadata score = getColumn(CLUSTERING_BACKEND, UTF8Type.instance.decompose("score"));
DecoratedKey key = buildKey(CLUSTERING_BACKEND, "0");
Unfiltered row = buildRow(Clustering.make(UTF8Type.instance.fromString("US"), Int32Type.instance.decompose(27)),
Row row = buildRow(Clustering.make(UTF8Type.instance.fromString("US"), Int32Type.instance.decompose(27)),
buildCell(height, instance.decompose(182), System.currentTimeMillis()),
buildCell(score, DoubleType.instance.decompose(1.0d), System.currentTimeMillis()));
Row staticRow = buildRow(Clustering.STATIC_CLUSTERING);
@ -280,46 +282,46 @@ public class OperationTest
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(27))));
node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.EQ, Int32Type.instance.decompose(182))));
assertTrue(node.buildFilter(controllerClustering).isSatisfiedBy(key, row, staticRow));
assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.EQ, Int32Type.instance.decompose(28))));
node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.EQ, Int32Type.instance.decompose(182))));
assertFalse(node.buildFilter(controllerClustering).isSatisfiedBy(key, row, staticRow));
assertFalse(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("US"))));
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GTE, Int32Type.instance.decompose(27))));
assertTrue(node.buildFilter(controllerClustering).isSatisfiedBy(key, row, staticRow));
assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("BY"))));
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.GTE, Int32Type.instance.decompose(28))));
assertFalse(node.buildFilter(controllerClustering).isSatisfiedBy(key, row, staticRow));
assertFalse(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("US"))));
node.add(new Operation.ExpressionNode(new SimpleExpression(age, Operator.LTE, Int32Type.instance.decompose(27))));
node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.GTE, Int32Type.instance.decompose(182))));
assertTrue(node.buildFilter(controllerClustering).isSatisfiedBy(key, row, staticRow));
assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(location, Operator.EQ, UTF8Type.instance.decompose("US"))));
node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.GTE, Int32Type.instance.decompose(182))));
node.add(new Operation.ExpressionNode(new SimpleExpression(score, Operator.EQ, DoubleType.instance.decompose(1.0d))));
assertTrue(node.buildFilter(controllerClustering).isSatisfiedBy(key, row, staticRow));
assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(height, Operator.GTE, Int32Type.instance.decompose(182))));
node.add(new Operation.ExpressionNode(new SimpleExpression(score, Operator.EQ, DoubleType.instance.decompose(1.0d))));
assertTrue(node.buildFilter(controllerClustering).isSatisfiedBy(key, row, staticRow));
assertTrue(node.buildFilter(controllerClustering, true).isSatisfiedBy(key, row, staticRow));
}
private Map<Expression.IndexOperator, Expression> convert(Multimap<ColumnMetadata, Expression> expressions)
@ -342,7 +344,7 @@ public class OperationTest
final ColumnMetadata value = getColumn(STATIC_BACKEND, UTF8Type.instance.decompose("value"));
DecoratedKey key = buildKey(STATIC_BACKEND, 0);
Unfiltered row = buildRow(Clustering.make(UTF8Type.instance.fromString("date"), LongType.instance.decompose(20160401L)),
Row row = buildRow(Clustering.make(UTF8Type.instance.fromString("date"), LongType.instance.decompose(20160401L)),
buildCell(value, DoubleType.instance.decompose(24.56), System.currentTimeMillis()));
Row staticRow = buildRow(Clustering.STATIC_CLUSTERING,
buildCell(sensorType, UTF8Type.instance.decompose("TEMPERATURE"), System.currentTimeMillis()));
@ -352,14 +354,14 @@ public class OperationTest
node.add(new Operation.ExpressionNode(new SimpleExpression(sensorType, Operator.EQ, UTF8Type.instance.decompose("TEMPERATURE"))));
node.add(new Operation.ExpressionNode(new SimpleExpression(value, Operator.EQ, DoubleType.instance.decompose(24.56))));
assertTrue(node.buildFilter(controllerStatic).isSatisfiedBy(key, row, staticRow));
assertTrue(node.buildFilter(controllerStatic, true).isSatisfiedBy(key, row, staticRow));
// sensor_type ='TEMPERATURE' AND value = 30
node = new Operation.AndNode();
node.add(new Operation.ExpressionNode(new SimpleExpression(sensorType, Operator.EQ, UTF8Type.instance.decompose("TEMPERATURE"))));
node.add(new Operation.ExpressionNode(new SimpleExpression(value, Operator.EQ, DoubleType.instance.decompose(30.00))));
assertFalse(node.buildFilter(controllerStatic).isSatisfiedBy(key, row, staticRow));
assertFalse(node.buildFilter(controllerStatic, true).isSatisfiedBy(key, row, staticRow));
}
public static TableMetadata.Builder skinnySAITableMetadata(String keyspace, String table)

4
test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java
View File

@ -1494,7 +1494,7 @@ public class SASIIndexTest
ColumnFamilyStore store = loadData(data1, true);
RowFilter filter = RowFilter.create();
RowFilter filter = RowFilter.create(true);
filter.add(store.metadata().getColumn(firstName), Operator.LIKE_CONTAINS, AsciiType.instance.fromString("a"));
ReadCommand command =
@ -2726,7 +2726,7 @@ public class SASIIndexTest
? DataRange.allData(PARTITIONER)
: DataRange.forKeyRange(new Range<>(startKey, PARTITIONER.getMinimumToken().maxKeyBound()));
RowFilter filter = RowFilter.create();
RowFilter filter = RowFilter.create(true);
for (Expression e : expressions)
filter.add(store.metadata().getColumn(e.name), e.op, e.value);