diff --git a/CHANGES.txt b/CHANGES.txt index 99625f2675..f92eb05dd9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -134,6 +134,7 @@ * Duplicate the buffer before passing it to analyser in SASI operation (CASSANDRA-13512) * Properly evict pstmts from prepared statements cache (CASSANDRA-13641) Merged from 3.0: + * Fix race condition in read command serialization (CASSANDRA-13363) * Fix AssertionError in short read protection (CASSANDRA-13747) * Don't skip corrupted sstables on startup (CASSANDRA-13620) * Fix the merging of cells with different user type versions (CASSANDRA-13776) diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 77eebcf2ef..84fef5e271 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -569,18 +569,10 @@ public class SelectStatement implements CQLStatement if (keyBounds == null) return ReadQuery.EMPTY; - PartitionRangeReadCommand command = new PartitionRangeReadCommand(table, - nowInSec, - columnFilter, - rowFilter, - limit, - new DataRange(keyBounds, clusteringIndexFilter), - Optional.empty()); - // If there's a secondary index that the command can use, have it validate - // the request parameters. Note that as a side effect, if a viable Index is - // identified by the CFS's index manager, it will be cached in the command - // and serialized during distribution to replicas in order to avoid performing - // further lookups. + PartitionRangeReadCommand command = + PartitionRangeReadCommand.create(table, nowInSec, columnFilter, rowFilter, limit, new DataRange(keyBounds, clusteringIndexFilter)); + + // If there's a secondary index that the command can use, have it validate the request parameters. command.maybeValidateIndex(); return command; diff --git a/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java b/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java index 8ced1c7dd1..481e906934 100644 --- a/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java +++ b/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java @@ -336,7 +336,7 @@ public abstract class AbstractReadCommandBuilder else bounds = new ExcludingBounds<>(start, end); - return new PartitionRangeReadCommand(cfs.metadata(), nowInSeconds, makeColumnFilter(), filter, makeLimits(), new DataRange(bounds, makeFilter()), Optional.empty()); + return PartitionRangeReadCommand.create(cfs.metadata(), nowInSeconds, makeColumnFilter(), filter, makeLimits(), new DataRange(bounds, makeFilter())); } static DecoratedKey makeKey(TableMetadata metadata, Object... partitionKey) diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java index da7daa74e3..e88f7fb70c 100644 --- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java +++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java @@ -20,8 +20,8 @@ package org.apache.cassandra.db; import java.io.IOException; import java.util.ArrayList; import java.util.List; -import java.util.Optional; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Iterables; import org.apache.cassandra.schema.TableMetadata; @@ -59,7 +59,7 @@ public class PartitionRangeReadCommand extends ReadCommand private final DataRange dataRange; private int oldestUnrepairedTombstone = Integer.MAX_VALUE; - public PartitionRangeReadCommand(boolean isDigest, + private PartitionRangeReadCommand(boolean isDigest, int digestVersion, TableMetadata metadata, int nowInSec, @@ -67,22 +67,28 @@ public class PartitionRangeReadCommand extends ReadCommand RowFilter rowFilter, DataLimits limits, DataRange dataRange, - Optional index) + IndexMetadata index) { - super(Kind.PARTITION_RANGE, isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits); + super(Kind.PARTITION_RANGE, isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits, index); this.dataRange = dataRange; - this.index = index; } - public PartitionRangeReadCommand(TableMetadata metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DataRange dataRange, - Optional index) + public static PartitionRangeReadCommand create(TableMetadata metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + DataRange dataRange) { - this(false, 0, metadata, nowInSec, columnFilter, rowFilter, limits, dataRange, index); + return new PartitionRangeReadCommand(false, + 0, + metadata, + nowInSec, + columnFilter, + rowFilter, + limits, + dataRange, + findIndex(metadata, rowFilter)); } /** @@ -95,13 +101,15 @@ public class PartitionRangeReadCommand extends ReadCommand */ public static PartitionRangeReadCommand allDataRead(TableMetadata metadata, int nowInSec) { - return new PartitionRangeReadCommand(metadata, + return new PartitionRangeReadCommand(false, + 0, + metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, DataRange.allData(metadata.partitioner), - Optional.empty()); + null); } public DataRange dataRange() @@ -136,24 +144,72 @@ public class PartitionRangeReadCommand extends ReadCommand */ public PartitionRangeReadCommand forSubRange(AbstractBounds range, boolean isRangeContinuation) { - DataRange newRange = dataRange().forSubRange(range); // If we're not a continuation of whatever range we've previously queried, we should ignore the states of the // DataLimits as it's either useless, or misleading. This is particularly important for GROUP BY queries, where // DataLimits.CQLGroupByLimits.GroupByAwareCounter assumes that if GroupingState.hasClustering(), then we're in // the middle of a group, but we can't make that assumption if we query and range "in advance" of where we are // on the ring. - DataLimits newLimits = isRangeContinuation ? limits() : limits().withoutState(); - return new PartitionRangeReadCommand(isDigestQuery(), digestVersion(), metadata(), nowInSec(), columnFilter(), rowFilter(), newLimits, newRange, index); + return new PartitionRangeReadCommand(isDigestQuery(), + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + isRangeContinuation ? limits() : limits().withoutState(), + dataRange().forSubRange(range), + indexMetadata()); } public PartitionRangeReadCommand copy() { - return new PartitionRangeReadCommand(isDigestQuery(), digestVersion(), metadata(), nowInSec(), columnFilter(), rowFilter(), limits(), dataRange(), index); + return new PartitionRangeReadCommand(isDigestQuery(), + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + limits(), + dataRange(), + indexMetadata()); } - public PartitionRangeReadCommand withUpdatedLimit(DataLimits newLimits) + public PartitionRangeReadCommand copyAsDigestQuery() { - return new PartitionRangeReadCommand(metadata(), nowInSec(), columnFilter(), rowFilter(), newLimits, dataRange(), index); + return new PartitionRangeReadCommand(true, + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + limits(), + dataRange(), + indexMetadata()); + } + + public ReadCommand withUpdatedLimit(DataLimits newLimits) + { + return new PartitionRangeReadCommand(isDigestQuery(), + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + newLimits, + dataRange(), + indexMetadata()); + } + + public PartitionRangeReadCommand withUpdatedLimitsAndDataRange(DataLimits newLimits, DataRange newDataRange) + { + return new PartitionRangeReadCommand(isDigestQuery(), + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + newLimits, + newDataRange, + indexMetadata()); } public long getTimeout() @@ -194,7 +250,8 @@ public class PartitionRangeReadCommand extends ReadCommand metric.rangeLatency.addNano(latencyNanos); } - protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController executionController) + @VisibleForTesting + public UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController executionController) { ColumnFamilyStore.ViewFragment view = cfs.select(View.selectLive(dataRange().keyRange())); Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().partitionKeyType)); @@ -356,7 +413,16 @@ public class PartitionRangeReadCommand extends ReadCommand private static class Deserializer extends SelectionDeserializer { - public ReadCommand deserialize(DataInputPlus in, int version, boolean isDigest, int digestVersion, TableMetadata metadata, int nowInSec, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits, Optional index) + public ReadCommand deserialize(DataInputPlus in, + int version, + boolean isDigest, + int digestVersion, + TableMetadata metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + IndexMetadata index) throws IOException { DataRange range = DataRange.serializer.deserialize(in, version, metadata); diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 08224bfca3..e135902481 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -18,9 +18,10 @@ package org.apache.cassandra.db; import java.io.IOException; -import java.util.*; import java.util.function.Predicate; +import javax.annotation.Nullable; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -32,6 +33,7 @@ import org.apache.cassandra.db.partitions.*; import org.apache.cassandra.db.rows.*; import org.apache.cassandra.db.transform.StoppingTransformation; import org.apache.cassandra.db.transform.Transformation; +import org.apache.cassandra.exceptions.UnknownIndexException; import org.apache.cassandra.index.Index; import org.apache.cassandra.index.IndexNotAvailableException; import org.apache.cassandra.io.IVersionedSerializer; @@ -44,7 +46,6 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; -import org.apache.cassandra.exceptions.UnknownIndexException; import org.apache.cassandra.service.ClientWarn; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; @@ -69,23 +70,25 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery private final RowFilter rowFilter; private final DataLimits limits; - // SecondaryIndexManager will attempt to provide the most selective of any available indexes - // during execution. Here we also store an the results of that lookup to repeating it over - // the lifetime of the command. - protected Optional index = Optional.empty(); - - // Flag to indicate whether the index manager has been queried to select an index for this - // command. This is necessary as the result of that lookup may be null, in which case we - // still don't want to repeat it. - private boolean indexManagerQueried = false; - - private boolean isDigestQuery; + private final boolean isDigestQuery; // if a digest query, the version for which the digest is expected. Ignored if not a digest. private int digestVersion; + @Nullable + private final IndexMetadata index; + protected static abstract class SelectionDeserializer { - public abstract ReadCommand deserialize(DataInputPlus in, int version, boolean isDigest, int digestVersion, TableMetadata metadata, int nowInSec, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits, Optional index) throws IOException; + public abstract ReadCommand deserialize(DataInputPlus in, + int version, + boolean isDigest, + int digestVersion, + TableMetadata metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + IndexMetadata index) throws IOException; } protected enum Kind @@ -108,7 +111,8 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery int nowInSec, ColumnFilter columnFilter, RowFilter rowFilter, - DataLimits limits) + DataLimits limits, + IndexMetadata index) { this.kind = kind; this.isDigestQuery = isDigestQuery; @@ -118,6 +122,7 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery this.columnFilter = columnFilter; this.rowFilter = rowFilter; this.limits = limits; + this.index = index; } protected abstract void serializeSelection(DataOutputPlus out, int version) throws IOException; @@ -220,18 +225,6 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery return digestVersion; } - /** - * Sets whether this command should be a digest one or not. - * - * @param isDigestQuery whether the command should be set as a digest one or not. - * @return this read command. - */ - public ReadCommand setIsDigestQuery(boolean isDigestQuery) - { - this.isDigestQuery = isDigestQuery; - return this; - } - /** * Sets the digest version, for when digest for that command is requested. *

