cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jmcken...@apache.org
Subject cassandra git commit: Fix ordering on SSTableScanner / SSTableReader close
Date Thu, 13 Nov 2014 18:07:27 GMT
Repository: cassandra
Updated Branches:
  refs/heads/cassandra-2.1 5615a79cf -> 3484181e2


Fix ordering on SSTableScanner / SSTableReader close

Patch by jmckenzie; reviewed by marcuse for CASSANDRA-8019


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/3484181e
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/3484181e
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/3484181e

Branch: refs/heads/cassandra-2.1
Commit: 3484181e2b68ce04acfe4e9028d5d1d76335094a
Parents: 5615a79
Author: Joshua McKenzie <jmckenzie@apache.org>
Authored: Thu Nov 13 11:56:34 2014 -0600
Committer: Joshua McKenzie <jmckenzie@apache.org>
Committed: Thu Nov 13 11:56:34 2014 -0600

----------------------------------------------------------------------
 .../cassandra/io/sstable/SSTableScanner.java    |   3 +
 .../io/sstable/SSTableRewriterTest.java         | 190 ++++++++++---------
 2 files changed, 106 insertions(+), 87 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/3484181e/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
index 62ac175..3f1f1f0 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
@@ -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()

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3484181e/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 4d248bd..8a494a6 100644
--- a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
@@ -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)
-            {
-                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)
+            while(scanner.hasNext())
             {
-                //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();


Mime
View raw message