cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1094789 - in /cassandra/branches/cassandra-0.7: ./ src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/streaming/ test/unit/org/apache/cassandra/streaming/
Date Mon, 18 Apr 2011 21:58:21 GMT
Author: jbellis
Date: Mon Apr 18 21:58:20 2011
New Revision: 1094789

URL: http://svn.apache.org/viewvc?rev=1094789&view=rev
Log:
preserve version when streaming data from old sstables
patch by jbellis; reviewed by Stu Hood for CASSANDRA-2283

Modified:
    cassandra/branches/cassandra-0.7/CHANGES.txt
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamIn.java
    cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/BootstrapTest.java
    cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/SerializationsTest.java

Modified: cassandra/branches/cassandra-0.7/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/CHANGES.txt?rev=1094789&r1=1094788&r2=1094789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.7/CHANGES.txt Mon Apr 18 21:58:20 2011
@@ -30,6 +30,7 @@
  * use 64KB flush buffer instead of in_memory_compaction_limit (CASSANDRA-2463)
  * fix duplicate results from CFS.scan (CASSANDRA-2406)
  * avoid caching token-only decoratedkeys (CASSANDRA-2416)
+ * preserve version when streaming data from old sstables (CASSANDRA-2283)
 
 
 0.7.4

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=1094789&r1=1094788&r2=1094789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
Mon Apr 18 21:58:20 2011
@@ -657,23 +657,29 @@ public class ColumnFamilyStore implement
         return columnFamily;
     }
 
+    public String getFlushPath()
+    {
+        return getFlushPath(Descriptor.CURRENT_VERSION);
+    }
+
     /*
      * @return a temporary file name for an sstable.
      * When the sstable object is closed, it will be renamed to a non-temporary
      * format, so incomplete sstables can be recognized and removed on startup.
      */
-    public String getFlushPath()
+    public String getFlushPath(String version)
     {
         long guessedSize = 2L * memsize.value() * 1024*1024; // 2* adds room for keys, column
indexes
         String location = DatabaseDescriptor.getDataFileLocationForTable(table.name, guessedSize);
         if (location == null)
             throw new RuntimeException("Insufficient disk space to flush");
-        return getTempSSTablePath(location);
+        return getTempSSTablePath(location, version);
     }
 
