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

merge from 1.2

This commit is contained in:
Jonathan Ellis 2013-07-30 15:08:18 -05:00
parent 369415c262 ab8a28e365
commit 1d6bed3ba8
7 changed files with 112 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

2
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -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();
}
/**

11
src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
View File

@ -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;

9
src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java
View File

@ -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();
}
}
}

7
src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java
View File

@ -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()
{
}

2
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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);

89
test/unit/org/apache/cassandra/io/sstable/SSTableLoaderTest.java Normal file
View File

@ -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<Token> 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<Row> 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());
}
}