cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject [1/3] git commit: fix bulk-loading compressed sstables patch by Tyler Hobbs and yukim; reviewed by jbellis for CASSANDRA-5820
Date Tue, 30 Jul 2013 20:08:16 GMT
Updated Branches:
  refs/heads/cassandra-1.2 e8732139d -> ab8a28e36
  refs/heads/trunk 369415c26 -> 1d6bed3ba


fix bulk-loading compressed sstables
patch by Tyler Hobbs and yukim; reviewed by jbellis for CASSANDRA-5820


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

Branch: refs/heads/cassandra-1.2
Commit: ab8a28e365da8ef515a40e838e773d67ad92a282
Parents: e873213
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Tue Jul 30 14:51:07 2013 -0500
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Tue Jul 30 14:54:08 2013 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../cassandra/io/sstable/SSTableReader.java     |  2 +-
 .../io/sstable/SSTableSimpleUnsortedWriter.java | 11 ++-
 .../io/util/CompressedPoolingSegmentedFile.java |  7 +-
 .../io/util/CompressedSegmentedFile.java        |  9 +-
 .../cassandra/io/sstable/SSTableLoaderTest.java | 98 ++++++++++++++++++++
 6 files changed, 120 insertions(+), 8 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8a28e3/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index bd85e88..8578855 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 1.2.9
+ * fix bulk-loading compressed sstables (CASSANDRA-5820)
  * (Hadoop) fix quoting in CqlPagingRecordReader and CqlRecordWriter 
    (CASSANDRA-5824)
  * update default LCS sstable size to 160MB (CASSANDRA-5727)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8a28e3/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index 7f52bcf..e9a03c8 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -571,7 +571,7 @@ public class SSTableReader extends SSTable
         if (!compression)
             throw new IllegalStateException(this + " is not compressed");
 
-        return ((CompressedPoolingSegmentedFile)dfile).metadata;
+        return ((ICompressedFile) dfile).getMetadata();
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8a28e3/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
index 9207276..48770d3 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
@@ -77,9 +77,7 @@ public class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
                                        int bufferSizeInMB,
                                        CompressionParameters compressParameters)
     {
-        super(directory, new CFMetaData(keyspace, columnFamily, subComparator == null ? ColumnFamilyType.Standard
: ColumnFamilyType.Super, comparator, subComparator).compressionParameters(compressParameters),
partitioner);
-        this.bufferSize = bufferSizeInMB * 1024L * 1024L;
-        this.diskWriter.start();
+        this(directory, new CFMetaData(keyspace, columnFamily, subComparator == null ? ColumnFamilyType.Standard
: ColumnFamilyType.Super, comparator, subComparator).compressionParameters(compressParameters),
partitioner, bufferSizeInMB);
     }
 
     public SSTableSimpleUnsortedWriter(File directory,
@@ -93,6 +91,13 @@ public class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
         this(directory, partitioner, keyspace, columnFamily, comparator, subComparator, bufferSizeInMB,
new CompressionParameters(null));
     }
 
