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

Fix ordering on SSTableScanner / SSTableReader close 

Patch by jmckenzie; reviewed by marcuse for CASSANDRA-8019
This commit is contained in:
Joshua McKenzie 2014-11-13 11:56:34 -06:00
parent 5615a79cf0
commit 3484181e2b
2 changed files with 105 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
View File

@ -63,6 +63,7 @@ public class SSTableScanner implements ICompactionScanner
SSTableScanner(SSTableReader sstable, DataRange dataRange, RateLimiter limiter)
{
assert sstable != null;
sstable.acquireReference();
this.dfile = limiter == null ? sstable.openDataReader() : sstable.openDataReader(limiter);
this.ifile = sstable.openIndexReader();
@ -92,6 +93,7 @@ public class SSTableScanner implements ICompactionScanner
SSTableScanner(SSTableReader sstable, Collection<Range<Token>> tokenRanges, RateLimiter limiter)
{
assert sstable != null;
sstable.acquireReference();
this.dfile = limiter == null ? sstable.openDataReader() : sstable.openDataReader(limiter);
this.ifile = sstable.openIndexReader();
@ -159,6 +161,7 @@ public class SSTableScanner implements ICompactionScanner
public void close() throws IOException
{
FileUtils.close(dfile, ifile);
sstable.releaseReference();
}
public long getLengthInBytes()

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

@ -69,14 +69,16 @@ public class SSTableRewriterTest extends SchemaLoader
Set<SSTableReader> sstables = new HashSet<>(cfs.getSSTables());
assertEquals(1, sstables.size());
SSTableRewriter writer = new SSTableRewriter(cfs, sstables, 1000, false);
AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);
ICompactionScanner scanner = scanners.scanners.get(0);
CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
while(scanner.hasNext())
try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);)
{
AbstractCompactedRow row = new LazilyCompactedRow(controller, Arrays.asList(scanner.next()));
writer.append(row);
ICompactionScanner scanner = scanners.scanners.get(0);
CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
while(scanner.hasNext())
{
AbstractCompactedRow row = new LazilyCompactedRow(controller, Arrays.asList(scanner.next()));
writer.append(row);
}
}
cfs.getDataTracker().markCompactedSSTablesReplaced(sstables, writer.finish(), OperationType.COMPACTION);
@ -144,7 +146,7 @@ public class SSTableRewriterTest extends SchemaLoader
@Test
public void testNumberOfFilesAndSizes() throws InterruptedException
public void testNumberOfFilesAndSizes() throws Exception
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
@ -158,20 +160,22 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableRewriter rewriter = new SSTableRewriter(cfs, compacting, 1000, false);
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
int files = 1;
while(scanner.hasNext())
try (ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0))
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
while(scanner.hasNext())
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
assertEquals(s.bytesOnDisk(), cfs.metric.liveDiskSpaceUsed.count());
assertEquals(s.bytesOnDisk(), cfs.metric.totalDiskSpaceUsed.count());
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
assertEquals(s.bytesOnDisk(), cfs.metric.liveDiskSpaceUsed.count());
assertEquals(s.bytesOnDisk(), cfs.metric.totalDiskSpaceUsed.count());
}
}
}
List<SSTableReader> sstables = rewriter.finish();
@ -191,7 +195,7 @@ public class SSTableRewriterTest extends SchemaLoader
}
@Test
public void testNumberOfFiles_dont_clean_readers() throws InterruptedException
public void testNumberOfFiles_dont_clean_readers() throws Exception
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
@ -205,17 +209,19 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableRewriter rewriter = new SSTableRewriter(cfs, compacting, 1000, false);
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
int files = 1;
while(scanner.hasNext())
try (ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0))
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
while(scanner.hasNext())
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
}
}
List<SSTableReader> sstables = rewriter.finish();
@ -230,7 +236,7 @@ public class SSTableRewriterTest extends SchemaLoader
@Test
public void testNumberOfFiles_abort() throws InterruptedException
public void testNumberOfFiles_abort() throws Exception
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
@ -246,17 +252,19 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableRewriter rewriter = new SSTableRewriter(cfs, compacting, 1000, false);
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
int files = 1;
while(scanner.hasNext())
try (ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0))
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
while(scanner.hasNext())
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
}
}
rewriter.abort();
@ -271,7 +279,7 @@ public class SSTableRewriterTest extends SchemaLoader
}
@Test
public void testNumberOfFiles_abort2() throws InterruptedException
public void testNumberOfFiles_abort2() throws Exception
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
@ -287,23 +295,25 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableRewriter rewriter = new SSTableRewriter(cfs, compacting, 1000, false);
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
int files = 1;
while(scanner.hasNext())
try (ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0))
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
while(scanner.hasNext())
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
if (files == 3)
{
//testing to abort when we have nothing written in the new file
rewriter.abort();
break;
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
if (files == 3)
{
//testing to abort when we have nothing written in the new file
rewriter.abort();
break;
}
}
}
Thread.sleep(1000);
@ -316,7 +326,7 @@ public class SSTableRewriterTest extends SchemaLoader
}
@Test
public void testNumberOfFiles_finish_empty_new_writer() throws InterruptedException
public void testNumberOfFiles_finish_empty_new_writer() throws Exception
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
@ -330,24 +340,26 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableRewriter rewriter = new SSTableRewriter(cfs, compacting, 1000, false);
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
int files = 1;
while(scanner.hasNext())
try (ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0))
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
while(scanner.hasNext())
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
if (files == 3)
{
//testing to finish when we have nothing written in the new file
List<SSTableReader> sstables = rewriter.finish();
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
break;
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
if (files == 3)
{
//testing to finish when we have nothing written in the new file
List<SSTableReader> sstables = rewriter.finish();
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
break;
}
}
}
Thread.sleep(1000);
@ -357,7 +369,7 @@ public class SSTableRewriterTest extends SchemaLoader
}
@Test
public void testNumberOfFiles_truncate() throws InterruptedException
public void testNumberOfFiles_truncate() throws Exception
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
@ -371,17 +383,19 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableRewriter rewriter = new SSTableRewriter(cfs, compacting, 1000, false);
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
int files = 1;
while(scanner.hasNext())
try (ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0))
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
while(scanner.hasNext())
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
}
}
List<SSTableReader> sstables = rewriter.finish();
@ -393,7 +407,7 @@ public class SSTableRewriterTest extends SchemaLoader
}
@Test
public void testSmallFiles() throws InterruptedException
public void testSmallFiles() throws Exception
{
Keyspace keyspace = Keyspace.open(KEYSPACE);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
@ -408,18 +422,20 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableRewriter rewriter = new SSTableRewriter(cfs, compacting, 1000, false);
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0);
int files = 1;
while(scanner.hasNext())
try (ICompactionScanner scanner = s.getScanner();
CompactionController controller = new CompactionController(cfs, compacting, 0))
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 2500000)
while(scanner.hasNext())
{
assertEquals(1, cfs.getSSTables().size()); // we dont open small files early ...
assertEquals(origFirst, cfs.getSSTables().iterator().next().first); // ... and the first key should stay the same
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 2500000)
{
assertEquals(1, cfs.getSSTables().size()); // we dont open small files early ...
assertEquals(origFirst, cfs.getSSTables().iterator().next().first); // ... and the first key should stay the same
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
}
}
}
List<SSTableReader> sstables = rewriter.finish();