diff --git a/CHANGES.txt b/CHANGES.txt index d650f0cf34..9808945582 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java index dc9324e03a..088ac2c94d 100644 --- a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java +++ b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java @@ -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); diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index d0a890a303..f934232d23 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -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. - * + *

* 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. - * + *

* 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); diff --git a/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java b/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java index 7b2c27b40a..30d1eda450 100644 --- a/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java +++ b/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java @@ -43,7 +43,7 @@ public abstract class AbstractReadCommandBuilder protected boolean reversed = false; protected Set columns; - protected final RowFilter filter = RowFilter.create(); + protected final RowFilter filter = RowFilter.create(true); private ClusteringBound lowerClusteringBound; private ClusteringBound upperClusteringBound; diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java index 843ccb9249..7c21c1287a 100644 --- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java +++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java @@ -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)) diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index b8ac0a2229..6d3447a366 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -1011,6 +1011,12 @@ public abstract class ReadCommand extends AbstractReadQuery @VisibleForTesting public static class Serializer implements IVersionedSerializer { + 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 diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 22b67c5802..fc842db925 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -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. diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java index cefd941622..588db7fb46 100644 --- a/src/java/org/apache/cassandra/db/filter/RowFilter.java +++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java @@ -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 +public class RowFilter implements Iterable { 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 expressions; - protected RowFilter(List expressions) + private final boolean needsReconciliation; + + protected RowFilter(List 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 expressions.add(expression); } - public void addUserExpression(UserExpression e) - { - expressions.add(e); - } - public List 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 CASSANDRA-19018 + */ + 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 return false; } - protected abstract Transformation> 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 CASSANDRA-19007 + */ + protected Transformation> filter(TableMetadata metadata, long nowInSec) + { + List partitionLevelExpressions = new ArrayList<>(); + List 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 return withNewExpressions(Collections.emptyList()); } - public RowFilter restrict(Predicate filter) + protected RowFilter withNewExpressions(List expressions) { - return withNewExpressions(expressions.stream().filter(filter).collect(Collectors.toList())); + return new RowFilter(expressions, needsReconciliation); } - protected abstract RowFilter withNewExpressions(List expressions); - public boolean isEmpty() { return expressions.isEmpty(); @@ -312,80 +424,6 @@ public abstract class RowFilter implements Iterable return sb.toString(); } - private static class CQLFilter extends RowFilter - { - static CQLFilter NONE = new CQLFilter(Collections.emptyList()); - - private CQLFilter(List expressions) - { - super(expressions); - } - - protected Transformation> filter(TableMetadata metadata, long nowInSec) - { - List partitionLevelExpressions = new ArrayList<>(); - List 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; - - 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 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 } - 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 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) diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java index 099a47eafd..7ccc6ff970 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java @@ -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 must 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 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 { diff --git a/src/java/org/apache/cassandra/index/sai/QueryContext.java b/src/java/org/apache/cassandra/index/sai/QueryContext.java index c2fba5a018..e319730359 100644 --- a/src/java/org/apache/cassandra/index/sai/QueryContext.java +++ b/src/java/org/apache/cassandra/index/sai/QueryContext.java @@ -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. - * + *

* 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) diff --git a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java index 49d93bd643..d32176a9aa 100644 --- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java +++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndex.java @@ -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 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> groups = groupBySize(nonIndexed, DatabaseDescriptor.getConcurrentIndexBuilders()); - List> futures = new ArrayList<>(); + List> futures = new ArrayList<>(); for (List 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() diff --git a/src/java/org/apache/cassandra/index/sai/disk/IndexSearchResultIterator.java b/src/java/org/apache/cassandra/index/sai/disk/IndexSearchResultIterator.java index 2904bb2585..66c6b253d9 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/IndexSearchResultIterator.java +++ b/src/java/org/apache/cassandra/index/sai/disk/IndexSearchResultIterator.java @@ -61,13 +61,17 @@ public class IndexSearchResultIterator extends KeyRangeIterator public static IndexSearchResultIterator build(Expression expression, Collection sstableIndexes, AbstractBounds keyRange, - QueryContext queryContext) + QueryContext queryContext, + boolean includeMemtables) { - List subIterators = new ArrayList<>(1 + sstableIndexes.size()); + List 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) { diff --git a/src/java/org/apache/cassandra/index/sai/disk/PerColumnIndexWriter.java b/src/java/org/apache/cassandra/index/sai/disk/PerColumnIndexWriter.java index 4ce75cbe08..8f1d657515 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/PerColumnIndexWriter.java +++ b/src/java/org/apache/cassandra/index/sai/disk/PerColumnIndexWriter.java @@ -31,6 +31,8 @@ public interface PerColumnIndexWriter { /** * Adds a row to this index. + *

+ * 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. - * + *

* Note: Implementations should be idempotent, i.e. safe to call multiple times without producing undesirable side-effects. */ void abort(Throwable cause); diff --git a/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java b/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java index 43bb79b91b..2b91bc304b 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java +++ b/src/java/org/apache/cassandra/index/sai/disk/RowMapping.java @@ -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; - } } diff --git a/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java b/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java index 2bf6639e16..0650e9b9d6 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java +++ b/src/java/org/apache/cassandra/index/sai/disk/v1/MemtableIndexWriter.java @@ -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); diff --git a/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeIterator.java b/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeIterator.java index da1b8c2a04..85beafb3c3 100644 --- a/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeIterator.java +++ b/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeIterator.java @@ -54,7 +54,7 @@ public abstract class KeyRangeIterator extends AbstractGuavaIterator { 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; diff --git a/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeUnionIterator.java b/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeUnionIterator.java index f2db621692..fd228b7e6d 100644 --- a/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeUnionIterator.java +++ b/src/java/org/apache/cassandra/index/sai/iterators/KeyRangeUnionIterator.java @@ -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 diff --git a/src/java/org/apache/cassandra/index/sai/plan/FilterTree.java b/src/java/org/apache/cassandra/index/sai/plan/FilterTree.java index 8c1ab0fe03..4107fad2d2 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/FilterTree.java +++ b/src/java/org/apache/cassandra/index/sai/plan/FilterTree.java @@ -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 expressions; protected final List children = new ArrayList<>(); + private final boolean isStrict; + private final QueryContext context; - FilterTree(BooleanOperator operation, ListMultimap expressions) + FilterTree(BooleanOperator baseOperator, ListMultimap 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 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 valueIterator = filter.getIndexTermType().valuesOf(row, now); - result = op.apply(result, collectionMatch(valueIterator, filter)); + Iterator 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; diff --git a/src/java/org/apache/cassandra/index/sai/plan/Operation.java b/src/java/org/apache/cassandra/index/sai/plan/Operation.java index eb56b9c132..c7f313863c 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/Operation.java +++ b/src/java/org/apache/cassandra/index/sai/plan/Operation.java @@ -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 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 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 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) diff --git a/src/java/org/apache/cassandra/index/sai/plan/QueryController.java b/src/java/org/apache/cassandra/index/sai/plan/QueryController.java index d844304812..aecc02377f 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/QueryController.java +++ b/src/java/org/apache/cassandra/index/sai/plan/QueryController.java @@ -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 ranges; private final AbstractBounds 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. *

* 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. *

- * 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. + *

+ * 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 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> 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> 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> queryViewPair : queryView.view) + { + // The initial sizes here reflect little more than an effort to avoid resizing for + // partition-restricted searches w/ LCS: + List repaired = new ArrayList<>(5); + List 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) diff --git a/src/java/org/apache/cassandra/index/sai/plan/QueryViewBuilder.java b/src/java/org/apache/cassandra/index/sai/plan/QueryViewBuilder.java index 0a9691fb3c..81aa2ab1e9 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/QueryViewBuilder.java +++ b/src/java/org/apache/cassandra/index/sai/plan/QueryViewBuilder.java @@ -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>> getQueryView(Collection expressions) { - // first let's determine the most selective expression - Pair> mostSelective = calculateMostSelective(expressions); - List>> 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 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 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> calculateMostSelective(Collection expressions) - { - Expression mostSelectiveExpression = null; - NavigableSet mostSelectiveIndexes = null; - - for (Expression expression : expressions) - { - if (expression.isNotIndexed()) - continue; - - View view = expression.getIndex().view(); - - NavigableSet 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 selectIndexesInRange(Collection 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) diff --git a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexQueryPlan.java b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexQueryPlan.java index ff55011986..e352f7fcdd 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexQueryPlan.java +++ b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexQueryPlan.java @@ -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 indexes; private final boolean isTopK; private StorageAttachedIndexQueryPlan(ColumnFamilyStore cfs, TableQueryMetrics queryMetrics, RowFilter postIndexFilter, - RowFilter filterOperation, + RowFilter indexFilter, ImmutableSet 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 indexes, - RowFilter rowFilter) + RowFilter filter) { ImmutableSet.Builder 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)); } diff --git a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java index a61fb18aca..279f0d82e2 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java +++ b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java @@ -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 clusters = new ArrayList<>(); + List 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 rows; - public PartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator content) + public PartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator 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; } diff --git a/src/java/org/apache/cassandra/service/reads/DataResolver.java b/src/java/org/apache/cassandra/service/reads/DataResolver.java index 2c44160fcd..684d5c9813 100644 --- a/src/java/org/apache/cassandra/service/reads/DataResolver.java +++ b/src/java/org/apache/cassandra/service/reads/DataResolver.java @@ -121,7 +121,7 @@ public class DataResolver, 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, 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, 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, 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, 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 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, P extends ReplicaPlan.ForRead< private PartitionIterator resolveInternal(ResolveContext context, UnfilteredPartitionIterators.MergeListener mergeListener, ResponseProvider responseProvider, - UnaryOperator preCountFilter) + @Nullable UnaryOperator preCountFilter) { int count = context.replicas.size(); List results = new ArrayList<>(count); @@ -306,7 +310,10 @@ public class DataResolver, 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, 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, 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. diff --git a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java index 67815ac925..fa14c38f16 100644 --- a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java +++ b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java @@ -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}). *

- * 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. - *

- * 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 CASSANDRA-8272 + * @see CASSANDRA-8273 + * @see CASSANDRA-15907 + * @see CASSANDRA-19018 */ public class ReplicaFilteringProtection> { @@ -166,13 +170,8 @@ public class ReplicaFilteringProtection> } /** - * 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. - *

- * 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> 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> } @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> 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> private BTreeSet.Builder> 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> // 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> // 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(), diff --git a/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java b/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java index 079080ab6f..353200088f 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java +++ b/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java @@ -80,9 +80,9 @@ public class RowIteratorMergeListener> // 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 readRepair; - public RowIteratorMergeListener(DecoratedKey partitionKey, RegularAndStaticColumns columns, boolean isReversed, ReplicaPlan.ForRead readPlan, ReadCommand command, ReadRepair readRepair) + public RowIteratorMergeListener(DecoratedKey partitionKey, RegularAndStaticColumns columns, boolean isReversed, ReplicaPlan.ForRead readPlan, ReadCommand command, ReadRepair readRepair) { this.partitionKey = partitionKey; this.columns = columns; @@ -204,13 +204,13 @@ public class RowIteratorMergeListener> } } - 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> } } Arrays.fill(currentRows, null); - - return merged; } private DeletionTime currentDeletion() diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java index 099ea053ae..9eab100f2c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java @@ -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) diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/ConcurrencyFactorTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/ConcurrencyFactorTest.java index 6c39591c2f..c846749c0f 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/ConcurrencyFactorTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/ConcurrencyFactorTest.java @@ -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(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/PartialUpdateHandlingTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/PartialUpdateHandlingTest.java new file mode 100644 index 0000000000..b4825adbb0 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/PartialUpdateHandlingTest.java @@ -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. + *

+ * The variables that affect the behavior of these queries are, at minimum, the following: + *

+ * 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> previousRows = new ArrayList<>(PARTITIONS_PER_TEST); + final List> 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 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 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 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 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> 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 restricted = Lists.newArrayList(specification.columns); + + if (specification.restrictPartitionKey) + { + restricted.add("pk"); + restricted.add("pk2"); + } + + List clauses = new ArrayList<>(); + boolean needsAllowFiltering = false; + + if (specification.validationMode == EQ) + { + Map 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 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 parameters() + { + List 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(); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringProtectionTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringProtectionTest.java deleted file mode 100644 index 95ed4c7946..0000000000 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringProtectionTest.java +++ /dev/null @@ -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()); - } - } -} diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java new file mode 100644 index 0000000000..301336f862 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java @@ -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 CASSANDRA-19018 + */ +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 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(); + } +} diff --git a/test/unit/org/apache/cassandra/db/CleanupTest.java b/test/unit/org/apache/cassandra/db/CleanupTest.java index 23fb0a3a62..ce17562fe3 100644 --- a/test/unit/org/apache/cassandra/db/CleanupTest.java +++ b/test/unit/org/apache/cassandra/db/CleanupTest.java @@ -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()); diff --git a/test/unit/org/apache/cassandra/db/ReadCommandTest.java b/test/unit/org/apache/cassandra/db/ReadCommandTest.java index 46eb5e7078..3701ae4193 100644 --- a/test/unit/org/apache/cassandra/db/ReadCommandTest.java +++ b/test/unit/org/apache/cassandra/db/ReadCommandTest.java @@ -359,7 +359,7 @@ public class ReadCommandTest List 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 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 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); diff --git a/test/unit/org/apache/cassandra/index/sai/cql/intersection/RandomIntersectionTester.java b/test/unit/org/apache/cassandra/index/sai/cql/intersection/RandomIntersectionTester.java index 3b1436ae39..721e69e927 100644 --- a/test/unit/org/apache/cassandra/index/sai/cql/intersection/RandomIntersectionTester.java +++ b/test/unit/org/apache/cassandra/index/sai/cql/intersection/RandomIntersectionTester.java @@ -80,7 +80,8 @@ public abstract class RandomIntersectionTester extends SAIRandomizedTester Map> 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> 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(); diff --git a/test/unit/org/apache/cassandra/index/sai/plan/OperationTest.java b/test/unit/org/apache/cassandra/index/sai/plan/OperationTest.java index b2b65454d7..8b0acaaf2b 100644 --- a/test/unit/org/apache/cassandra/index/sai/plan/OperationTest.java +++ b/test/unit/org/apache/cassandra/index/sai/plan/OperationTest.java @@ -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 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 convert(Multimap 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) diff --git a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java index 4163e27069..6cd82b3a23 100644 --- a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java +++ b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java @@ -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);