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

Merge branch 'cassandra-3.11' into trunk

This commit is contained in:
Aleksey Yeschenko 2017-08-30 17:32:09 +01:00
parent 76efcc699e 7ad1945ee7
commit 3e4d000c9e
17 changed files with 337 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

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

@ -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;

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

@ -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)

112
src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
View File

@ -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<IndexMetadata> 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<IndexMetadata> 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<PartitionPosition> 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<IndexMetadata> 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);

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

@ -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<IndexMetadata> 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<IndexMetadata> 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.
* <p>
@ -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.
* <p>
@ -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<IndexMetadata> 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<IndexMetadata> 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;
}
}

148
src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
View File

@ -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<Clustering> 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<IndexMetadata> 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);
}
}

8
src/java/org/apache/cassandra/index/SecondaryIndexManager.java
View File

@ -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<Index> searchableIndexes = new HashSet<>();
for (RowFilter.Expression expression : command.rowFilter())
for (RowFilter.Expression expression : rowFilter)
{
if (expression.isCustom())
{

3
src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java
View File

@ -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

3
src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java
View File

@ -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

4
src/java/org/apache/cassandra/service/AbstractReadExecutor.java
View File

@ -91,7 +91,7 @@ public abstract class AbstractReadExecutor
protected void makeDigestRequests(Iterable<InetAddress> endpoints)
{
makeRequests(command.copy().setIsDigestQuery(true), endpoints);
makeRequests(command.copyAsDigestQuery(), endpoints);
}
private void makeRequests(ReadCommand readCommand, Iterable<InetAddress> 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)

8
src/java/org/apache/cassandra/service/pager/PartitionRangeQueryPager.java
View File

@ -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> 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)

25
test/unit/org/apache/cassandra/Util.java
View File

@ -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)

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

@ -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);

29
test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java
View File

@ -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());

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

@ -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());
}

2
test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java
View File

@ -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())
{

2
test/unit/org/apache/cassandra/service/ReadExecutorTest.java
View File

@ -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;
}