diff --git a/CHANGES.txt b/CHANGES.txt index 83310ed39b..a9cb47cfbc 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -7,6 +7,7 @@ * DC-local CAS (CASSANDRA-5797) * Add a native_protocol_version column to the system.local table (CASSANRDA-5819) Merged from 1.2: + * 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) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index 9f1006159a..e30e4171a2 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -568,7 +568,7 @@ public class SSTableReader extends SSTable if (!compression) throw new IllegalStateException(this + " is not compressed"); - return ((CompressedPoolingSegmentedFile)dfile).metadata; + return ((ICompressedFile) dfile).getMetadata(); } /** diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java index c3a6212116..52e5a03054 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; diff --git a/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java index ef7b3b371d..121bdb2223 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,10 +48,15 @@ public class CompressedPoolingSegmentedFile extends PoolingSegmentedFile return CompressedRandomAccessReader.open(path, metadata, this); } + public CompressionMetadata getMetadata() + { + return metadata; + } + @Override public void cleanup() { super.cleanup(); metadata.close(); } -} \ No newline at end of file +} diff --git a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java index 4a1e755579..f41b1aab0f 100644 --- a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.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 CompressedSegmentedFile extends SegmentedFile +public class CompressedSegmentedFile extends SegmentedFile implements ICompressedFile { public final CompressionMetadata metadata; @@ -50,6 +50,11 @@ public class CompressedSegmentedFile extends SegmentedFile return reader; } + public CompressionMetadata getMetadata() + { + return metadata; + } + public void cleanup() { } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index bec8c8da85..d0581fb346 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -3526,7 +3526,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE SSTableLoader.Client client = new SSTableLoader.Client() { - @Override public void init(String keyspace) { try @@ -3545,7 +3544,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } - @Override public CFMetaData getCFMetaData(String keyspace, String cfName) { return Schema.instance.getCFMetaData(keyspace, cfName); 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 0000000000..236ee2d824 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableLoaderTest.java @@ -0,0 +1,89 @@ +/* + * 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.Keyspace; +import org.apache.cassandra.db.Row; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +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() + { + public void init(String keyspace) + { + for (Range range : StorageService.instance.getLocalRanges("Keyspace1")) + addRangeForEndpoint(range, FBUtilities.getBroadcastAddress()); + setPartitioner(StorageService.getPartitioner()); + } + + public CFMetaData getCFMetaData(String keyspace, String cfName) + { + return Schema.instance.getCFMetaData(keyspace, cfName); + } + }, new OutputHandler.SystemOutput(false, false)); + + loader.stream().get(); + + List rows = Util.getRangeSlice(Keyspace.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()); + } +}