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 2018/06/13 20:09:13 UTC
cassandra git commit: Abort compactions quicker
Repository: cassandra
Updated Branches:
refs/heads/trunk 2aeed037e -> bdb52801c
Abort compactions quicker
Patch by marcuse; reviewed by Alex Petrov for CASSANDRA-14397
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/bdb52801
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/bdb52801
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/bdb52801
Branch: refs/heads/trunk
Commit: bdb52801c7384ef07f7fc0b4f3b965bdf35d821d
Parents: 2aeed03
Author: Marcus Eriksson <ma...@apache.org>
Authored: Fri Apr 13 15:15:03 2018 +0200
Committer: Marcus Eriksson <ma...@apache.org>
Committed: Wed Jun 13 13:06:59 2018 -0700
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../db/compaction/CompactionIterator.java | 38 +++++++++++-
.../db/compaction/CompactionManager.java | 3 -
.../cassandra/db/compaction/CompactionTask.java | 3 -
.../db/repair/CassandraValidationIterator.java | 8 ---
.../db/compaction/CompactionIteratorTest.java | 61 ++++++++++++++++++++
6 files changed, 99 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdb52801/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 629df0c..49738cd 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0
+ * Abort compactions quicker (CASSANDRA-14397)
* Support light-weight transactions in cassandra-stress (CASSANDRA-13529)
* Make AsyncOneResponse use the correct timeout (CASSANDRA-14509)
* Add option to sanity check tombstones on reads/compactions (CASSANDRA-14467)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdb52801/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index dfbb6cc..c9d7e52 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -104,7 +104,8 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
? EmptyIterators.unfilteredPartition(controller.cfs.metadata())
: UnfilteredPartitionIterators.merge(scanners, nowInSec, listener());
merged = Transformation.apply(merged, new GarbageSkipper(controller, nowInSec));
- this.compacted = Transformation.apply(merged, new Purger(controller, nowInSec));
+ merged = Transformation.apply(merged, new Purger(controller, nowInSec));
+ compacted = Transformation.apply(merged, new AbortableUnfilteredPartitionTransformation(this));
}
public TableMetadata metadata()
@@ -542,4 +543,39 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
return new GarbageSkippingUnfilteredRowIterator(partition, UnfilteredRowIterators.merge(iters, nowInSec), nowInSec, cellLevelGC);
}
}
+
+ private static class AbortableUnfilteredPartitionTransformation extends Transformation<UnfilteredRowIterator>
+ {
+ private final AbortableUnfilteredRowTransformation abortableIter;
+
+ private AbortableUnfilteredPartitionTransformation(CompactionIterator iter)
+ {
+ this.abortableIter = new AbortableUnfilteredRowTransformation(iter);
+ }
+
+ @Override
+ protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
+ {
+ if (abortableIter.iter.isStopRequested())
+ throw new CompactionInterruptedException(abortableIter.iter.getCompactionInfo());
+ return Transformation.apply(partition, abortableIter);
+ }
+ }
+
+ private static class AbortableUnfilteredRowTransformation extends Transformation
+ {
+ private final CompactionIterator iter;
+
+ private AbortableUnfilteredRowTransformation(CompactionIterator iter)
+ {
+ this.iter = iter;
+ }
+
+ public Row applyToRow(Row row)
+ {
+ if (iter.isStopRequested())
+ throw new CompactionInterruptedException(iter.getCompactionInfo());
+ return row;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdb52801/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 5c61982..a872fea 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -1155,9 +1155,6 @@ public class CompactionManager implements CompactionManagerMBean
while (ci.hasNext())
{
- if (ci.isStopRequested())
- throw new CompactionInterruptedException(ci.getCompactionInfo());
-
try (UnfilteredRowIterator partition = ci.next();
UnfilteredRowIterator notCleaned = cleanupStrategy.cleanup(partition))
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdb52801/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index 5697df2..662384c 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -197,9 +197,6 @@ public class CompactionTask extends AbstractCompactionTask
estimatedKeys = writer.estimatedKeys();
while (ci.hasNext())
{
- if (ci.isStopRequested())
- throw new CompactionInterruptedException(ci.getCompactionInfo());
-
if (writer.append(ci.next()))
totalKeysWritten++;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdb52801/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java b/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
index 6fa0be2..caf1b8e 100644
--- a/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
+++ b/src/java/org/apache/cassandra/db/repair/CassandraValidationIterator.java
@@ -281,23 +281,15 @@ public class CassandraValidationIterator extends ValidationPartitionIterator
return cfs.metadata.get();
}
- private void throwIfStopRequested()
- {
- if (ci.isStopRequested())
- throw new CompactionInterruptedException(ci.getCompactionInfo());
- }
-
@Override
public boolean hasNext()
{
- throwIfStopRequested();
return ci.hasNext();
}
@Override
public UnfilteredRowIterator next()
{
- throwIfStopRequested();
return ci.next();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/bdb52801/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java
index 99df52f..d5ea56c 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java
@@ -310,6 +310,67 @@ public class CompactionIteratorTest
}
}
+ @Test
+ public void transformTest()
+ {
+ UnfilteredRowsGenerator generator = new UnfilteredRowsGenerator(metadata.comparator, false);
+ List<List<Unfiltered>> inputLists = parse(new String[] {"10[100] 11[100] 12[100]"}, generator);
+ List<List<Unfiltered>> tombstoneLists = parse(new String[] {}, generator);
+ List<Iterable<UnfilteredRowIterator>> content = ImmutableList.copyOf(Iterables.transform(inputLists, list -> ImmutableList.of(listToIterator(list, kk))));
+ Map<DecoratedKey, Iterable<UnfilteredRowIterator>> transformedSources = new TreeMap<>();
+ transformedSources.put(kk, Iterables.transform(tombstoneLists, list -> listToIterator(list, kk)));
+ try (CompactionController controller = new Controller(Keyspace.openAndGetStore(metadata), transformedSources, GC_BEFORE);
+ CompactionIterator iter = new CompactionIterator(OperationType.COMPACTION,
+ Lists.transform(content, x -> new Scanner(x)),
+ controller, NOW, null))
+ {
+ assertTrue(iter.hasNext());
+ UnfilteredRowIterator rows = iter.next();
+ assertTrue(rows.hasNext());
+ assertNotNull(rows.next());
+
+ iter.stop();
+ try
+ {
+ // Will call Transformation#applyToRow
+ rows.hasNext();
+ fail("Should have thrown CompactionInterruptedException");
+ }
+ catch (CompactionInterruptedException e)
+ {
+ // ignore
+ }
+ }
+ }
+
+ @Test
+ public void transformPartitionTest()
+ {
+ UnfilteredRowsGenerator generator = new UnfilteredRowsGenerator(metadata.comparator, false);
+ List<List<Unfiltered>> inputLists = parse(new String[] {"10[100] 11[100] 12[100]"}, generator);
+ List<List<Unfiltered>> tombstoneLists = parse(new String[] {}, generator);
+ List<Iterable<UnfilteredRowIterator>> content = ImmutableList.copyOf(Iterables.transform(inputLists, list -> ImmutableList.of(listToIterator(list, kk))));
+ Map<DecoratedKey, Iterable<UnfilteredRowIterator>> transformedSources = new TreeMap<>();
+ transformedSources.put(kk, Iterables.transform(tombstoneLists, list -> listToIterator(list, kk)));
+ try (CompactionController controller = new Controller(Keyspace.openAndGetStore(metadata), transformedSources, GC_BEFORE);
+ CompactionIterator iter = new CompactionIterator(OperationType.COMPACTION,
+ Lists.transform(content, x -> new Scanner(x)),
+ controller, NOW, null))
+ {
+ iter.stop();
+ try
+ {
+ // Will call Transformation#applyToPartition
+ iter.hasNext();
+ fail("Should have thrown CompactionInterruptedException");
+ }
+ catch (CompactionInterruptedException e)
+ {
+ // ignore
+ }
+ }
+ }
+
class Controller extends CompactionController
{
private final Map<DecoratedKey, Iterable<UnfilteredRowIterator>> tombstoneSources;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org