-    public String getTempSSTablePath(String directory)
+    public String getTempSSTablePath(String directory, String version)
     {
-        Descriptor desc = new Descriptor(new File(directory),
+        Descriptor desc = new Descriptor(version,
+                                         new File(directory),
                                          table.name,
                                          columnFamily,
                                          fileIndexGenerator.incrementAndGet(),
@@ -681,6 +687,11 @@ public class ColumnFamilyStore implement
         return desc.filenameFor(Component.DATA);
     }
 
+    public String getTempSSTablePath(String directory)
+    {
+        return getTempSSTablePath(directory, Descriptor.CURRENT_VERSION);
+    }
+
     /** flush the given memtable and swap in a new one for its CFS, if it hasn't been frozen
already.  threadsafe. */
     Future<?> maybeSwitchMemtable(Memtable oldMemtable, final boolean writeCommitLog)
     {

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamIn.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamIn.java?rev=1094789&r1=1094788&r2=1094789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamIn.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamIn.java
Mon Apr 18 21:58:20 2011
@@ -49,11 +49,6 @@ public class StreamIn
     /**
      * Request ranges to be transferred from source to local node
      */
-    public static void requestRanges(InetAddress source, String tableName, Collection<Range>
ranges)
-    {
-        requestRanges(source, tableName, ranges, null);
-    }
-
     public static void requestRanges(InetAddress source, String tableName, Collection<Range>
ranges, Runnable callback)
     {
         assert ranges.size() > 0;
@@ -74,7 +69,7 @@ public class StreamIn
         // new local sstable
         Table table = Table.open(remotedesc.ksname);
         ColumnFamilyStore cfStore = table.getColumnFamilyStore(remotedesc.cfname);
-        Descriptor localdesc = Descriptor.fromFilename(cfStore.getFlushPath());
+        Descriptor localdesc = Descriptor.fromFilename(cfStore.getFlushPath(remote.desc.version));
 
         return new PendingFile(localdesc, remote);
      }

Modified: cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/BootstrapTest.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/BootstrapTest.java?rev=1094789&r1=1094788&r2=1094789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/BootstrapTest.java
(original)
+++ cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/BootstrapTest.java
Mon Apr 18 21:58:20 2011
@@ -37,6 +37,7 @@ public class BootstrapTest extends Schem
     public void testGetNewNames() throws IOException
     {
         Descriptor desc = Descriptor.fromFilename(new File("Keyspace1", "Standard1-500-Data.db").toString());
+        assert !desc.isLatestVersion; // deliberately test old version; see CASSANDRA-2283
         PendingFile inContext = new PendingFile(null, desc, "Data.db", Arrays.asList(new
Pair<Long,Long>(0L, 1L)));
 
         PendingFile outContext = StreamIn.getContextMapping(inContext);
@@ -45,7 +46,8 @@ public class BootstrapTest extends Schem
 
         // nothing else should
         assertEquals(inContext.component, outContext.component);
-        assertEquals(inContext.desc.ksname, outContext.desc.ksname);
-        assertEquals(inContext.desc.cfname, outContext.desc.cfname);
+        assertEquals(desc.ksname, outContext.desc.ksname);
+        assertEquals(desc.cfname, outContext.desc.cfname);
+        assertEquals(desc.version, outContext.desc.version);
     }
 }

Modified: cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/SerializationsTest.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/SerializationsTest.java?rev=1094789&r1=1094788&r2=1094789&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/SerializationsTest.java
(original)
+++ cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/streaming/SerializationsTest.java
Mon Apr 18 21:58:20 2011
@@ -50,9 +50,9 @@ public class SerializationsTest extends 
     private void testPendingFileWrite() throws IOException
     {
         // make sure to test serializing null and a pf with no sstable.
-        PendingFile normal = makePendingFile(true, "fake_component", 100);
-        PendingFile noSections = makePendingFile(true, "not_real", 0);
-        PendingFile noSST = makePendingFile(false, "also_fake", 100);
+        PendingFile normal = makePendingFile(true, 100);
+        PendingFile noSections = makePendingFile(true, 0);
+        PendingFile noSST = makePendingFile(false, 100);
         
         DataOutputStream out = getOutput("streaming.PendingFile.bin");
         PendingFile.serializer().serialize(normal, out);
@@ -78,14 +78,14 @@ public class SerializationsTest extends 
     
     private void testStreamHeaderWrite() throws IOException
     {
-        StreamHeader sh0 = new StreamHeader("Keyspace1", 123L, makePendingFile(true, "zz",
100));
-        StreamHeader sh1 = new StreamHeader("Keyspace1", 124L, makePendingFile(false, "zz",
100));
+        StreamHeader sh0 = new StreamHeader("Keyspace1", 123L, makePendingFile(true, 100));
+        StreamHeader sh1 = new StreamHeader("Keyspace1", 124L, makePendingFile(false, 100));
         Collection<PendingFile> files = new ArrayList<PendingFile>();
         for (int i = 0; i < 50; i++)
-            files.add(makePendingFile(i % 2 == 0, "aa", 100));
-        StreamHeader sh2 = new StreamHeader("Keyspace1", 125L, makePendingFile(true, "bb",
100), files);
+            files.add(makePendingFile(i % 2 == 0, 100));
+        StreamHeader sh2 = new StreamHeader("Keyspace1", 125L, makePendingFile(true, 100),
files);
         StreamHeader sh3 = new StreamHeader("Keyspace1", 125L, null, files);
-        StreamHeader sh4 = new StreamHeader("Keyspace1", 125L, makePendingFile(true, "bb",
100), new ArrayList<PendingFile>());
+        StreamHeader sh4 = new StreamHeader("Keyspace1", 125L, makePendingFile(true, 100),
new ArrayList<PendingFile>());
         
         DataOutputStream out = getOutput("streaming.StreamHeader.bin");
         StreamHeader.serializer().serialize(sh0, out);
@@ -132,13 +132,13 @@ public class SerializationsTest extends 
         in.close();
     }
     
-    private static PendingFile makePendingFile(boolean sst, String comp, int numSecs)
+    private static PendingFile makePendingFile(boolean sst, int numSecs)
     {
         Descriptor desc = new Descriptor("z", new File("path/doesn't/matter"), "Keyspace1",
"Standard1", 23, false);
         List<Pair<Long, Long>> sections = new ArrayList<Pair<Long, Long>>();
         for (int i = 0; i < numSecs; i++)
             sections.add(new Pair<Long, Long>(new Long(i), new Long(i * i)));
-        return new PendingFile(sst ? makeSSTable() : null, desc, comp, sections);
+        return new PendingFile(sst ? makeSSTable() : null, desc, SSTable.COMPONENT_DATA,
sections);
     }
     
     private void testStreamRequestMessageWrite() throws IOException
@@ -147,8 +147,8 @@ public class SerializationsTest extends 
         for (int i = 0; i < 5; i++)
             ranges.add(new Range(new BytesToken(ByteBufferUtil.bytes(Integer.toString(10*i))),
new BytesToken(ByteBufferUtil.bytes(Integer.toString(10*i+5)))));
         StreamRequestMessage msg0 = new StreamRequestMessage(FBUtilities.getLocalAddress(),
ranges, "Keyspace1", 123L);
-        StreamRequestMessage msg1 = new StreamRequestMessage(FBUtilities.getLocalAddress(),
makePendingFile(true, "aa", 100), 124L);
-        StreamRequestMessage msg2 = new StreamRequestMessage(FBUtilities.getLocalAddress(),
makePendingFile(false, "aa", 100), 124L);
+        StreamRequestMessage msg1 = new StreamRequestMessage(FBUtilities.getLocalAddress(),
makePendingFile(true, 100), 124L);
+        StreamRequestMessage msg2 = new StreamRequestMessage(FBUtilities.getLocalAddress(),
makePendingFile(false, 100), 124L);
         
         DataOutputStream out = getOutput("streaming.StreamRequestMessage.bin");
         StreamRequestMessage.serializer().serialize(msg0, out);



Mime
View raw message