diff --git a/CHANGES.txt b/CHANGES.txt index 71330e190d..3b33f5db7a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -5,6 +5,7 @@ 3.0 + * Don't use 'names query' read path for counters (CASSANDRA-10572) * Fix backward compatibility for counters (CASSANDRA-10470) * Remove memory_allocator paramter from cassandra.yaml (CASSANDRA-10581) * Execute the metadata reload task of all registered indexes on CFS::reload (CASSANDRA-10604) diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java index 41aef83c0c..9564005fef 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java +++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java @@ -133,7 +133,7 @@ public class CQL3CasRequest implements CASRequest return conditionColumns; } - public SinglePartitionReadCommand readCommand(int nowInSec) + public SinglePartitionReadCommand readCommand(int nowInSec) { assert !conditions.isEmpty(); Slices.Builder builder = new Slices.Builder(cfm.comparator, conditions.size()); diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 89d5165589..08b85277dc 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -345,7 +345,7 @@ public abstract class ModificationStatement implements CQLStatement throw new InvalidRequestException(String.format("Write operation require a read but consistency %s is not supported on reads", cl)); } - List> commands = new ArrayList<>(partitionKeys.size()); + List commands = new ArrayList<>(partitionKeys.size()); int nowInSec = FBUtilities.nowInSeconds(); for (ByteBuffer key : partitionKeys) commands.add(SinglePartitionReadCommand.create(cfm, @@ -574,7 +574,7 @@ public abstract class ModificationStatement implements CQLStatement { UUID ballot = UUIDGen.getTimeUUIDFromMicros(state.getTimestamp()); - SinglePartitionReadCommand readCommand = request.readCommand(FBUtilities.nowInSeconds()); + SinglePartitionReadCommand readCommand = request.readCommand(FBUtilities.nowInSeconds()); FilteredPartition current; try (ReadExecutionController executionController = readCommand.executionController(); PartitionIterator iter = readCommand.executeInternal(executionController)) diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index a0e0be8960..03dca2d2b4 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -440,7 +440,7 @@ public class SelectStatement implements CQLStatement // Note that we use the total limit for every key, which is potentially inefficient. // However, IN + LIMIT is not a very sensible choice. - List> commands = new ArrayList<>(keys.size()); + List commands = new ArrayList<>(keys.size()); for (ByteBuffer key : keys) { QueryProcessor.validateKey(key); diff --git a/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java b/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java index 9bb89a663f..dab22c70e1 100644 --- a/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java +++ b/src/java/org/apache/cassandra/db/AbstractReadCommandBuilder.java @@ -255,7 +255,7 @@ public abstract class AbstractReadCommandBuilder @Override public ReadCommand build() { - return SinglePartitionSliceCommand.create(cfs.metadata, nowInSeconds, makeColumnFilter(), filter, makeLimits(), partitionKey, makeFilter()); + return SinglePartitionReadCommand.create(cfs.metadata, nowInSeconds, makeColumnFilter(), filter, makeLimits(), partitionKey, makeFilter()); } } diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 6675428681..4f402d4df9 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -1194,9 +1194,9 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery out.writeLong(singleReadCommand.nowInSec() * 1000L); // convert from seconds to millis if (singleReadCommand.clusteringIndexFilter().kind() == ClusteringIndexFilter.Kind.SLICE) - serializeSliceCommand((SinglePartitionSliceCommand) singleReadCommand, out); + serializeSliceCommand(singleReadCommand, out); else - serializeNamesCommand((SinglePartitionNamesCommand) singleReadCommand, out); + serializeNamesCommand(singleReadCommand, out); } public ReadCommand deserialize(DataInputPlus in, int version) throws IOException @@ -1242,14 +1242,14 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery size += TypeSizes.sizeof((long) command.nowInSec()); if (singleReadCommand.clusteringIndexFilter().kind() == ClusteringIndexFilter.Kind.SLICE) - return size + serializedSliceCommandSize((SinglePartitionSliceCommand) singleReadCommand); + return size + serializedSliceCommandSize(singleReadCommand); else - return size + serializedNamesCommandSize((SinglePartitionNamesCommand) singleReadCommand); + return size + serializedNamesCommandSize(singleReadCommand); } - private void serializeNamesCommand(SinglePartitionNamesCommand command, DataOutputPlus out) throws IOException + private void serializeNamesCommand(SinglePartitionReadCommand command, DataOutputPlus out) throws IOException { - serializeNamesFilter(command, command.clusteringIndexFilter(), out); + serializeNamesFilter(command, (ClusteringIndexNamesFilter)command.clusteringIndexFilter(), out); } private static void serializeNamesFilter(ReadCommand command, ClusteringIndexNamesFilter filter, DataOutputPlus out) throws IOException @@ -1307,12 +1307,12 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery return size + TypeSizes.sizeof(true); // countCql3Rows } - private SinglePartitionNamesCommand deserializeNamesCommand(DataInputPlus in, boolean isDigest, CFMetaData metadata, DecoratedKey key, int nowInSeconds, int version) throws IOException + private SinglePartitionReadCommand deserializeNamesCommand(DataInputPlus in, boolean isDigest, CFMetaData metadata, DecoratedKey key, int nowInSeconds, int version) throws IOException { Pair selectionAndFilter = deserializeNamesSelectionAndFilter(in, metadata); // messages from old nodes will expect the thrift format, so always use 'true' for isForThrift - return new SinglePartitionNamesCommand( + return new SinglePartitionReadCommand( isDigest, version, true, metadata, nowInSeconds, selectionAndFilter.left, RowFilter.NONE, DataLimits.NONE, key, selectionAndFilter.right); } @@ -1362,17 +1362,17 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery return Pair.create(selectionBuilder.build(), filter); } - private long serializedNamesCommandSize(SinglePartitionNamesCommand command) + private long serializedNamesCommandSize(SinglePartitionReadCommand command) { - ClusteringIndexNamesFilter filter = command.clusteringIndexFilter(); + ClusteringIndexNamesFilter filter = (ClusteringIndexNamesFilter)command.clusteringIndexFilter(); PartitionColumns columns = command.columnFilter().fetchedColumns(); return serializedNamesFilterSize(filter, command.metadata(), columns); } - private void serializeSliceCommand(SinglePartitionSliceCommand command, DataOutputPlus out) throws IOException + private void serializeSliceCommand(SinglePartitionReadCommand command, DataOutputPlus out) throws IOException { CFMetaData metadata = command.metadata(); - ClusteringIndexSliceFilter filter = command.clusteringIndexFilter(); + ClusteringIndexSliceFilter filter = (ClusteringIndexSliceFilter)command.clusteringIndexFilter(); Slices slices = filter.requestedSlices(); boolean makeStaticSlice = !command.columnFilter().fetchedColumns().statics.isEmpty() && !slices.selects(Clustering.STATIC_CLUSTERING); @@ -1399,7 +1399,7 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery out.writeInt(compositesToGroup); } - private SinglePartitionSliceCommand deserializeSliceCommand(DataInputPlus in, boolean isDigest, CFMetaData metadata, DecoratedKey key, int nowInSeconds, int version) throws IOException + private SinglePartitionReadCommand deserializeSliceCommand(DataInputPlus in, boolean isDigest, CFMetaData metadata, DecoratedKey key, int nowInSeconds, int version) throws IOException { Pair p = deserializeSlicePartitionFilter(in, metadata); ClusteringIndexSliceFilter filter = p.left; @@ -1420,13 +1420,13 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery limits = DataLimits.cqlLimits(count); // messages from old nodes will expect the thrift format, so always use 'true' for isForThrift - return new SinglePartitionSliceCommand(isDigest, version, true, metadata, nowInSeconds, columnFilter, RowFilter.NONE, limits, key, filter); + return new SinglePartitionReadCommand(isDigest, version, true, metadata, nowInSeconds, columnFilter, RowFilter.NONE, limits, key, filter); } - private long serializedSliceCommandSize(SinglePartitionSliceCommand command) + private long serializedSliceCommandSize(SinglePartitionReadCommand command) { CFMetaData metadata = command.metadata(); - ClusteringIndexSliceFilter filter = command.clusteringIndexFilter(); + ClusteringIndexSliceFilter filter = (ClusteringIndexSliceFilter)command.clusteringIndexFilter(); Slices slices = filter.requestedSlices(); boolean makeStaticSlice = !command.columnFilter().fetchedColumns().statics.isEmpty() && !slices.selects(Clustering.STATIC_CLUSTERING); @@ -1600,9 +1600,9 @@ public abstract class ReadCommand extends MonitorableImpl implements ReadQuery if (!shouldConvertNamesToSlice(metadata, command.columnFilter().fetchedColumns())) return command; - ClusteringIndexNamesFilter filter = ((SinglePartitionNamesCommand) command).clusteringIndexFilter(); + ClusteringIndexNamesFilter filter = (ClusteringIndexNamesFilter)command.clusteringIndexFilter(); ClusteringIndexSliceFilter sliceFilter = convertNamesFilterToSliceFilter(filter, metadata); - return new SinglePartitionSliceCommand( + return new SinglePartitionReadCommand( command.isDigestQuery(), command.digestVersion(), command.isForThrift(), metadata, command.nowInSec(), command.columnFilter(), command.rowFilter(), command.limits(), command.partitionKey(), sliceFilter); } diff --git a/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java deleted file mode 100644 index de572d6743..0000000000 --- a/src/java/org/apache/cassandra/db/SinglePartitionNamesCommand.java +++ /dev/null @@ -1,276 +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.db; - -import java.nio.ByteBuffer; -import java.util.*; - -import com.google.common.collect.Sets; - -import org.apache.cassandra.concurrent.Stage; -import org.apache.cassandra.concurrent.StageManager; -import org.apache.cassandra.config.CFMetaData; -import org.apache.cassandra.config.ColumnDefinition; -import org.apache.cassandra.db.lifecycle.SSTableSet; -import org.apache.cassandra.db.lifecycle.View; -import org.apache.cassandra.db.rows.*; -import org.apache.cassandra.db.partitions.*; -import org.apache.cassandra.db.filter.*; -import org.apache.cassandra.io.sstable.format.SSTableReader; -import org.apache.cassandra.metrics.TableMetrics; -import org.apache.cassandra.thrift.ThriftResultsMerger; -import org.apache.cassandra.tracing.Tracing; -import org.apache.cassandra.utils.SearchIterator; -import org.apache.cassandra.utils.btree.BTreeSet; -import org.apache.cassandra.utils.memory.HeapAllocator; - -/** - * General interface for storage-engine read queries. - */ -public class SinglePartitionNamesCommand extends SinglePartitionReadCommand -{ - private int oldestUnrepairedDeletionTime = Integer.MAX_VALUE; - - protected SinglePartitionNamesCommand(boolean isDigest, - int digestVersion, - boolean isForThrift, - CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - ClusteringIndexNamesFilter clusteringIndexFilter) - { - super(isDigest, digestVersion, isForThrift, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter); - } - - public SinglePartitionNamesCommand(CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - ClusteringIndexNamesFilter clusteringIndexFilter) - { - this(false, 0, false, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter); - } - - public SinglePartitionNamesCommand(CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - ByteBuffer key, - ClusteringIndexNamesFilter clusteringIndexFilter) - { - this(metadata, nowInSec, columnFilter, rowFilter, limits, metadata.decorateKey(key), clusteringIndexFilter); - } - - public SinglePartitionNamesCommand copy() - { - return new SinglePartitionNamesCommand(isDigestQuery(), digestVersion(), isForThrift(), metadata(), nowInSec(), columnFilter(), rowFilter(), limits(), partitionKey(), clusteringIndexFilter()); - } - - @Override - protected int oldestUnrepairedTombstone() - { - return oldestUnrepairedDeletionTime; - } - - protected UnfilteredRowIterator queryMemtableAndDiskInternal(ColumnFamilyStore cfs, boolean copyOnHeap) - { - Tracing.trace("Acquiring sstable references"); - ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, partitionKey())); - - ImmutableBTreePartition result = null; - ClusteringIndexNamesFilter filter = clusteringIndexFilter(); - - Tracing.trace("Merging memtable contents"); - for (Memtable memtable : view.memtables) - { - Partition partition = memtable.getPartition(partitionKey()); - if (partition == null) - continue; - - try (UnfilteredRowIterator iter = filter.getUnfilteredRowIterator(columnFilter(), partition)) - { - if (iter.isEmpty()) - continue; - - UnfilteredRowIterator clonedFilter = copyOnHeap - ? UnfilteredRowIterators.cloningIterator(iter, HeapAllocator.instance) - : iter; - result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(clonedFilter, nowInSec()) : clonedFilter, result, false); - } - } - - /* add the SSTables on disk */ - Collections.sort(view.sstables, SSTableReader.maxTimestampComparator); - int sstablesIterated = 0; - - // read sorted sstables - for (SSTableReader sstable : view.sstables) - { - // if we've already seen a partition tombstone with a timestamp greater - // than the most recent update to this sstable, we're done, since the rest of the sstables - // will also be older - if (result != null && sstable.getMaxTimestamp() < result.partitionLevelDeletion().markedForDeleteAt()) - break; - - long currentMaxTs = sstable.getMaxTimestamp(); - filter = reduceFilter(filter, result, currentMaxTs); - if (filter == null) - break; - - Tracing.trace("Merging data from sstable {}", sstable.descriptor.generation); - sstable.incrementReadCount(); - try (UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift()))) - { - if (iter.isEmpty()) - continue; - - sstablesIterated++; - result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter, result, sstable.isRepaired()); - } - } - - cfs.metric.updateSSTableIterated(sstablesIterated); - - if (result == null || result.isEmpty()) - return EmptyIterators.unfilteredRow(metadata(), partitionKey(), false); - - DecoratedKey key = result.partitionKey(); - cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1); - - // "hoist up" the requested data into a more recent sstable - if (sstablesIterated > cfs.getMinimumCompactionThreshold() - && !cfs.isAutoCompactionDisabled() - && cfs.getCompactionStrategyManager().shouldDefragment()) - { - // !!WARNING!! if we stop copying our data to a heap-managed object, - // we will need to track the lifetime of this mutation as well - Tracing.trace("Defragmenting requested data"); - - try (UnfilteredRowIterator iter = result.unfilteredIterator(columnFilter(), Slices.ALL, false)) - { - final Mutation mutation = new Mutation(PartitionUpdate.fromIterator(iter)); - StageManager.getStage(Stage.MUTATION).execute(new Runnable() - { - public void run() - { - // skipping commitlog and index updates is fine since we're just de-fragmenting existing data - Keyspace.open(mutation.getKeyspaceName()).apply(mutation, false, false); - } - }); - } - } - - return withStateTracking(result.unfilteredIterator(columnFilter(), Slices.ALL, clusteringIndexFilter().isReversed())); - } - - private ImmutableBTreePartition add(UnfilteredRowIterator iter, ImmutableBTreePartition result, boolean isRepaired) - { - if (!isRepaired) - oldestUnrepairedDeletionTime = Math.min(oldestUnrepairedDeletionTime, iter.stats().minLocalDeletionTime); - - int maxRows = Math.max(clusteringIndexFilter().requestedRows().size(), 1); - if (result == null) - return ImmutableBTreePartition.create(iter, maxRows); - - try (UnfilteredRowIterator merged = UnfilteredRowIterators.merge(Arrays.asList(iter, result.unfilteredIterator(columnFilter(), Slices.ALL, clusteringIndexFilter().isReversed())), nowInSec())) - { - return ImmutableBTreePartition.create(merged, maxRows); - } - } - - private ClusteringIndexNamesFilter reduceFilter(ClusteringIndexNamesFilter filter, Partition result, long sstableTimestamp) - { - if (result == null) - return filter; - - SearchIterator searchIter = result.searchIterator(columnFilter(), false); - - PartitionColumns columns = columnFilter().fetchedColumns(); - NavigableSet clusterings = filter.requestedRows(); - - // We want to remove rows for which we have values for all requested columns. We have to deal with both static and regular rows. - // TODO: we could also remove a selected column if we've found values for every requested row but we'll leave - // that for later. - - boolean removeStatic = false; - if (!columns.statics.isEmpty()) - { - Row staticRow = searchIter.next(Clustering.STATIC_CLUSTERING); - removeStatic = staticRow != null && canRemoveRow(staticRow, columns.statics, sstableTimestamp); - } - - NavigableSet toRemove = null; - for (Clustering clustering : clusterings) - { - if (!searchIter.hasNext()) - break; - - Row row = searchIter.next(clustering); - if (row == null || !canRemoveRow(row, columns.regulars, sstableTimestamp)) - continue; - - if (toRemove == null) - toRemove = new TreeSet<>(result.metadata().comparator); - toRemove.add(clustering); - } - - if (!removeStatic && toRemove == null) - return filter; - - // Check if we have everything we need - boolean hasNoMoreStatic = columns.statics.isEmpty() || removeStatic; - boolean hasNoMoreClusterings = clusterings.isEmpty() || (toRemove != null && toRemove.size() == clusterings.size()); - if (hasNoMoreStatic && hasNoMoreClusterings) - return null; - - if (toRemove != null) - { - BTreeSet.Builder newClusterings = BTreeSet.builder(result.metadata().comparator); - newClusterings.addAll(Sets.difference(clusterings, toRemove)); - clusterings = newClusterings.build(); - } - return new ClusteringIndexNamesFilter(clusterings, filter.isReversed()); - } - - private boolean canRemoveRow(Row row, Columns requestedColumns, long sstableTimestamp) - { - // We can remove a row if it has data that is more recent that the next sstable to consider for the data that the query - // cares about. And the data we care about is 1) the row timestamp (since every query cares if the row exists or not) - // and 2) the requested columns. - if (row.primaryKeyLivenessInfo().isEmpty() || row.primaryKeyLivenessInfo().timestamp() <= sstableTimestamp) - return false; - - for (ColumnDefinition column : requestedColumns) - { - // We can never be sure we have all of a collection, so never remove rows in that case. - if (column.type.isCollection()) - return false; - - Cell cell = row.getCell(column); - if (cell == null || cell.timestamp() <= sstableTimestamp) - return false; - } - return true; - } -} diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java index 448eaa6e0f..e7c763ebbf 100644 --- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java +++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java @@ -22,17 +22,22 @@ import java.nio.ByteBuffer; import java.util.*; import com.google.common.collect.Iterables; +import com.google.common.collect.Sets; + import org.apache.cassandra.cache.IRowCacheEntry; import org.apache.cassandra.cache.RowCacheKey; import org.apache.cassandra.cache.RowCacheSentinel; +import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ColumnDefinition; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.lifecycle.*; import org.apache.cassandra.db.filter.*; import org.apache.cassandra.db.partitions.*; -import org.apache.cassandra.db.rows.UnfilteredRowIterator; -import org.apache.cassandra.db.rows.UnfilteredRowIterators; +import org.apache.cassandra.db.rows.*; import org.apache.cassandra.exceptions.RequestExecutionException; +import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.metrics.TableMetrics; @@ -43,29 +48,37 @@ import org.apache.cassandra.service.CacheService; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.StorageProxy; import org.apache.cassandra.service.pager.*; +import org.apache.cassandra.thrift.ThriftResultsMerger; import org.apache.cassandra.tracing.Tracing; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.SearchIterator; +import org.apache.cassandra.utils.btree.BTreeSet; import org.apache.cassandra.utils.concurrent.OpOrder; +import org.apache.cassandra.utils.memory.HeapAllocator; + /** * A read command that selects a (part of a) single partition. */ -public abstract class SinglePartitionReadCommand extends ReadCommand +public class SinglePartitionReadCommand extends ReadCommand { protected static final SelectionDeserializer selectionDeserializer = new Deserializer(); private final DecoratedKey partitionKey; - private final F clusteringIndexFilter; + private final ClusteringIndexFilter clusteringIndexFilter; - protected SinglePartitionReadCommand(boolean isDigest, - int digestVersion, - boolean isForThrift, - CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - F clusteringIndexFilter) + private int oldestUnrepairedTombstone = Integer.MAX_VALUE; + + public SinglePartitionReadCommand(boolean isDigest, + int digestVersion, + boolean isForThrift, + CFMetaData metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + DecoratedKey partitionKey, + ClusteringIndexFilter clusteringIndexFilter) { super(Kind.SINGLE_PARTITION, isDigest, digestVersion, isForThrift, metadata, nowInSec, columnFilter, rowFilter, limits); assert partitionKey.getPartitioner() == metadata.partitioner; @@ -86,13 +99,13 @@ public abstract class SinglePartitionReadCommand create(CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - ClusteringIndexFilter clusteringIndexFilter) + public static SinglePartitionReadCommand create(CFMetaData metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + DecoratedKey partitionKey, + ClusteringIndexFilter clusteringIndexFilter) { return create(false, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter); } @@ -111,20 +124,16 @@ public abstract class SinglePartitionReadCommand create(boolean isForThrift, - CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - ClusteringIndexFilter clusteringIndexFilter) + public static SinglePartitionReadCommand create(boolean isForThrift, + CFMetaData metadata, + int nowInSec, + ColumnFilter columnFilter, + RowFilter rowFilter, + DataLimits limits, + DecoratedKey partitionKey, + ClusteringIndexFilter clusteringIndexFilter) { - if (clusteringIndexFilter instanceof ClusteringIndexSliceFilter) - return new SinglePartitionSliceCommand(false, 0, isForThrift, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, (ClusteringIndexSliceFilter) clusteringIndexFilter); - - assert clusteringIndexFilter instanceof ClusteringIndexNamesFilter; - return new SinglePartitionNamesCommand(false, 0, isForThrift, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, (ClusteringIndexNamesFilter) clusteringIndexFilter); + return new SinglePartitionReadCommand(false, 0, isForThrift, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter); } /** @@ -138,7 +147,7 @@ public abstract class SinglePartitionReadCommand create(CFMetaData metadata, int nowInSec, DecoratedKey key, ColumnFilter columnFilter, ClusteringIndexFilter filter) + public static SinglePartitionReadCommand create(CFMetaData metadata, int nowInSec, DecoratedKey key, ColumnFilter columnFilter, ClusteringIndexFilter filter) { return create(metadata, nowInSec, columnFilter, RowFilter.NONE, DataLimits.NONE, key, filter); } @@ -154,7 +163,7 @@ public abstract class SinglePartitionReadCommand iterators = new ArrayList<>(Iterables.size(view.memtables) + view.sstables.size()); + ClusteringIndexFilter filter = clusteringIndexFilter(); + + try + { + for (Memtable memtable : view.memtables) + { + Partition partition = memtable.getPartition(partitionKey()); + if (partition == null) + continue; + + @SuppressWarnings("resource") // 'iter' is added to iterators which is closed on exception, or through the closing of the final merged iterator + UnfilteredRowIterator iter = filter.getUnfilteredRowIterator(columnFilter(), partition); + @SuppressWarnings("resource") // same as above + UnfilteredRowIterator maybeCopied = copyOnHeap ? UnfilteredRowIterators.cloningIterator(iter, HeapAllocator.instance) : iter; + oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, partition.stats().minLocalDeletionTime); + iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(maybeCopied, nowInSec()) : maybeCopied); + } + /* + * We can't eliminate full sstables based on the timestamp of what we've already read like + * in collectTimeOrderedData, but we still want to eliminate sstable whose maxTimestamp < mostRecentTombstone + * we've read. We still rely on the sstable ordering by maxTimestamp since if + * maxTimestamp_s1 > maxTimestamp_s0, + * we're guaranteed that s1 cannot have a row tombstone such that + * timestamp(tombstone) > maxTimestamp_s0 + * since we necessarily have + * timestamp(tombstone) <= maxTimestamp_s1 + * In other words, iterating in maxTimestamp order allow to do our mostRecentPartitionTombstone elimination + * in one pass, and minimize the number of sstables for which we read a partition tombstone. + */ + int sstablesIterated = 0; + Collections.sort(view.sstables, SSTableReader.maxTimestampComparator); + List skippedSSTables = null; + long mostRecentPartitionTombstone = Long.MIN_VALUE; + long minTimestamp = Long.MAX_VALUE; + int nonIntersectingSSTables = 0; + + for (SSTableReader sstable : view.sstables) + { + minTimestamp = Math.min(minTimestamp, sstable.getMinTimestamp()); + // if we've already seen a partition tombstone with a timestamp greater + // than the most recent update to this sstable, we can skip it + if (sstable.getMaxTimestamp() < mostRecentPartitionTombstone) + break; + + if (!filter.shouldInclude(sstable)) + { + nonIntersectingSSTables++; + // sstable contains no tombstone if maxLocalDeletionTime == Integer.MAX_VALUE, so we can safely skip those entirely + if (sstable.getSSTableMetadata().maxLocalDeletionTime != Integer.MAX_VALUE) + { + if (skippedSSTables == null) + skippedSSTables = new ArrayList<>(); + skippedSSTables.add(sstable); + } + continue; + } + + sstable.incrementReadCount(); + @SuppressWarnings("resource") // 'iter' is added to iterators which is closed on exception, or through the closing of the final merged iterator + UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift())); + if (!sstable.isRepaired()) + oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime()); + + iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter); + mostRecentPartitionTombstone = Math.max(mostRecentPartitionTombstone, iter.partitionLevelDeletion().markedForDeleteAt()); + sstablesIterated++; + } + + int includedDueToTombstones = 0; + // Check for partition tombstones in the skipped sstables + if (skippedSSTables != null) + { + for (SSTableReader sstable : skippedSSTables) + { + if (sstable.getMaxTimestamp() <= minTimestamp) + continue; + + sstable.incrementReadCount(); + @SuppressWarnings("resource") // 'iter' is either closed right away, or added to iterators which is close on exception, or through the closing of the final merged iterator + UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift())); + if (iter.partitionLevelDeletion().markedForDeleteAt() > minTimestamp) + { + iterators.add(iter); + if (!sstable.isRepaired()) + oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime()); + includedDueToTombstones++; + sstablesIterated++; + } + else + { + iter.close(); + } + } + } + if (Tracing.isTracing()) + Tracing.trace("Skipped {}/{} non-slice-intersecting sstables, included {} due to tombstones", + nonIntersectingSSTables, view.sstables.size(), includedDueToTombstones); + + cfs.metric.updateSSTableIterated(sstablesIterated); + + if (iterators.isEmpty()) + return EmptyIterators.unfilteredRow(cfs.metadata, partitionKey(), filter.isReversed()); + + Tracing.trace("Merging data from memtables and {} sstables", sstablesIterated); + + @SuppressWarnings("resource") // Closed through the closing of the result of that method. + UnfilteredRowIterator merged = UnfilteredRowIterators.merge(iterators, nowInSec()); + if (!merged.isEmpty()) + { + DecoratedKey key = merged.partitionKey(); + cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1); + } + + return withStateTracking(merged); + } + catch (RuntimeException | Error e) + { + try + { + FBUtilities.closeAll(iterators); + } + catch (Exception suppressed) + { + e.addSuppressed(suppressed); + } + throw e; + } + } + + private boolean queryNeitherCountersNorCollections() + { + for (ColumnDefinition column : columnFilter().fetchedColumns()) + { + if (column.type.isCollection() || column.type.isCounter()) + return false; + } + return true; + } + + /** + * Do a read by querying the memtable(s) first, and then each relevant sstables sequentially by order of the sstable + * max timestamp. + * + * This is used for names query in the hope of only having to query the 1 or 2 most recent query and then knowing nothing + * more recent could be in the older sstables (which we can only guarantee if we know exactly which row we queries, and if + * no collection or counters are included). + * This method assumes the filter is a {@code ClusteringIndexNamesFilter}. + */ + private UnfilteredRowIterator queryMemtableAndSSTablesInTimestampOrder(ColumnFamilyStore cfs, boolean copyOnHeap, ClusteringIndexNamesFilter filter) + { + Tracing.trace("Acquiring sstable references"); + ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, partitionKey())); + + ImmutableBTreePartition result = null; + + Tracing.trace("Merging memtable contents"); + for (Memtable memtable : view.memtables) + { + Partition partition = memtable.getPartition(partitionKey()); + if (partition == null) + continue; + + try (UnfilteredRowIterator iter = filter.getUnfilteredRowIterator(columnFilter(), partition)) + { + if (iter.isEmpty()) + continue; + + UnfilteredRowIterator clonedFilter = copyOnHeap + ? UnfilteredRowIterators.cloningIterator(iter, HeapAllocator.instance) + : iter; + result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(clonedFilter, nowInSec()) : clonedFilter, result, filter, false); + } + } + + /* add the SSTables on disk */ + Collections.sort(view.sstables, SSTableReader.maxTimestampComparator); + int sstablesIterated = 0; + + // read sorted sstables + for (SSTableReader sstable : view.sstables) + { + // if we've already seen a partition tombstone with a timestamp greater + // than the most recent update to this sstable, we're done, since the rest of the sstables + // will also be older + if (result != null && sstable.getMaxTimestamp() < result.partitionLevelDeletion().markedForDeleteAt()) + break; + + long currentMaxTs = sstable.getMaxTimestamp(); + filter = reduceFilter(filter, result, currentMaxTs); + if (filter == null) + break; + + Tracing.trace("Merging data from sstable {}", sstable.descriptor.generation); + sstable.incrementReadCount(); + try (UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift()))) + { + if (iter.isEmpty()) + continue; + + sstablesIterated++; + result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter, result, filter, sstable.isRepaired()); + } + } + + cfs.metric.updateSSTableIterated(sstablesIterated); + + if (result == null || result.isEmpty()) + return EmptyIterators.unfilteredRow(metadata(), partitionKey(), false); + + DecoratedKey key = result.partitionKey(); + cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1); + + // "hoist up" the requested data into a more recent sstable + if (sstablesIterated > cfs.getMinimumCompactionThreshold() + && !cfs.isAutoCompactionDisabled() + && cfs.getCompactionStrategyManager().shouldDefragment()) + { + // !!WARNING!! if we stop copying our data to a heap-managed object, + // we will need to track the lifetime of this mutation as well + Tracing.trace("Defragmenting requested data"); + + try (UnfilteredRowIterator iter = result.unfilteredIterator(columnFilter(), Slices.ALL, false)) + { + final Mutation mutation = new Mutation(PartitionUpdate.fromIterator(iter)); + StageManager.getStage(Stage.MUTATION).execute(new Runnable() + { + public void run() + { + // skipping commitlog and index updates is fine since we're just de-fragmenting existing data + Keyspace.open(mutation.getKeyspaceName()).apply(mutation, false, false); + } + }); + } + } + + return withStateTracking(result.unfilteredIterator(columnFilter(), Slices.ALL, clusteringIndexFilter().isReversed())); + } + + private ImmutableBTreePartition add(UnfilteredRowIterator iter, ImmutableBTreePartition result, ClusteringIndexNamesFilter filter, boolean isRepaired) + { + if (!isRepaired) + oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, iter.stats().minLocalDeletionTime); + + int maxRows = Math.max(filter.requestedRows().size(), 1); + if (result == null) + return ImmutableBTreePartition.create(iter, maxRows); + + try (UnfilteredRowIterator merged = UnfilteredRowIterators.merge(Arrays.asList(iter, result.unfilteredIterator(columnFilter(), Slices.ALL, filter.isReversed())), nowInSec())) + { + return ImmutableBTreePartition.create(merged, maxRows); + } + } + + private ClusteringIndexNamesFilter reduceFilter(ClusteringIndexNamesFilter filter, Partition result, long sstableTimestamp) + { + if (result == null) + return filter; + + SearchIterator searchIter = result.searchIterator(columnFilter(), false); + + PartitionColumns columns = columnFilter().fetchedColumns(); + NavigableSet clusterings = filter.requestedRows(); + + // We want to remove rows for which we have values for all requested columns. We have to deal with both static and regular rows. + // TODO: we could also remove a selected column if we've found values for every requested row but we'll leave + // that for later. + + boolean removeStatic = false; + if (!columns.statics.isEmpty()) + { + Row staticRow = searchIter.next(Clustering.STATIC_CLUSTERING); + removeStatic = staticRow != null && canRemoveRow(staticRow, columns.statics, sstableTimestamp); + } + + NavigableSet toRemove = null; + for (Clustering clustering : clusterings) + { + if (!searchIter.hasNext()) + break; + + Row row = searchIter.next(clustering); + if (row == null || !canRemoveRow(row, columns.regulars, sstableTimestamp)) + continue; + + if (toRemove == null) + toRemove = new TreeSet<>(result.metadata().comparator); + toRemove.add(clustering); + } + + if (!removeStatic && toRemove == null) + return filter; + + // Check if we have everything we need + boolean hasNoMoreStatic = columns.statics.isEmpty() || removeStatic; + boolean hasNoMoreClusterings = clusterings.isEmpty() || (toRemove != null && toRemove.size() == clusterings.size()); + if (hasNoMoreStatic && hasNoMoreClusterings) + return null; + + if (toRemove != null) + { + BTreeSet.Builder newClusterings = BTreeSet.builder(result.metadata().comparator); + newClusterings.addAll(Sets.difference(clusterings, toRemove)); + clusterings = newClusterings.build(); + } + return new ClusteringIndexNamesFilter(clusterings, filter.isReversed()); + } + + private boolean canRemoveRow(Row row, Columns requestedColumns, long sstableTimestamp) + { + // We can remove a row if it has data that is more recent that the next sstable to consider for the data that the query + // cares about. And the data we care about is 1) the row timestamp (since every query cares if the row exists or not) + // and 2) the requested columns. + if (row.primaryKeyLivenessInfo().isEmpty() || row.primaryKeyLivenessInfo().timestamp() <= sstableTimestamp) + return false; + + for (ColumnDefinition column : requestedColumns) + { + Cell cell = row.getCell(column); + if (cell == null || cell.timestamp() <= sstableTimestamp) + return false; + } + return true; + } @Override public String toString() @@ -453,11 +864,11 @@ public abstract class SinglePartitionReadCommand> commands; + public final List commands; private final DataLimits limits; private final int nowInSec; - public Group(List> commands, DataLimits limits) + public Group(List commands, DataLimits limits) { assert !commands.isEmpty(); this.commands = commands; @@ -467,9 +878,9 @@ public abstract class SinglePartitionReadCommand command) + public static Group one(SinglePartitionReadCommand command) { - return new Group(Collections.>singletonList(command), command.limits()); + return new Group(Collections.singletonList(command), command.limits()); } public PartitionIterator execute(ConsistencyLevel consistency, ClientState clientState) throws RequestExecutionException @@ -541,10 +952,7 @@ public abstract class SinglePartitionReadCommand -{ - private int oldestUnrepairedTombstone = Integer.MAX_VALUE; - - public SinglePartitionSliceCommand(boolean isDigest, - int digestVersion, - boolean isForThrift, - CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - ClusteringIndexSliceFilter clusteringIndexFilter) - { - super(isDigest, digestVersion, isForThrift, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter); - } - - public SinglePartitionSliceCommand(CFMetaData metadata, - int nowInSec, - ColumnFilter columnFilter, - RowFilter rowFilter, - DataLimits limits, - DecoratedKey partitionKey, - ClusteringIndexSliceFilter clusteringIndexFilter) - { - this(false, 0, false, metadata, nowInSec, columnFilter, rowFilter, limits, partitionKey, clusteringIndexFilter); - } - - /** - * Creates a new single partition slice command for the provided single slice. - * - * @param metadata the table to query. - * @param nowInSec the time in seconds to use are "now" for this query. - * @param key the partition key for the partition to query. - * @param slice the slice of rows to query. - * - * @return a newly created read command that queries {@code slice} in {@code key}. The returned query will - * query every columns for the table (without limit or row filtering) and be in forward order. - */ - public static SinglePartitionReadCommand create(CFMetaData metadata, int nowInSec, DecoratedKey key, Slice slice) - { - return create(metadata, nowInSec, key, Slices.with(metadata.comparator, slice)); - } - - /** - * Creates a new single partition slice command for the provided slices. - * - * @param metadata the table to query. - * @param nowInSec the time in seconds to use are "now" for this query. - * @param key the partition key for the partition to query. - * @param slices the slices of rows to query. - * - * @return a newly created read command that queries the {@code slices} in {@code key}. The returned query will - * query every columns for the table (without limit or row filtering) and be in forward order. - */ - public static SinglePartitionReadCommand create(CFMetaData metadata, int nowInSec, DecoratedKey key, Slices slices) - { - ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(slices, false); - return new SinglePartitionSliceCommand(metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, key, filter); - } - - /** - * Creates a new single partition slice command for the provided slices. - * - * @param metadata the table to query. - * @param nowInSec the time in seconds to use are "now" for this query. - * @param key the partition key for the partition to query. - * @param slices the slices of rows to query. - * - * @return a newly created read command that queries the {@code slices} in {@code key}. The returned query will - * query every columns for the table (without limit or row filtering) and be in forward order. - */ - public static SinglePartitionReadCommand create(CFMetaData metadata, int nowInSec, ByteBuffer key, Slices slices) - { - return create(metadata, nowInSec, metadata.decorateKey(key), slices); - } - - public SinglePartitionSliceCommand copy() - { - return new SinglePartitionSliceCommand(isDigestQuery(), digestVersion(), isForThrift(), metadata(), nowInSec(), columnFilter(), rowFilter(), limits(), partitionKey(), clusteringIndexFilter()); - } - - @Override - protected int oldestUnrepairedTombstone() - { - return oldestUnrepairedTombstone; - } - - @SuppressWarnings("resource") // Iterators added to iterators list which is closed on exception or with the closing of the result of the method. - protected UnfilteredRowIterator queryMemtableAndDiskInternal(ColumnFamilyStore cfs, boolean copyOnHeap) - { - Tracing.trace("Acquiring sstable references"); - ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, partitionKey())); - - List iterators = new ArrayList<>(Iterables.size(view.memtables) + view.sstables.size()); - ClusteringIndexSliceFilter filter = clusteringIndexFilter(); - - try - { - for (Memtable memtable : view.memtables) - { - Partition partition = memtable.getPartition(partitionKey()); - if (partition == null) - continue; - - UnfilteredRowIterator iter = filter.getUnfilteredRowIterator(columnFilter(), partition); - UnfilteredRowIterator maybeCopied = copyOnHeap ? UnfilteredRowIterators.cloningIterator(iter, HeapAllocator.instance) : iter; - oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, partition.stats().minLocalDeletionTime); - iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(maybeCopied, nowInSec()) : maybeCopied); - } - /* - * We can't eliminate full sstables based on the timestamp of what we've already read like - * in collectTimeOrderedData, but we still want to eliminate sstable whose maxTimestamp < mostRecentTombstone - * we've read. We still rely on the sstable ordering by maxTimestamp since if - * maxTimestamp_s1 > maxTimestamp_s0, - * we're guaranteed that s1 cannot have a row tombstone such that - * timestamp(tombstone) > maxTimestamp_s0 - * since we necessarily have - * timestamp(tombstone) <= maxTimestamp_s1 - * In other words, iterating in maxTimestamp order allow to do our mostRecentPartitionTombstone elimination - * in one pass, and minimize the number of sstables for which we read a partition tombstone. - */ - int sstablesIterated = 0; - Collections.sort(view.sstables, SSTableReader.maxTimestampComparator); - List skippedSSTables = null; - long mostRecentPartitionTombstone = Long.MIN_VALUE; - long minTimestamp = Long.MAX_VALUE; - int nonIntersectingSSTables = 0; - - for (SSTableReader sstable : view.sstables) - { - minTimestamp = Math.min(minTimestamp, sstable.getMinTimestamp()); - // if we've already seen a partition tombstone with a timestamp greater - // than the most recent update to this sstable, we can skip it - if (sstable.getMaxTimestamp() < mostRecentPartitionTombstone) - break; - - if (!filter.shouldInclude(sstable)) - { - nonIntersectingSSTables++; - // sstable contains no tombstone if maxLocalDeletionTime == Integer.MAX_VALUE, so we can safely skip those entirely - if (sstable.getSSTableMetadata().maxLocalDeletionTime != Integer.MAX_VALUE) - { - if (skippedSSTables == null) - skippedSSTables = new ArrayList<>(); - skippedSSTables.add(sstable); - } - continue; - } - - sstable.incrementReadCount(); - UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift())); - if (!sstable.isRepaired()) - oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime()); - - iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter); - mostRecentPartitionTombstone = Math.max(mostRecentPartitionTombstone, iter.partitionLevelDeletion().markedForDeleteAt()); - sstablesIterated++; - } - - int includedDueToTombstones = 0; - // Check for partition tombstones in the skipped sstables - if (skippedSSTables != null) - { - for (SSTableReader sstable : skippedSSTables) - { - if (sstable.getMaxTimestamp() <= minTimestamp) - continue; - - sstable.incrementReadCount(); - - UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift())); - if (iter.partitionLevelDeletion().markedForDeleteAt() > minTimestamp) - { - iterators.add(iter); - if (!sstable.isRepaired()) - oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime()); - includedDueToTombstones++; - sstablesIterated++; - } - else - { - iter.close(); - } - } - } - if (Tracing.isTracing()) - Tracing.trace("Skipped {}/{} non-slice-intersecting sstables, included {} due to tombstones", - nonIntersectingSSTables, view.sstables.size(), includedDueToTombstones); - - cfs.metric.updateSSTableIterated(sstablesIterated); - - if (iterators.isEmpty()) - return EmptyIterators.unfilteredRow(cfs.metadata, partitionKey(), filter.isReversed()); - - Tracing.trace("Merging data from memtables and {} sstables", sstablesIterated); - - UnfilteredRowIterator merged = UnfilteredRowIterators.merge(iterators, nowInSec()); - if (!merged.isEmpty()) - { - DecoratedKey key = merged.partitionKey(); - cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1); - } - - return withStateTracking(merged); - } - catch (RuntimeException | Error e) - { - try - { - FBUtilities.closeAll(iterators); - } - catch (Exception suppressed) - { - e.addSuppressed(suppressed); - } - throw e; - } - } -} diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java index 0b43c19ee5..cc90c40ff9 100644 --- a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java +++ b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java @@ -33,7 +33,7 @@ public abstract class PartitionIterators private PartitionIterators() {} @SuppressWarnings("resource") // The created resources are returned right away - public static RowIterator getOnlyElement(final PartitionIterator iter, SinglePartitionReadCommand command) + public static RowIterator getOnlyElement(final PartitionIterator iter, SinglePartitionReadCommand command) { // If the query has no results, we'll get an empty iterator, but we still // want a RowIterator out of this method, so we return an empty one. diff --git a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java index afa731c500..a3f79814ac 100644 --- a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java +++ b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java @@ -51,7 +51,7 @@ public abstract class UnfilteredPartitionIterators } @SuppressWarnings("resource") // The created resources are returned right away - public static UnfilteredRowIterator getOnlyElement(final UnfilteredPartitionIterator iter, SinglePartitionReadCommand command) + public static UnfilteredRowIterator getOnlyElement(final UnfilteredPartitionIterator iter, SinglePartitionReadCommand command) { // If the query has no results, we'll get an empty iterator, but we still // want a RowIterator out of this method, so we return an empty one. 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 7303cbe2f9..5aaea40ce8 100644 --- a/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java +++ b/src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java @@ -134,13 +134,13 @@ public class CompositesSearcher extends CassandraIndexSearcher // Query the gathered index hits. We still need to filter stale hits from the resulting query. ClusteringIndexNamesFilter filter = new ClusteringIndexNamesFilter(clusterings.build(), false); - SinglePartitionReadCommand dataCmd = new SinglePartitionNamesCommand(index.baseCfs.metadata, - command.nowInSec(), - command.columnFilter(), - command.rowFilter(), - DataLimits.NONE, - partitionKey, - filter); + SinglePartitionReadCommand dataCmd = SinglePartitionReadCommand.create(index.baseCfs.metadata, + command.nowInSec(), + command.columnFilter(), + command.rowFilter(), + DataLimits.NONE, + partitionKey, + filter); @SuppressWarnings("resource") // We close right away if empty, and if it's assign to next it will be called either // by the next caller of next, or through closing this iterator is this come before. UnfilteredRowIterator dataIter = filterStaleEntries(dataCmd.queryMemtableAndDisk(index.baseCfs, diff --git a/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java b/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java index 39066da120..79f9e99a06 100644 --- a/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java +++ b/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java @@ -709,7 +709,7 @@ public final class LegacySchemaMigrator Slices slices = Slices.with(comparator, Slice.make(comparator, typeName)); int nowInSec = FBUtilities.nowInSeconds(); DecoratedKey key = store.metadata.decorateKey(AsciiType.instance.fromString(keyspaceName)); - SinglePartitionReadCommand command = SinglePartitionSliceCommand.create(store.metadata, nowInSec, key, slices); + SinglePartitionReadCommand command = SinglePartitionReadCommand.create(store.metadata, nowInSec, key, slices); try (OpOrder.Group op = store.readOrdering.start(); RowIterator partition = UnfilteredRowIterators.filter(command.queryMemtableAndDisk(store, op), nowInSec)) diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java index 77ae400718..e0c901df22 100644 --- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@ -459,8 +459,8 @@ public final class SchemaKeyspace Slices slices = Slices.with(comparator, Slice.make(comparator, tableName)); int nowInSec = FBUtilities.nowInSeconds(); try (OpOrder.Group op = store.readOrdering.start(); - RowIterator partition = UnfilteredRowIterators.filter(SinglePartitionSliceCommand.create(store.metadata, nowInSec, getSchemaKSKey(keyspaceName), slices) - .queryMemtableAndDisk(store, op), nowInSec)) + RowIterator partition = UnfilteredRowIterators.filter(SinglePartitionReadCommand.create(store.metadata, nowInSec, getSchemaKSKey(keyspaceName), slices) + .queryMemtableAndDisk(store, op), nowInSec)) { return fct.apply(partition); } diff --git a/src/java/org/apache/cassandra/service/DataResolver.java b/src/java/org/apache/cassandra/service/DataResolver.java index 2de02f6401..f3858d7121 100644 --- a/src/java/org/apache/cassandra/service/DataResolver.java +++ b/src/java/org/apache/cassandra/service/DataResolver.java @@ -362,18 +362,18 @@ public class DataResolver extends ResponseResolver DataLimits retryLimits = command.limits().forShortReadRetry(toQuery); ClusteringIndexFilter filter = command.clusteringIndexFilter(partitionKey); ClusteringIndexFilter retryFilter = lastClustering == null ? filter : filter.forPaging(metadata.comparator, lastClustering, false); - SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(), - command.nowInSec(), - command.columnFilter(), - command.rowFilter(), - retryLimits, - partitionKey, - retryFilter); + SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(), + command.nowInSec(), + command.columnFilter(), + command.rowFilter(), + retryLimits, + partitionKey, + retryFilter); return doShortReadRetry(cmd); } - private UnfilteredRowIterator doShortReadRetry(SinglePartitionReadCommand retryCommand) + private UnfilteredRowIterator doShortReadRetry(SinglePartitionReadCommand retryCommand) { DataResolver resolver = new DataResolver(keyspace, retryCommand, ConsistencyLevel.ONE, 1); ReadCallback handler = new ReadCallback(resolver, ConsistencyLevel.ONE, retryCommand, Collections.singletonList(source)); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 424909e629..35b2b61f5e 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -1542,7 +1542,7 @@ public class StorageProxy implements StorageProxyMBean * 4. If the digests (if any) match the data return the data * 5. else carry out read repair by getting data from all the nodes. */ - private static PartitionIterator fetchRows(List> commands, ConsistencyLevel consistencyLevel) + private static PartitionIterator fetchRows(List commands, ConsistencyLevel consistencyLevel) throws UnavailableException, ReadFailureException, ReadTimeoutException { int cmdCount = commands.size(); @@ -1576,14 +1576,14 @@ public class StorageProxy implements StorageProxyMBean private static class SinglePartitionReadLifecycle { - private final SinglePartitionReadCommand command; + private final SinglePartitionReadCommand command; private final AbstractReadExecutor executor; private final ConsistencyLevel consistency; private PartitionIterator result; private ReadCallback repairHandler; - SinglePartitionReadLifecycle(SinglePartitionReadCommand command, ConsistencyLevel consistency) + SinglePartitionReadLifecycle(SinglePartitionReadCommand command, ConsistencyLevel consistency) { this.command = command; this.executor = AbstractReadExecutor.getReadExecutor(command, consistency); diff --git a/src/java/org/apache/cassandra/service/pager/SinglePartitionPager.java b/src/java/org/apache/cassandra/service/pager/SinglePartitionPager.java index 70d4559b36..6f172841cb 100644 --- a/src/java/org/apache/cassandra/service/pager/SinglePartitionPager.java +++ b/src/java/org/apache/cassandra/service/pager/SinglePartitionPager.java @@ -35,11 +35,11 @@ public class SinglePartitionPager extends AbstractQueryPager { private static final Logger logger = LoggerFactory.getLogger(SinglePartitionPager.class); - private final SinglePartitionReadCommand command; + private final SinglePartitionReadCommand command; private volatile PagingState.RowMark lastReturned; - public SinglePartitionPager(SinglePartitionReadCommand command, PagingState state, int protocolVersion) + public SinglePartitionPager(SinglePartitionReadCommand command, PagingState state, int protocolVersion) { super(command, protocolVersion); this.command = command; diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 02c08710de..ee86f9dfd5 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -85,7 +85,7 @@ public class CassandraServer implements Cassandra.Iface return ThriftSessionManager.instance.currentSession(); } - protected PartitionIterator read(List> commands, org.apache.cassandra.db.ConsistencyLevel consistency_level, ClientState cState) + protected PartitionIterator read(List commands, org.apache.cassandra.db.ConsistencyLevel consistency_level, ClientState cState) throws org.apache.cassandra.exceptions.InvalidRequestException, UnavailableException, TimedOutException { try @@ -257,7 +257,7 @@ public class CassandraServer implements Cassandra.Iface : result; } - private Map> getSlice(List> commands, boolean subColumnsOnly, int cellLimit, org.apache.cassandra.db.ConsistencyLevel consistency_level, ClientState cState) + private Map> getSlice(List commands, boolean subColumnsOnly, int cellLimit, org.apache.cassandra.db.ConsistencyLevel consistency_level, ClientState cState) throws org.apache.cassandra.exceptions.InvalidRequestException, UnavailableException, TimedOutException { try (PartitionIterator results = read(commands, consistency_level, cState)) @@ -551,7 +551,7 @@ public class CassandraServer implements Cassandra.Iface org.apache.cassandra.db.ConsistencyLevel consistencyLevel = ThriftConversion.fromThrift(consistency_level); consistencyLevel.validateForRead(keyspace); - List> commands = new ArrayList<>(keys.size()); + List commands = new ArrayList<>(keys.size()); ColumnFilter columnFilter = makeColumnFilter(metadata, column_parent, predicate); ClusteringIndexFilter filter = toInternalFilter(metadata, column_parent, predicate); DataLimits limits = getLimits(1, metadata.isSuper() && !column_parent.isSetSuper_column(), predicate); @@ -641,7 +641,7 @@ public class CassandraServer implements Cassandra.Iface } DecoratedKey dk = metadata.decorateKey(key); - SinglePartitionReadCommand command = SinglePartitionReadCommand.create(true, metadata, FBUtilities.nowInSeconds(), columns, RowFilter.NONE, DataLimits.NONE, dk, filter); + SinglePartitionReadCommand command = SinglePartitionReadCommand.create(true, metadata, FBUtilities.nowInSeconds(), columns, RowFilter.NONE, DataLimits.NONE, dk, filter); try (RowIterator result = PartitionIterators.getOnlyElement(read(Arrays.asList(command), consistencyLevel, cState), command)) { @@ -2437,8 +2437,8 @@ public class CassandraServer implements Cassandra.Iface ThriftValidation.validateKey(metadata, request.key); DecoratedKey dk = metadata.decorateKey(request.key); - SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(true, metadata, FBUtilities.nowInSeconds(), columns, RowFilter.NONE, limits, dk, filter); - return getSlice(Collections.>singletonList(cmd), + SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(true, metadata, FBUtilities.nowInSeconds(), columns, RowFilter.NONE, limits, dk, filter); + return getSlice(Collections.singletonList(cmd), false, limits.perPartitionCount(), consistencyLevel, @@ -2525,7 +2525,7 @@ public class CassandraServer implements Cassandra.Iface // We want to know if the partition exists, so just fetch a single cell. ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(Slices.ALL, false); DataLimits limits = DataLimits.thriftLimits(1, 1); - return new SinglePartitionSliceCommand(false, 0, true, metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, limits, key, filter); + return new SinglePartitionReadCommand(false, 0, true, metadata, nowInSec, ColumnFilter.all(metadata), RowFilter.NONE, limits, key, filter); } // Gather the clustering for the expected values and query those. diff --git a/test/unit/org/apache/cassandra/db/KeyspaceTest.java b/test/unit/org/apache/cassandra/db/KeyspaceTest.java index e3243c74f4..c0c275325a 100644 --- a/test/unit/org/apache/cassandra/db/KeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/KeyspaceTest.java @@ -133,7 +133,7 @@ public class KeyspaceTest extends CQLTester Clustering endClustering = new Clustering(ByteBufferUtil.bytes(sliceEnd)); Slices slices = Slices.with(cfs.getComparator(), Slice.make(startClustering, endClustering)); ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(slices, reversed); - SinglePartitionSliceCommand command = singlePartitionSlice(cfs, key, filter, limit); + SinglePartitionReadCommand command = singlePartitionSlice(cfs, key, filter, limit); try (ReadExecutionController executionController = command.executionController(); PartitionIterator iterator = command.executeInternal(executionController)) @@ -208,7 +208,7 @@ public class KeyspaceTest extends CQLTester PartitionColumns columns = PartitionColumns.of(cfs.metadata.getColumnDefinition(new ColumnIdentifier("c", false))); ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(Slices.ALL, false); - SinglePartitionSliceCommand command = singlePartitionSlice(cfs, "0", filter, null); + SinglePartitionReadCommand command = singlePartitionSlice(cfs, "0", filter, null); try (ReadExecutionController executionController = command.executionController(); PartitionIterator iterator = command.executeInternal(executionController)) { @@ -261,12 +261,12 @@ public class KeyspaceTest extends CQLTester return new ClusteringIndexSliceFilter(slices, reversed); } - private static SinglePartitionSliceCommand singlePartitionSlice(ColumnFamilyStore cfs, String key, ClusteringIndexSliceFilter filter, Integer rowLimit) + private static SinglePartitionReadCommand singlePartitionSlice(ColumnFamilyStore cfs, String key, ClusteringIndexSliceFilter filter, Integer rowLimit) { DataLimits limit = rowLimit == null ? DataLimits.NONE : DataLimits.cqlLimits(rowLimit); - return new SinglePartitionSliceCommand( + return SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, limit, Util.dk(key), filter); } @@ -290,7 +290,7 @@ public class KeyspaceTest extends CQLTester { ClusteringIndexSliceFilter filter = slices(cfs, 5, null, false); - SinglePartitionSliceCommand command = singlePartitionSlice(cfs, "0", filter, 2); + SinglePartitionReadCommand command = singlePartitionSlice(cfs, "0", filter, 2); assertRowsInResult(cfs, command, 5, 7); command = singlePartitionSlice(cfs, "0", slices(cfs, 4, null, false), 2); @@ -332,7 +332,7 @@ public class KeyspaceTest extends CQLTester for (int round = 0; round < 2; round++) { - SinglePartitionSliceCommand command = singlePartitionSlice(cfs, "0", slices(cfs, null, null, false), 2); + SinglePartitionReadCommand command = singlePartitionSlice(cfs, "0", slices(cfs, null, null, false), 2); assertRowsInResult(cfs, command, 0, 1); command = singlePartitionSlice(cfs, "0", slices(cfs, 1, null, false), 1); @@ -361,7 +361,7 @@ public class KeyspaceTest extends CQLTester for (int round = 0; round < 2; round++) { - SinglePartitionSliceCommand command = singlePartitionSlice(cfs, "0", slices(cfs, 2, null, false), 3); + SinglePartitionReadCommand command = singlePartitionSlice(cfs, "0", slices(cfs, 2, null, false), 3); assertRowsInResult(cfs, command, -1, -1, 4); if (round == 0) @@ -412,7 +412,7 @@ public class KeyspaceTest extends CQLTester ((ClearableHistogram)cfs.metric.sstablesPerReadHistogram.cf).clear(); - SinglePartitionSliceCommand command = singlePartitionSlice(cfs, "0", slices(cfs, null, 1499, false), 1000); + SinglePartitionReadCommand command = singlePartitionSlice(cfs, "0", slices(cfs, null, 1499, false), 1000); int[] expectedValues = new int[500]; for (int i = 0; i < 500; i++) expectedValues[i] = i + 1000; @@ -473,17 +473,17 @@ public class KeyspaceTest extends CQLTester private void validateSliceLarge(ColumnFamilyStore cfs) { ClusteringIndexSliceFilter filter = slices(cfs, 1000, null, false); - SinglePartitionSliceCommand command = new SinglePartitionSliceCommand( + SinglePartitionReadCommand command = SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, DataLimits.cqlLimits(3), Util.dk("0"), filter); assertRowsInResult(cfs, command, 1000, 1001, 1002); filter = slices(cfs, 1195, null, false); - command = new SinglePartitionSliceCommand( + command = SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, DataLimits.cqlLimits(3), Util.dk("0"), filter); assertRowsInResult(cfs, command, 1195, 1196, 1197); filter = slices(cfs, null, 1996, true); - command = new SinglePartitionSliceCommand( + command = SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, DataLimits.cqlLimits(1000), Util.dk("0"), filter); int[] expectedValues = new int[997]; for (int i = 0, v = 1996; v >= 1000; i++, v--) @@ -491,22 +491,22 @@ public class KeyspaceTest extends CQLTester assertRowsInResult(cfs, command, expectedValues); filter = slices(cfs, 1990, null, false); - command = new SinglePartitionSliceCommand( + command = SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, DataLimits.cqlLimits(3), Util.dk("0"), filter); assertRowsInResult(cfs, command, 1990, 1991, 1992); filter = slices(cfs, null, null, true); - command = new SinglePartitionSliceCommand( + command = SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, DataLimits.cqlLimits(3), Util.dk("0"), filter); assertRowsInResult(cfs, command, 1999, 1998, 1997); filter = slices(cfs, null, 9000, true); - command = new SinglePartitionSliceCommand( + command = SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, DataLimits.cqlLimits(3), Util.dk("0"), filter); assertRowsInResult(cfs, command, 1999, 1998, 1997); filter = slices(cfs, 9000, null, false); - command = new SinglePartitionSliceCommand( + command = SinglePartitionReadCommand.create( cfs.metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(cfs.metadata), RowFilter.NONE, DataLimits.cqlLimits(3), Util.dk("0"), filter); assertRowsInResult(cfs, command); } diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java index a6ee48fd2f..d0cc8904fb 100644 --- a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java +++ b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java @@ -210,7 +210,7 @@ public class RangeTombstoneTest sb.add(Slice.Bound.create(cfs.getComparator(), true, true, 1), Slice.Bound.create(cfs.getComparator(), false, true, 10)); sb.add(Slice.Bound.create(cfs.getComparator(), true, true, 16), Slice.Bound.create(cfs.getComparator(), false, true, 20)); - partition = Util.getOnlyPartitionUnfiltered(SinglePartitionSliceCommand.create(cfs.metadata, FBUtilities.nowInSeconds(), Util.dk(key), sb.build())); + partition = Util.getOnlyPartitionUnfiltered(SinglePartitionReadCommand.create(cfs.metadata, FBUtilities.nowInSeconds(), Util.dk(key), sb.build())); rt = rangeTombstones(partition); assertEquals(2, rt.size()); } diff --git a/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java b/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java index dbcb1a4331..749ce2ebc4 100644 --- a/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java +++ b/test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java @@ -96,7 +96,7 @@ public class SinglePartitionSliceCommandTest ByteBuffer zero = ByteBufferUtil.bytes(0); Slices slices = Slices.with(cfm.comparator, Slice.make(Slice.Bound.inclusiveStartOf(zero), Slice.Bound.inclusiveEndOf(zero))); ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(slices, false); - ReadCommand cmd = new SinglePartitionSliceCommand(false, MessagingService.VERSION_30, true, cfm, + ReadCommand cmd = new SinglePartitionReadCommand(false, MessagingService.VERSION_30, true, cfm, FBUtilities.nowInSeconds(), columnFilter, RowFilter.NONE, @@ -146,13 +146,13 @@ public class SinglePartitionSliceCommandTest ColumnFilter columnFilter = ColumnFilter.selection(PartitionColumns.of(s)); ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.NONE, false); - ReadCommand cmd = new SinglePartitionSliceCommand(false, MessagingService.VERSION_30, true, cfm, - FBUtilities.nowInSeconds(), - columnFilter, - RowFilter.NONE, - DataLimits.NONE, - key, - sliceFilter); + ReadCommand cmd = new SinglePartitionReadCommand(false, MessagingService.VERSION_30, true, cfm, + 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)) diff --git a/test/unit/org/apache/cassandra/service/QueryPagerTest.java b/test/unit/org/apache/cassandra/service/QueryPagerTest.java index 4daf2e50a8..d8d74f3761 100644 --- a/test/unit/org/apache/cassandra/service/QueryPagerTest.java +++ b/test/unit/org/apache/cassandra/service/QueryPagerTest.java @@ -147,12 +147,12 @@ public class QueryPagerTest return builder.withPagingLimit(100).build(); } - private static SinglePartitionSliceCommand sliceQuery(String key, String start, String end, int count) + private static SinglePartitionReadCommand sliceQuery(String key, String start, String end, int count) { return sliceQuery(key, start, end, false, count); } - private static SinglePartitionSliceCommand sliceQuery(String key, String start, String end, boolean reversed, int count) + private static SinglePartitionReadCommand sliceQuery(String key, String start, String end, boolean reversed, int count) { ClusteringComparator cmp = cfs().getComparator(); CFMetaData metadata = cfs().metadata; @@ -160,9 +160,7 @@ public class QueryPagerTest Slice slice = Slice.make(cmp.make(start), cmp.make(end)); ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(Slices.with(cmp, slice), reversed); - SinglePartitionSliceCommand command = new SinglePartitionSliceCommand(cfs().metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, Util.dk(key), filter); - - return command; + return SinglePartitionReadCommand.create(cfs().metadata, FBUtilities.nowInSeconds(), ColumnFilter.all(metadata), RowFilter.NONE, DataLimits.NONE, Util.dk(key), filter); } private static ReadCommand rangeNamesQuery(String keyStart, String keyEnd, int count, String... names) @@ -306,7 +304,7 @@ public class QueryPagerTest public void multiQueryTest(boolean testPagingState, int protocolVersion) throws Exception { - ReadQuery command = new SinglePartitionReadCommand.Group(new ArrayList>() + ReadQuery command = new SinglePartitionReadCommand.Group(new ArrayList() {{ add(sliceQuery("k1", "c2", "c6", 10)); add(sliceQuery("k4", "c3", "c5", 10)); @@ -428,7 +426,7 @@ public class QueryPagerTest for (int i = 0; i < 5; i++) executeInternal(String.format("INSERT INTO %s.%s (k, c, v) VALUES ('k%d', 'c%d', null)", keyspace, table, 0, i)); - ReadCommand command = SinglePartitionSliceCommand.create(cfs.metadata, FBUtilities.nowInSeconds(), Util.dk("k0"), Slice.ALL); + ReadCommand command = SinglePartitionReadCommand.create(cfs.metadata, FBUtilities.nowInSeconds(), Util.dk("k0"), Slice.ALL); QueryPager pager = command.getPager(null, Server.CURRENT_VERSION);