cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ble...@apache.org
Subject [8/8] cassandra git commit: Merge branch cassandra-3.11 into trunk
Date Thu, 01 Jun 2017 08:24:19 GMT
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.lerer@gmail.com>
Authored: Thu Jun 1 10:20:29 2017 +0200
Committer: Benjamin Lerer <b.lerer@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


Mime
View raw message