@@ -248,6 +241,30 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery return this; } + /** + * Index (metadata) chosen for this query. Can be null. + * + * @return index (metadata) chosen for this query + */ + @Nullable + public IndexMetadata indexMetadata() + { + return index; + } + + /** + * Index instance chosen for this query. Can be null. + * + * @return Index instance chosen for this query. Can be null. + */ + @Nullable + public Index index() + { + return null == index + ? null + : Keyspace.openAndGetStore(metadata).indexManager.getIndex(index); + } + /** * The clustering index filter this command to use for the provided key. *

@@ -268,6 +285,11 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery */ public abstract ReadCommand copy(); + /** + * Returns a copy of this command with isDigestQuery set to true. + */ + public abstract ReadCommand copyAsDigestQuery(); + protected abstract UnfilteredPartitionIterator queryStorage(ColumnFamilyStore cfs, ReadExecutionController executionController); protected abstract int oldestUnrepairedTombstone(); @@ -279,35 +301,32 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery : ReadResponse.createDataResponse(iterator, this); } - public long indexSerializedSize(int version) + long indexSerializedSize(int version) { - if (index.isPresent()) - return IndexMetadata.serializer.serializedSize(index.get(), version); - else - return 0; + return null != index + ? IndexMetadata.serializer.serializedSize(index, version) + : 0; } public Index getIndex(ColumnFamilyStore cfs) { - // if we've already consulted the index manager, and it returned a valid index - // the result should be cached here. - if(index.isPresent()) - return cfs.indexManager.getIndex(index.get()); + return null != index + ? cfs.indexManager.getIndex(index) + : null; + } - // if no cached index is present, but we've already consulted the index manager - // then no registered index is suitable for this command, so just return null. - if (indexManagerQueried) + static IndexMetadata findIndex(TableMetadata table, RowFilter rowFilter) + { + if (table.indexes.isEmpty() || rowFilter.isEmpty()) return null; - // do the lookup, set the flag to indicate so and cache the result if not null - Index selected = cfs.indexManager.getBestIndexFor(this); - indexManagerQueried = true; + ColumnFamilyStore cfs = Keyspace.openAndGetStore(table); - if (selected == null) - return null; + Index index = cfs.indexManager.getBestIndexFor(rowFilter); - index = Optional.of(selected.getIndexMetadata()); - return selected; + return null != index + ? index.getIndexMetadata() + : null; } /** @@ -620,7 +639,7 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery public void serialize(ReadCommand command, DataOutputPlus out, int version) throws IOException { out.writeByte(command.kind.ordinal()); - out.writeByte(digestFlag(command.isDigestQuery()) | indexFlag(command.index.isPresent())); + out.writeByte(digestFlag(command.isDigestQuery()) | indexFlag(null != command.indexMetadata())); if (command.isDigestQuery()) out.writeUnsignedVInt(command.digestVersion()); command.metadata.id.serialize(out); @@ -628,8 +647,8 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery ColumnFilter.serializer.serialize(command.columnFilter(), out, version); RowFilter.serializer.serialize(command.rowFilter(), out, version); DataLimits.serializer.serialize(command.limits(), out, version, command.metadata.comparator); - if (command.index.isPresent()) - IndexMetadata.serializer.serialize(command.index.get(), out, version); + if (null != command.index) + IndexMetadata.serializer.serialize(command.index, out, version); command.serializeSelection(out, version); } @@ -653,19 +672,17 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery int nowInSec = in.readInt(); ColumnFilter columnFilter = ColumnFilter.serializer.deserialize(in, version, metadata); RowFilter rowFilter = RowFilter.serializer.deserialize(in, version, metadata); - DataLimits limits = DataLimits.serializer.deserialize(in, version, metadata.comparator); - Optional index = hasIndex - ? deserializeIndexMetadata(in, version, metadata) - : Optional.empty(); + DataLimits limits = DataLimits.serializer.deserialize(in, version, metadata.comparator); + IndexMetadata index = hasIndex ? deserializeIndexMetadata(in, version, metadata) : null; return kind.selectionDeserializer.deserialize(in, version, isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits, index); } - private Optional deserializeIndexMetadata(DataInputPlus in, int version, TableMetadata metadata) throws IOException + private IndexMetadata deserializeIndexMetadata(DataInputPlus in, int version, TableMetadata metadata) throws IOException { try { - return Optional.of(IndexMetadata.serializer.deserialize(in, version, metadata)); + return IndexMetadata.serializer.deserialize(in, version, metadata); } catch (UnknownIndexException e) { @@ -674,7 +691,7 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery "being fully propagated. Local read will proceed without using the " + "index. Please wait for schema agreement after index creation.", metadata.keyspace, metadata.name, e.indexId); - return Optional.empty(); + return null; } } diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java index bd65535f4a..f4f36d8c2e 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java @@ -22,6 +22,7 @@ import java.nio.ByteBuffer; import java.util.*; import java.util.stream.Collectors; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Iterables; import com.google.common.collect.Sets; @@ -71,22 +72,59 @@ public class SinglePartitionReadCommand extends ReadCommand private int oldestUnrepairedTombstone = Integer.MAX_VALUE; - public SinglePartitionReadCommand(boolean isDigest, - int digestVersion, - TableMetadata metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - ClusteringIndexFilter clusteringIndexFilter) + @VisibleForTesting + protected SinglePartitionReadCommand(boolean isDigest, + int digestVersion, + TableMetadata metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + DecoratedKey partitionKey, + ClusteringIndexFilter clusteringIndexFilter, + IndexMetadata index) { - super(Kind.SINGLE_PARTITION, isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits); + super(Kind.SINGLE_PARTITION, isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits, index); assert partitionKey.getPartitioner() == metadata.partitioner; this.partitionKey = partitionKey; this.clusteringIndexFilter = clusteringIndexFilter; } + /** + * Creates a new read command on a single partition. + * + * @param metadata the table to query. + * @param nowInSec the time in seconds to use are "now" for this query. + * @param columnFilter the column filter to use for the query. + * @param rowFilter the row filter to use for the query. + * @param limits the limits to use for the query. + * @param partitionKey the partition key for the partition to query. + * @param clusteringIndexFilter the clustering index filter to use for the query. + * @param indexMetadata explicitly specified index to use for the query + * + * @return a newly created read command. + */ + public static SinglePartitionReadCommand create(TableMetadata metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + DecoratedKey partitionKey, + ClusteringIndexFilter clusteringIndexFilter, + IndexMetadata indexMetadata) + { + return new SinglePartitionReadCommand(false, + 0, + metadata, + nowInSec, + columnFilter, + rowFilter, + limits, + partitionKey, + clusteringIndexFilter, + indexMetadata); + } + /** * Creates a new read command on a single partition. * @@ -108,7 +146,14 @@ public class SinglePartitionReadCommand extends ReadCommand DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter) { - return new SinglePartitionReadCommand(false, 0, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter); + return create(metadata, + nowInSec, + columnFilter, + rowFilter, + limits, + partitionKey, + clusteringIndexFilter, + findIndex(metadata, rowFilter)); } /** @@ -122,7 +167,11 @@ public class SinglePartitionReadCommand extends ReadCommand * * @return a newly created read command. The returned command will use no row filter and have no limits. */ - public static SinglePartitionReadCommand create(TableMetadata metadata, int nowInSec, DecoratedKey key, ColumnFilter columnFilter, ClusteringIndexFilter filter) + public static SinglePartitionReadCommand create(TableMetadata metadata, + int nowInSec, + DecoratedKey key, + ColumnFilter columnFilter, + ClusteringIndexFilter filter) { return create(metadata, nowInSec, columnFilter, RowFilter.NONE, DataLimits.NONE, key, filter); } @@ -138,7 +187,7 @@ public class SinglePartitionReadCommand extends ReadCommand */ public static SinglePartitionReadCommand fullPartitionRead(TableMetadata metadata, int nowInSec, DecoratedKey key) { - return SinglePartitionReadCommand.create(metadata, nowInSec, key, Slices.ALL); + return create(metadata, nowInSec, key, Slices.ALL); } /** @@ -152,7 +201,7 @@ public class SinglePartitionReadCommand extends ReadCommand */ public static SinglePartitionReadCommand fullPartitionRead(TableMetadata metadata, int nowInSec, ByteBuffer key) { - return SinglePartitionReadCommand.create(metadata, nowInSec, metadata.partitioner.decorateKey(key), Slices.ALL); + return create(metadata, nowInSec, metadata.partitioner.decorateKey(key), Slices.ALL); } /** @@ -185,7 +234,7 @@ public class SinglePartitionReadCommand extends ReadCommand public static SinglePartitionReadCommand create(TableMetadata metadata, int nowInSec, DecoratedKey key, Slices slices) { ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(slices, false); - return SinglePartitionReadCommand.create(metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, key, filter); + return create(metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, key, filter); } /** @@ -218,7 +267,7 @@ public class SinglePartitionReadCommand extends ReadCommand public static SinglePartitionReadCommand create(TableMetadata metadata, int nowInSec, DecoratedKey key, NavigableSet names) { ClusteringIndexNamesFilter filter = new ClusteringIndexNamesFilter(names, false); - return SinglePartitionReadCommand.create(metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, key, filter); + return create(metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, key, filter); } /** @@ -239,7 +288,44 @@ public class SinglePartitionReadCommand extends ReadCommand public SinglePartitionReadCommand copy() { - return new SinglePartitionReadCommand(isDigestQuery(), digestVersion(), metadata(), nowInSec(), columnFilter(), rowFilter(), limits(), partitionKey(), clusteringIndexFilter()); + return new SinglePartitionReadCommand(isDigestQuery(), + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + limits(), + partitionKey(), + clusteringIndexFilter(), + indexMetadata()); + } + + public SinglePartitionReadCommand copyAsDigestQuery() + { + return new SinglePartitionReadCommand(true, + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + limits(), + partitionKey(), + clusteringIndexFilter(), + indexMetadata()); + } + + public SinglePartitionReadCommand withUpdatedLimit(DataLimits newLimits) + { + return new SinglePartitionReadCommand(isDigestQuery(), + digestVersion(), + metadata(), + nowInSec(), + columnFilter(), + rowFilter(), + newLimits, + partitionKey(), + clusteringIndexFilter(), + indexMetadata()); } public DecoratedKey partitionKey() @@ -304,19 +390,6 @@ public class SinglePartitionReadCommand extends ReadCommand lastReturned == null ? clusteringIndexFilter() : clusteringIndexFilter.forPaging(metadata().comparator, lastReturned, false)); } - public SinglePartitionReadCommand withUpdatedLimit(DataLimits newLimits) - { - return new SinglePartitionReadCommand(isDigestQuery(), - digestVersion(), - metadata(), - nowInSec(), - columnFilter(), - rowFilter(), - newLimits, - partitionKey, - clusteringIndexFilter); - } - public PartitionIterator execute(ConsistencyLevel consistency, ClientState clientState, long queryStartNanoTime) throws RequestExecutionException { return StorageProxy.read(Group.one(this), consistency, clientState, queryStartNanoTime); @@ -418,7 +491,7 @@ public class SinglePartitionReadCommand extends ReadCommand final int rowsToCache = metadata().params.caching.rowsPerPartitionToCache(); @SuppressWarnings("resource") // we close on exception or upon closing the result of this method - UnfilteredRowIterator iter = SinglePartitionReadCommand.fullPartitionRead(metadata(), nowInSec(), partitionKey()).queryMemtableAndDisk(cfs, executionController); + UnfilteredRowIterator iter = fullPartitionRead(metadata(), nowInSec(), partitionKey()).queryMemtableAndDisk(cfs, executionController); try { // Use a custom iterator instead of DataLimits to avoid stopping the original iterator @@ -1094,12 +1167,21 @@ public class SinglePartitionReadCommand extends ReadCommand private static class Deserializer extends SelectionDeserializer { - public ReadCommand deserialize(DataInputPlus in, int version, boolean isDigest, int digestVersion, TableMetadata metadata, int nowInSec, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits, Optional index) + public ReadCommand deserialize(DataInputPlus in, + int version, + boolean isDigest, + int digestVersion, + TableMetadata metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + IndexMetadata index) throws IOException { DecoratedKey key = metadata.partitioner.decorateKey(metadata.partitionKeyType.readValue(in, DatabaseDescriptor.getMaxValueSize())); ClusteringIndexFilter filter = ClusteringIndexFilter.serializer.deserialize(in, version, metadata); - return new SinglePartitionReadCommand(isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits, key, filter); + return new SinglePartitionReadCommand(isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits, key, filter, index); } } diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java index bc17e19092..27a9b166ec 100644 --- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java +++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java @@ -986,17 +986,17 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum * cached for future use when obtaining a Searcher, getting the index's underlying CFS for * ReadOrderGroup, or an estimate of the result size from an average index query. * - * @param command ReadCommand to be executed + * @param rowFilter RowFilter of the command to be executed * @return an Index instance, ready to use during execution of the command, or null if none * of the registered indexes can support the command. */ - public Index getBestIndexFor(ReadCommand command) + public Index getBestIndexFor(RowFilter rowFilter) { - if (indexes.isEmpty() || command.rowFilter().isEmpty()) + if (indexes.isEmpty() || rowFilter.isEmpty()) return null; Set searchableIndexes = new HashSet<>(); - for (RowFilter.Expression expression : command.rowFilter()) + for (RowFilter.Expression expression : rowFilter) { if (expression.isCustom()) { diff --git a/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java b/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java index 852ebecc12..f9ae705107 100644 --- a/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java +++ b/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java @@ -160,7 +160,8 @@ public class CompositesSearcher extends CassandraIndexSearcher command.rowFilter(), DataLimits.NONE, partitionKey, - filter); + filter, + null); } @SuppressWarnings("resource") // We close right away if empty, and if it's assign to next it will be called either diff --git a/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java b/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java index 2ab5345c5c..0f4fcb2e61 100644 --- a/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java +++ b/src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java @@ -91,7 +91,8 @@ public class KeysSearcher extends CassandraIndexSearcher command.rowFilter(), DataLimits.NONE, key, - command.clusteringIndexFilter(key)); + command.clusteringIndexFilter(key), + null); @SuppressWarnings("resource") // filterIfStale closes it's iterator if either it materialize it or if it returns null. // Otherwise, we close right away if empty, and if it's assigned to next it will be called either diff --git a/src/java/org/apache/cassandra/service/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/AbstractReadExecutor.java index 956a40a81a..7ff0b9b4ee 100644 --- a/src/java/org/apache/cassandra/service/AbstractReadExecutor.java +++ b/src/java/org/apache/cassandra/service/AbstractReadExecutor.java @@ -91,7 +91,7 @@ public abstract class AbstractReadExecutor protected void makeDigestRequests(Iterable endpoints) { - makeRequests(command.copy().setIsDigestQuery(true), endpoints); + makeRequests(command.copyAsDigestQuery(), endpoints); } private void makeRequests(ReadCommand readCommand, Iterable endpoints) @@ -345,7 +345,7 @@ public abstract class AbstractReadExecutor // Could be waiting on the data, or on enough digests. ReadCommand retryCommand = command; if (handler.resolver.isDataPresent()) - retryCommand = command.copy().setIsDigestQuery(true); + retryCommand = command.copyAsDigestQuery(); InetAddress extraReplica = Iterables.getLast(targetReplicas); if (traceState != null) diff --git a/src/java/org/apache/cassandra/service/pager/PartitionRangeQueryPager.java b/src/java/org/apache/cassandra/service/pager/PartitionRangeQueryPager.java index 4f09f975dd..ba6862d017 100644 --- a/src/java/org/apache/cassandra/service/pager/PartitionRangeQueryPager.java +++ b/src/java/org/apache/cassandra/service/pager/PartitionRangeQueryPager.java @@ -17,15 +17,11 @@ */ package org.apache.cassandra.service.pager; -import java.util.Optional; - import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.DataLimits; import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.dht.*; import org.apache.cassandra.exceptions.RequestExecutionException; -import org.apache.cassandra.index.Index; -import org.apache.cassandra.schema.IndexMetadata; import org.apache.cassandra.transport.ProtocolVersion; /** @@ -106,9 +102,7 @@ public class PartitionRangeQueryPager extends AbstractQueryPager } } - Index index = command.getIndex(Keyspace.openAndGetStore(command.metadata())); - Optional indexMetadata = index != null ? Optional.of(index.getIndexMetadata()) : Optional.empty(); - return new PartitionRangeReadCommand(command.metadata(), command.nowInSec(), command.columnFilter(), command.rowFilter(), limits, pageRange, indexMetadata); + return ((PartitionRangeReadCommand) command).withUpdatedLimitsAndDataRange(limits, pageRange); } protected void recordLast(DecoratedKey key, Row last) diff --git a/test/unit/org/apache/cassandra/Util.java b/test/unit/org/apache/cassandra/Util.java index a01aab676b..93115c058e 100644 --- a/test/unit/org/apache/cassandra/Util.java +++ b/test/unit/org/apache/cassandra/Util.java @@ -655,30 +655,7 @@ public class Util ColumnFamilyStore cfs, ReadExecutionController controller) { - return new InternalPartitionRangeReadCommand(command).queryStorageInternal(cfs, controller); - } - - private static final class InternalPartitionRangeReadCommand extends PartitionRangeReadCommand - { - - private InternalPartitionRangeReadCommand(PartitionRangeReadCommand original) - { - super(original.isDigestQuery(), - original.digestVersion(), - original.metadata(), - original.nowInSec(), - original.columnFilter(), - original.rowFilter(), - original.limits(), - original.dataRange(), - Optional.empty()); - } - - private UnfilteredPartitionIterator queryStorageInternal(ColumnFamilyStore cfs, - ReadExecutionController controller) - { - return queryStorage(cfs, controller); - } + return command.queryStorage(cfs, controller); } public static Closeable markDirectoriesUnwriteable(ColumnFamilyStore cfs) diff --git a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java index eecb55e5be..df5e7cecf2 100644 --- a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java +++ b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java @@ -117,7 +117,7 @@ public class SecondaryIndexTest .filterOn("birthdate", Operator.EQ, 1L) .build(); - Index.Searcher searcher = cfs.indexManager.getBestIndexFor(rc).searcherFor(rc); + Index.Searcher searcher = rc.index().searcherFor(rc); try (ReadExecutionController executionController = rc.executionController(); UnfilteredPartitionIterator pi = searcher.search(executionController)) { @@ -204,7 +204,7 @@ public class SecondaryIndexTest // verify that it's not being indexed under any other value either ReadCommand rc = Util.cmd(cfs).build(); - assertNull(cfs.indexManager.getBestIndexFor(rc)); + assertNull(rc.index()); // resurrect w/ a newer timestamp new RowUpdateBuilder(cfs.metadata(), 2, "k1").clustering("c").add("birthdate", 1L).build().apply();; @@ -222,13 +222,13 @@ public class SecondaryIndexTest // todo - checking the # of index searchers for the command is probably not the best thing to test here RowUpdateBuilder.deleteRow(cfs.metadata(), 3, "k1", "c").applyUnsafe(); rc = Util.cmd(cfs).build(); - assertNull(cfs.indexManager.getBestIndexFor(rc)); + assertNull(rc.index()); // make sure obsolete mutations don't generate an index entry // todo - checking the # of index searchers for the command is probably not the best thing to test here new RowUpdateBuilder(cfs.metadata(), 3, "k1").clustering("c").add("birthdate", 1L).build().apply();; rc = Util.cmd(cfs).build(); - assertNull(cfs.indexManager.getBestIndexFor(rc)); + assertNull(rc.index()); } @Test @@ -536,7 +536,7 @@ public class SecondaryIndexTest ColumnMetadata cdef = cfs.metadata().getColumn(col); ReadCommand rc = Util.cmd(cfs).filterOn(cdef.name.toString(), Operator.EQ, ((AbstractType) cdef.cellValueType()).decompose(val)).build(); - Index.Searcher searcher = cfs.indexManager.getBestIndexFor(rc).searcherFor(rc); + Index.Searcher searcher = rc.index().searcherFor(rc); if (count != 0) assertNotNull(searcher); diff --git a/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java b/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java index 1d4bdb6ca8..f79066b745 100644 --- a/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java +++ b/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java @@ -117,13 +117,13 @@ public class SinglePartitionSliceCommandTest ColumnFilter columnFilter = ColumnFilter.selection(RegularAndStaticColumns.of(s)); ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.NONE, false); - ReadCommand cmd = new SinglePartitionReadCommand(false, MessagingService.VERSION_30, metadata, - FBUtilities.nowInSeconds(), - columnFilter, - RowFilter.NONE, - DataLimits.NONE, - key, - sliceFilter); + ReadCommand cmd = SinglePartitionReadCommand.create(metadata, + FBUtilities.nowInSeconds(), + columnFilter, + RowFilter.NONE, + DataLimits.NONE, + key, + sliceFilter); // check raw iterator for static cell try (ReadExecutionController executionController = cmd.executionController(); UnfilteredPartitionIterator pi = cmd.executeLocally(executionController)) @@ -175,14 +175,13 @@ public class SinglePartitionSliceCommandTest ColumnFilter columnFilter = ColumnFilter.selection(RegularAndStaticColumns.of(s)); Slice slice = Slice.make(ClusteringBound.BOTTOM, ClusteringBound.inclusiveEndOf(ByteBufferUtil.bytes("i1"))); ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.with(metadata.comparator, slice), false); - ReadCommand cmd = new SinglePartitionReadCommand(false, MessagingService.VERSION_30, metadata, - FBUtilities.nowInSeconds(), - columnFilter, - RowFilter.NONE, - DataLimits.NONE, - key, - sliceFilter); - + ReadCommand cmd = SinglePartitionReadCommand.create(metadata, + FBUtilities.nowInSeconds(), + columnFilter, + RowFilter.NONE, + DataLimits.NONE, + key, + sliceFilter); String ret = cmd.toCQLString(); Assert.assertNotNull(ret); Assert.assertFalse(ret.isEmpty()); diff --git a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java index 986e6042fc..406832a552 100644 --- a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java +++ b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java @@ -1306,14 +1306,13 @@ public class SASIIndexTest RowFilter filter = RowFilter.create(); filter.add(store.metadata().getColumn(firstName), Operator.LIKE_CONTAINS, AsciiType.instance.fromString("a")); - ReadCommand command = new PartitionRangeReadCommand(store.metadata(), - FBUtilities.nowInSeconds(), - ColumnFilter.all(store.metadata()), - filter, - DataLimits.NONE, - DataRange.allData(store.metadata().partitioner), - Optional.empty()); - + ReadCommand command = + PartitionRangeReadCommand.create(store.metadata(), + FBUtilities.nowInSeconds(), + ColumnFilter.all(store.metadata()), + filter, + DataLimits.NONE, + DataRange.allData(store.metadata().partitioner)); try { new QueryPlan(store, command, 0).execute(ReadExecutionController.empty()); @@ -2270,13 +2269,13 @@ public class SASIIndexTest ColumnIndex index = ((SASIIndex) store.indexManager.getIndexByName(store.name + "_first_name")).getIndex(); IndexMemtable beforeFlushMemtable = index.getCurrentMemtable(); - PartitionRangeReadCommand command = new PartitionRangeReadCommand(store.metadata(), - FBUtilities.nowInSeconds(), - ColumnFilter.all(store.metadata()), - RowFilter.NONE, - DataLimits.NONE, - DataRange.allData(store.getPartitioner()), - Optional.empty()); + PartitionRangeReadCommand command = + PartitionRangeReadCommand.create(store.metadata(), + FBUtilities.nowInSeconds(), + ColumnFilter.all(store.metadata()), + RowFilter.NONE, + DataLimits.NONE, + DataRange.allData(store.getPartitioner())); QueryController controller = new QueryController(store, command, Integer.MAX_VALUE); org.apache.cassandra.index.sasi.plan.Expression expression = @@ -2410,13 +2409,13 @@ public class SASIIndexTest for (Expression e : expressions) filter.add(store.metadata().getColumn(e.name), e.op, e.value); - ReadCommand command = new PartitionRangeReadCommand(store.metadata(), - FBUtilities.nowInSeconds(), - columnFilter, - filter, - DataLimits.cqlLimits(maxResults), - range, - Optional.empty()); + ReadCommand command = + PartitionRangeReadCommand.create(store.metadata(), + FBUtilities.nowInSeconds(), + columnFilter, + filter, + DataLimits.cqlLimits(maxResults), + range); return command.executeLocally(command.executionController()); } diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java index 64d0252d64..8a7b666965 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java @@ -595,7 +595,7 @@ public class SSTableReaderTest .columns("birthdate") .filterOn("birthdate", Operator.EQ, 1L) .build(); - Index.Searcher searcher = indexedCFS.indexManager.getBestIndexFor(rc).searcherFor(rc); + Index.Searcher searcher = rc.index().searcherFor(rc); assertNotNull(searcher); try (ReadExecutionController executionController = rc.executionController()) { diff --git a/test/unit/org/apache/cassandra/service/ReadExecutorTest.java b/test/unit/org/apache/cassandra/service/ReadExecutorTest.java index fca8ecad5c..7630cc6b65 100644 --- a/test/unit/org/apache/cassandra/service/ReadExecutorTest.java +++ b/test/unit/org/apache/cassandra/service/ReadExecutorTest.java @@ -187,7 +187,7 @@ public class ReadExecutorTest MockSinglePartitionReadCommand(long timeout) { - super(false, 0, cfs.metadata(), 0, null, null, null, Util.dk("ry@n_luvs_teh_y@nk33z"), null); + super(false, 0, cfs.metadata(), 0, null, null, null, Util.dk("ry@n_luvs_teh_y@nk33z"), null, null); this.timeout = timeout; }