You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2016/07/06 06:17:26 UTC
[1/9] cassandra git commit: Always select the live sstables when
getting sstables in bounds
Repository: cassandra
Updated Branches:
refs/heads/cassandra-3.0 221769516 -> bc23632f2
refs/heads/cassandra-3.9 78f077de0 -> 574e8df20
refs/heads/trunk 2444349e3 -> 60359948d
Always select the live sstables when getting sstables in bounds
Patch by marcuse; reviewed by benedict for CASSANDRA-11944
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5b0566a7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5b0566a7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5b0566a7
Branch: refs/heads/cassandra-3.0
Commit: 5b0566a70f373d6eb537c89c0db2a2e224706916
Parents: 2217695
Author: Marcus Eriksson <ma...@apache.org>
Authored: Thu Jun 2 09:37:06 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 07:52:28 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/db/ColumnFamilyStore.java | 8 ++++----
.../apache/cassandra/db/PartitionRangeReadCommand.java | 2 +-
.../org/apache/cassandra/db/SizeEstimatesRecorder.java | 5 +++--
.../db/compaction/AbstractCompactionStrategy.java | 2 +-
.../cassandra/db/compaction/CompactionController.java | 2 +-
.../db/compaction/DateTieredCompactionStrategy.java | 2 +-
.../db/compaction/TimeWindowCompactionStrategy.java | 2 +-
src/java/org/apache/cassandra/db/lifecycle/View.java | 10 +++++-----
.../org/apache/cassandra/streaming/StreamSession.java | 8 +++++++-
test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java | 2 +-
11 files changed, 26 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 7f8a3a1..99ac3ad 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.9
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
those static columns in query results (CASSANDRA-12123)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 3264327..1be3175 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1231,7 +1231,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @return sstables whose key range overlaps with that of the given sstables, not including itself.
* (The given sstables may or may not overlap with each other.)
*/
- public Collection<SSTableReader> getOverlappingSSTables(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Collection<SSTableReader> getOverlappingLiveSSTables(Iterable<SSTableReader> sstables)
{
logger.trace("Checking for sstables overlapping {}", sstables);
@@ -1282,7 +1282,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Set<SSTableReader> results = new HashSet<>();
for (AbstractBounds<PartitionPosition> bound : bounds)
- Iterables.addAll(results, view.sstablesInBounds(sstableSet, bound.left, bound.right));
+ Iterables.addAll(results, view.liveSSTablesInBounds(bound.left, bound.right));
return Sets.difference(results, ImmutableSet.copyOf(sstables));
}
@@ -1290,11 +1290,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
/**
* like getOverlappingSSTables, but acquires references before returning
*/
- public Refs<SSTableReader> getAndReferenceOverlappingSSTables(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Refs<SSTableReader> getAndReferenceOverlappingLiveSSTables(Iterable<SSTableReader> sstables)
{
while (true)
{
- Iterable<SSTableReader> overlapped = getOverlappingSSTables(sstableSet, sstables);
+ Iterable<SSTableReader> overlapped = getOverlappingLiveSSTables(sstables);
Refs<SSTableReader> refs = Refs.tryRef(overlapped);
if (refs != null)
return refs;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index 9585b59..842ad5f 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -175,7 +175,7 @@ public class PartitionRangeReadCommand extends ReadCommand
protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
{
- ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, dataRange().keyRange()));
+ ColumnFamilyStore.ViewFragment view = cfs.select(View.selectLive(dataRange().keyRange()));
Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().getKeyValidator()));
// fetch data from current memtable, historical memtables, and SSTables in the correct order.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 2a74ea9..3461aef 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -103,8 +103,9 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable
{
while (refs == null)
{
- ColumnFamilyStore.ViewFragment view = table.select(View.select(SSTableSet.CANONICAL, Range.makeRowRange(range)));
- refs = Refs.tryRef(view.sstables);
+ // note that this is not guaranteed to return all sstables within the ranges, but for an estimated size, that should be fine
+ Iterable<SSTableReader> canonicalSSTables = table.getTracker().getView().select(SSTableSet.CANONICAL, table.select(View.selectLive(Range.makeRowRange(range))).sstables);
+ refs = Refs.tryRef(canonicalSSTables);
}
// calculate the estimates.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
index e30e4f7..0dce52b 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
@@ -380,7 +380,7 @@ public abstract class AbstractCompactionStrategy
if (uncheckedTombstoneCompaction)
return true;
- Collection<SSTableReader> overlaps = cfs.getOverlappingSSTables(SSTableSet.CANONICAL, Collections.singleton(sstable));
+ Collection<SSTableReader> overlaps = cfs.getOverlappingLiveSSTables(Collections.singleton(sstable));
if (overlaps.isEmpty())
{
// there is no overlap, tombstones are safely droppable
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/CompactionController.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionController.java b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
index a5b8308..fbf29e3 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
@@ -103,7 +103,7 @@ public class CompactionController implements AutoCloseable
if (compacting == null)
overlappingSSTables = Refs.tryRef(Collections.<SSTableReader>emptyList());
else
- overlappingSSTables = cfs.getAndReferenceOverlappingSSTables(SSTableSet.LIVE, compacting);
+ overlappingSSTables = cfs.getAndReferenceOverlappingLiveSSTables(compacting);
this.overlapIterator = new OverlapIterator<>(buildIntervals(overlappingSSTables));
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
index 8571906..3e6ae61 100644
--- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
@@ -99,7 +99,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
// Find fully expired SSTables. Those will be included no matter what.
- expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(SSTableSet.CANONICAL, uncompacting), gcBefore);
+ expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore);
lastExpiredCheck = System.currentTimeMillis();
}
Set<SSTableReader> candidates = Sets.newHashSet(filterSuspectSSTables(uncompacting));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
index d1630c5..e2ab7dc 100644
--- a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
@@ -104,7 +104,7 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
logger.debug("TWCS expired check sufficiently far in the past, checking for fully expired SSTables");
- expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(SSTableSet.CANONICAL, uncompacting), gcBefore);
+ expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore);
lastExpiredCheck = System.currentTimeMillis();
}
else
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java
index 99903fc..96aaa49 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/View.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/View.java
@@ -136,7 +136,7 @@ public class View
return Iterables.concat(sstables, filterOut(compacting, sstables));
}
- private Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
{
switch (sstableSet)
{
@@ -182,7 +182,7 @@ public class View
* Returns the sstables that have any partition between {@code left} and {@code right}, when both bounds are taken inclusively.
* The interval formed by {@code left} and {@code right} shouldn't wrap.
*/
- public Iterable<SSTableReader> sstablesInBounds(SSTableSet sstableSet, PartitionPosition left, PartitionPosition right)
+ public Iterable<SSTableReader> liveSSTablesInBounds(PartitionPosition left, PartitionPosition right)
{
assert !AbstractBounds.strictlyWrapsAround(left, right);
@@ -190,7 +190,7 @@ public class View
return Collections.emptyList();
PartitionPosition stopInTree = right.isMinimum() ? intervalTree.max() : right;
- return select(sstableSet, intervalTree.search(Interval.create(left, stopInTree)));
+ return select(SSTableSet.LIVE, intervalTree.search(Interval.create(left, stopInTree)));
}
public static List<SSTableReader> sstablesInBounds(PartitionPosition left, PartitionPosition right, SSTableIntervalTree intervalTree)
@@ -228,14 +228,14 @@ public class View
* @return a ViewFragment containing the sstables and memtables that may need to be merged
* for rows within @param rowBounds, inclusive, according to the interval tree.
*/
- public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet, AbstractBounds<PartitionPosition> rowBounds)
+ public static Function<View, Iterable<SSTableReader>> selectLive(AbstractBounds<PartitionPosition> rowBounds)
{
// Note that View.sstablesInBounds always includes it's bound while rowBounds may not. This is ok however
// because the fact we restrict the sstables returned by this function is an optimization in the first
// place and the returned sstables will (almost) never cover *exactly* rowBounds anyway. It's also
// *very* unlikely that a sstable is included *just* because we consider one of the bound inclusively
// instead of exclusively, so the performance impact is negligible in practice.
- return (view) -> view.sstablesInBounds(sstableSet, rowBounds.left, rowBounds.right);
+ return (view) -> view.liveSSTablesInBounds(rowBounds.left, rowBounds.right);
}
// METHODS TO CONSTRUCT FUNCTIONS FOR MODIFYING A VIEW:
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/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 d5c060e..a14f815 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -29,8 +29,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.*;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
-import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
+import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.slf4j.Logger;
@@ -332,6 +332,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber
SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.sstables(SSTableSet.CANONICAL));
for (Range<PartitionPosition> keyRange : keyRanges)
{
+ // keyRange excludes its start, while sstableInBounds is inclusive (of both start and end).
+ // This is fine however, because keyRange has been created from a token range through Range.makeRowRange (see above).
+ // And that later method uses the Token.maxKeyBound() method to creates the range, which return a "fake" key that
+ // sort after all keys having the token. That "fake" key cannot however be equal to any real key, so that even
+ // including keyRange.left will still exclude any key having the token of the original token range, and so we're
+ // still actually selecting what we wanted.
for (SSTableReader sstable : View.sstablesInBounds(keyRange.left, keyRange.right, intervalTree))
{
if (!isIncremental || !sstable.isRepaired())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
index 8a5e00e..98f9300 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
@@ -71,7 +71,7 @@ public class ViewTest
continue;
AbstractBounds<PartitionPosition> bounds = AbstractBounds.bounds(min, minInc, max, maxInc);
- List<SSTableReader> r = ImmutableList.copyOf(initialView.sstablesInBounds(SSTableSet.LIVE,bounds.left, bounds.right));
+ List<SSTableReader> r = ImmutableList.copyOf(initialView.liveSSTablesInBounds(bounds.left, bounds.right));
Assert.assertEquals(String.format("%d(%s) %d(%s)", i, minInc, j, maxInc), j - i + (minInc ? 0 : -1) + (maxInc ? 1 : 0), r.size());
}
}
[4/9] cassandra git commit: Avoid missing sstables when getting the
canonical sstables
Posted by ma...@apache.org.
Avoid missing sstables when getting the canonical sstables
Patch by marcuse; reviewed by Stefania Alborghetti for CASSANDRA-11996
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/bc23632f
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/bc23632f
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/bc23632f
Branch: refs/heads/cassandra-3.0
Commit: bc23632f201f760147d8bd1fbee68533fc3f6dfa
Parents: 5b0566a
Author: Marcus Eriksson <ma...@apache.org>
Authored: Mon Jun 13 15:29:08 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 07:57:24 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/ColumnFamilyStore.java | 6 +--
.../cassandra/db/SizeEstimatesRecorder.java | 7 +++-
.../apache/cassandra/db/lifecycle/Tracker.java | 2 +-
.../org/apache/cassandra/db/lifecycle/View.java | 30 +++++++-------
.../apache/cassandra/db/view/ViewBuilder.java | 4 +-
.../cassandra/index/SecondaryIndexManager.java | 2 +-
.../index/internal/CassandraIndex.java | 2 +-
.../io/sstable/IndexSummaryManager.java | 2 +-
.../cassandra/streaming/StreamSession.java | 4 +-
.../apache/cassandra/db/lifecycle/ViewTest.java | 6 +--
.../index/internal/CustomCassandraIndex.java | 2 +-
.../io/sstable/SSTableRewriterTest.java | 41 +++++++++++++++++++-
13 files changed, 78 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 99ac3ad..b3063b4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.9
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
* Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 1be3175..b95e88d 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1482,7 +1482,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public Iterable<SSTableReader> getSSTables(SSTableSet sstableSet)
{
- return data.getView().sstables(sstableSet);
+ return data.getView().select(sstableSet);
}
public Iterable<SSTableReader> getUncompactingSSTables()
@@ -1916,7 +1916,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public Iterable<DecoratedKey> keySamples(Range<Token> range)
{
- try (RefViewFragment view = selectAndReference(View.select(SSTableSet.CANONICAL)))
+ try (RefViewFragment view = selectAndReference(View.selectFunction(SSTableSet.CANONICAL)))
{
Iterable<DecoratedKey>[] samples = new Iterable[view.sstables.size()];
int i = 0;
@@ -1930,7 +1930,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public long estimatedKeysForRange(Range<Token> range)
{
- try (RefViewFragment view = selectAndReference(View.select(SSTableSet.CANONICAL)))
+ try (RefViewFragment view = selectAndReference(View.selectFunction(SSTableSet.CANONICAL)))
{
long count = 0;
for (SSTableReader sstable : view.sstables)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 3461aef..0b31b87 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.dht.Range;
@@ -103,8 +104,10 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable
{
while (refs == null)
{
- // note that this is not guaranteed to return all sstables within the ranges, but for an estimated size, that should be fine
- Iterable<SSTableReader> canonicalSSTables = table.getTracker().getView().select(SSTableSet.CANONICAL, table.select(View.selectLive(Range.makeRowRange(range))).sstables);
+ Iterable<SSTableReader> sstables = table.getTracker().getView().select(SSTableSet.CANONICAL);
+ SSTableIntervalTree tree = SSTableIntervalTree.build(sstables);
+ Range<PartitionPosition> r = Range.makeRowRange(range);
+ Iterable<SSTableReader> canonicalSSTables = View.sstablesInBounds(r.left, r.right, tree);
refs = Refs.tryRef(canonicalSSTables);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
index 16090a1..c94b88f 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
@@ -401,7 +401,7 @@ public class Tracker
public Iterable<SSTableReader> getUncompacting()
{
- return view.get().sstables(SSTableSet.NONCOMPACTING);
+ return view.get().select(SSTableSet.NONCOMPACTING);
}
public Iterable<SSTableReader> getUncompacting(Iterable<SSTableReader> candidates)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java
index 96aaa49..3fa197f 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/View.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/View.java
@@ -118,14 +118,9 @@ public class View
return sstables;
}
- public Iterable<SSTableReader> sstables(SSTableSet sstableSet)
- {
- return select(sstableSet, sstables);
- }
-
public Iterable<SSTableReader> sstables(SSTableSet sstableSet, Predicate<SSTableReader> filter)
{
- return select(sstableSet, filter(sstables, filter));
+ return filter(select(sstableSet), filter);
}
// any sstable known by this tracker in any form; we have a special method here since it's only used for testing/debug
@@ -136,7 +131,7 @@ public class View
return Iterables.concat(sstables, filterOut(compacting, sstables));
}
- public Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Iterable<SSTableReader> select(SSTableSet sstableSet)
{
switch (sstableSet)
{
@@ -145,9 +140,18 @@ public class View
case NONCOMPACTING:
return filter(sstables, (s) -> !compacting.contains(s));
case CANONICAL:
- return transform(filter(sstables,
- (s) -> s.openReason != SSTableReader.OpenReason.EARLY),
- (s) -> s.openReason != SSTableReader.OpenReason.MOVED_START ? s : compactingMap.get(s));
+ Set<SSTableReader> canonicalSSTables = new HashSet<>();
+ for (SSTableReader sstable : compacting)
+ if (sstable.openReason != SSTableReader.OpenReason.EARLY)
+ canonicalSSTables.add(sstable);
+ // reason for checking if compacting contains the sstable is that if compacting has an EARLY version
+ // of a NORMAL sstable, we still have the canonical version of that sstable in sstables.
+ // note that the EARLY version is equal, but not == since it is a different instance of the same sstable.
+ for (SSTableReader sstable : sstables)
+ if (!compacting.contains(sstable) && sstable.openReason != SSTableReader.OpenReason.EARLY)
+ canonicalSSTables.add(sstable);
+
+ return canonicalSSTables;
default:
throw new IllegalStateException();
}
@@ -190,7 +194,7 @@ public class View
return Collections.emptyList();
PartitionPosition stopInTree = right.isMinimum() ? intervalTree.max() : right;
- return select(SSTableSet.LIVE, intervalTree.search(Interval.create(left, stopInTree)));
+ return intervalTree.search(Interval.create(left, stopInTree));
}
public static List<SSTableReader> sstablesInBounds(PartitionPosition left, PartitionPosition right, SSTableIntervalTree intervalTree)
@@ -204,9 +208,9 @@ public class View
return intervalTree.search(Interval.create(left, stopInTree));
}
- public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet)
+ public static Function<View, Iterable<SSTableReader>> selectFunction(SSTableSet sstableSet)
{
- return (view) -> view.sstables(sstableSet);
+ return (view) -> view.select(sstableSet);
}
public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet, Predicate<SSTableReader> filter)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/view/ViewBuilder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/view/ViewBuilder.java b/src/java/org/apache/cassandra/db/view/ViewBuilder.java
index b2b409b..b55eda0 100644
--- a/src/java/org/apache/cassandra/db/view/ViewBuilder.java
+++ b/src/java/org/apache/cassandra/db/view/ViewBuilder.java
@@ -108,7 +108,7 @@ public class ViewBuilder extends CompactionInfo.Holder
if (buildStatus == null)
{
baseCfs.forceBlockingFlush();
- function = org.apache.cassandra.db.lifecycle.View.select(SSTableSet.CANONICAL);
+ function = org.apache.cassandra.db.lifecycle.View.selectFunction(SSTableSet.CANONICAL);
int generation = Integer.MIN_VALUE;
try (Refs<SSTableReader> temp = baseCfs.selectAndReference(function).refs)
@@ -129,7 +129,7 @@ public class ViewBuilder extends CompactionInfo.Holder
@Nullable
public Iterable<SSTableReader> apply(org.apache.cassandra.db.lifecycle.View view)
{
- Iterable<SSTableReader> readers = org.apache.cassandra.db.lifecycle.View.select(SSTableSet.CANONICAL).apply(view);
+ Iterable<SSTableReader> readers = org.apache.cassandra.db.lifecycle.View.selectFunction(SSTableSet.CANONICAL).apply(view);
if (readers != null)
return Iterables.filter(readers, ssTableReader -> ssTableReader.descriptor.generation <= buildStatus.left);
return null;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
index 9635c59..6dfdeee 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
@@ -271,7 +271,7 @@ public class SecondaryIndexManager implements IndexRegistry
{
if (index.shouldBuildBlocking())
{
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
buildIndexesBlocking(sstables, Collections.singleton(index));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
index 9d997a7..2a0dec0 100644
--- a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
+++ b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
@@ -699,7 +699,7 @@ public abstract class CassandraIndex implements Index
{
baseCfs.forceBlockingFlush();
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
if (sstables.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
index aed35c9..ddda430 100644
--- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
+++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
@@ -208,7 +208,7 @@ public class IndexSummaryManager implements IndexSummaryManagerMBean
do
{
View view = cfStore.getTracker().getView();
- allSSTables = ImmutableSet.copyOf(view.sstables(SSTableSet.CANONICAL));
+ allSSTables = ImmutableSet.copyOf(view.select(SSTableSet.CANONICAL));
nonCompacting = ImmutableSet.copyOf(view.getUncompacting(allSSTables));
}
while (null == (txn = cfStore.getTracker().tryModify(nonCompacting, OperationType.UNKNOWN)));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/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 a14f815..bfbedc7 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -329,7 +329,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
keyRanges.add(Range.makeRowRange(range));
refs.addAll(cfStore.selectAndReference(view -> {
Set<SSTableReader> sstables = Sets.newHashSet();
- SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.sstables(SSTableSet.CANONICAL));
+ SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.select(SSTableSet.CANONICAL));
for (Range<PartitionPosition> keyRange : keyRanges)
{
// keyRange excludes its start, while sstableInBounds is inclusive (of both start and end).
@@ -346,7 +346,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
}
if (logger.isDebugEnabled())
- logger.debug("ViewFilter for {}/{} sstables", sstables.size(), Iterables.size(view.sstables(SSTableSet.CANONICAL)));
+ logger.debug("ViewFilter for {}/{} sstables", sstables.size(), Iterables.size(view.select(SSTableSet.CANONICAL)));
return sstables;
}).refs);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
index 98f9300..a5dceca 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
@@ -100,8 +100,8 @@ public class ViewTest
Assert.assertFalse(View.permitCompacting(readers.subList(1, 2)).apply(cur));
Assert.assertTrue(readers.subList(2, 5).containsAll(copyOf(cur.getUncompacting(readers))));
Assert.assertEquals(3, copyOf(cur.getUncompacting(readers)).size());
- Assert.assertTrue(ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING)).containsAll(readers.subList(2, 5)));
- Assert.assertEquals(3, ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING)).size());
+ Assert.assertTrue(ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).containsAll(readers.subList(2, 5)));
+ Assert.assertEquals(3, ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).size());
// check marking already compacting readers fails with an exception
testFailure(View.updateCompacting(emptySet(), readers.subList(0, 1)), cur);
@@ -129,7 +129,7 @@ public class ViewTest
testFailure(View.updateCompacting(copyOf(readers.subList(0, 2)), emptySet()), cur);
Assert.assertTrue(copyOf(concat(readers.subList(0, 1), readers.subList(2, 5))).containsAll(copyOf(cur.getUncompacting(readers))));
Assert.assertEquals(4, copyOf(cur.getUncompacting(readers)).size());
- Set<SSTableReader> nonCompacting = ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING));
+ Set<SSTableReader> nonCompacting = ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING));
Assert.assertTrue(nonCompacting.containsAll(readers.subList(2, 5)));
Assert.assertTrue(nonCompacting.containsAll(readers.subList(0, 1)));
Assert.assertEquals(4, nonCompacting.size());
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
index 6aaefb7..2124abe 100644
--- a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
+++ b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
@@ -623,7 +623,7 @@ public class CustomCassandraIndex implements Index
{
baseCfs.forceBlockingFlush();
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
if (sstables.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
index 18bc760..c842b7f 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
@@ -759,7 +759,7 @@ public class SSTableRewriterTest extends SSTableWriterTestBase
if (!checked && writer.currentWriter().getFilePointer() > 15000000)
{
checked = true;
- ColumnFamilyStore.ViewFragment viewFragment = cfs.select(View.select(SSTableSet.CANONICAL));
+ ColumnFamilyStore.ViewFragment viewFragment = cfs.select(View.selectFunction(SSTableSet.CANONICAL));
// canonical view should have only one SSTable which is not opened early.
assertEquals(1, viewFragment.sstables.size());
SSTableReader sstable = viewFragment.sstables.get(0);
@@ -872,6 +872,45 @@ public class SSTableRewriterTest extends SSTableWriterTestBase
validateCFS(cfs);
}
+ @Test
+ public void testCanonicalSSTables() throws ExecutionException, InterruptedException
+ {
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
+ truncate(cfs);
+
+ cfs.addSSTable(writeFile(cfs, 100));
+ Collection<SSTableReader> allSSTables = cfs.getSSTables();
+ assertEquals(1, allSSTables.size());
+ final AtomicBoolean done = new AtomicBoolean(false);
+ final AtomicBoolean failed = new AtomicBoolean(false);
+ Runnable r = () -> {
+ while (!done.get())
+ {
+ Iterable<SSTableReader> sstables = cfs.getSSTables(SSTableSet.CANONICAL);
+ if (Iterables.size(sstables) != 1)
+ {
+ failed.set(true);
+ return;
+ }
+ }
+ };
+ Thread t = new Thread(r);
+ try
+ {
+ t.start();
+ cfs.forceMajorCompaction();
+ }
+ finally
+ {
+ done.set(true);
+ t.join(20);
+ }
+ assertFalse(failed.get());
+
+
+ }
+
private void validateKeys(Keyspace ks)
{
for (int i = 0; i < 100; i++)
[9/9] cassandra git commit: Merge branch 'cassandra-3.9' into trunk
Posted by ma...@apache.org.
Merge branch 'cassandra-3.9' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/60359948
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/60359948
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/60359948
Branch: refs/heads/trunk
Commit: 60359948d889d73158005cbfa137e6217324a42b
Parents: 2444349 574e8df
Author: Marcus Eriksson <ma...@apache.org>
Authored: Wed Jul 6 08:08:50 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 08:08:50 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../apache/cassandra/db/ColumnFamilyStore.java | 14 +++----
.../cassandra/db/PartitionRangeReadCommand.java | 2 +-
.../cassandra/db/SizeEstimatesRecorder.java | 8 +++-
.../compaction/AbstractCompactionStrategy.java | 2 +-
.../db/compaction/CompactionController.java | 2 +-
.../DateTieredCompactionStrategy.java | 2 +-
.../TimeWindowCompactionStrategy.java | 2 +-
.../apache/cassandra/db/lifecycle/Tracker.java | 2 +-
.../org/apache/cassandra/db/lifecycle/View.java | 36 +++++++++--------
.../apache/cassandra/db/view/ViewBuilder.java | 4 +-
.../cassandra/index/SecondaryIndexManager.java | 2 +-
.../index/internal/CassandraIndex.java | 2 +-
.../io/sstable/IndexSummaryManager.java | 2 +-
.../cassandra/streaming/StreamSession.java | 12 ++++--
.../apache/cassandra/db/lifecycle/ViewTest.java | 8 ++--
.../index/internal/CustomCassandraIndex.java | 2 +-
.../io/sstable/SSTableRewriterTest.java | 41 +++++++++++++++++++-
18 files changed, 100 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/60359948/CHANGES.txt
----------------------------------------------------------------------
[6/9] cassandra git commit: Avoid missing sstables when getting the
canonical sstables
Posted by ma...@apache.org.
Avoid missing sstables when getting the canonical sstables
Patch by marcuse; reviewed by Stefania Alborghetti for CASSANDRA-11996
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/bc23632f
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/bc23632f
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/bc23632f
Branch: refs/heads/trunk
Commit: bc23632f201f760147d8bd1fbee68533fc3f6dfa
Parents: 5b0566a
Author: Marcus Eriksson <ma...@apache.org>
Authored: Mon Jun 13 15:29:08 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 07:57:24 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/ColumnFamilyStore.java | 6 +--
.../cassandra/db/SizeEstimatesRecorder.java | 7 +++-
.../apache/cassandra/db/lifecycle/Tracker.java | 2 +-
.../org/apache/cassandra/db/lifecycle/View.java | 30 +++++++-------
.../apache/cassandra/db/view/ViewBuilder.java | 4 +-
.../cassandra/index/SecondaryIndexManager.java | 2 +-
.../index/internal/CassandraIndex.java | 2 +-
.../io/sstable/IndexSummaryManager.java | 2 +-
.../cassandra/streaming/StreamSession.java | 4 +-
.../apache/cassandra/db/lifecycle/ViewTest.java | 6 +--
.../index/internal/CustomCassandraIndex.java | 2 +-
.../io/sstable/SSTableRewriterTest.java | 41 +++++++++++++++++++-
13 files changed, 78 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 99ac3ad..b3063b4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.9
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
* Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 1be3175..b95e88d 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1482,7 +1482,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public Iterable<SSTableReader> getSSTables(SSTableSet sstableSet)
{
- return data.getView().sstables(sstableSet);
+ return data.getView().select(sstableSet);
}
public Iterable<SSTableReader> getUncompactingSSTables()
@@ -1916,7 +1916,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public Iterable<DecoratedKey> keySamples(Range<Token> range)
{
- try (RefViewFragment view = selectAndReference(View.select(SSTableSet.CANONICAL)))
+ try (RefViewFragment view = selectAndReference(View.selectFunction(SSTableSet.CANONICAL)))
{
Iterable<DecoratedKey>[] samples = new Iterable[view.sstables.size()];
int i = 0;
@@ -1930,7 +1930,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public long estimatedKeysForRange(Range<Token> range)
{
- try (RefViewFragment view = selectAndReference(View.select(SSTableSet.CANONICAL)))
+ try (RefViewFragment view = selectAndReference(View.selectFunction(SSTableSet.CANONICAL)))
{
long count = 0;
for (SSTableReader sstable : view.sstables)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 3461aef..0b31b87 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.dht.Range;
@@ -103,8 +104,10 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable
{
while (refs == null)
{
- // note that this is not guaranteed to return all sstables within the ranges, but for an estimated size, that should be fine
- Iterable<SSTableReader> canonicalSSTables = table.getTracker().getView().select(SSTableSet.CANONICAL, table.select(View.selectLive(Range.makeRowRange(range))).sstables);
+ Iterable<SSTableReader> sstables = table.getTracker().getView().select(SSTableSet.CANONICAL);
+ SSTableIntervalTree tree = SSTableIntervalTree.build(sstables);
+ Range<PartitionPosition> r = Range.makeRowRange(range);
+ Iterable<SSTableReader> canonicalSSTables = View.sstablesInBounds(r.left, r.right, tree);
refs = Refs.tryRef(canonicalSSTables);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
index 16090a1..c94b88f 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
@@ -401,7 +401,7 @@ public class Tracker
public Iterable<SSTableReader> getUncompacting()
{
- return view.get().sstables(SSTableSet.NONCOMPACTING);
+ return view.get().select(SSTableSet.NONCOMPACTING);
}
public Iterable<SSTableReader> getUncompacting(Iterable<SSTableReader> candidates)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java
index 96aaa49..3fa197f 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/View.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/View.java
@@ -118,14 +118,9 @@ public class View
return sstables;
}
- public Iterable<SSTableReader> sstables(SSTableSet sstableSet)
- {
- return select(sstableSet, sstables);
- }
-
public Iterable<SSTableReader> sstables(SSTableSet sstableSet, Predicate<SSTableReader> filter)
{
- return select(sstableSet, filter(sstables, filter));
+ return filter(select(sstableSet), filter);
}
// any sstable known by this tracker in any form; we have a special method here since it's only used for testing/debug
@@ -136,7 +131,7 @@ public class View
return Iterables.concat(sstables, filterOut(compacting, sstables));
}
- public Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Iterable<SSTableReader> select(SSTableSet sstableSet)
{
switch (sstableSet)
{
@@ -145,9 +140,18 @@ public class View
case NONCOMPACTING:
return filter(sstables, (s) -> !compacting.contains(s));
case CANONICAL:
- return transform(filter(sstables,
- (s) -> s.openReason != SSTableReader.OpenReason.EARLY),
- (s) -> s.openReason != SSTableReader.OpenReason.MOVED_START ? s : compactingMap.get(s));
+ Set<SSTableReader> canonicalSSTables = new HashSet<>();
+ for (SSTableReader sstable : compacting)
+ if (sstable.openReason != SSTableReader.OpenReason.EARLY)
+ canonicalSSTables.add(sstable);
+ // reason for checking if compacting contains the sstable is that if compacting has an EARLY version
+ // of a NORMAL sstable, we still have the canonical version of that sstable in sstables.
+ // note that the EARLY version is equal, but not == since it is a different instance of the same sstable.
+ for (SSTableReader sstable : sstables)
+ if (!compacting.contains(sstable) && sstable.openReason != SSTableReader.OpenReason.EARLY)
+ canonicalSSTables.add(sstable);
+
+ return canonicalSSTables;
default:
throw new IllegalStateException();
}
@@ -190,7 +194,7 @@ public class View
return Collections.emptyList();
PartitionPosition stopInTree = right.isMinimum() ? intervalTree.max() : right;
- return select(SSTableSet.LIVE, intervalTree.search(Interval.create(left, stopInTree)));
+ return intervalTree.search(Interval.create(left, stopInTree));
}
public static List<SSTableReader> sstablesInBounds(PartitionPosition left, PartitionPosition right, SSTableIntervalTree intervalTree)
@@ -204,9 +208,9 @@ public class View
return intervalTree.search(Interval.create(left, stopInTree));
}
- public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet)
+ public static Function<View, Iterable<SSTableReader>> selectFunction(SSTableSet sstableSet)
{
- return (view) -> view.sstables(sstableSet);
+ return (view) -> view.select(sstableSet);
}
public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet, Predicate<SSTableReader> filter)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/view/ViewBuilder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/view/ViewBuilder.java b/src/java/org/apache/cassandra/db/view/ViewBuilder.java
index b2b409b..b55eda0 100644
--- a/src/java/org/apache/cassandra/db/view/ViewBuilder.java
+++ b/src/java/org/apache/cassandra/db/view/ViewBuilder.java
@@ -108,7 +108,7 @@ public class ViewBuilder extends CompactionInfo.Holder
if (buildStatus == null)
{
baseCfs.forceBlockingFlush();
- function = org.apache.cassandra.db.lifecycle.View.select(SSTableSet.CANONICAL);
+ function = org.apache.cassandra.db.lifecycle.View.selectFunction(SSTableSet.CANONICAL);
int generation = Integer.MIN_VALUE;
try (Refs<SSTableReader> temp = baseCfs.selectAndReference(function).refs)
@@ -129,7 +129,7 @@ public class ViewBuilder extends CompactionInfo.Holder
@Nullable
public Iterable<SSTableReader> apply(org.apache.cassandra.db.lifecycle.View view)
{
- Iterable<SSTableReader> readers = org.apache.cassandra.db.lifecycle.View.select(SSTableSet.CANONICAL).apply(view);
+ Iterable<SSTableReader> readers = org.apache.cassandra.db.lifecycle.View.selectFunction(SSTableSet.CANONICAL).apply(view);
if (readers != null)
return Iterables.filter(readers, ssTableReader -> ssTableReader.descriptor.generation <= buildStatus.left);
return null;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
index 9635c59..6dfdeee 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
@@ -271,7 +271,7 @@ public class SecondaryIndexManager implements IndexRegistry
{
if (index.shouldBuildBlocking())
{
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
buildIndexesBlocking(sstables, Collections.singleton(index));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
index 9d997a7..2a0dec0 100644
--- a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
+++ b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
@@ -699,7 +699,7 @@ public abstract class CassandraIndex implements Index
{
baseCfs.forceBlockingFlush();
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
if (sstables.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
index aed35c9..ddda430 100644
--- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
+++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
@@ -208,7 +208,7 @@ public class IndexSummaryManager implements IndexSummaryManagerMBean
do
{
View view = cfStore.getTracker().getView();
- allSSTables = ImmutableSet.copyOf(view.sstables(SSTableSet.CANONICAL));
+ allSSTables = ImmutableSet.copyOf(view.select(SSTableSet.CANONICAL));
nonCompacting = ImmutableSet.copyOf(view.getUncompacting(allSSTables));
}
while (null == (txn = cfStore.getTracker().tryModify(nonCompacting, OperationType.UNKNOWN)));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/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 a14f815..bfbedc7 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -329,7 +329,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
keyRanges.add(Range.makeRowRange(range));
refs.addAll(cfStore.selectAndReference(view -> {
Set<SSTableReader> sstables = Sets.newHashSet();
- SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.sstables(SSTableSet.CANONICAL));
+ SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.select(SSTableSet.CANONICAL));
for (Range<PartitionPosition> keyRange : keyRanges)
{
// keyRange excludes its start, while sstableInBounds is inclusive (of both start and end).
@@ -346,7 +346,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
}
if (logger.isDebugEnabled())
- logger.debug("ViewFilter for {}/{} sstables", sstables.size(), Iterables.size(view.sstables(SSTableSet.CANONICAL)));
+ logger.debug("ViewFilter for {}/{} sstables", sstables.size(), Iterables.size(view.select(SSTableSet.CANONICAL)));
return sstables;
}).refs);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
index 98f9300..a5dceca 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
@@ -100,8 +100,8 @@ public class ViewTest
Assert.assertFalse(View.permitCompacting(readers.subList(1, 2)).apply(cur));
Assert.assertTrue(readers.subList(2, 5).containsAll(copyOf(cur.getUncompacting(readers))));
Assert.assertEquals(3, copyOf(cur.getUncompacting(readers)).size());
- Assert.assertTrue(ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING)).containsAll(readers.subList(2, 5)));
- Assert.assertEquals(3, ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING)).size());
+ Assert.assertTrue(ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).containsAll(readers.subList(2, 5)));
+ Assert.assertEquals(3, ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).size());
// check marking already compacting readers fails with an exception
testFailure(View.updateCompacting(emptySet(), readers.subList(0, 1)), cur);
@@ -129,7 +129,7 @@ public class ViewTest
testFailure(View.updateCompacting(copyOf(readers.subList(0, 2)), emptySet()), cur);
Assert.assertTrue(copyOf(concat(readers.subList(0, 1), readers.subList(2, 5))).containsAll(copyOf(cur.getUncompacting(readers))));
Assert.assertEquals(4, copyOf(cur.getUncompacting(readers)).size());
- Set<SSTableReader> nonCompacting = ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING));
+ Set<SSTableReader> nonCompacting = ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING));
Assert.assertTrue(nonCompacting.containsAll(readers.subList(2, 5)));
Assert.assertTrue(nonCompacting.containsAll(readers.subList(0, 1)));
Assert.assertEquals(4, nonCompacting.size());
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
index 6aaefb7..2124abe 100644
--- a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
+++ b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
@@ -623,7 +623,7 @@ public class CustomCassandraIndex implements Index
{
baseCfs.forceBlockingFlush();
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
if (sstables.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
index 18bc760..c842b7f 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
@@ -759,7 +759,7 @@ public class SSTableRewriterTest extends SSTableWriterTestBase
if (!checked && writer.currentWriter().getFilePointer() > 15000000)
{
checked = true;
- ColumnFamilyStore.ViewFragment viewFragment = cfs.select(View.select(SSTableSet.CANONICAL));
+ ColumnFamilyStore.ViewFragment viewFragment = cfs.select(View.selectFunction(SSTableSet.CANONICAL));
// canonical view should have only one SSTable which is not opened early.
assertEquals(1, viewFragment.sstables.size());
SSTableReader sstable = viewFragment.sstables.get(0);
@@ -872,6 +872,45 @@ public class SSTableRewriterTest extends SSTableWriterTestBase
validateCFS(cfs);
}
+ @Test
+ public void testCanonicalSSTables() throws ExecutionException, InterruptedException
+ {
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
+ truncate(cfs);
+
+ cfs.addSSTable(writeFile(cfs, 100));
+ Collection<SSTableReader> allSSTables = cfs.getSSTables();
+ assertEquals(1, allSSTables.size());
+ final AtomicBoolean done = new AtomicBoolean(false);
+ final AtomicBoolean failed = new AtomicBoolean(false);
+ Runnable r = () -> {
+ while (!done.get())
+ {
+ Iterable<SSTableReader> sstables = cfs.getSSTables(SSTableSet.CANONICAL);
+ if (Iterables.size(sstables) != 1)
+ {
+ failed.set(true);
+ return;
+ }
+ }
+ };
+ Thread t = new Thread(r);
+ try
+ {
+ t.start();
+ cfs.forceMajorCompaction();
+ }
+ finally
+ {
+ done.set(true);
+ t.join(20);
+ }
+ assertFalse(failed.get());
+
+
+ }
+
private void validateKeys(Keyspace ks)
{
for (int i = 0; i < 100; i++)
[7/9] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.9
Posted by ma...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.9
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/574e8df2
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/574e8df2
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/574e8df2
Branch: refs/heads/cassandra-3.9
Commit: 574e8df207f46d1af361a4a651e6b6cdde18688c
Parents: 78f077d bc23632
Author: Marcus Eriksson <ma...@apache.org>
Authored: Wed Jul 6 08:08:42 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 08:08:42 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../apache/cassandra/db/ColumnFamilyStore.java | 14 +++----
.../cassandra/db/PartitionRangeReadCommand.java | 2 +-
.../cassandra/db/SizeEstimatesRecorder.java | 8 +++-
.../compaction/AbstractCompactionStrategy.java | 2 +-
.../db/compaction/CompactionController.java | 2 +-
.../DateTieredCompactionStrategy.java | 2 +-
.../TimeWindowCompactionStrategy.java | 2 +-
.../apache/cassandra/db/lifecycle/Tracker.java | 2 +-
.../org/apache/cassandra/db/lifecycle/View.java | 36 +++++++++--------
.../apache/cassandra/db/view/ViewBuilder.java | 4 +-
.../cassandra/index/SecondaryIndexManager.java | 2 +-
.../index/internal/CassandraIndex.java | 2 +-
.../io/sstable/IndexSummaryManager.java | 2 +-
.../cassandra/streaming/StreamSession.java | 12 ++++--
.../apache/cassandra/db/lifecycle/ViewTest.java | 8 ++--
.../index/internal/CustomCassandraIndex.java | 2 +-
.../io/sstable/SSTableRewriterTest.java | 41 +++++++++++++++++++-
18 files changed, 100 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index aa3e939,b3063b4..afeef2c
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,7 -1,6 +1,9 @@@
-3.0.9
+3.9
+ * Fix SASI PREFIX search in CONTAINS mode with partial terms (CASSANDRA-12073)
+ * Increase size of flushExecutor thread pool (CASSANDRA-12071)
+Merged from 3.0:
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
those static columns in query results (CASSANDRA-12123)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index bca1877,842ad5f..a8959c2
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@@ -173,9 -173,9 +173,9 @@@ public class PartitionRangeReadCommand
metric.rangeLatency.addNano(latencyNanos);
}
- protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
+ protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController executionController)
{
- ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, dataRange().keyRange()));
+ ColumnFamilyStore.ViewFragment view = cfs.select(View.selectLive(dataRange().keyRange()));
Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().getKeyValidator()));
// fetch data from current memtable, historical memtables, and SSTables in the correct order.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/CompactionController.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/view/ViewBuilder.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
----------------------------------------------------------------------
[2/9] cassandra git commit: Always select the live sstables when
getting sstables in bounds
Posted by ma...@apache.org.
Always select the live sstables when getting sstables in bounds
Patch by marcuse; reviewed by benedict for CASSANDRA-11944
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5b0566a7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5b0566a7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5b0566a7
Branch: refs/heads/cassandra-3.9
Commit: 5b0566a70f373d6eb537c89c0db2a2e224706916
Parents: 2217695
Author: Marcus Eriksson <ma...@apache.org>
Authored: Thu Jun 2 09:37:06 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 07:52:28 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/db/ColumnFamilyStore.java | 8 ++++----
.../apache/cassandra/db/PartitionRangeReadCommand.java | 2 +-
.../org/apache/cassandra/db/SizeEstimatesRecorder.java | 5 +++--
.../db/compaction/AbstractCompactionStrategy.java | 2 +-
.../cassandra/db/compaction/CompactionController.java | 2 +-
.../db/compaction/DateTieredCompactionStrategy.java | 2 +-
.../db/compaction/TimeWindowCompactionStrategy.java | 2 +-
src/java/org/apache/cassandra/db/lifecycle/View.java | 10 +++++-----
.../org/apache/cassandra/streaming/StreamSession.java | 8 +++++++-
test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java | 2 +-
11 files changed, 26 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 7f8a3a1..99ac3ad 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.9
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
those static columns in query results (CASSANDRA-12123)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 3264327..1be3175 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1231,7 +1231,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @return sstables whose key range overlaps with that of the given sstables, not including itself.
* (The given sstables may or may not overlap with each other.)
*/
- public Collection<SSTableReader> getOverlappingSSTables(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Collection<SSTableReader> getOverlappingLiveSSTables(Iterable<SSTableReader> sstables)
{
logger.trace("Checking for sstables overlapping {}", sstables);
@@ -1282,7 +1282,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Set<SSTableReader> results = new HashSet<>();
for (AbstractBounds<PartitionPosition> bound : bounds)
- Iterables.addAll(results, view.sstablesInBounds(sstableSet, bound.left, bound.right));
+ Iterables.addAll(results, view.liveSSTablesInBounds(bound.left, bound.right));
return Sets.difference(results, ImmutableSet.copyOf(sstables));
}
@@ -1290,11 +1290,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
/**
* like getOverlappingSSTables, but acquires references before returning
*/
- public Refs<SSTableReader> getAndReferenceOverlappingSSTables(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Refs<SSTableReader> getAndReferenceOverlappingLiveSSTables(Iterable<SSTableReader> sstables)
{
while (true)
{
- Iterable<SSTableReader> overlapped = getOverlappingSSTables(sstableSet, sstables);
+ Iterable<SSTableReader> overlapped = getOverlappingLiveSSTables(sstables);
Refs<SSTableReader> refs = Refs.tryRef(overlapped);
if (refs != null)
return refs;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index 9585b59..842ad5f 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -175,7 +175,7 @@ public class PartitionRangeReadCommand extends ReadCommand
protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
{
- ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, dataRange().keyRange()));
+ ColumnFamilyStore.ViewFragment view = cfs.select(View.selectLive(dataRange().keyRange()));
Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().getKeyValidator()));
// fetch data from current memtable, historical memtables, and SSTables in the correct order.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 2a74ea9..3461aef 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -103,8 +103,9 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable
{
while (refs == null)
{
- ColumnFamilyStore.ViewFragment view = table.select(View.select(SSTableSet.CANONICAL, Range.makeRowRange(range)));
- refs = Refs.tryRef(view.sstables);
+ // note that this is not guaranteed to return all sstables within the ranges, but for an estimated size, that should be fine
+ Iterable<SSTableReader> canonicalSSTables = table.getTracker().getView().select(SSTableSet.CANONICAL, table.select(View.selectLive(Range.makeRowRange(range))).sstables);
+ refs = Refs.tryRef(canonicalSSTables);
}
// calculate the estimates.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
index e30e4f7..0dce52b 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
@@ -380,7 +380,7 @@ public abstract class AbstractCompactionStrategy
if (uncheckedTombstoneCompaction)
return true;
- Collection<SSTableReader> overlaps = cfs.getOverlappingSSTables(SSTableSet.CANONICAL, Collections.singleton(sstable));
+ Collection<SSTableReader> overlaps = cfs.getOverlappingLiveSSTables(Collections.singleton(sstable));
if (overlaps.isEmpty())
{
// there is no overlap, tombstones are safely droppable
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/CompactionController.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionController.java b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
index a5b8308..fbf29e3 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
@@ -103,7 +103,7 @@ public class CompactionController implements AutoCloseable
if (compacting == null)
overlappingSSTables = Refs.tryRef(Collections.<SSTableReader>emptyList());
else
- overlappingSSTables = cfs.getAndReferenceOverlappingSSTables(SSTableSet.LIVE, compacting);
+ overlappingSSTables = cfs.getAndReferenceOverlappingLiveSSTables(compacting);
this.overlapIterator = new OverlapIterator<>(buildIntervals(overlappingSSTables));
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
index 8571906..3e6ae61 100644
--- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
@@ -99,7 +99,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
// Find fully expired SSTables. Those will be included no matter what.
- expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(SSTableSet.CANONICAL, uncompacting), gcBefore);
+ expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore);
lastExpiredCheck = System.currentTimeMillis();
}
Set<SSTableReader> candidates = Sets.newHashSet(filterSuspectSSTables(uncompacting));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
index d1630c5..e2ab7dc 100644
--- a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
@@ -104,7 +104,7 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
logger.debug("TWCS expired check sufficiently far in the past, checking for fully expired SSTables");
- expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(SSTableSet.CANONICAL, uncompacting), gcBefore);
+ expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore);
lastExpiredCheck = System.currentTimeMillis();
}
else
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java
index 99903fc..96aaa49 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/View.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/View.java
@@ -136,7 +136,7 @@ public class View
return Iterables.concat(sstables, filterOut(compacting, sstables));
}
- private Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
{
switch (sstableSet)
{
@@ -182,7 +182,7 @@ public class View
* Returns the sstables that have any partition between {@code left} and {@code right}, when both bounds are taken inclusively.
* The interval formed by {@code left} and {@code right} shouldn't wrap.
*/
- public Iterable<SSTableReader> sstablesInBounds(SSTableSet sstableSet, PartitionPosition left, PartitionPosition right)
+ public Iterable<SSTableReader> liveSSTablesInBounds(PartitionPosition left, PartitionPosition right)
{
assert !AbstractBounds.strictlyWrapsAround(left, right);
@@ -190,7 +190,7 @@ public class View
return Collections.emptyList();
PartitionPosition stopInTree = right.isMinimum() ? intervalTree.max() : right;
- return select(sstableSet, intervalTree.search(Interval.create(left, stopInTree)));
+ return select(SSTableSet.LIVE, intervalTree.search(Interval.create(left, stopInTree)));
}
public static List<SSTableReader> sstablesInBounds(PartitionPosition left, PartitionPosition right, SSTableIntervalTree intervalTree)
@@ -228,14 +228,14 @@ public class View
* @return a ViewFragment containing the sstables and memtables that may need to be merged
* for rows within @param rowBounds, inclusive, according to the interval tree.
*/
- public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet, AbstractBounds<PartitionPosition> rowBounds)
+ public static Function<View, Iterable<SSTableReader>> selectLive(AbstractBounds<PartitionPosition> rowBounds)
{
// Note that View.sstablesInBounds always includes it's bound while rowBounds may not. This is ok however
// because the fact we restrict the sstables returned by this function is an optimization in the first
// place and the returned sstables will (almost) never cover *exactly* rowBounds anyway. It's also
// *very* unlikely that a sstable is included *just* because we consider one of the bound inclusively
// instead of exclusively, so the performance impact is negligible in practice.
- return (view) -> view.sstablesInBounds(sstableSet, rowBounds.left, rowBounds.right);
+ return (view) -> view.liveSSTablesInBounds(rowBounds.left, rowBounds.right);
}
// METHODS TO CONSTRUCT FUNCTIONS FOR MODIFYING A VIEW:
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/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 d5c060e..a14f815 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -29,8 +29,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.*;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
-import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
+import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.slf4j.Logger;
@@ -332,6 +332,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber
SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.sstables(SSTableSet.CANONICAL));
for (Range<PartitionPosition> keyRange : keyRanges)
{
+ // keyRange excludes its start, while sstableInBounds is inclusive (of both start and end).
+ // This is fine however, because keyRange has been created from a token range through Range.makeRowRange (see above).
+ // And that later method uses the Token.maxKeyBound() method to creates the range, which return a "fake" key that
+ // sort after all keys having the token. That "fake" key cannot however be equal to any real key, so that even
+ // including keyRange.left will still exclude any key having the token of the original token range, and so we're
+ // still actually selecting what we wanted.
for (SSTableReader sstable : View.sstablesInBounds(keyRange.left, keyRange.right, intervalTree))
{
if (!isIncremental || !sstable.isRepaired())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
index 8a5e00e..98f9300 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
@@ -71,7 +71,7 @@ public class ViewTest
continue;
AbstractBounds<PartitionPosition> bounds = AbstractBounds.bounds(min, minInc, max, maxInc);
- List<SSTableReader> r = ImmutableList.copyOf(initialView.sstablesInBounds(SSTableSet.LIVE,bounds.left, bounds.right));
+ List<SSTableReader> r = ImmutableList.copyOf(initialView.liveSSTablesInBounds(bounds.left, bounds.right));
Assert.assertEquals(String.format("%d(%s) %d(%s)", i, minInc, j, maxInc), j - i + (minInc ? 0 : -1) + (maxInc ? 1 : 0), r.size());
}
}
[8/9] cassandra git commit: Merge branch 'cassandra-3.0' into
cassandra-3.9
Posted by ma...@apache.org.
Merge branch 'cassandra-3.0' into cassandra-3.9
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/574e8df2
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/574e8df2
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/574e8df2
Branch: refs/heads/trunk
Commit: 574e8df207f46d1af361a4a651e6b6cdde18688c
Parents: 78f077d bc23632
Author: Marcus Eriksson <ma...@apache.org>
Authored: Wed Jul 6 08:08:42 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 08:08:42 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../apache/cassandra/db/ColumnFamilyStore.java | 14 +++----
.../cassandra/db/PartitionRangeReadCommand.java | 2 +-
.../cassandra/db/SizeEstimatesRecorder.java | 8 +++-
.../compaction/AbstractCompactionStrategy.java | 2 +-
.../db/compaction/CompactionController.java | 2 +-
.../DateTieredCompactionStrategy.java | 2 +-
.../TimeWindowCompactionStrategy.java | 2 +-
.../apache/cassandra/db/lifecycle/Tracker.java | 2 +-
.../org/apache/cassandra/db/lifecycle/View.java | 36 +++++++++--------
.../apache/cassandra/db/view/ViewBuilder.java | 4 +-
.../cassandra/index/SecondaryIndexManager.java | 2 +-
.../index/internal/CassandraIndex.java | 2 +-
.../io/sstable/IndexSummaryManager.java | 2 +-
.../cassandra/streaming/StreamSession.java | 12 ++++--
.../apache/cassandra/db/lifecycle/ViewTest.java | 8 ++--
.../index/internal/CustomCassandraIndex.java | 2 +-
.../io/sstable/SSTableRewriterTest.java | 41 +++++++++++++++++++-
18 files changed, 100 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index aa3e939,b3063b4..afeef2c
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,7 -1,6 +1,9 @@@
-3.0.9
+3.9
+ * Fix SASI PREFIX search in CONTAINS mode with partial terms (CASSANDRA-12073)
+ * Increase size of flushExecutor thread pool (CASSANDRA-12071)
+Merged from 3.0:
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
those static columns in query results (CASSANDRA-12123)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index bca1877,842ad5f..a8959c2
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@@ -173,9 -173,9 +173,9 @@@ public class PartitionRangeReadCommand
metric.rangeLatency.addNano(latencyNanos);
}
- protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
+ protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController executionController)
{
- ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, dataRange().keyRange()));
+ ColumnFamilyStore.ViewFragment view = cfs.select(View.selectLive(dataRange().keyRange()));
Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().getKeyValidator()));
// fetch data from current memtable, historical memtables, and SSTables in the correct order.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/CompactionController.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/db/view/ViewBuilder.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/src/java/org/apache/cassandra/streaming/StreamSession.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/574e8df2/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
----------------------------------------------------------------------
[5/9] cassandra git commit: Avoid missing sstables when getting the
canonical sstables
Posted by ma...@apache.org.
Avoid missing sstables when getting the canonical sstables
Patch by marcuse; reviewed by Stefania Alborghetti for CASSANDRA-11996
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/bc23632f
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/bc23632f
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/bc23632f
Branch: refs/heads/cassandra-3.9
Commit: bc23632f201f760147d8bd1fbee68533fc3f6dfa
Parents: 5b0566a
Author: Marcus Eriksson <ma...@apache.org>
Authored: Mon Jun 13 15:29:08 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 07:57:24 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../apache/cassandra/db/ColumnFamilyStore.java | 6 +--
.../cassandra/db/SizeEstimatesRecorder.java | 7 +++-
.../apache/cassandra/db/lifecycle/Tracker.java | 2 +-
.../org/apache/cassandra/db/lifecycle/View.java | 30 +++++++-------
.../apache/cassandra/db/view/ViewBuilder.java | 4 +-
.../cassandra/index/SecondaryIndexManager.java | 2 +-
.../index/internal/CassandraIndex.java | 2 +-
.../io/sstable/IndexSummaryManager.java | 2 +-
.../cassandra/streaming/StreamSession.java | 4 +-
.../apache/cassandra/db/lifecycle/ViewTest.java | 6 +--
.../index/internal/CustomCassandraIndex.java | 2 +-
.../io/sstable/SSTableRewriterTest.java | 41 +++++++++++++++++++-
13 files changed, 78 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 99ac3ad..b3063b4 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.9
+ * Avoid missing sstables when getting the canonical sstables (CASSANDRA-11996)
* Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 1be3175..b95e88d 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1482,7 +1482,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public Iterable<SSTableReader> getSSTables(SSTableSet sstableSet)
{
- return data.getView().sstables(sstableSet);
+ return data.getView().select(sstableSet);
}
public Iterable<SSTableReader> getUncompactingSSTables()
@@ -1916,7 +1916,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public Iterable<DecoratedKey> keySamples(Range<Token> range)
{
- try (RefViewFragment view = selectAndReference(View.select(SSTableSet.CANONICAL)))
+ try (RefViewFragment view = selectAndReference(View.selectFunction(SSTableSet.CANONICAL)))
{
Iterable<DecoratedKey>[] samples = new Iterable[view.sstables.size()];
int i = 0;
@@ -1930,7 +1930,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public long estimatedKeysForRange(Range<Token> range)
{
- try (RefViewFragment view = selectAndReference(View.select(SSTableSet.CANONICAL)))
+ try (RefViewFragment view = selectAndReference(View.selectFunction(SSTableSet.CANONICAL)))
{
long count = 0;
for (SSTableReader sstable : view.sstables)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 3461aef..0b31b87 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.dht.Range;
@@ -103,8 +104,10 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable
{
while (refs == null)
{
- // note that this is not guaranteed to return all sstables within the ranges, but for an estimated size, that should be fine
- Iterable<SSTableReader> canonicalSSTables = table.getTracker().getView().select(SSTableSet.CANONICAL, table.select(View.selectLive(Range.makeRowRange(range))).sstables);
+ Iterable<SSTableReader> sstables = table.getTracker().getView().select(SSTableSet.CANONICAL);
+ SSTableIntervalTree tree = SSTableIntervalTree.build(sstables);
+ Range<PartitionPosition> r = Range.makeRowRange(range);
+ Iterable<SSTableReader> canonicalSSTables = View.sstablesInBounds(r.left, r.right, tree);
refs = Refs.tryRef(canonicalSSTables);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
index 16090a1..c94b88f 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java
@@ -401,7 +401,7 @@ public class Tracker
public Iterable<SSTableReader> getUncompacting()
{
- return view.get().sstables(SSTableSet.NONCOMPACTING);
+ return view.get().select(SSTableSet.NONCOMPACTING);
}
public Iterable<SSTableReader> getUncompacting(Iterable<SSTableReader> candidates)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java
index 96aaa49..3fa197f 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/View.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/View.java
@@ -118,14 +118,9 @@ public class View
return sstables;
}
- public Iterable<SSTableReader> sstables(SSTableSet sstableSet)
- {
- return select(sstableSet, sstables);
- }
-
public Iterable<SSTableReader> sstables(SSTableSet sstableSet, Predicate<SSTableReader> filter)
{
- return select(sstableSet, filter(sstables, filter));
+ return filter(select(sstableSet), filter);
}
// any sstable known by this tracker in any form; we have a special method here since it's only used for testing/debug
@@ -136,7 +131,7 @@ public class View
return Iterables.concat(sstables, filterOut(compacting, sstables));
}
- public Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Iterable<SSTableReader> select(SSTableSet sstableSet)
{
switch (sstableSet)
{
@@ -145,9 +140,18 @@ public class View
case NONCOMPACTING:
return filter(sstables, (s) -> !compacting.contains(s));
case CANONICAL:
- return transform(filter(sstables,
- (s) -> s.openReason != SSTableReader.OpenReason.EARLY),
- (s) -> s.openReason != SSTableReader.OpenReason.MOVED_START ? s : compactingMap.get(s));
+ Set<SSTableReader> canonicalSSTables = new HashSet<>();
+ for (SSTableReader sstable : compacting)
+ if (sstable.openReason != SSTableReader.OpenReason.EARLY)
+ canonicalSSTables.add(sstable);
+ // reason for checking if compacting contains the sstable is that if compacting has an EARLY version
+ // of a NORMAL sstable, we still have the canonical version of that sstable in sstables.
+ // note that the EARLY version is equal, but not == since it is a different instance of the same sstable.
+ for (SSTableReader sstable : sstables)
+ if (!compacting.contains(sstable) && sstable.openReason != SSTableReader.OpenReason.EARLY)
+ canonicalSSTables.add(sstable);
+
+ return canonicalSSTables;
default:
throw new IllegalStateException();
}
@@ -190,7 +194,7 @@ public class View
return Collections.emptyList();
PartitionPosition stopInTree = right.isMinimum() ? intervalTree.max() : right;
- return select(SSTableSet.LIVE, intervalTree.search(Interval.create(left, stopInTree)));
+ return intervalTree.search(Interval.create(left, stopInTree));
}
public static List<SSTableReader> sstablesInBounds(PartitionPosition left, PartitionPosition right, SSTableIntervalTree intervalTree)
@@ -204,9 +208,9 @@ public class View
return intervalTree.search(Interval.create(left, stopInTree));
}
- public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet)
+ public static Function<View, Iterable<SSTableReader>> selectFunction(SSTableSet sstableSet)
{
- return (view) -> view.sstables(sstableSet);
+ return (view) -> view.select(sstableSet);
}
public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet, Predicate<SSTableReader> filter)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/db/view/ViewBuilder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/view/ViewBuilder.java b/src/java/org/apache/cassandra/db/view/ViewBuilder.java
index b2b409b..b55eda0 100644
--- a/src/java/org/apache/cassandra/db/view/ViewBuilder.java
+++ b/src/java/org/apache/cassandra/db/view/ViewBuilder.java
@@ -108,7 +108,7 @@ public class ViewBuilder extends CompactionInfo.Holder
if (buildStatus == null)
{
baseCfs.forceBlockingFlush();
- function = org.apache.cassandra.db.lifecycle.View.select(SSTableSet.CANONICAL);
+ function = org.apache.cassandra.db.lifecycle.View.selectFunction(SSTableSet.CANONICAL);
int generation = Integer.MIN_VALUE;
try (Refs<SSTableReader> temp = baseCfs.selectAndReference(function).refs)
@@ -129,7 +129,7 @@ public class ViewBuilder extends CompactionInfo.Holder
@Nullable
public Iterable<SSTableReader> apply(org.apache.cassandra.db.lifecycle.View view)
{
- Iterable<SSTableReader> readers = org.apache.cassandra.db.lifecycle.View.select(SSTableSet.CANONICAL).apply(view);
+ Iterable<SSTableReader> readers = org.apache.cassandra.db.lifecycle.View.selectFunction(SSTableSet.CANONICAL).apply(view);
if (readers != null)
return Iterables.filter(readers, ssTableReader -> ssTableReader.descriptor.generation <= buildStatus.left);
return null;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
index 9635c59..6dfdeee 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
@@ -271,7 +271,7 @@ public class SecondaryIndexManager implements IndexRegistry
{
if (index.shouldBuildBlocking())
{
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
buildIndexesBlocking(sstables, Collections.singleton(index));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
index 9d997a7..2a0dec0 100644
--- a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
+++ b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
@@ -699,7 +699,7 @@ public abstract class CassandraIndex implements Index
{
baseCfs.forceBlockingFlush();
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
if (sstables.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
index aed35c9..ddda430 100644
--- a/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
+++ b/src/java/org/apache/cassandra/io/sstable/IndexSummaryManager.java
@@ -208,7 +208,7 @@ public class IndexSummaryManager implements IndexSummaryManagerMBean
do
{
View view = cfStore.getTracker().getView();
- allSSTables = ImmutableSet.copyOf(view.sstables(SSTableSet.CANONICAL));
+ allSSTables = ImmutableSet.copyOf(view.select(SSTableSet.CANONICAL));
nonCompacting = ImmutableSet.copyOf(view.getUncompacting(allSSTables));
}
while (null == (txn = cfStore.getTracker().tryModify(nonCompacting, OperationType.UNKNOWN)));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/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 a14f815..bfbedc7 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -329,7 +329,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
keyRanges.add(Range.makeRowRange(range));
refs.addAll(cfStore.selectAndReference(view -> {
Set<SSTableReader> sstables = Sets.newHashSet();
- SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.sstables(SSTableSet.CANONICAL));
+ SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.select(SSTableSet.CANONICAL));
for (Range<PartitionPosition> keyRange : keyRanges)
{
// keyRange excludes its start, while sstableInBounds is inclusive (of both start and end).
@@ -346,7 +346,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
}
if (logger.isDebugEnabled())
- logger.debug("ViewFilter for {}/{} sstables", sstables.size(), Iterables.size(view.sstables(SSTableSet.CANONICAL)));
+ logger.debug("ViewFilter for {}/{} sstables", sstables.size(), Iterables.size(view.select(SSTableSet.CANONICAL)));
return sstables;
}).refs);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
index 98f9300..a5dceca 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
@@ -100,8 +100,8 @@ public class ViewTest
Assert.assertFalse(View.permitCompacting(readers.subList(1, 2)).apply(cur));
Assert.assertTrue(readers.subList(2, 5).containsAll(copyOf(cur.getUncompacting(readers))));
Assert.assertEquals(3, copyOf(cur.getUncompacting(readers)).size());
- Assert.assertTrue(ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING)).containsAll(readers.subList(2, 5)));
- Assert.assertEquals(3, ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING)).size());
+ Assert.assertTrue(ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).containsAll(readers.subList(2, 5)));
+ Assert.assertEquals(3, ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING)).size());
// check marking already compacting readers fails with an exception
testFailure(View.updateCompacting(emptySet(), readers.subList(0, 1)), cur);
@@ -129,7 +129,7 @@ public class ViewTest
testFailure(View.updateCompacting(copyOf(readers.subList(0, 2)), emptySet()), cur);
Assert.assertTrue(copyOf(concat(readers.subList(0, 1), readers.subList(2, 5))).containsAll(copyOf(cur.getUncompacting(readers))));
Assert.assertEquals(4, copyOf(cur.getUncompacting(readers)).size());
- Set<SSTableReader> nonCompacting = ImmutableSet.copyOf(cur.sstables(SSTableSet.NONCOMPACTING));
+ Set<SSTableReader> nonCompacting = ImmutableSet.copyOf(cur.select(SSTableSet.NONCOMPACTING));
Assert.assertTrue(nonCompacting.containsAll(readers.subList(2, 5)));
Assert.assertTrue(nonCompacting.containsAll(readers.subList(0, 1)));
Assert.assertEquals(4, nonCompacting.size());
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
index 6aaefb7..2124abe 100644
--- a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
+++ b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
@@ -623,7 +623,7 @@ public class CustomCassandraIndex implements Index
{
baseCfs.forceBlockingFlush();
- try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.select(SSTableSet.CANONICAL));
+ try (ColumnFamilyStore.RefViewFragment viewFragment = baseCfs.selectAndReference(View.selectFunction(SSTableSet.CANONICAL));
Refs<SSTableReader> sstables = viewFragment.refs)
{
if (sstables.isEmpty())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bc23632f/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
index 18bc760..c842b7f 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
@@ -759,7 +759,7 @@ public class SSTableRewriterTest extends SSTableWriterTestBase
if (!checked && writer.currentWriter().getFilePointer() > 15000000)
{
checked = true;
- ColumnFamilyStore.ViewFragment viewFragment = cfs.select(View.select(SSTableSet.CANONICAL));
+ ColumnFamilyStore.ViewFragment viewFragment = cfs.select(View.selectFunction(SSTableSet.CANONICAL));
// canonical view should have only one SSTable which is not opened early.
assertEquals(1, viewFragment.sstables.size());
SSTableReader sstable = viewFragment.sstables.get(0);
@@ -872,6 +872,45 @@ public class SSTableRewriterTest extends SSTableWriterTestBase
validateCFS(cfs);
}
+ @Test
+ public void testCanonicalSSTables() throws ExecutionException, InterruptedException
+ {
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
+ truncate(cfs);
+
+ cfs.addSSTable(writeFile(cfs, 100));
+ Collection<SSTableReader> allSSTables = cfs.getSSTables();
+ assertEquals(1, allSSTables.size());
+ final AtomicBoolean done = new AtomicBoolean(false);
+ final AtomicBoolean failed = new AtomicBoolean(false);
+ Runnable r = () -> {
+ while (!done.get())
+ {
+ Iterable<SSTableReader> sstables = cfs.getSSTables(SSTableSet.CANONICAL);
+ if (Iterables.size(sstables) != 1)
+ {
+ failed.set(true);
+ return;
+ }
+ }
+ };
+ Thread t = new Thread(r);
+ try
+ {
+ t.start();
+ cfs.forceMajorCompaction();
+ }
+ finally
+ {
+ done.set(true);
+ t.join(20);
+ }
+ assertFalse(failed.get());
+
+
+ }
+
private void validateKeys(Keyspace ks)
{
for (int i = 0; i < 100; i++)
[3/9] cassandra git commit: Always select the live sstables when
getting sstables in bounds
Posted by ma...@apache.org.
Always select the live sstables when getting sstables in bounds
Patch by marcuse; reviewed by benedict for CASSANDRA-11944
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5b0566a7
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5b0566a7
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5b0566a7
Branch: refs/heads/trunk
Commit: 5b0566a70f373d6eb537c89c0db2a2e224706916
Parents: 2217695
Author: Marcus Eriksson <ma...@apache.org>
Authored: Thu Jun 2 09:37:06 2016 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jul 6 07:52:28 2016 +0200
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/db/ColumnFamilyStore.java | 8 ++++----
.../apache/cassandra/db/PartitionRangeReadCommand.java | 2 +-
.../org/apache/cassandra/db/SizeEstimatesRecorder.java | 5 +++--
.../db/compaction/AbstractCompactionStrategy.java | 2 +-
.../cassandra/db/compaction/CompactionController.java | 2 +-
.../db/compaction/DateTieredCompactionStrategy.java | 2 +-
.../db/compaction/TimeWindowCompactionStrategy.java | 2 +-
src/java/org/apache/cassandra/db/lifecycle/View.java | 10 +++++-----
.../org/apache/cassandra/streaming/StreamSession.java | 8 +++++++-
test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java | 2 +-
11 files changed, 26 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 7f8a3a1..99ac3ad 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
3.0.9
+ * Always select the live sstables when getting sstables in bounds (CASSANDRA-11944)
* Fix column ordering of results with static columns for Thrift requests in
a mixed 2.x/3.x cluster, also fix potential non-resolved duplication of
those static columns in query results (CASSANDRA-12123)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 3264327..1be3175 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1231,7 +1231,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @return sstables whose key range overlaps with that of the given sstables, not including itself.
* (The given sstables may or may not overlap with each other.)
*/
- public Collection<SSTableReader> getOverlappingSSTables(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Collection<SSTableReader> getOverlappingLiveSSTables(Iterable<SSTableReader> sstables)
{
logger.trace("Checking for sstables overlapping {}", sstables);
@@ -1282,7 +1282,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Set<SSTableReader> results = new HashSet<>();
for (AbstractBounds<PartitionPosition> bound : bounds)
- Iterables.addAll(results, view.sstablesInBounds(sstableSet, bound.left, bound.right));
+ Iterables.addAll(results, view.liveSSTablesInBounds(bound.left, bound.right));
return Sets.difference(results, ImmutableSet.copyOf(sstables));
}
@@ -1290,11 +1290,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
/**
* like getOverlappingSSTables, but acquires references before returning
*/
- public Refs<SSTableReader> getAndReferenceOverlappingSSTables(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Refs<SSTableReader> getAndReferenceOverlappingLiveSSTables(Iterable<SSTableReader> sstables)
{
while (true)
{
- Iterable<SSTableReader> overlapped = getOverlappingSSTables(sstableSet, sstables);
+ Iterable<SSTableReader> overlapped = getOverlappingLiveSSTables(sstables);
Refs<SSTableReader> refs = Refs.tryRef(overlapped);
if (refs != null)
return refs;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index 9585b59..842ad5f 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -175,7 +175,7 @@ public class PartitionRangeReadCommand extends ReadCommand
protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
{
- ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, dataRange().keyRange()));
+ ColumnFamilyStore.ViewFragment view = cfs.select(View.selectLive(dataRange().keyRange()));
Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().getKeyValidator()));
// fetch data from current memtable, historical memtables, and SSTables in the correct order.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 2a74ea9..3461aef 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -103,8 +103,9 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable
{
while (refs == null)
{
- ColumnFamilyStore.ViewFragment view = table.select(View.select(SSTableSet.CANONICAL, Range.makeRowRange(range)));
- refs = Refs.tryRef(view.sstables);
+ // note that this is not guaranteed to return all sstables within the ranges, but for an estimated size, that should be fine
+ Iterable<SSTableReader> canonicalSSTables = table.getTracker().getView().select(SSTableSet.CANONICAL, table.select(View.selectLive(Range.makeRowRange(range))).sstables);
+ refs = Refs.tryRef(canonicalSSTables);
}
// calculate the estimates.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
index e30e4f7..0dce52b 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
@@ -380,7 +380,7 @@ public abstract class AbstractCompactionStrategy
if (uncheckedTombstoneCompaction)
return true;
- Collection<SSTableReader> overlaps = cfs.getOverlappingSSTables(SSTableSet.CANONICAL, Collections.singleton(sstable));
+ Collection<SSTableReader> overlaps = cfs.getOverlappingLiveSSTables(Collections.singleton(sstable));
if (overlaps.isEmpty())
{
// there is no overlap, tombstones are safely droppable
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/CompactionController.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionController.java b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
index a5b8308..fbf29e3 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
@@ -103,7 +103,7 @@ public class CompactionController implements AutoCloseable
if (compacting == null)
overlappingSSTables = Refs.tryRef(Collections.<SSTableReader>emptyList());
else
- overlappingSSTables = cfs.getAndReferenceOverlappingSSTables(SSTableSet.LIVE, compacting);
+ overlappingSSTables = cfs.getAndReferenceOverlappingLiveSSTables(compacting);
this.overlapIterator = new OverlapIterator<>(buildIntervals(overlappingSSTables));
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
index 8571906..3e6ae61 100644
--- a/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/DateTieredCompactionStrategy.java
@@ -99,7 +99,7 @@ public class DateTieredCompactionStrategy extends AbstractCompactionStrategy
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
// Find fully expired SSTables. Those will be included no matter what.
- expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(SSTableSet.CANONICAL, uncompacting), gcBefore);
+ expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore);
lastExpiredCheck = System.currentTimeMillis();
}
Set<SSTableReader> candidates = Sets.newHashSet(filterSuspectSSTables(uncompacting));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
index d1630c5..e2ab7dc 100644
--- a/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
+++ b/src/java/org/apache/cassandra/db/compaction/TimeWindowCompactionStrategy.java
@@ -104,7 +104,7 @@ public class TimeWindowCompactionStrategy extends AbstractCompactionStrategy
if (System.currentTimeMillis() - lastExpiredCheck > options.expiredSSTableCheckFrequency)
{
logger.debug("TWCS expired check sufficiently far in the past, checking for fully expired SSTables");
- expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingSSTables(SSTableSet.CANONICAL, uncompacting), gcBefore);
+ expired = CompactionController.getFullyExpiredSSTables(cfs, uncompacting, cfs.getOverlappingLiveSSTables(uncompacting), gcBefore);
lastExpiredCheck = System.currentTimeMillis();
}
else
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/src/java/org/apache/cassandra/db/lifecycle/View.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java
index 99903fc..96aaa49 100644
--- a/src/java/org/apache/cassandra/db/lifecycle/View.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/View.java
@@ -136,7 +136,7 @@ public class View
return Iterables.concat(sstables, filterOut(compacting, sstables));
}
- private Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
+ public Iterable<SSTableReader> select(SSTableSet sstableSet, Iterable<SSTableReader> sstables)
{
switch (sstableSet)
{
@@ -182,7 +182,7 @@ public class View
* Returns the sstables that have any partition between {@code left} and {@code right}, when both bounds are taken inclusively.
* The interval formed by {@code left} and {@code right} shouldn't wrap.
*/
- public Iterable<SSTableReader> sstablesInBounds(SSTableSet sstableSet, PartitionPosition left, PartitionPosition right)
+ public Iterable<SSTableReader> liveSSTablesInBounds(PartitionPosition left, PartitionPosition right)
{
assert !AbstractBounds.strictlyWrapsAround(left, right);
@@ -190,7 +190,7 @@ public class View
return Collections.emptyList();
PartitionPosition stopInTree = right.isMinimum() ? intervalTree.max() : right;
- return select(sstableSet, intervalTree.search(Interval.create(left, stopInTree)));
+ return select(SSTableSet.LIVE, intervalTree.search(Interval.create(left, stopInTree)));
}
public static List<SSTableReader> sstablesInBounds(PartitionPosition left, PartitionPosition right, SSTableIntervalTree intervalTree)
@@ -228,14 +228,14 @@ public class View
* @return a ViewFragment containing the sstables and memtables that may need to be merged
* for rows within @param rowBounds, inclusive, according to the interval tree.
*/
- public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet, AbstractBounds<PartitionPosition> rowBounds)
+ public static Function<View, Iterable<SSTableReader>> selectLive(AbstractBounds<PartitionPosition> rowBounds)
{
// Note that View.sstablesInBounds always includes it's bound while rowBounds may not. This is ok however
// because the fact we restrict the sstables returned by this function is an optimization in the first
// place and the returned sstables will (almost) never cover *exactly* rowBounds anyway. It's also
// *very* unlikely that a sstable is included *just* because we consider one of the bound inclusively
// instead of exclusively, so the performance impact is negligible in practice.
- return (view) -> view.sstablesInBounds(sstableSet, rowBounds.left, rowBounds.right);
+ return (view) -> view.liveSSTablesInBounds(rowBounds.left, rowBounds.right);
}
// METHODS TO CONSTRUCT FUNCTIONS FOR MODIFYING A VIEW:
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/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 d5c060e..a14f815 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -29,8 +29,8 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.*;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
-import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.SSTableIntervalTree;
+import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.slf4j.Logger;
@@ -332,6 +332,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber
SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.sstables(SSTableSet.CANONICAL));
for (Range<PartitionPosition> keyRange : keyRanges)
{
+ // keyRange excludes its start, while sstableInBounds is inclusive (of both start and end).
+ // This is fine however, because keyRange has been created from a token range through Range.makeRowRange (see above).
+ // And that later method uses the Token.maxKeyBound() method to creates the range, which return a "fake" key that
+ // sort after all keys having the token. That "fake" key cannot however be equal to any real key, so that even
+ // including keyRange.left will still exclude any key having the token of the original token range, and so we're
+ // still actually selecting what we wanted.
for (SSTableReader sstable : View.sstablesInBounds(keyRange.left, keyRange.right, intervalTree))
{
if (!isIncremental || !sstable.isRepaired())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5b0566a7/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
index 8a5e00e..98f9300 100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java
@@ -71,7 +71,7 @@ public class ViewTest
continue;
AbstractBounds<PartitionPosition> bounds = AbstractBounds.bounds(min, minInc, max, maxInc);
- List<SSTableReader> r = ImmutableList.copyOf(initialView.sstablesInBounds(SSTableSet.LIVE,bounds.left, bounds.right));
+ List<SSTableReader> r = ImmutableList.copyOf(initialView.liveSSTablesInBounds(bounds.left, bounds.right));
Assert.assertEquals(String.format("%d(%s) %d(%s)", i, minInc, j, maxInc), j - i + (minInc ? 0 : -1) + (maxInc ? 1 : 0), r.size());
}
}