+    public SSTableSimpleUnsortedWriter(File directory, CFMetaData metadata, IPartitioner
partitioner, long bufferSizeInMB)
+    {
+        super(directory, metadata, partitioner);
+        this.bufferSize = bufferSizeInMB * 1024L * 1024L;
+        this.diskWriter.start();
+    }
+
     protected void writeRow(DecoratedKey key, ColumnFamily columnFamily) throws IOException
     {
         currentSize += key.key.remaining() + ColumnFamily.serializer.serializedSize(columnFamily,
MessagingService.current_version) * 1.2;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8a28e3/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java
index 6d78b3e..e142f2e 100644
--- a/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java
+++ b/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java
@@ -20,7 +20,7 @@ package org.apache.cassandra.io.util;
 import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
 import org.apache.cassandra.io.compress.CompressionMetadata;
 
-public class CompressedPoolingSegmentedFile extends PoolingSegmentedFile
+public class CompressedPoolingSegmentedFile extends PoolingSegmentedFile implements ICompressedFile
 {
     public final CompressionMetadata metadata;
 
@@ -48,6 +48,11 @@ public class CompressedPoolingSegmentedFile extends PoolingSegmentedFile
         return CompressedRandomAccessReader.open(path, metadata, this);
     }
 
+    public CompressionMetadata getMetadata()
+    {
+        return metadata;
+    }
+
     @Override
     public void cleanup()
     {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8a28e3/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java
index f861e6b..f41b1aa 100644
--- a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java
+++ b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java
@@ -17,12 +17,10 @@
  */
 package org.apache.cassandra.io.util;
 
-import java.io.File;
-
 import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
 import org.apache.cassandra.io.compress.CompressionMetadata;
 
-public class CompressedSegmentedFile extends SegmentedFile
+public class CompressedSegmentedFile extends SegmentedFile implements ICompressedFile
 {
     public final CompressionMetadata metadata;
 
@@ -52,6 +50,11 @@ public class CompressedSegmentedFile extends SegmentedFile
         return reader;
     }
 
+    public CompressionMetadata getMetadata()
+    {
+        return metadata;
+    }
+
     public void cleanup()
     {
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8a28e3/test/unit/org/apache/cassandra/io/sstable/SSTableLoaderTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableLoaderTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableLoaderTest.java
new file mode 100644
index 0000000..8fa886e
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/sstable/SSTableLoaderTest.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.io.sstable;
+
+import java.io.File;
+import java.util.List;
+
+import com.google.common.io.Files;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.Schema;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Row;
+import org.apache.cassandra.db.Table;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.OutputHandler;
+
+import static junit.framework.Assert.assertEquals;
+
+public class SSTableLoaderTest extends SchemaLoader
+{
+    @BeforeClass
+    public static void setup() throws Exception
+    {
+        StorageService.instance.initServer();
+    }
+
+    @Test
+    public void testLoadingSSTable() throws Exception
+    {
+        File tempdir = Files.createTempDir();
+        File dataDir = new File(tempdir.getAbsolutePath() + File.separator + "Keyspace1"
+ File.separator + "Standard1");
+        assert dataDir.mkdirs();
+        CFMetaData cfmeta = Schema.instance.getCFMetaData("Keyspace1", "Standard1");
+        SSTableSimpleUnsortedWriter writer = new SSTableSimpleUnsortedWriter(dataDir,
+                                                                             cfmeta,
+                                                                             StorageService.getPartitioner(),
+                                                                             1);
+        DecoratedKey key = Util.dk("key1");
+        writer.newRow(key.key);
+        writer.addColumn(ByteBufferUtil.bytes("col1"), ByteBufferUtil.bytes(100), 1);
+        writer.close();
+
+        SSTableLoader loader = new SSTableLoader(dataDir, new SSTableLoader.Client()
+        {
+            @Override
+            public void init(String keyspace)
+            {
+                try
+                {
+                    for (Range<Token> range : StorageService.instance.getLocalRanges("Keyspace1"))
+                        addRangeForEndpoint(range, FBUtilities.getBroadcastAddress());
+                    setPartitioner(StorageService.getPartitioner());
+                } catch (ConfigurationException e)
+                {
+                    throw new RuntimeException(e);
+                }
+            }
+
+            @Override
+            public boolean validateColumnFamily(String keyspace, String cfName)
+            {
+                return true;
+            }
+        }, new OutputHandler.SystemOutput(false, false));
+
+        loader.stream().get();
+
+        List<Row> rows = Util.getRangeSlice(Table.open("Keyspace1").getColumnFamilyStore("Standard1"));
+        assertEquals(1, rows.size());
+        assertEquals(key, rows.get(0).key);
+        assertEquals(ByteBufferUtil.bytes(100), rows.get(0).cf.getColumn(ByteBufferUtil.bytes("col1")).value());
+    }
+}


Mime
View raw message