You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bl...@apache.org on 2017/06/01 08:24:19 UTC
[8/8] cassandra git commit: Merge branch cassandra-3.11 into trunk
Merge branch cassandra-3.11 into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/96899bbb
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/96899bbb
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/96899bbb
Branch: refs/heads/trunk
Commit: 96899bbb60ebba05408c3248e756ed33605e8075
Parents: a741efd 5c9db9a
Author: Benjamin Lerer <b....@gmail.com>
Authored: Thu Jun 1 10:20:29 2017 +0200
Committer: Benjamin Lerer <b....@gmail.com>
Committed: Thu Jun 1 10:22:54 2017 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
NEWS.txt | 2 +
doc/source/operating/metrics.rst | 2 +-
.../cassandra/db/PartitionRangeReadCommand.java | 20 ++++-
.../db/SinglePartitionReadCommand.java | 83 ++++++++++++++------
.../org/apache/cassandra/db/StorageHook.java | 53 ++++++++-----
.../UnfilteredRowIteratorWithLowerBound.java | 10 ++-
.../io/sstable/format/SSTableReader.java | 62 +++++++++------
.../io/sstable/format/SSTableReadsListener.java | 81 +++++++++++++++++++
.../io/sstable/format/big/BigTableReader.java | 33 +++++---
.../io/sstable/format/big/BigTableScanner.java | 23 ++++--
.../miscellaneous/SSTablesIteratedTest.java | 69 +++++++++++++++-
.../cassandra/db/compaction/TTLExpiryTest.java | 5 +-
.../sstable/SSTableCorruptionDetectionTest.java | 7 +-
.../io/sstable/SSTableScannerTest.java | 5 +-
.../cassandra/io/sstable/SSTableWriterTest.java | 7 +-
16 files changed, 369 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/NEWS.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/doc/source/operating/metrics.rst
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index a47302b,aa8271d..bc80907
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@@ -206,16 -209,17 +207,17 @@@ public class PartitionRangeReadCommand
for (Memtable memtable : view.memtables)
{
@SuppressWarnings("resource") // We close on exception and on closing the result returned by this method
- Memtable.MemtableUnfilteredPartitionIterator iter = memtable.makePartitionIterator(columnFilter(), dataRange(), isForThrift());
+ Memtable.MemtableUnfilteredPartitionIterator iter = memtable.makePartitionIterator(columnFilter(), dataRange());
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, iter.getMinLocalDeletionTime());
- iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, metadata(), nowInSec()) : iter);
+ iterators.add(iter);
}
+ SSTableReadsListener readCountUpdater = newReadCountUpdater();
for (SSTableReader sstable : view.sstables)
{
@SuppressWarnings("resource") // We close on exception and on closing the result returned by this method
- UnfilteredPartitionIterator iter = sstable.getScanner(columnFilter(), dataRange());
- UnfilteredPartitionIterator iter = sstable.getScanner(columnFilter(), dataRange(), isForThrift(), readCountUpdater);
- iterators.add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, metadata(), nowInSec()) : iter);
++ UnfilteredPartitionIterator iter = sstable.getScanner(columnFilter(), dataRange(), readCountUpdater);
+ iterators.add(iter);
if (!sstable.isRepaired())
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime());
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index d5d2901,47c426e..5ebfdcc
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@@ -567,7 -600,7 +569,7 @@@ public class SinglePartitionReadComman
@SuppressWarnings("resource") // 'iter' is added to iterators which is closed on exception,
// or through the closing of the final merged iterator
- UnfilteredRowIteratorWithLowerBound iter = makeIterator(cfs, sstable);
- UnfilteredRowIteratorWithLowerBound iter = makeIterator(cfs, sstable, true, metricsCollector);
++ UnfilteredRowIteratorWithLowerBound iter = makeIterator(cfs, sstable, metricsCollector);
if (!sstable.isRepaired())
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime());
@@@ -587,7 -620,7 +589,7 @@@
@SuppressWarnings("resource") // 'iter' is added to iterators which is close on exception,
// or through the closing of the final merged iterator
- UnfilteredRowIteratorWithLowerBound iter = makeIterator(cfs, sstable);
- UnfilteredRowIteratorWithLowerBound iter = makeIterator(cfs, sstable, false, metricsCollector);
++ UnfilteredRowIteratorWithLowerBound iter = makeIterator(cfs, sstable, metricsCollector);
if (!sstable.isRepaired())
oldestUnrepairedTombstone = Math.min(oldestUnrepairedTombstone, sstable.getMinLocalDeletionTime());
@@@ -600,10 -633,10 +602,10 @@@
nonIntersectingSSTables, view.sstables.size(), includedDueToTombstones);
if (iterators.isEmpty())
- return EmptyIterators.unfilteredRow(cfs.metadata, partitionKey(), filter.isReversed());
+ return EmptyIterators.unfilteredRow(cfs.metadata(), partitionKey(), filter.isReversed());
- StorageHook.instance.reportRead(cfs.metadata.cfId, partitionKey());
+ StorageHook.instance.reportRead(cfs.metadata().id, partitionKey());
- return withSSTablesIterated(iterators, cfs.metric);
+ return withSSTablesIterated(iterators, cfs.metric, metricsCollector);
}
catch (RuntimeException | Error e)
{
@@@ -630,13 -663,20 +632,16 @@@
return clusteringIndexFilter().shouldInclude(sstable);
}
- private UnfilteredRowIteratorWithLowerBound makeIterator(ColumnFamilyStore cfs, final SSTableReader sstable)
+ private UnfilteredRowIteratorWithLowerBound makeIterator(ColumnFamilyStore cfs,
- final SSTableReader sstable,
- boolean applyThriftTransformation,
++ SSTableReader sstable,
+ SSTableReadsListener listener)
{
return StorageHook.instance.makeRowIteratorWithLowerBound(cfs,
partitionKey(),
sstable,
clusteringIndexFilter(),
- columnFilter());
+ columnFilter(),
- isForThrift(),
- nowInSec(),
- applyThriftTransformation,
+ listener);
}
@@@ -743,10 -780,15 +745,14 @@@
continue; // no tombstone at all, we can skip that sstable
// We need to get the partition deletion and include it if it's live. In any case though, we're done with that sstable.
- sstable.incrementReadCount();
- try (UnfilteredRowIterator iter = StorageHook.instance.makeRowIterator(cfs, sstable, partitionKey(), filter.getSlices(metadata()), columnFilter(), filter.isReversed()))
+ try (UnfilteredRowIterator iter = StorageHook.instance.makeRowIterator(cfs,
+ sstable,
+ partitionKey(),
+ filter.getSlices(metadata()),
+ columnFilter(),
+ filter.isReversed(),
- isForThrift(),
+ metricsCollector))
{
- sstablesIterated++;
if (!iter.partitionLevelDeletion().isLive())
result = add(UnfilteredRowIterators.noRowsIterator(iter.metadata(), iter.partitionKey(), Rows.EMPTY_STATIC_ROW, iter.partitionLevelDeletion(), filter.isReversed()), result, filter, sstable.isRepaired());
else
@@@ -755,17 -797,21 +761,20 @@@
continue;
}
- Tracing.trace("Merging data from sstable {}", sstable.descriptor.generation);
- sstable.incrementReadCount();
- try (UnfilteredRowIterator iter = StorageHook.instance.makeRowIterator(cfs, sstable, partitionKey(), filter.getSlices(metadata()), columnFilter(), filter.isReversed()))
+ try (UnfilteredRowIterator iter = StorageHook.instance.makeRowIterator(cfs,
+ sstable,
+ partitionKey(),
+ filter.getSlices(metadata()),
+ columnFilter(),
+ filter.isReversed(),
- isForThrift(),
+ metricsCollector))
{
if (iter.isEmpty())
continue;
if (sstable.isRepaired())
onlyUnrepaired = false;
- sstablesIterated++;
- result = add(isForThrift() ? ThriftResultsMerger.maybeWrap(iter, nowInSec()) : iter, result, filter, sstable.isRepaired());
+ result = add(iter, result, filter, sstable.isRepaired());
}
}
@@@ -776,10 -822,10 +785,10 @@@
DecoratedKey key = result.partitionKey();
cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1);
- StorageHook.instance.reportRead(cfs.metadata.cfId, partitionKey());
+ StorageHook.instance.reportRead(cfs.metadata.id, partitionKey());
// "hoist up" the requested data into a more recent sstable
- if (sstablesIterated > cfs.getMinimumCompactionThreshold()
+ if (metricsCollector.getMergedSSTables() > cfs.getMinimumCompactionThreshold()
&& onlyUnrepaired
&& !cfs.isAutoCompactionDisabled()
&& cfs.getCompactionStrategyManager().shouldDefragment())
@@@ -1041,12 -1088,40 +1050,40 @@@
private static class Deserializer extends SelectionDeserializer
{
- public ReadCommand deserialize(DataInputPlus in, int version, boolean isDigest, int digestVersion, boolean isForThrift, CFMetaData 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, Optional<IndexMetadata> index)
throws IOException
{
- DecoratedKey key = metadata.decorateKey(metadata.getKeyValidator().readValue(in, DatabaseDescriptor.getMaxValueSize()));
+ DecoratedKey key = metadata.partitioner.decorateKey(metadata.partitionKeyType.readValue(in, DatabaseDescriptor.getMaxValueSize()));
ClusteringIndexFilter filter = ClusteringIndexFilter.serializer.deserialize(in, version, metadata);
- return new SinglePartitionReadCommand(isDigest, digestVersion, isForThrift, metadata, nowInSec, columnFilter, rowFilter, limits, key, filter);
+ return new SinglePartitionReadCommand(isDigest, digestVersion, metadata, nowInSec, columnFilter, rowFilter, limits, key, filter);
}
}
+
+ /**
+ * {@code SSTableReaderListener} used to collect metrics about SSTable read access.
+ */
+ private static final class SSTableReadMetricsCollector implements SSTableReadsListener
+ {
+ /**
+ * The number of SSTables that need to be merged. This counter is only updated for single partition queries
+ * since this has been the behavior so far.
+ */
+ private int mergedSSTables;
+
+ @Override
+ public void onSSTableSelected(SSTableReader sstable, RowIndexEntry<?> indexEntry, SelectionReason reason)
+ {
+ sstable.incrementReadCount();
+ mergedSSTables++;
+ }
+
+ /**
+ * Returns the number of SSTables that need to be merged.
+ * @return the number of SSTables that need to be merged.
+ */
+ public int getMergedSSTables()
+ {
+ return mergedSSTables;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/db/StorageHook.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/StorageHook.java
index 3df8805,48d7ede..be1d0bf
--- a/src/java/org/apache/cassandra/db/StorageHook.java
+++ b/src/java/org/apache/cassandra/db/StorageHook.java
@@@ -24,7 -26,7 +24,8 @@@ import org.apache.cassandra.db.partitio
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIteratorWithLowerBound;
import org.apache.cassandra.io.sstable.format.SSTableReader;
+ import org.apache.cassandra.io.sstable.format.SSTableReadsListener;
+import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.FBUtilities;
public interface StorageHook
@@@ -37,13 -39,20 +38,15 @@@
DecoratedKey partitionKey,
SSTableReader sstable,
ClusteringIndexFilter filter,
- ColumnFilter selectedColumns);
+ ColumnFilter selectedColumns,
- boolean isForThrift,
- int nowInSec,
- boolean applyThriftTransformation,
+ SSTableReadsListener listener);
-
public UnfilteredRowIterator makeRowIterator(ColumnFamilyStore cfs,
SSTableReader sstable,
DecoratedKey key,
Slices slices,
ColumnFilter selectedColumns,
- boolean reversed);
+ boolean reversed,
- boolean isForThrift,
+ SSTableReadsListener listener);
static StorageHook createHook()
{
@@@ -52,27 -61,44 +55,37 @@@
{
return FBUtilities.construct(className, StorageHook.class.getSimpleName());
}
- else
+
+ return new StorageHook()
{
- return new StorageHook()
- {
- public void reportWrite(TableId tableId, PartitionUpdate partitionUpdate) {}
- public void reportWrite(UUID cfid, PartitionUpdate partitionUpdate) {}
++ public void reportWrite(TableId tableId, PartitionUpdate partitionUpdate) {}
- public void reportRead(TableId tableId, DecoratedKey key) {}
- public void reportRead(UUID cfid, DecoratedKey key) {}
++ public void reportRead(TableId tableId, DecoratedKey key) {}
- public UnfilteredRowIteratorWithLowerBound makeRowIteratorWithLowerBound(ColumnFamilyStore cfs, DecoratedKey partitionKey, SSTableReader sstable, ClusteringIndexFilter filter, ColumnFilter selectedColumns)
- {
- return new UnfilteredRowIteratorWithLowerBound(partitionKey,
- sstable,
- filter,
- selectedColumns);
- }
+ public UnfilteredRowIteratorWithLowerBound makeRowIteratorWithLowerBound(ColumnFamilyStore cfs,
+ DecoratedKey partitionKey,
+ SSTableReader sstable,
+ ClusteringIndexFilter filter,
+ ColumnFilter selectedColumns,
- boolean isForThrift,
- int nowInSec,
- boolean applyThriftTransformation,
+ SSTableReadsListener listener)
+ {
+ return new UnfilteredRowIteratorWithLowerBound(partitionKey,
+ sstable,
+ filter,
+ selectedColumns,
- isForThrift,
- nowInSec,
- applyThriftTransformation,
+ listener);
+ }
- public UnfilteredRowIterator makeRowIterator(ColumnFamilyStore cfs, SSTableReader sstable, DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed)
- {
- return sstable.iterator(key, slices, selectedColumns, reversed);
- }
- };
- }
+ public UnfilteredRowIterator makeRowIterator(ColumnFamilyStore cfs,
+ SSTableReader sstable,
+ DecoratedKey key,
+ Slices slices,
+ ColumnFilter selectedColumns,
+ boolean reversed,
- boolean isForThrift,
+ SSTableReadsListener listener)
+ {
- return sstable.iterator(key, slices, selectedColumns, reversed, isForThrift, listener);
++ return sstable.iterator(key, slices, selectedColumns, reversed, listener);
+ }
+ };
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorWithLowerBound.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorWithLowerBound.java
index d23c37c,84a742b..71c3d7f
--- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorWithLowerBound.java
+++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorWithLowerBound.java
@@@ -31,7 -31,9 +31,8 @@@ import org.apache.cassandra.db.filter.C
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.io.sstable.IndexInfo;
import org.apache.cassandra.io.sstable.format.SSTableReader;
+ import org.apache.cassandra.io.sstable.format.SSTableReadsListener;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
-import org.apache.cassandra.thrift.ThriftResultsMerger;
import org.apache.cassandra.utils.IteratorWithLowerBound;
/**
@@@ -47,18 -49,30 +48,21 @@@ public class UnfilteredRowIteratorWithL
private final SSTableReader sstable;
private final ClusteringIndexFilter filter;
private final ColumnFilter selectedColumns;
- private final boolean isForThrift;
- private final int nowInSec;
- private final boolean applyThriftTransformation;
+ private final SSTableReadsListener listener;
private ClusteringBound lowerBound;
private boolean firstItemRetrieved;
public UnfilteredRowIteratorWithLowerBound(DecoratedKey partitionKey,
SSTableReader sstable,
ClusteringIndexFilter filter,
- ColumnFilter selectedColumns)
+ ColumnFilter selectedColumns,
- boolean isForThrift,
- int nowInSec,
- boolean applyThriftTransformation,
+ SSTableReadsListener listener)
{
super(partitionKey);
this.sstable = sstable;
this.filter = filter;
this.selectedColumns = selectedColumns;
- this.isForThrift = isForThrift;
- this.nowInSec = nowInSec;
- this.applyThriftTransformation = applyThriftTransformation;
+ this.listener = listener;
this.lowerBound = null;
this.firstItemRetrieved = false;
}
@@@ -89,11 -103,11 +93,9 @@@
@Override
protected UnfilteredRowIterator initializeIterator()
{
- 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 = sstable.iterator(partitionKey(), filter.getSlices(metadata()), selectedColumns, filter.isReversed());
- UnfilteredRowIterator iter = sstable.iterator(partitionKey(), filter.getSlices(metadata()), selectedColumns, filter.isReversed(), isForThrift, listener);
- return isForThrift && applyThriftTransformation
- ? ThriftResultsMerger.maybeWrap(iter, nowInSec)
- : iter;
++ UnfilteredRowIterator iter = sstable.iterator(partitionKey(), filter.getSlices(metadata()), selectedColumns, filter.isReversed(), listener);
+ return iter;
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index 4495edf,e9b2491..25cfddb
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@@ -1512,35 -1523,55 +1512,62 @@@ public abstract class SSTableReader ext
return null;
}
+ private boolean keyCacheEnabled()
+ {
+ return keyCache != null && keyCache.getCapacity() > 0 && metadata().params.caching.cacheKeys();
+ }
+
/**
- * Get position updating key cache and stats.
- * @see #getPosition(PartitionPosition, SSTableReader.Operator, boolean)
+ * Retrieves the position while updating the key cache and the stats.
+ * @param key The key to apply as the rhs to the given Operator. A 'fake' key is allowed to
+ * allow key selection by token bounds but only if op != * EQ
+ * @param op The Operator defining matching keys: the nearest key to the target matching the operator wins.
*/
- public RowIndexEntry getPosition(PartitionPosition key, Operator op)
+ public final RowIndexEntry getPosition(PartitionPosition key, Operator op)
{
- return getPosition(key, op, true, false);
+ return getPosition(key, op, SSTableReadsListener.NOOP_LISTENER);
}
- public RowIndexEntry getPosition(PartitionPosition key, Operator op, boolean updateCacheAndStats)
+ /**
+ * Retrieves the position while updating the key cache and the stats.
+ * @param key The key to apply as the rhs to the given Operator. A 'fake' key is allowed to
+ * allow key selection by token bounds but only if op != * EQ
+ * @param op The Operator defining matching keys: the nearest key to the target matching the operator wins.
+ * @param listener the {@code SSTableReaderListener} that must handle the notifications.
+ */
+ public final RowIndexEntry getPosition(PartitionPosition key, Operator op, SSTableReadsListener listener)
{
- return getPosition(key, op, updateCacheAndStats, false);
+ return getPosition(key, op, true, false, listener);
}
+
- public final RowIndexEntry getPosition(PartitionPosition key, Operator op, boolean updateCacheAndStats)
++ public final RowIndexEntry getPosition(PartitionPosition key,
++ Operator op,
++ boolean updateCacheAndStats)
+ {
+ return getPosition(key, op, updateCacheAndStats, false, SSTableReadsListener.NOOP_LISTENER);
+ }
++
/**
* @param key The key to apply as the rhs to the given Operator. A 'fake' key is allowed to
* allow key selection by token bounds but only if op != * EQ
* @param op The Operator defining matching keys: the nearest key to the target matching the operator wins.
* @param updateCacheAndStats true if updating stats and cache
+ * @param listener a listener used to handle internal events
* @return The index entry corresponding to the key, or null if the key is not present
*/
- protected abstract RowIndexEntry getPosition(PartitionPosition key, Operator op, boolean updateCacheAndStats, boolean permitMatchPastLast);
+ protected abstract RowIndexEntry getPosition(PartitionPosition key,
+ Operator op,
+ boolean updateCacheAndStats,
+ boolean permitMatchPastLast,
+ SSTableReadsListener listener);
+
+ public abstract UnfilteredRowIterator iterator(DecoratedKey key,
+ Slices slices,
+ ColumnFilter selectedColumns,
+ boolean reversed,
- boolean isForThrift,
+ SSTableReadsListener listener);
- public abstract UnfilteredRowIterator iterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed);
- public abstract UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed, boolean isForThrift);
+ public abstract UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed);
public abstract UnfilteredRowIterator simpleIterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, boolean tombstoneOnly);
@@@ -1698,9 -1753,14 +1725,10 @@@
/**
* @param columns the columns to return.
* @param dataRange filter to use when reading the columns
+ * @param listener a listener used to handle internal read events
* @return A Scanner for seeking over the rows of the SSTable.
*/
- public abstract ISSTableScanner getScanner(ColumnFilter columns, DataRange dataRange);
- public abstract ISSTableScanner getScanner(ColumnFilter columns,
- DataRange dataRange,
- RateLimiter limiter,
- boolean isForThrift,
- SSTableReadsListener listener);
++ public abstract ISSTableScanner getScanner(ColumnFilter columns, DataRange dataRange, SSTableReadsListener listener);
public FileDataInput getFileDataInput(long position)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/io/sstable/format/SSTableReadsListener.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/format/SSTableReadsListener.java
index 0000000,8f6e3c0..6d384bf
mode 000000,100644..100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReadsListener.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReadsListener.java
@@@ -1,0 -1,82 +1,81 @@@
+ /*
+ * 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.io.sstable.format;
+
+ import org.apache.cassandra.db.RowIndexEntry;
-import org.apache.cassandra.io.sstable.format.SSTableReadsListener.SelectionReason;
+
+ /**
+ * Listener for receiving notifications associated with reading SSTables.
+ */
+ public interface SSTableReadsListener
+ {
+ /**
+ * The reasons for skipping an SSTable
+ */
+ enum SkippingReason
+ {
+ BLOOM_FILTER,
+ MIN_MAX_KEYS,
+ PARTITION_INDEX_LOOKUP,
+ INDEX_ENTRY_NOT_FOUND;
+ }
+
+ /**
+ * The reasons for selecting an SSTable
+ */
+ enum SelectionReason
+ {
+ KEY_CACHE_HIT,
+ INDEX_ENTRY_FOUND;
+ }
+
+ /**
+ * Listener that does nothing.
+ */
+ static final SSTableReadsListener NOOP_LISTENER = new SSTableReadsListener() {};
+
+ /**
+ * Handles notification that the specified SSTable has been skipped during a single partition query.
+ *
+ * @param sstable the SSTable reader
+ * @param reason the reason for which the SSTable has been skipped
+ */
+ default void onSSTableSkipped(SSTableReader sstable, SkippingReason reason)
+ {
+ }
+
+ /**
+ * Handles notification that the specified SSTable has been selected during a single partition query.
+ *
+ * @param sstable the SSTable reader
+ * @param indexEntry the index entry
+ * @param reason the reason for which the SSTable has been selected
+ */
+ default void onSSTableSelected(SSTableReader sstable, RowIndexEntry<?> indexEntry, SelectionReason reason)
+ {
+ }
+
+ /**
+ * Handles notification that the specified SSTable is being scanned during a partition range query.
+ *
+ * @param sstable the SSTable reader of the SSTable being scanned.
+ */
+ default void onScanningStarted(SSTableReader sstable)
+ {
+ }
+ }
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java
index c29bc5d,8551819..1b6a299
--- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java
@@@ -37,11 -32,19 +37,14 @@@ import org.apache.cassandra.dht.Range
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.io.sstable.format.SSTableReader;
+ import org.apache.cassandra.io.sstable.format.SSTableReadsListener;
-import org.apache.cassandra.io.sstable.format.SSTableReadsListener.SkippingReason;
+ import org.apache.cassandra.io.sstable.format.SSTableReadsListener.SelectionReason;
++import org.apache.cassandra.io.sstable.format.SSTableReadsListener.SkippingReason;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.io.util.FileDataInput;
+import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.ByteBufferUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.*;
/**
* SSTableReaders are open()ed by Keyspace.onStart; after that they are created by SSTableWriter.renameAndOpen.
@@@ -56,29 -59,29 +59,29 @@@ public class BigTableReader extends SST
super(desc, components, metadata, maxDataAge, sstableMetadata, openReason, header);
}
- public UnfilteredRowIterator iterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed)
- public UnfilteredRowIterator iterator(DecoratedKey key, Slices slices, ColumnFilter selectedColumns, boolean reversed, boolean isForThrift, SSTableReadsListener listener)
++ public UnfilteredRowIterator iterator(DecoratedKey key,
++ Slices slices,
++ ColumnFilter selectedColumns,
++ boolean reversed,
++ SSTableReadsListener listener)
{
- RowIndexEntry rie = getPosition(key, SSTableReader.Operator.EQ);
+ RowIndexEntry rie = getPosition(key, SSTableReader.Operator.EQ, listener);
- return iterator(null, key, rie, slices, selectedColumns, reversed, isForThrift);
+ return iterator(null, key, rie, slices, selectedColumns, reversed);
}
- public UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed, boolean isForThrift)
+ public UnfilteredRowIterator iterator(FileDataInput file, DecoratedKey key, RowIndexEntry indexEntry, Slices slices, ColumnFilter selectedColumns, boolean reversed)
{
if (indexEntry == null)
- return UnfilteredRowIterators.noRowsIterator(metadata, key, Rows.EMPTY_STATIC_ROW, DeletionTime.LIVE, reversed);
+ return UnfilteredRowIterators.noRowsIterator(metadata(), key, Rows.EMPTY_STATIC_ROW, DeletionTime.LIVE, reversed);
return reversed
- ? new SSTableReversedIterator(this, file, key, indexEntry, slices, selectedColumns, isForThrift, ifile)
- : new SSTableIterator(this, file, key, indexEntry, slices, selectedColumns, isForThrift, ifile);
+ ? new SSTableReversedIterator(this, file, key, indexEntry, slices, selectedColumns, ifile)
+ : new SSTableIterator(this, file, key, indexEntry, slices, selectedColumns, ifile);
}
- /**
- * @param columns the columns to return.
- * @param dataRange filter to use when reading the columns
- * @return A Scanner for seeking over the rows of the SSTable.
- */
- public ISSTableScanner getScanner(ColumnFilter columns, DataRange dataRange)
+ @Override
- public ISSTableScanner getScanner(ColumnFilter columns,
- DataRange dataRange,
- RateLimiter limiter,
- boolean isForThrift,
- SSTableReadsListener listener)
++ public ISSTableScanner getScanner(ColumnFilter columns, DataRange dataRange, SSTableReadsListener listener)
{
- return BigTableScanner.getScanner(this, columns, dataRange);
- return BigTableScanner.getScanner(this, columns, dataRange, limiter, isForThrift, listener);
++ return BigTableScanner.getScanner(this, columns, dataRange, listener);
}
/**
@@@ -124,14 -127,12 +127,18 @@@
return SSTableIdentityIterator.create(this, dfile, position, key, tombstoneOnly);
}
- @Override
+ /**
+ * @param key The key to apply as the rhs to the given Operator. A 'fake' key is allowed to
+ * allow key selection by token bounds but only if op != * EQ
+ * @param op The Operator defining matching keys: the nearest key to the target matching the operator wins.
+ * @param updateCacheAndStats true if updating stats and cache
+ * @return The index entry corresponding to the key, or null if the key is not present
+ */
- protected RowIndexEntry getPosition(PartitionPosition key, Operator op, boolean updateCacheAndStats, boolean permitMatchPastLast)
+ protected RowIndexEntry getPosition(PartitionPosition key,
+ Operator op,
+ boolean updateCacheAndStats,
+ boolean permitMatchPastLast,
+ SSTableReadsListener listener)
{
if (op == Operator.EQ)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java
index 716ef4c,f4bd1ea..b01573c
--- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java
@@@ -60,38 -62,65 +61,46 @@@ public class BigTableScanner implement
private final ColumnFilter columns;
private final DataRange dataRange;
private final RowIndexEntry.IndexSerializer rowIndexEntrySerializer;
- private final boolean isForThrift;
+ private final SSTableReadsListener listener;
private long startScan = -1;
private long bytesScanned = 0;
protected Iterator<UnfilteredRowIterator> iterator;
// Full scan of the sstables
- public static ISSTableScanner getScanner(SSTableReader sstable, RateLimiter limiter)
+ public static ISSTableScanner getScanner(SSTableReader sstable)
{
- return new BigTableScanner(sstable, ColumnFilter.all(sstable.metadata()), null, Iterators.singletonIterator(fullRange(sstable)));
- return new BigTableScanner(sstable,
- ColumnFilter.all(sstable.metadata),
- limiter,
- Iterators.singletonIterator(fullRange(sstable)));
++ return getScanner(sstable, Iterators.singletonIterator(fullRange(sstable)));
}
- public static ISSTableScanner getScanner(SSTableReader sstable, ColumnFilter columns, DataRange dataRange)
+ public static ISSTableScanner getScanner(SSTableReader sstable,
+ ColumnFilter columns,
+ DataRange dataRange,
- RateLimiter limiter,
- boolean isForThrift,
+ SSTableReadsListener listener)
{
- return new BigTableScanner(sstable, columns, dataRange, makeBounds(sstable, dataRange).iterator());
- return new BigTableScanner(sstable, columns, dataRange, limiter, isForThrift, makeBounds(sstable, dataRange).iterator(), listener);
++ return new BigTableScanner(sstable, columns, dataRange, makeBounds(sstable, dataRange).iterator(), listener);
}
- public static ISSTableScanner getScanner(SSTableReader sstable, Collection<Range<Token>> tokenRanges, RateLimiter limiter)
+ public static ISSTableScanner getScanner(SSTableReader sstable, Collection<Range<Token>> tokenRanges)
{
// We want to avoid allocating a SSTableScanner if the range don't overlap the sstable (#5249)
List<Pair<Long, Long>> positions = sstable.getPositionsForRanges(tokenRanges);
if (positions.isEmpty())
return new EmptySSTableScanner(sstable);
- return new BigTableScanner(sstable, ColumnFilter.all(sstable.metadata()), null, makeBounds(sstable, tokenRanges).iterator());
- return new BigTableScanner(sstable,
- ColumnFilter.all(sstable.metadata),
- limiter,
- makeBounds(sstable, tokenRanges).iterator());
++ return getScanner(sstable, makeBounds(sstable, tokenRanges).iterator());
}
public static ISSTableScanner getScanner(SSTableReader sstable, Iterator<AbstractBounds<PartitionPosition>> rangeIterator)
{
- return new BigTableScanner(sstable, ColumnFilter.all(sstable.metadata()), null, rangeIterator);
- return new BigTableScanner(sstable, ColumnFilter.all(sstable.metadata), null, rangeIterator);
- }
-
- private BigTableScanner(SSTableReader sstable,
- ColumnFilter columns,
- RateLimiter limiter,
- Iterator<AbstractBounds<PartitionPosition>> rangeIterator)
- {
- this(sstable, columns, null, limiter, false, rangeIterator, SSTableReadsListener.NOOP_LISTENER);
++ return new BigTableScanner(sstable, ColumnFilter.all(sstable.metadata()), null, rangeIterator, SSTableReadsListener.NOOP_LISTENER);
}
- private BigTableScanner(SSTableReader sstable, ColumnFilter columns, DataRange dataRange, Iterator<AbstractBounds<PartitionPosition>> rangeIterator)
+ private BigTableScanner(SSTableReader sstable,
+ ColumnFilter columns,
+ DataRange dataRange,
- RateLimiter limiter,
- boolean isForThrift,
+ Iterator<AbstractBounds<PartitionPosition>> rangeIterator,
+ SSTableReadsListener listener)
{
assert sstable != null;
@@@ -100,10 -129,12 +109,11 @@@
this.sstable = sstable;
this.columns = columns;
this.dataRange = dataRange;
- this.rowIndexEntrySerializer = sstable.descriptor.version.getSSTableFormat().getIndexSerializer(sstable.metadata,
+ this.rowIndexEntrySerializer = sstable.descriptor.version.getSSTableFormat().getIndexSerializer(sstable.metadata(),
sstable.descriptor.version,
sstable.header);
- this.isForThrift = isForThrift;
this.rangeIterator = rangeIterator;
+ this.listener = listener;
}
private static List<AbstractBounds<PartitionPosition>> makeBounds(SSTableReader sstable, Collection<Range<Token>> tokenRanges)
@@@ -264,6 -299,7 +274,7 @@@
private Iterator<UnfilteredRowIterator> createIterator()
{
- listener.onScanningStarted(sstable);
++ this.listener.onScanningStarted(sstable);
return new KeyScanningIterator();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/test/unit/org/apache/cassandra/db/compaction/TTLExpiryTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/compaction/TTLExpiryTest.java
index 0a970e1,9fafc74..a2352fc
--- a/test/unit/org/apache/cassandra/db/compaction/TTLExpiryTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/TTLExpiryTest.java
@@@ -243,7 -243,10 +244,9 @@@ public class TTLExpiryTes
cfs.enableAutoCompaction(true);
assertEquals(1, cfs.getLiveSSTables().size());
SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
- ISSTableScanner scanner = sstable.getScanner(ColumnFilter.all(cfs.metadata()), DataRange.allData(cfs.getPartitioner()));
- ISSTableScanner scanner = sstable.getScanner(ColumnFilter.all(sstable.metadata),
++ ISSTableScanner scanner = sstable.getScanner(ColumnFilter.all(cfs.metadata()),
+ DataRange.allData(cfs.getPartitioner()),
- false,
+ SSTableReadsListener.NOOP_LISTENER);
assertTrue(scanner.hasNext());
while(scanner.hasNext())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java
index bc82128,f7ced23..581109c
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java
@@@ -208,7 -211,12 +209,11 @@@ public class SSTableCorruptionDetection
for (int i = 0; i < numberOfPks; i++)
{
DecoratedKey dk = Util.dk(String.format("pkvalue_%07d", i));
- try (UnfilteredRowIterator rowIter = sstable.iterator(dk, Slices.ALL, ColumnFilter.all(cfs.metadata()), false))
+ try (UnfilteredRowIterator rowIter = sstable.iterator(dk,
+ Slices.ALL,
- ColumnFilter.all(cfs.metadata),
- false,
++ ColumnFilter.all(cfs.metadata()),
+ false,
+ SSTableReadsListener.NOOP_LISTENER))
{
while (rowIter.hasNext())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java
index 353b1ad,d1db09a..eff95fc
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java
@@@ -179,9 -181,9 +180,11 @@@ public class SSTableScannerTes
private static void assertScanMatches(SSTableReader sstable, int scanStart, int scanEnd, int ... boundaries)
{
assert boundaries.length % 2 == 0;
- for (DataRange range : dataRanges(sstable.metadata, scanStart, scanEnd))
+ for (DataRange range : dataRanges(sstable.metadata(), scanStart, scanEnd))
{
- try(ISSTableScanner scanner = sstable.getScanner(ColumnFilter.all(sstable.metadata()), range))
- try(ISSTableScanner scanner = sstable.getScanner(ColumnFilter.all(sstable.metadata), range, false, SSTableReadsListener.NOOP_LISTENER))
++ try(ISSTableScanner scanner = sstable.getScanner(ColumnFilter.all(sstable.metadata()),
++ range,
++ SSTableReadsListener.NOOP_LISTENER))
{
for (int b = 0; b < boundaries.length; b += 2)
for (int i = boundaries[b]; i <= boundaries[b + 1]; i++)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/96899bbb/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java
index fd93ca1,391927c..5d62cdb
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java
@@@ -223,7 -224,12 +224,11 @@@ public class SSTableWriterTest extends
try
{
DecoratedKey dk = Util.dk("large_value");
- UnfilteredRowIterator rowIter = sstable.iterator(dk, Slices.ALL, ColumnFilter.all(cfs.metadata()), false);
+ UnfilteredRowIterator rowIter = sstable.iterator(dk,
+ Slices.ALL,
- ColumnFilter.all(cfs.metadata),
- false,
++ ColumnFilter.all(cfs.metadata()),
+ false,
+ SSTableReadsListener.NOOP_LISTENER);
while (rowIter.hasNext())
{
rowIter.next();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org