natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-2.1' into cassandra-2.2 

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/db/compaction/CompactionManager.java
	src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
	test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
This commit is contained in:
Marcus Eriksson 2015-05-29 09:57:54 +02:00
parent 7d0676215e eaeabffc8e
commit 180130a8e5
5 changed files with 88 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -11,6 +11,7 @@
* Add ability to stop compaction by ID (CASSANDRA-7207)
* Let CassandraVersion handle SNAPSHOT version (CASSANDRA-9438)
Merged from 2.1:
* Avoid getting unreadable keys during anticompaction (CASSANDRA-9508)
* (cqlsh) Better float precision by default (CASSANDRA-9224)
* Improve estimated row count (CASSANDRA-9107)
* Optimize range tombstone memory footprint (CASSANDRA-8603)

4
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -1207,8 +1207,8 @@ public class CompactionManager implements CompactionManagerMBean
long repairedKeyCount = 0;
long unrepairedKeyCount = 0;
AbstractCompactionStrategy strategy = cfs.getCompactionStrategy();
try (SSTableRewriter repairedSSTableWriter = new SSTableRewriter(cfs, anticompactionGroup, groupMaxDataAge, false);
SSTableRewriter unRepairedSSTableWriter = new SSTableRewriter(cfs, anticompactionGroup, groupMaxDataAge, false);
try (SSTableRewriter repairedSSTableWriter = new SSTableRewriter(cfs, anticompactionGroup, groupMaxDataAge, false, false);
SSTableRewriter unRepairedSSTableWriter = new SSTableRewriter(cfs, anticompactionGroup, groupMaxDataAge, false, false);
AbstractCompactionStrategy.ScannerList scanners = strategy.getScanners(anticompactionGroup.originals());
CompactionController controller = new CompactionController(cfs, sstableAsSet, getDefaultGcBefore(cfs)))
{

41
src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
View File

@ -49,28 +49,11 @@ import org.apache.cassandra.utils.concurrent.Transactional;
*/
public class SSTableRewriter extends Transactional.AbstractTransactional implements Transactional
{
private static long preemptiveOpenInterval;
static
{
long interval = DatabaseDescriptor.getSSTablePreempiveOpenIntervalInMB() * (1L << 20);
if (interval < 0)
interval = Long.MAX_VALUE;
preemptiveOpenInterval = interval;
}
@VisibleForTesting
public static void overrideOpenInterval(long size)
{
preemptiveOpenInterval = size;
}
@VisibleForTesting
public static long getOpenInterval()
{
return preemptiveOpenInterval;
}
public static boolean disableEarlyOpeningForTests = false;
private final ColumnFamilyStore cfs;
private final long preemptiveOpenInterval;
private final long maxAge;
private long repairedAt = -1;
// the set of final readers we will expose on commit
@ -90,6 +73,17 @@ public class SSTableRewriter extends Transactional.AbstractTransactional impleme
private boolean throwEarly, throwLate;
public SSTableRewriter(ColumnFamilyStore cfs, LifecycleTransaction transaction, long maxAge, boolean isOffline)
{
this(cfs, transaction, maxAge, isOffline, true);
}
public SSTableRewriter(ColumnFamilyStore cfs, LifecycleTransaction transaction, long maxAge, boolean isOffline, boolean shouldOpenEarly)
{
this(cfs, transaction, maxAge, isOffline, calculateOpenInterval(shouldOpenEarly));
}
@VisibleForTesting
public SSTableRewriter(ColumnFamilyStore cfs, LifecycleTransaction transaction, long maxAge, boolean isOffline, long preemptiveOpenInterval)
{
this.transaction = transaction;
for (SSTableReader sstable : this.transaction.originals())
@ -97,6 +91,15 @@ public class SSTableRewriter extends Transactional.AbstractTransactional impleme
this.cfs = cfs;
this.maxAge = maxAge;
this.isOffline = isOffline;
this.preemptiveOpenInterval = preemptiveOpenInterval;
}
private static long calculateOpenInterval(boolean shouldOpenEarly)
{
long interval = DatabaseDescriptor.getSSTablePreempiveOpenIntervalInMB() * (1L << 20);
if (disableEarlyOpeningForTests || !shouldOpenEarly || interval < 0)
interval = Long.MAX_VALUE;
return interval;
}
public SSTableWriter currentWriter()

6
test/unit/org/apache/cassandra/db/ScrubTest.java
View File

@ -275,15 +275,15 @@ public class ScrubTest
@Test
public void testScrubCorruptedCounterRowNoEarlyOpen() throws IOException, WriteTimeoutException
{
long oldOpenVal = SSTableRewriter.getOpenInterval();
boolean oldDisabledVal = SSTableRewriter.disableEarlyOpeningForTests;
try
{
SSTableRewriter.overrideOpenInterval(Long.MAX_VALUE);
SSTableRewriter.disableEarlyOpeningForTests = true;
testScrubCorruptedCounterRow();
}
finally
{
SSTableRewriter.overrideOpenInterval(oldOpenVal);
SSTableRewriter.disableEarlyOpeningForTests = oldDisabledVal;
}
}

82
test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.io.sstable;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
@ -130,10 +131,10 @@ public class SSTableRewriterTest extends SchemaLoader
cfs.addSSTable(s);
Set<SSTableReader> sstables = new HashSet<>(cfs.getSSTables());
assertEquals(1, sstables.size());
SSTableRewriter.overrideOpenInterval(10000000);
try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);
LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
SSTableRewriter writer = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter writer = new SSTableRewriter(cfs, txn, 1000, false, 10000000);)
{
ISSTableScanner scanner = scanners.scanners.get(0);
CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
@ -164,11 +165,11 @@ public class SSTableRewriterTest extends SchemaLoader
cfs.addSSTable(s);
Set<SSTableReader> sstables = new HashSet<>(cfs.getSSTables());
assertEquals(1, sstables.size());
SSTableRewriter.overrideOpenInterval(10000000);
boolean checked = false;
try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);
LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
SSTableRewriter writer = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter writer = new SSTableRewriter(cfs, txn, 1000, false, 10000000))
{
ISSTableScanner scanner = scanners.scanners.get(0);
CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
@ -264,14 +265,13 @@ public class SSTableRewriterTest extends SchemaLoader
long startStorageMetricsLoad = StorageMetrics.load.getCount();
long sBytesOnDisk = s.bytesOnDisk();
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(10000000);
List<SSTableReader> sstables;
int files = 1;
try (ISSTableScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false, 10000000))
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
while(scanner.hasNext())
@ -315,14 +315,13 @@ public class SSTableRewriterTest extends SchemaLoader
cfs.addSSTable(s);
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(10000000);
List<SSTableReader> sstables;
int files = 1;
try (ISSTableScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false, 10000000))
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
while(scanner.hasNext())
@ -439,12 +438,11 @@ public class SSTableRewriterTest extends SchemaLoader
DecoratedKey origLast = s.last;
long startSize = cfs.metric.liveDiskSpaceUsed.getCount();
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(10000000);
try (ISSTableScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false, 10000000);)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
test.run(scanner, controller, s, cfs, rewriter);
@ -471,13 +469,12 @@ public class SSTableRewriterTest extends SchemaLoader
cfs.addSSTable(s);
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(10000000);
int files = 1;
try (ISSTableScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false, 10000000))
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
while(scanner.hasNext())
@ -516,14 +513,13 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableReader s = writeFile(cfs, 1000);
cfs.addSSTable(s);
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(10000000);
List<SSTableReader> sstables;
int files = 1;
try (ISSTableScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false, 10000000))
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
while(scanner.hasNext())
@ -558,14 +554,13 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableReader s = writeFile(cfs, 400);
cfs.addSSTable(s);
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(1000000);
List<SSTableReader> sstables;
int files = 1;
try (ISSTableScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false);)
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false, 1000000);)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
while(scanner.hasNext())
@ -648,12 +643,11 @@ public class SSTableRewriterTest extends SchemaLoader
if (!offline)
cfs.addSSTable(s);
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(10000000);
try (ISSTableScanner scanner = compacting.iterator().next().getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = offline ? LifecycleTransaction.offline(OperationType.UNKNOWN, compacting)
: cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, offline);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, offline, 10000000);
)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
@ -730,12 +724,11 @@ public class SSTableRewriterTest extends SchemaLoader
Set<SSTableReader> compacting = new HashSet<>();
compacting.add(s);
SSTableRewriter.overrideOpenInterval(1);
int keyCount = 0;
try (ISSTableScanner scanner = compacting.iterator().next().getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(compacting, OperationType.UNKNOWN);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false);
SSTableRewriter rewriter = new SSTableRewriter(cfs, txn, 1000, false, 1);
)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
@ -767,12 +760,11 @@ public class SSTableRewriterTest extends SchemaLoader
cfs.addSSTable(s);
Set<SSTableReader> sstables = Sets.newHashSet(s);
assertEquals(1, sstables.size());
SSTableRewriter.overrideOpenInterval(10000000);
boolean checked = false;
try (ISSTableScanner scanner = sstables.iterator().next().getScanner();
CompactionController controller = new CompactionController(cfs, sstables, 0);
LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
SSTableRewriter writer = new SSTableRewriter(cfs, txn, 1000, false);
SSTableRewriter writer = new SSTableRewriter(cfs, txn, 1000, false, 10000000);
)
{
writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
@ -796,6 +788,52 @@ public class SSTableRewriterTest extends SchemaLoader
validateCFS(cfs);
}
/**
* emulates anticompaction - writing from one source sstable to two new sstables
*
* @throws IOException
*/
@Test
public void testTwoWriters() throws IOException
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
cfs.truncateBlocking();
SSTableReader s = writeFile(cfs, 1000);
cfs.addSSTable(s);
Set<SSTableReader> sstables = Sets.newHashSet(s);
assertEquals(1, sstables.size());
try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);
LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
SSTableRewriter writer = new SSTableRewriter(cfs, txn, 1000, false, false);
SSTableRewriter writer2 = new SSTableRewriter(cfs, txn, 1000, false, false))
{
ISSTableScanner scanner = scanners.scanners.get(0);
CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
writer2.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
while (scanner.hasNext())
{
AbstractCompactedRow row = new LazilyCompactedRow(controller, Collections.singletonList(scanner.next()));
if (writer.currentWriter().getFilePointer() < 15000000)
writer.append(row);
else
writer2.append(row);
}
for (int i = 0; i < 5000; i++)
{
DecoratedKey key = Util.dk(ByteBufferUtil.bytes(i));
ColumnFamily cf = Util.getColumnFamily(keyspace, key, CF);
assertTrue(cf != null);
}
}
truncateCF();
validateCFS(cfs);
}
private void validateKeys(Keyspace ks)
{
for (int i = 0; i < 100; i++)