You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2015/07/02 20:04:44 UTC
[4/4] cassandra git commit: Ensure StreamSession uses canonical
sstable instances
Ensure StreamSession uses canonical sstable instances
patch by benedict; reviewed by yukim for CASSANDRA-9700
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/61aeb910
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/61aeb910
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/61aeb910
Branch: refs/heads/trunk
Commit: 61aeb910930faa6caeac98a184e1909e69d7abb0
Parents: 7813dee
Author: Benedict Elliott Smith <be...@apache.org>
Authored: Thu Jul 2 18:05:09 2015 +0100
Committer: Benedict Elliott Smith <be...@apache.org>
Committed: Thu Jul 2 19:04:19 2015 +0100
----------------------------------------------------------------------
.../cassandra/streaming/StreamSession.java | 21 +++++++++++---------
1 file changed, 12 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/61aeb910/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java
index d27c4e2..c9ebdad 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -321,18 +321,21 @@ public class StreamSession implements IEndpointStateChangeSubscriber
{
public List<SSTableReader> apply(View view)
{
- List<SSTableReader> filteredSSTables = ColumnFamilyStore.CANONICAL_SSTABLES.apply(view);
+ Map<SSTableReader, SSTableReader> permittedInstances = new HashMap<>();
+ for (SSTableReader reader : ColumnFamilyStore.CANONICAL_SSTABLES.apply(view))
+ permittedInstances.put(reader, reader);
+
Set<SSTableReader> sstables = Sets.newHashSet();
- if (filteredSSTables != null)
+ for (AbstractBounds<PartitionPosition> rowBounds : rowBoundsList)
{
- for (AbstractBounds<PartitionPosition> rowBounds : rowBoundsList)
+ // sstableInBounds may contain early opened sstables
+ for (SSTableReader sstable : view.sstablesInBounds(rowBounds))
{
- // sstableInBounds may contain early opened sstables
- for (SSTableReader sstable : view.sstablesInBounds(rowBounds))
- {
- if (filteredSSTables.contains(sstable) && (!isIncremental || !sstable.isRepaired()))
- sstables.add(sstable);
- }
+ if (isIncremental && sstable.isRepaired())
+ continue;
+ sstable = permittedInstances.get(sstable);
+ if (sstable != null)
+ sstables.add(sstable);
}
}