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

split sstable into data, index, and bloom filter files. this allows us to avoid saving up index chunks 

in memory until the sstable data is completely written, while still allowing fast scanning of the index
on startup.  it also simplifies the sstable/sequencefile code considerably.
patch by jbellis; reviewed by Jun Rao for  CASSANDRA-208

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@783077 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-06-09 17:46:42 +00:00
parent f685f4dec8
commit 72e6eea9e4
19 changed files with 403 additions and 905 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/java/org/apache/cassandra/db/ColumnFamily.java
View File

@ -246,6 +246,7 @@ public final class ColumnFamily
void clear()
{
logger_.debug("clearing");
columns_.clear();
size_.set(0);
}

5
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -434,7 +434,10 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
});
f.get();
assert oldMemtable.isFlushed() || oldMemtable.isClean();
/* this assert is not threadsafe -- the memtable could have been clean when forceFlush
checked it, but dirty now thanks to another thread. But as long as we are only
calling this from single-threaded test code it is useful to have as a sanity check. */
assert oldMemtable.isFlushed() || oldMemtable.isClean();
}
void forceFlushBinary()

32
src/java/org/apache/cassandra/db/FileStruct.java
View File

@ -85,26 +85,14 @@ public class FileStruct implements Comparable<FileStruct>, Iterator<String>
{
try
{
Coordinate range = SSTable.getCoordinates(seekKey, reader, partitioner);
reader.seek(range.end_);
long position = reader.getPositionFromBlockIndex(seekKey);
if (position == -1)
long position = SSTable.getNearestPosition(seekKey, reader, partitioner);
if (position < 0)
{
reader.seek(range.start_);
}
else
{
reader.seek(position);
}
while (!exhausted)
{
advance();
if (key.compareTo(seekKey) >= 0)
{
break;
}
exhausted = true;
return;
}
reader.seek(position);
advance();
}
catch (IOException e)
{
@ -144,12 +132,6 @@ public class FileStruct implements Comparable<FileStruct>, Iterator<String>
bufIn.reset(bufOut.getData(), bufOut.getLength());
key = bufIn.readUTF();
/* If the key we read is the Block Index Key then omit and read the next key. */
if (key.equals(SSTable.blockIndexKey_))
{
reader.close();
exhausted = true;
}
}
public boolean hasNext()
@ -199,7 +181,7 @@ public class FileStruct implements Comparable<FileStruct>, Iterator<String>
protected String computeNext()
{
if (key.equals(SSTable.blockIndexKey_))
if (isExhausted())
{
return endOfData();
}

9
src/java/org/apache/cassandra/db/Table.java
View File

@ -268,8 +268,7 @@ public class Table
* list of the associated Column Family. Also merge the CBF into the
* sampler.
*/
SSTable ssTable = new SSTable(streamContext.getTargetFile(), StorageService.getPartitioner());
ssTable.close();
new SSTable(streamContext.getTargetFile(), StorageService.getPartitioner());
logger_.debug("Merging the counting bloom filter in the sampler ...");
String[] peices = FBUtilities.strip(fileName, "-");
Table.open(peices[0]).getColumnFamilyStore(peices[1]).addToList(streamContext.getTargetFile());
@ -756,11 +755,13 @@ public class Table
{
Row row = new Row(key);
String[] values = RowMutation.getColumnAndColumnFamily(cf);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
String cfName = values[0];
String startWith = values.length > 1 ? values[1] : "";
ColumnFamilyStore cfStore = columnFamilyStores_.get(cfName);
long start1 = System.currentTimeMillis();
try
{
ColumnFamily columnFamily = cfStore.getSliceFrom(key, values[0], values[1], isAscending, count);
ColumnFamily columnFamily = cfStore.getSliceFrom(key, cfName, startWith, isAscending, count);
if (columnFamily != null)
row.addColumnFamily(columnFamily);
long timeTaken = System.currentTimeMillis() - start1;

5
src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java
View File

@ -248,12 +248,13 @@ public final class BufferedRandomAccessFile extends RandomAccessFile
{
return this.curr_;
}
public long length() throws IOException
{
// max accounts for the case where we have written past the old file length, but not yet flushed our buffer
return Math.max(this.curr_, super.length());
}
public int read() throws IOException
{
if (this.curr_ >= this.hi_)

26
src/java/org/apache/cassandra/io/IFileReader.java
View File

@ -50,15 +50,7 @@ public interface IFileReader
* @throws IOException
*/
public long readLong() throws IOException;
/**
* This method helps is retrieving the offset of the specified
* key in the file using the block index.
*
* @param key key whose position we need in the block index.
*/
public long getPositionFromBlockIndex(String key) throws IOException;
/**
* This method dumps the next key/value into the DataOuputStream
* passed in.
@ -69,18 +61,6 @@ public interface IFileReader
*/
public long next(DataOutputBuffer bufOut) throws IOException;
/**
* This method dumps the next key/value into the DataOuputStream
* passed in.
*
* @param key key we are interested in.
* @param bufOut DataOutputStream that needs to be filled.
* @param section region of the file that needs to be read
* @throws IOException
* @return the number of bytes read.
*/
public long next(String key, DataOutputBuffer bufOut, Coordinate section) throws IOException;
/**
* This method dumps the next key/value into the DataOuputStream
* passed in. Always use this method to query for application
@ -93,12 +73,12 @@ public interface IFileReader
* that we want to return
* OR
* @param timeRange - time range we are interested in
* @param section region of the file that needs to be read
* @param position
* @throws IOException
* @return number of bytes read.
*
*/
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, List<String> columnNames, IndexHelper.TimeRange timeRange, Coordinate section) throws IOException;
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, List<String> columnNames, IndexHelper.TimeRange timeRange, long position) throws IOException;
/**
* Close the file after reading.

649
src/java/org/apache/cassandra/io/SSTable.java
View File

@ -18,54 +18,31 @@
package org.apache.cassandra.io;
import java.io.File;
import java.io.IOException;
import java.io.*;
import java.util.*;
import org.apache.log4j.Logger;
import org.apache.commons.lang.StringUtils;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.SequenceFile.ColumnGroupReader;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.BasicUtilities;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.utils.LogUtil;
import org.apache.cassandra.io.SequenceFile.ColumnGroupReader;
import org.apache.log4j.Logger;
/**
* This class is built on top of the SequenceFile. It stores
* data on disk in sorted fashion. However the sorting is upto
* the application. This class expects keys to be handed to it
* in sorted order. SSTable is broken up into blocks where each
* block contains 128 keys. At the end of the file the block
* index is written which contains the offsets to the keys in the
* block. SSTable also maintains an index file to which every 128th
* key is written with a pointer to the block index which is the block
* that actually contains the key. This index file is then read and
* maintained in memory. SSTable is append only and immutable. SSTable
* on disk looks as follows:
* <p/>
* -------------------------
* |------------------------|<-------|
* | | | BLOCK-INDEX PTR
* | | |
* |------------------------|--------
* |------------------------|<-------|
* | | |
* | | | BLOCK-INDEX PTR
* | | |
* |------------------------|---------
* |------------------------|<--------|
* | | |
* | | |
* | | | BLOCK-INDEX PTR
* | | |
* |------------------------| |
* |------------------------|----------
* |------------------------|-----------------> BLOOM-FILTER
* version-info <--|----------|-------------|-------> relative offset to last block index.
* <p/>
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
* in sorted order.
*
* A separate index file is maintained as well, containing the
* SSTable keys and the offset into the SSTable at which they are found.
* Every 1/indexInterval key is read into memory when the SSTable is opened.
*
* Finally, a bloom filter file is also kept for the keys in each SSTable.
*/
public class SSTable
@ -75,12 +52,8 @@ public class SSTable
private static Object indexLoadLock_ = new Object();
/* Every 128th key is an index. */
private static final int indexInterval_ = 128;
/* Key associated with block index written to disk */
public static final String blockIndexKey_ = "BLOCK-INDEX";
/* Required extension for temporary files created during compactions. */
public static final String temporaryFile_ = "tmp";
/* Use this long as a 64 bit entity to turn on some bits for various settings */
private static final long version_ = 0L;
/*
* This map has the SSTable as key and a BloomFilter as value. This
* BloomFilter will tell us if a key/column pair is in the SSTable.
@ -95,9 +68,14 @@ public class SSTable
}
/*
* Maintains a list of KeyPositionInfo objects per SSTable file loaded.
* Maintains a list of KeyPosition objects per SSTable file loaded.
* We do this so that we don't read the index file into memory multiple
* times.
*
* The positions here refer to positions _in the index file_, not in
* the sstable file directly. So looking up a data location is a two
* step process: use this map to find the location of the original index
* entry, then read the data file at the location given by the index.
*/
static IndexMap indexMetadataMap_ = new IndexMap();
@ -105,28 +83,26 @@ public class SSTable
* This method deletes both the specified data file
* and the associated index file
*
* @param dataFileName - data file associated with the SSTable
* @param dataFile - data file associated with the SSTable
*/
public static void delete(String dataFileName)
public static void delete(String dataFile) throws IOException
{
/* remove the cached index table from memory */
indexMetadataMap_.remove(dataFileName);
indexMetadataMap_.remove(dataFile);
/* Delete the checksum file associated with this data file */
try
{
ChecksumManager.onFileDelete(dataFileName);
}
catch (IOException ex)
{
logger_.info(LogUtil.throwableToString(ex));
}
ChecksumManager.onFileDelete(dataFile);
File file = new File(dataFileName);
assert file.exists() : "attempted to delete non-existing file " + dataFileName;
/* delete the data file */
deleteWithConfirm(new File(dataFile));
deleteWithConfirm(new File(indexFilename(dataFile)));
deleteWithConfirm(new File(filterFilename(dataFile)));
}
private static void deleteWithConfirm(File file) throws IOException
{
assert file.exists() : "attempted to delete non-existing file " + file.getName();
if (!file.delete())
{
logger_.error("Failed to delete " + dataFileName);
throw new IOException("Failed to delete " + file.getName());
}
}
@ -136,7 +112,7 @@ public class SSTable
for (String dataFile : dataFiles)
{
List<KeyPositionInfo> index = indexMetadataMap_.get(dataFile);
List<KeyPosition> index = indexMetadataMap_.get(dataFile);
if (index != null)
{
int indexKeyCount = index.size();
@ -154,17 +130,17 @@ public class SSTable
public static List<String> getIndexedKeys()
{
Set<String> indexFiles = indexMetadataMap_.keySet();
List<KeyPositionInfo> KeyPositions = new ArrayList<KeyPositionInfo>();
List<KeyPosition> positions = new ArrayList<KeyPosition>();
for (String indexFile : indexFiles)
{
KeyPositions.addAll(indexMetadataMap_.get(indexFile));
positions.addAll(indexMetadataMap_.get(indexFile));
}
List<String> indexedKeys = new ArrayList<String>();
for (KeyPositionInfo position : KeyPositions)
for (KeyPosition kp : positions)
{
indexedKeys.add(position.decoratedKey);
indexedKeys.add(kp.key);
}
Collections.sort(indexedKeys);
@ -180,10 +156,9 @@ public class SSTable
{
for (String filename : filenames)
{
SSTable ssTable = null;
try
{
ssTable = new SSTable(filename, StorageService.getPartitioner());
new SSTable(filename, StorageService.getPartitioner());
}
catch (IOException ex)
{
@ -191,13 +166,6 @@ public class SSTable
FileUtils.delete(filename);
logger_.warn(LogUtil.throwableToString(ex));
}
finally
{
if (ssTable != null)
{
ssTable.close();
}
}
}
}
@ -233,14 +201,10 @@ public class SSTable
}
String dataFile_;
private long keysWritten;
private IFileWriter dataWriter_;
private BufferedRandomAccessFile indexRAF_;
private String lastWrittenKey_;
private long firstBlockPosition_ = 0L;
private int indexKeysWritten_ = 0;
/* Holds the keys and their respective positions of the current block index */
private SortedMap<String, SSTableIndex.BlockMetadata> blockIndex_;
/* Holds all the block indicies for this SSTable */
private List<SortedMap<String, SSTableIndex.BlockMetadata>> blockIndexes_;
private IPartitioner partitioner_;
/**
@ -252,145 +216,6 @@ public class SSTable
{
dataFile_ = dataFileName;
partitioner_ = partitioner;
init();
}
/**
* This ctor is used for writing data into the SSTable. Use this
* version for non DB writes to the SSTable.
*/
public SSTable(String directory, String filename, IPartitioner partitioner) throws IOException
{
dataFile_ = directory + System.getProperty("file.separator") + filename + "-Data.db";
partitioner_ = partitioner;
blockIndex_ = new TreeMap<String, SSTableIndex.BlockMetadata>(partitioner_.getReverseDecoratedKeyComparator());
blockIndexes_ = new ArrayList<SortedMap<String, SSTableIndex.BlockMetadata>>();
dataWriter_ = SequenceFile.bufferedWriter(dataFile_, 4 * 1024 * 1024);
}
private void loadBloomFilter(IFileReader indexReader, long size) throws IOException
{
/* read the position of the bloom filter */
indexReader.seek(size - 8);
byte[] bytes = new byte[8];
long currentPosition = indexReader.getCurrentPosition();
indexReader.readDirect(bytes);
long position = BasicUtilities.byteArrayToLong(bytes);
/* seek to the position of the bloom filter */
indexReader.seek(currentPosition - position);
DataOutputBuffer bufOut = new DataOutputBuffer();
DataInputBuffer bufIn = new DataInputBuffer();
/* read the bloom filter from disk */
indexReader.next(bufOut);
bufOut.close();
bufIn.reset(bufOut.getData(), bufOut.getLength());
String clientKey = bufIn.readUTF();
if (clientKey.equals(SequenceFile.marker_))
{
/*
* We are now reading the serialized Bloom Filter. We read
* the length and then pass the bufIn to the serializer of
* the BloomFilter. We then store the Bloom filter in the
* map. However if the Bloom Filter already exists then we
* need not read the rest of the file.
*/
bufIn.readInt();
if (bfs_.get(dataFile_) == null)
{
bfs_.put(dataFile_, BloomFilter.serializer().deserialize(bufIn));
}
}
}
private void loadIndexFile() throws IOException
{
logger_.debug("Loading indexes from " + dataFile_);
IFileReader indexReader = null;
/* Read all block indexes to maintain an index in memory */
try
{
indexReader = SequenceFile.bufferedReader(dataFile_, 4 * 1024 * 1024);
long size = indexReader.getEOF();
/* load the bloom filter into memory */
loadBloomFilter(indexReader, size);
/* read the position of the last block index */
byte[] bytes = new byte[8];
/* seek to the position to read the relative position of the first block index */
indexReader.seek(size - 16L);
/* the beginning of the first block index */
long currentPosition = indexReader.getCurrentPosition();
indexReader.readDirect(bytes);
long firstBlockIndexPosition = BasicUtilities.byteArrayToLong(bytes);
List<KeyPositionInfo> KeyPositions = new ArrayList<KeyPositionInfo>();
indexMetadataMap_.put(dataFile_, KeyPositions);
DataOutputBuffer bufOut = new DataOutputBuffer();
DataInputBuffer bufIn = new DataInputBuffer();
long nextPosition = currentPosition - firstBlockIndexPosition;
indexReader.seek(nextPosition);
/* read the block indexes from the end of the file till we hit the first one. */
while (nextPosition > 0)
{
bufOut.reset();
/* position @ the current block index being processed */
currentPosition = indexReader.getCurrentPosition();
long bytesRead = indexReader.next(bufOut);
if (bytesRead != -1)
{
bufIn.reset(bufOut.getData(), bufOut.getLength());
/* read the block key. */
String blockIndexKey = bufIn.readUTF();
if (!blockIndexKey.equals(SSTable.blockIndexKey_))
{
break;
}
/* read the size of the block index */
bufIn.readInt();
/* Number of keys in the block. */
int keys = bufIn.readInt();
String largestKeyInBlock;
for (int i = 0; i < keys; ++i)
{
String keyInBlock = bufIn.readUTF();
if (i == 0)
{
largestKeyInBlock = keyInBlock;
/* relative offset in the block for the key*/
bufIn.readLong();
/* size of data associated with the key */
bufIn.readLong();
/* load the actual position of the block index into the index map */
KeyPositions.add(new KeyPositionInfo(largestKeyInBlock, currentPosition, partitioner_));
}
else
{
/*
* This is not the key we are looking for. So read its position
* and the size of the data associated with it. This was stored
* as the BlockMetadata.
*/
bufIn.readLong();
bufIn.readLong();
}
}
}
}
bufIn.close();
bufOut.close();
Collections.sort(KeyPositions);
}
finally
{
if (indexReader != null)
{
indexReader.close();
}
}
}
private void init() throws IOException
{
/*
* this is to prevent multiple threads from
* loading the same index files multiple times
@ -402,11 +227,79 @@ public class SSTable
{
long start = System.currentTimeMillis();
loadIndexFile();
loadBloomFilter();
logger_.debug("INDEX LOAD TIME: " + (System.currentTimeMillis() - start) + " ms.");
}
}
}
/**
* This ctor is used for writing data into the SSTable. Use this
* version for non DB writes to the SSTable.
*/
public SSTable(String directory, String filename, IPartitioner partitioner) throws IOException
{
dataFile_ = directory + System.getProperty("file.separator") + filename + "-Data.db";
partitioner_ = partitioner;
dataWriter_ = SequenceFile.bufferedWriter(dataFile_, 4 * 1024 * 1024);
indexRAF_ = new BufferedRandomAccessFile(indexFilename(), "rw", 1024 * 1024);
}
private void loadBloomFilter() throws IOException
{
assert bfs_.get(dataFile_) == null;
DataInputStream stream = new DataInputStream(new FileInputStream(filterFilename()));
bfs_.put(dataFile_, BloomFilter.serializer().deserialize(stream));
}
private void loadIndexFile() throws IOException
{
BufferedRandomAccessFile input = new BufferedRandomAccessFile(indexFilename(), "r");
ArrayList<KeyPosition> indexEntries = new ArrayList<KeyPosition>();
int i = 0;
long indexSize = input.length();
while (true)
{
long indexPosition = input.getFilePointer();
if (indexPosition == indexSize)
{
break;
}
String decoratedKey = input.readUTF();
input.readLong();
if (i++ % indexInterval_ == 0)
{
indexEntries.add(new KeyPosition(decoratedKey, indexPosition, partitioner_));
}
}
indexMetadataMap_.put(dataFile_, indexEntries);
}
private static String indexFilename(String dataFile)
{
String[] parts = dataFile.split("-");
parts[parts.length - 1] = "Index.db";
return StringUtils.join(parts, "-");
}
private String indexFilename()
{
return indexFilename(dataFile_);
}
private static String filterFilename(String dataFile)
{
String[] parts = dataFile.split("-");
parts[parts.length - 1] = "Filter.db";
return StringUtils.join(parts, "-");
}
private String filterFilename()
{
return filterFilename(dataFile_);
}
private String getFile(String name) throws IOException
{
File file = new File(name);
@ -439,132 +332,138 @@ public class SSTable
return (lastWrittenKey_ == null) ? 0 : dataWriter_.getCurrentPosition();
}
private void afterAppend(String decoratedKey, long position, long size) throws IOException
private void afterAppend(String decoratedKey, long position) throws IOException
{
++indexKeysWritten_;
lastWrittenKey_ = decoratedKey;
blockIndex_.put(decoratedKey, new SSTableIndex.BlockMetadata(position, size));
if (indexKeysWritten_ == indexInterval_)
{
blockIndexes_.add(blockIndex_);
blockIndex_ = new TreeMap<String, SSTableIndex.BlockMetadata>(partitioner_.getReverseDecoratedKeyComparator());
indexKeysWritten_ = 0;
}
}
long indexPosition = indexRAF_.getFilePointer();
indexRAF_.writeUTF(decoratedKey);
indexRAF_.writeLong(position);
logger_.trace("wrote " + decoratedKey + " at " + position);
/**
* Dumps all the block indicies for this SSTable
* at the end of the file.
*
* @throws IOException
*/
private void dumpBlockIndexes() throws IOException
{
firstBlockPosition_ = dataWriter_.getCurrentPosition();
for (SortedMap<String, SSTableIndex.BlockMetadata> block : blockIndexes_)
{
dumpBlockIndex(block);
}
}
private void dumpBlockIndex(SortedMap<String, SSTableIndex.BlockMetadata> blockIndex) throws IOException
{
/* Block Index is empty so bail. */
if (blockIndex.size() == 0)
{
if (keysWritten++ % indexInterval_ != 0)
return;
}
DataOutputBuffer bufOut = new DataOutputBuffer();
/*
* Record the position where we start writing the block index. This is will be
* used as the position of the lastWrittenKey in the block in the index file
*/
long position = dataWriter_.getCurrentPosition();
Set<String> keys = blockIndex.keySet();
/* Number of keys in this block */
bufOut.writeInt(keys.size());
for (String decoratedKey : keys)
List<KeyPosition> indexEntries = SSTable.indexMetadataMap_.get(dataFile_);
if (indexEntries == null)
{
bufOut.writeUTF(decoratedKey);
SSTableIndex.BlockMetadata blockMetadata = blockIndex.get(decoratedKey);
/* position of the key as a relative offset */
bufOut.writeLong(position - blockMetadata.position_);
bufOut.writeLong(blockMetadata.size_);
indexEntries = new ArrayList<KeyPosition>();
SSTable.indexMetadataMap_.put(dataFile_, indexEntries);
}
/* Write out the block index. */
dataWriter_.append(SSTable.blockIndexKey_, bufOut);
/* Load this index into the in memory index map */
List<KeyPositionInfo> KeyPositions = SSTable.indexMetadataMap_.get(dataFile_);
if (KeyPositions == null)
{
KeyPositions = new ArrayList<KeyPositionInfo>();
SSTable.indexMetadataMap_.put(dataFile_, KeyPositions);
}
KeyPositions.add(new KeyPositionInfo(blockIndex.firstKey(), position, partitioner_));
blockIndex.clear();
indexEntries.add(new KeyPosition(decoratedKey, indexPosition, partitioner_));
logger_.trace("wrote index of " + decoratedKey + " at " + indexPosition);
}
// TODO make this take a DataOutputStream and wrap the byte[] version to combine them
public void append(String decoratedKey, DataOutputBuffer buffer) throws IOException
{
long currentPosition = beforeAppend(decoratedKey);
dataWriter_.append(decoratedKey, buffer);
afterAppend(decoratedKey, currentPosition, buffer.getLength());
afterAppend(decoratedKey, currentPosition);
}
public void append(String decoratedKey, byte[] value) throws IOException
{
long currentPosition = beforeAppend(decoratedKey);
dataWriter_.append(decoratedKey, value);
afterAppend(decoratedKey, currentPosition, value.length);
afterAppend(decoratedKey, currentPosition);
}
/*
TODO only the end_ part of the returned Coordinate is ever used. Apparently this code works, but it's definitely due for some cleanup
since the code fooling about with start_ appears to be irrelevant.
*/
public static Coordinate getCoordinates(String decoratedKey, IFileReader dataReader, IPartitioner partitioner) throws IOException
/** get the position in the index file to start scanning to find the given key (at most indexInterval keys away) */
private static long getIndexScanPosition(String decoratedKey, IFileReader dataReader, IPartitioner partitioner)
{
List<KeyPositionInfo> indexInfo = indexMetadataMap_.get(dataReader.getFileName());
assert indexInfo != null && indexInfo.size() > 0;
long start = 0L;
long end;
int index = Collections.binarySearch(indexInfo, new KeyPositionInfo(decoratedKey, -1, partitioner));
List<KeyPosition> positions = indexMetadataMap_.get(dataReader.getFileName());
assert positions != null && positions.size() > 0;
int index = Collections.binarySearch(positions, new KeyPosition(decoratedKey, -1, partitioner));
if (index < 0)
{
/*
* We are here which means that the requested
* key is not an index.
*/
index = (++index) * (-1);
/*
* This means key is not present at all. Hence
* a scan is in order.
*/
start = (index == 0) ? 0 : indexInfo.get(index - 1).position;
if (index < indexInfo.size())
{
end = indexInfo.get(index).position;
}
else
{
/* This is the Block Index in the file. */
end = start;
}
// binary search gives us the first index _greater_ than the key searched for,
// i.e., its insertion position
int greaterThan = (index + 1) * -1;
if (greaterThan == 0)
return -1;
return positions.get(greaterThan - 1).position;
}
else
{
/*
* If we are here that means the key is in the index file
* and we can retrieve it w/o a scan. In reality we would
* like to have a retreive(key, fromPosition) but for now
* we use scan(start, start + 1) - a hack.
*/
start = indexInfo.get(index).position;
end = start;
return positions.get(index).position;
}
}
/**
* returns the position in the data file to find the given key, or -1 if the key is not present
*/
/* TODO having this static means we have to keep re-opening the index file, which sucks. Need to move towards
greater encapsulation. */
public static long getPosition(String decoratedKey, IFileReader dataReader, IPartitioner partitioner) throws IOException
{
long start = getIndexScanPosition(decoratedKey, dataReader, partitioner);
if (start < 0)
{
return -1;
}
BufferedRandomAccessFile input = new BufferedRandomAccessFile(indexFilename(dataReader.getFileName()), "r");
input.seek(start);
int i = 0;
try
{
do
{
String indexDecoratedKey;
try
{
indexDecoratedKey = input.readUTF();
}
catch (EOFException e)
{
return -1;
}
long position = input.readLong();
int v = partitioner.getDecoratedKeyComparator().compare(indexDecoratedKey, decoratedKey);
if (v == 0)
return position;
if (v > 0)
return -1;
} while (++i < indexInterval_);
}
finally
{
input.close();
}
return -1;
}
/** like getPosition, but if key is not found will return the location of the first key _greater_ than the desired one, or -1 if no such key exists. */
public static long getNearestPosition(String decoratedKey, IFileReader dataReader, IPartitioner partitioner) throws IOException
{
long start = getIndexScanPosition(decoratedKey, dataReader, partitioner);
if (start < 0)
{
return 0;
}
BufferedRandomAccessFile input = new BufferedRandomAccessFile(indexFilename(dataReader.getFileName()), "r");
input.seek(start);
try
{
while (true)
{
String indexDecoratedKey;
try
{
indexDecoratedKey = input.readUTF();
}
catch (EOFException e)
{
return -1;
}
long position = input.readLong();
int v = partitioner.getDecoratedKeyComparator().compare(indexDecoratedKey, decoratedKey);
if (v >= 0)
return position;
}
}
finally
{
input.close();
}
return new Coordinate(start, end);
}
public DataInputBuffer next(final String clientKey, String cfName, List<String> columnNames) throws IOException
@ -578,19 +477,12 @@ public class SSTable
try
{
dataReader = SequenceFile.reader(dataFile_);
// dataReader = SequenceFile.chksumReader(dataFile_, 4*1024*1024);
/* Morph key into actual key based on the partition type. */
String decoratedKey = partitioner_.decorateKey(clientKey);
Coordinate fileCoordinate = getCoordinates(decoratedKey, dataReader, partitioner_);
/*
* we have the position we have to read from in order to get the
* column family, get the column family and column(s) needed.
*/
long position = getPosition(decoratedKey, dataReader, partitioner_);
DataOutputBuffer bufOut = new DataOutputBuffer();
DataInputBuffer bufIn = new DataInputBuffer();
long bytesRead = dataReader.next(decoratedKey, bufOut, cfName, columnNames, timeRange, fileCoordinate);
long bytesRead = dataReader.next(decoratedKey, bufOut, cfName, columnNames, timeRange, position);
if (bytesRead != -1L)
{
if (bufOut.getLength() > 0)
@ -620,72 +512,51 @@ public class SSTable
return next(clientKey, columnFamilyName, cnNames);
}
public void close() throws IOException
{
close(new byte[0], 0);
}
public void close(BloomFilter bf) throws IOException
{
/* Any remnants in the blockIndex should be added to the dump */
blockIndexes_.add(blockIndex_);
dumpBlockIndexes();
// bloom filter
FileOutputStream fos = new FileOutputStream(filterFilename());
DataOutputStream stream = new DataOutputStream(fos);
BloomFilter.serializer().serialize(bf, stream);
stream.flush();
fos.getFD().sync();
stream.close();
/* reset the buffer and serialize the Bloom Filter. */
DataOutputBuffer bufOut = new DataOutputBuffer();
BloomFilter.serializer().serialize(bf, bufOut);
close(bufOut.getData(), bufOut.getLength());
bufOut.close();
// byte[] bytes = new byte[bufOut.getLength()];
// System.arraycopy(bufOut.getData(), 0, bytes, 0, bufOut.getLength());
// close(bytes, bytes.length);
// index
indexRAF_.getChannel().force(true);
indexRAF_.close();
// main data
dataWriter_.close(); // calls force
}
private static String rename(String tmpFilename)
{
String filename = tmpFilename.replace("-" + temporaryFile_, "");
new File(tmpFilename).renameTo(new File(filename));
return filename;
}
/**
* Renames a temporary SSTable file to a valid data and index file
* Renames temporary SSTable files to valid data, index, and bloom filter files
*/
public void closeRename(BloomFilter bf) throws IOException
{
close(bf);
String tmpDataFile = dataFile_;
String dataFileName = dataFile_.replace("-" + temporaryFile_, "");
File dataFile = new File(dataFile_);
dataFile.renameTo(new File(dataFileName));
dataFile_ = dataFileName;
/* Now repair the in memory index associated with the old name */
List<KeyPositionInfo> KeyPositions = SSTable.indexMetadataMap_.remove(tmpDataFile);
SSTable.indexMetadataMap_.put(dataFile_, KeyPositions);
}
close(bf);
private void close(byte[] footer, int size) throws IOException
{
/*
* Write the bloom filter for this SSTable.
* Then write three longs one which is a version
* and one which is a pointer to the last written
* block index and the last one is the position of
* the Bloom Filter.
*/
assert dataWriter_ != null;
long bloomFilterPosition = dataWriter_.getCurrentPosition();
dataWriter_.close(footer, size);
/* write the version field into the SSTable */
dataWriter_.writeDirect(BasicUtilities.longToByteArray(version_));
/* write the relative position of the first block index from current position */
long blockPosition = dataWriter_.getCurrentPosition() - firstBlockPosition_;
dataWriter_.writeDirect(BasicUtilities.longToByteArray(blockPosition));
String oldDataFileName = dataFile_;
rename(indexFilename());
rename(filterFilename());
dataFile_ = rename(dataFile_); // important to do this last since index & filter file names are derived from it
/* write the position of the bloom filter */
long bloomFilterRelativePosition = dataWriter_.getCurrentPosition() - bloomFilterPosition;
dataWriter_.writeDirect(BasicUtilities.longToByteArray(bloomFilterRelativePosition));
dataWriter_.close();
List<KeyPosition> positions = SSTable.indexMetadataMap_.remove(oldDataFileName);
SSTable.indexMetadataMap_.put(dataFile_, positions);
}
/**
* obtain a BlockReader for the getColumnSlice call.
*/
public ColumnGroupReader getColumnGroupReader(String key, String cfName,
String startColumn, boolean isAscending) throws IOException
public ColumnGroupReader getColumnGroupReader(String key, String cfName, String startColumn, boolean isAscending) throws IOException
{
ColumnGroupReader reader = null;
IFileReader dataReader = SequenceFile.reader(dataFile_);
@ -694,8 +565,8 @@ public class SSTable
{
/* Morph key into actual key based on the partition type. */
String decoratedKey = partitioner_.decorateKey(key);
Coordinate fileCoordinate = getCoordinates(decoratedKey, dataReader, partitioner_);
reader = new ColumnGroupReader(dataFile_, decoratedKey, cfName, startColumn, isAscending, fileCoordinate);
long position = getPosition(decoratedKey, dataReader, partitioner_);
reader = new ColumnGroupReader(dataFile_, decoratedKey, cfName, startColumn, isAscending, position);
}
finally
{
@ -704,6 +575,12 @@ public class SSTable
}
return reader;
}
/** obviously only for testing */
public static void forceBloomFilterFailures(String filename)
{
SSTable.bfs_.put(filename, BloomFilter.alwaysMatchingBloomFilter());
}
}
@ -736,27 +613,27 @@ class TouchedKeyCache extends LinkedHashMap<String, Long>
*
* All keys are decorated.
*/
class KeyPositionInfo implements Comparable<KeyPositionInfo>
class KeyPosition implements Comparable<KeyPosition>
{
public final String decoratedKey;
public final String key;
public final long position;
private final IPartitioner partitioner; // TODO rip out the static uses of KP so we can just use the parent SSTable's partitioner, when necessary
public KeyPositionInfo(String decoratedKey, long position, IPartitioner partitioner)
public KeyPosition(String key, long position, IPartitioner partitioner)
{
this.decoratedKey = decoratedKey;
this.key = key;
this.position = position;
this.partitioner = partitioner;
}
public int compareTo(KeyPositionInfo kp)
public int compareTo(KeyPosition kp)
{
return partitioner.getDecoratedKeyComparator().compare(decoratedKey, kp.decoratedKey);
return partitioner.getDecoratedKeyComparator().compare(key, kp.key);
}
public String toString()
{
return decoratedKey + ":" + position;
return key + ":" + position;
}
}
@ -766,7 +643,7 @@ class KeyPositionInfo implements Comparable<KeyPositionInfo>
*/
class IndexMap
{
private final Hashtable<String, List<KeyPositionInfo>> hashtable = new Hashtable<String, List<KeyPositionInfo>>();
private final Hashtable<String, List<KeyPosition>> hashtable = new Hashtable<String, List<KeyPosition>>();
private String cannonicalize(String filename)
{
@ -780,12 +657,12 @@ class IndexMap
}
}
public List<KeyPositionInfo> get(String filename)
public List<KeyPosition> get(String filename)
{
return hashtable.get(cannonicalize(filename));
}
public List<KeyPositionInfo> put(String filename, List<KeyPositionInfo> value)
public List<KeyPosition> put(String filename, List<KeyPosition> value)
{
return hashtable.put(cannonicalize(filename), value);
}
@ -800,7 +677,7 @@ class IndexMap
return hashtable.keySet();
}
public List<KeyPositionInfo> remove(String filename)
public List<KeyPosition> remove(String filename)
{
return hashtable.remove(cannonicalize(filename));
}

22
src/java/org/apache/cassandra/io/SSTableIndex.java
View File

@ -1,22 +0,0 @@
package org.apache.cassandra.io;
public class SSTableIndex
{
/**
* This class holds the position of a key in a block
* and the size of the data associated with this key.
*/
public static class BlockMetadata
{
protected static final BlockMetadata NULL = new BlockMetadata(-1L, -1L);
long position_;
long size_;
BlockMetadata(long position, long size)
{
position_ = position;
size_ = size;
}
}
}

137
src/java/org/apache/cassandra/io/SequenceFile.java
View File

@ -180,6 +180,7 @@ public class SequenceFile
public void close() throws IOException
{
file_.getChannel().force(true);
file_.close();
}
@ -552,13 +553,13 @@ public class SequenceFile
private int localDeletionTime_;
private long markedForDeleteAt_;
ColumnGroupReader(String filename, String key, String cfName, String startColumn, boolean isAscending, Coordinate section) throws IOException
ColumnGroupReader(String filename, String key, String cfName, String startColumn, boolean isAscending, long position) throws IOException
{
super(filename, 128 * 1024);
this.cfName_ = cfName;
this.key_ = key;
this.isAscending_ = isAscending;
init(startColumn, section);
init(startColumn, position);
}
/**
@ -593,10 +594,10 @@ public class SequenceFile
return fullColIndexList;
}
private void init(String startColumn, Coordinate section) throws IOException
private void init(String startColumn, long position) throws IOException
{
String keyInDisk = null;
if (seekTo(key_, section) >= 0)
if (seekTo(position) >= 0)
keyInDisk = file_.readUTF();
if ( keyInDisk != null && keyInDisk.equals(key_))
@ -629,7 +630,7 @@ public class SequenceFile
int index = Collections.binarySearch(columnIndexList_, new IndexHelper.ColumnNameIndexInfo(startColumn));
curRangeIndex_ = index < 0 ? (++index) * (-1) - 1 : index;
}
else
else
{
/* no keys found in this file because of a false positive in BF */
curRangeIndex_ = -1;
@ -689,70 +690,8 @@ public class SequenceFile
return filename_;
}
/**
* Return the position of the given key from the block index.
*
* @param key the key whose offset is to be extracted from the current block index
*/
public long getPositionFromBlockIndex(String key) throws IOException
long seekTo(long position) throws IOException
{
long position = -1L;
/* note the beginning of the block index */
long blockIndexPosition = file_.getFilePointer();
/* read the block key. */
String blockIndexKey = file_.readUTF();
if (!blockIndexKey.equals(SSTable.blockIndexKey_))
throw new IOException("Unexpected position to be reading the block index from.");
/* read the size of the block index */
int size = file_.readInt();
/* Read the entire block index. */
byte[] bytes = new byte[size];
file_.readFully(bytes);
DataInputBuffer bufIn = new DataInputBuffer();
bufIn.reset(bytes, bytes.length);
/* Number of keys in the block. */
int keys = bufIn.readInt();
for (int i = 0; i < keys; ++i)
{
String keyInBlock = bufIn.readUTF();
if (keyInBlock.equals(key))
{
position = bufIn.readLong();
break;
}
else
{
/*
* This is not the key we are looking for. So read its position
* and the size of the data associated with it. This was strored
* as the BlockMetadata.
*/
bufIn.readLong();
bufIn.readLong();
}
}
/* we do this because relative position of the key within a block is stored. */
if (position != -1L)
position = blockIndexPosition - position;
return position;
}
/**
* This method seek the disk head to the block index, finds
* the offset of the key within the block and seeks to that
* offset.
*
* @param key we are interested in.
* @param section indicates the location of the block index.
* @throws IOException
*/
protected long seekTo(String key, Coordinate section) throws IOException
{
seek(section.end_);
long position = getPositionFromBlockIndex(key);
if (position >= 0)
seek(position);
return position;
@ -842,18 +781,19 @@ public class SequenceFile
* @param columnNames columnNames we are interested in
* OR
* @param timeRange time range we are interested in
* @param section region of the file that needs to be read
* @param position
* @return number of bytes that were read.
* @throws IOException
*/
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, List<String> columnNames, IndexHelper.TimeRange timeRange, Coordinate section) throws IOException
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, List<String> columnNames, IndexHelper.TimeRange timeRange, long position) throws IOException
{
assert !columnFamilyName.contains(":");
assert timeRange == null || columnNames == null; // at most one may be non-null
long bytesRead = -1L;
if (isEOF() || seekTo(key, section) < 0)
if (isEOF() || seekTo(position) < 0)
return bytesRead;
/* note the position where the key starts */
long startPosition = file_.getFilePointer();
String keyInDisk = file_.readUTF();
@ -1094,61 +1034,6 @@ public class SequenceFile
bytesRead = -1L;
return bytesRead;
}
/**
* This method dumps the next key/value into the DataOuputStream
* passed in.
*
* @param key - key we are interested in.
* @param bufOut DataOutputStream that needs to be filled.
* @param section region of the file that needs to be read
* @return total number of bytes read/considered
*/
public long next(String key, DataOutputBuffer bufOut, Coordinate section) throws IOException
{
long bytesRead = -1L;
if (isEOF() || seekTo(key, section) < 0)
return bytesRead;
/* note the position where the key starts */
long startPosition = file_.getFilePointer();
String keyInDisk = file_.readUTF();
if (keyInDisk != null)
{
/*
* If key on disk is greater than requested key
* we can bail out since we exploit the property
* of the SSTable format.
*/
if (keyInDisk.compareTo(key) > 0)
return bytesRead;
/*
* If we found the key then we populate the buffer that
* is passed in. If not then we skip over this key and
* position ourselves to read the next one.
*/
int dataSize = file_.readInt();
if (keyInDisk.equals(key))
{
/* write the key into buffer */
bufOut.writeUTF(keyInDisk);
/* write data size into buffer */
bufOut.writeInt(dataSize);
/* write the data into buffer */
bufOut.write(file_, dataSize);
}
else
{
/* skip over data portion */
file_.seek(dataSize + file_.getFilePointer());
}
long endPosition = file_.getFilePointer();
bytesRead = endPosition - startPosition;
}
return bytesRead;
}
}
public static class Reader extends AbstractReader

39
src/java/org/apache/cassandra/loader/Loader.java
View File

@ -1,21 +1,21 @@
/*
* 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.
*/
/*
* 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.
*/
/**
*
*/
@ -315,8 +315,7 @@ public class Loader
/* Load the indexes into memory */
for ( String df : ssTables )
{
SSTable ssTable = new SSTable(df, StorageService.getPartitioner());
ssTable.close();
new SSTable(df, StorageService.getPartitioner());
}
/* We should have only one file since we just compacted. */
List<String> indexedKeys = SSTable.getIndexedKeys();

3
src/java/org/apache/cassandra/loader/PreLoad.java
View File

@ -98,8 +98,7 @@ public class PreLoad
/* Load the indexes into memory */
for ( String df : ssTables )
{
SSTable ssTable = new SSTable(df, StorageService.getPartitioner());
ssTable.close();
new SSTable(df, StorageService.getPartitioner());
}
/* We should have only one file since we just compacted. */
List<String> indexedKeys = SSTable.getIndexedKeys();

167
src/java/org/apache/cassandra/tools/IndexBuilder.java
View File

@ -1,167 +0,0 @@
/**
* 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.tools;
import java.io.IOException;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.DataOutputBuffer;
import org.apache.cassandra.io.IFileReader;
import org.apache.cassandra.io.IFileWriter;
import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.io.SequenceFile;
import org.apache.cassandra.utils.BasicUtilities;
import org.apache.cassandra.utils.BloomFilter;
public class IndexBuilder
{
private static final int bufferSize_ = 64*1024;
public static void main(String[] args)
{
if ( args.length != 1 )
{
System.out.println("Usage : java com.facebook.infrastructure.tools.IndexBuilder <full path to the data file>");
System.exit(1);
}
try
{
int blockCount = getBlockCount(args[0]);
System.out.println("Number of keys in the data file : " + (blockCount + 1)*SSTable.indexInterval());
buildIndex(args[0], blockCount);
}
catch(Throwable t)
{
System.err.println("Exception: " + t.getMessage());
t.printStackTrace(System.err);
}
}
private static int getBlockCount(String dataFile) throws IOException
{
IFileReader dataReader = SequenceFile.bufferedReader(dataFile, bufferSize_);
DataOutputBuffer bufOut = new DataOutputBuffer();
DataInputBuffer bufIn = new DataInputBuffer();
int blockCount = 0;
try
{
while ( !dataReader.isEOF() )
{
bufOut.reset();
dataReader.next(bufOut);
bufIn.reset(bufOut.getData(), bufOut.getLength());
/* Key just read */
String key = bufIn.readUTF();
if ( key.equals(SSTable.blockIndexKey_) )
{
++blockCount;
}
}
}
finally
{
dataReader.close();
}
return blockCount;
}
private static void buildIndex(String dataFile, int blockCount) throws IOException
{
String indexFile = dataFile.replace("-Data.", "-Index.");
final int bufferSize = 64*1024;
IFileWriter indexWriter = SequenceFile.bufferedWriter(indexFile, bufferSize);
IFileReader dataReader = SequenceFile.bufferedReader(dataFile, bufferSize);
DataOutputBuffer bufOut = new DataOutputBuffer();
DataInputBuffer bufIn = new DataInputBuffer();
/* BloomFilter of all data in the data file */
BloomFilter bf = new BloomFilter((SSTable.indexInterval() + 1)*blockCount, 8);
try
{
while ( !dataReader.isEOF() )
{
bufOut.reset();
/* Record the position of the key. */
long blockIndexOffset = dataReader.getCurrentPosition();
dataReader.next(bufOut);
bufIn.reset(bufOut.getData(), bufOut.getLength());
/* Key just read */
String key = bufIn.readUTF();
if ( key.equals(SSTable.blockIndexKey_) )
{
/* Ignore the size of the data associated with the block index */
bufIn.readInt();
/* Number of keys in the block. */
int blockSize = bufIn.readInt();
/* Largest key in the block */
String largestKey = null;
/*
* Read the keys in this block and find the largest key in
* this block. This is the key that gets written into the
* index file.
*/
for ( int i = 0; i < blockSize; ++i )
{
String currentKey = bufIn.readUTF();
bf.add(currentKey);
if ( largestKey == null )
{
largestKey = currentKey;
}
else
{
if ( currentKey.compareTo(largestKey) > 0 )
{
/* record this key */
largestKey = currentKey;
}
}
/* read the position of the key and the size of key data and throws it away. */
bufIn.readLong();
bufIn.readLong();
}
/*
* Write into the index file the largest key in the block
* and the offset of the block index in the data file.
*/
indexWriter.append(largestKey, BasicUtilities.longToByteArray(blockIndexOffset));
}
}
}
finally
{
dataReader.close();
/* Cache the bloom filter */
SSTable.storeBloomFilter(dataFile, bf);
/* Write the bloom filter into the index file */
bufOut.reset();
BloomFilter.serializer().serialize(bf, bufOut);
byte[] bytes = new byte[bufOut.getLength()];
System.arraycopy(bufOut.getData(), 0, bytes, 0, bytes.length);
indexWriter.close(bytes, bytes.length);
bufOut.close();
}
}
}

100
src/java/org/apache/cassandra/tools/KeyExtracter.java
View File

@ -1,100 +0,0 @@
/**
* 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.tools;
import java.io.IOException;
import java.io.RandomAccessFile;
import org.apache.cassandra.io.*;
public class KeyExtracter
{
private static final int bufferSize_ = 64*1024;
public static void main(String[] args) throws Throwable
{
if ( args.length != 3 )
{
System.out.println("Usage : java com.facebook.infrastructure.tools.IndexBuilder <key to extract> <data file> <output file>");
System.exit(1);
}
String keyToExtract = args[0];
String dataFile = args[1];
String outputFile = args[2];
extractKeyIntoFile(keyToExtract, dataFile, outputFile);
}
public static boolean extractKeyIntoFile(String keyToExtract, String dataFile, String outputFile) throws IOException
{
IFileReader dataReader = SequenceFile.bufferedReader(dataFile, bufferSize_);
DataOutputBuffer bufOut = new DataOutputBuffer();
DataInputBuffer bufIn = new DataInputBuffer();
try
{
while ( !dataReader.isEOF() )
{
bufOut.reset();
dataReader.next(bufOut);
bufIn.reset(bufOut.getData(), bufOut.getLength());
/* Key just read */
String key = bufIn.readUTF();
/* check if we want this key */
if ( key.equals(keyToExtract) )
{
int keySize = bufIn.readInt();
byte[] keyData = new byte[keySize];
bufIn.read(keyData, 0, keySize);
/* write the key data into a file */
RandomAccessFile raf = new RandomAccessFile(outputFile, "rw");
raf.writeUTF(key);
raf.writeInt(keySize);
raf.write(keyData);
dumpBlockIndex(keyToExtract, 0L, keySize, raf);
raf.close();
return true;
}
}
}
finally
{
dataReader.close();
}
return false;
}
private static void dumpBlockIndex(String key, long position, long size, RandomAccessFile raf) throws IOException
{
DataOutputBuffer bufOut = new DataOutputBuffer();
/* Number of keys in this block */
bufOut.writeInt(1);
bufOut.writeUTF(key);
bufOut.writeLong(position);
bufOut.writeLong(size);
/* Write out the block index. */
raf.writeUTF(SSTable.blockIndexKey_);
raf.writeInt(bufOut.getLength());
raf.write(bufOut.getData(), 0, bufOut.getLength());
}
}

8
src/java/org/apache/cassandra/utils/BloomFilter.java
View File

@ -120,6 +120,14 @@ public class BloomFilter extends Filter
}
return n;
}
/** @return a BloomFilter that always returns a positive match, for testing */
public static BloomFilter alwaysMatchingBloomFilter()
{
BitSet set = new BitSet(64);
set.set(0, 64);
return new BloomFilter(1, set);
}
}
class BloomFilterSerializer implements ICompactSerializer<BloomFilter>

9
test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -115,10 +115,9 @@ public class ColumnFamilyStoreTest extends CleanupHelper
store.forceBlockingFlush();
List<String> ssTables = table.getAllSSTablesOnDisk();
/* the following call can happen if BF is wrong. Should return an empty buffer. */
IFilter filter = new IdentityFilter();
SSTable ssTable = new SSTable(ssTables.get(0), StorageService.getPartitioner());
DataInputBuffer bufIn = filter.next("key2", "Standard1:Column1", ssTable);
assertEquals(bufIn.getLength(), 0);
assertEquals(1, ssTables.size());
SSTable.forceBloomFilterFailures(ssTables.get(0));
ColumnFamily cf = store.getColumnFamily("key2", "Standard1:Column1", new IdentityFilter());
assertNull(cf);
}
}

15
test/unit/org/apache/cassandra/db/NameSortTest.java
View File

@ -20,15 +20,14 @@ package org.apache.cassandra.db;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.SortedSet;
import java.util.Iterator;
import java.util.Collection;
import java.util.Arrays;
import org.apache.commons.lang.ArrayUtils;
import org.junit.Test;
public class NameSortTest extends ColumnFamilyStoreTest
import org.apache.cassandra.CleanupHelper;
public class NameSortTest extends CleanupHelper
{
@Test
public void testNameSort1() throws IOException, ExecutionException, InterruptedException
@ -64,7 +63,7 @@ public class NameSortTest extends ColumnFamilyStoreTest
// standard
for (int j = 0; j < 8; ++j)
{
byte[] bytes = j % 2 == 0 ? bytes1 : bytes2;
byte[] bytes = j % 2 == 0 ? "a".getBytes() : "b".getBytes();
rm = new RowMutation("Table1", key);
rm.add("Standard1:" + "Column-" + j, bytes, j);
rm.apply();
@ -75,7 +74,7 @@ public class NameSortTest extends ColumnFamilyStoreTest
{
for (int k = 0; k < 4; ++k)
{
byte[] bytes = (j + k) % 2 == 0 ? bytes1 : bytes2;
byte[] bytes = (j + k) % 2 == 0 ? "a".getBytes() : "b".getBytes();
rm = new RowMutation("Table1", key);
rm.add("Super1:" + "SuperColumn-" + j + ":Column-" + k, bytes, k);
rm.apply();
@ -102,7 +101,7 @@ public class NameSortTest extends ColumnFamilyStoreTest
for (IColumn column : columns)
{
int j = Integer.valueOf(column.name().split("-")[1]);
byte[] bytes = j % 2 == 0 ? bytes1 : bytes2;
byte[] bytes = j % 2 == 0 ? "a".getBytes() : "b".getBytes();
assert Arrays.equals(bytes, column.value());
}
@ -118,7 +117,7 @@ public class NameSortTest extends ColumnFamilyStoreTest
for (IColumn subColumn : subColumns)
{
int k = Integer.valueOf(subColumn.name().split("-")[1]);
byte[] bytes = (j + k) % 2 == 0 ? bytes1 : bytes2;
byte[] bytes = (j + k) % 2 == 0 ? "a".getBytes() : "b".getBytes();
assert Arrays.equals(bytes, subColumn.value());
}
}

59
test/unit/org/apache/cassandra/db/TableTest.java
View File

@ -22,11 +22,14 @@ import java.util.SortedSet;
import java.util.Arrays;
import java.util.List;
import java.util.ArrayList;
import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import org.junit.Test;
import static junit.framework.Assert.*;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.io.SSTable;
public class TableTest extends CleanupHelper
{
@ -200,7 +203,7 @@ public class TableTest extends CleanupHelper
private RowMutation makeSimpleRowMutation()
{
RowMutation rm = new RowMutation(TABLE_NAME,TEST_KEY);
ColumnFamily cf = new ColumnFamily("Standard1","Standard");
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
cf.addColumn(new Column("col1","val1".getBytes(), 1L));
cf.addColumn(new Column("col2","val2".getBytes(), 1L));
cf.addColumn(new Column("col3","val3".getBytes(), 1L));
@ -208,9 +211,46 @@ public class TableTest extends CleanupHelper
return rm;
}
@Test
public void testGetSliceNoMatch() throws Throwable
{
Table table = Table.open(TABLE_NAME);
RowMutation rm = new RowMutation(TABLE_NAME, "row1000");
ColumnFamily cf = new ColumnFamily("Standard2", "Standard");
cf.addColumn(new Column("col1", "val1".getBytes(), 1));
rm.add(cf);
rm.apply();
validateGetSliceNoMatch(table);
table.getColumnFamilyStore("Standard2").forceBlockingFlush();
validateGetSliceNoMatch(table);
SortedSet<String> ssTables = table.getColumnFamilyStore("Standard2").getSSTableFilenames();
assertEquals(1, ssTables.size());
SSTable.forceBloomFilterFailures(ssTables.iterator().next());
validateGetSliceNoMatch(table);
}
private void validateGetSliceNoMatch(Table table) throws IOException
{
Row result;
ColumnFamily cf;
// key before the rows that exists
result = table.getSliceFrom("a", "Standard2", true, 0);
cf = result.getColumnFamily("Standard2");
assertColumns(cf);
// key after the rows that exist
result = table.getSliceFrom("z", "Standard2", true, 0);
cf = result.getColumnFamily("Standard2");
assertColumns(cf);
}
@Test
public void testGetSliceFromBasic() throws Throwable
{
// tests slicing against data from one row in a memtable and then flushed to an sstable
Table table = Table.open(TABLE_NAME);
String ROW = "row1";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
@ -227,9 +267,8 @@ public class TableTest extends CleanupHelper
rm = new RowMutation(TABLE_NAME, ROW);
rm.delete("Standard1:col4", 2L);
rm.apply();
validateGetSliceFromBasic(table, ROW);
// flush to disk
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
validateGetSliceFromBasic(table, ROW);
}
@ -237,6 +276,7 @@ public class TableTest extends CleanupHelper
@Test
public void testGetSliceFromAdvanced() throws Throwable
{
// tests slicing against data from one row spread across two sstables
Table table = Table.open(TABLE_NAME);
String ROW = "row2";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
@ -249,7 +289,6 @@ public class TableTest extends CleanupHelper
cf.addColumn(new Column("col6", "val6".getBytes(), 1L));
rm.add(cf);
rm.apply();
// flush to disk
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
rm = new RowMutation(TABLE_NAME, ROW);
@ -259,9 +298,8 @@ public class TableTest extends CleanupHelper
cf.addColumn(new Column("col3", "valx".getBytes(), 2L));
rm.add(cf);
rm.apply();
validateGetSliceFromAdvanced(table, ROW);
// flush to disk
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
validateGetSliceFromAdvanced(table, ROW);
}
@ -269,6 +307,7 @@ public class TableTest extends CleanupHelper
@Test
public void testGetSliceFromLarge() throws Throwable
{
// tests slicing against 1000 columns in an sstable
Table table = Table.open(TABLE_NAME);
String ROW = "row3";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
@ -277,7 +316,6 @@ public class TableTest extends CleanupHelper
cf.addColumn(new Column("col" + i, ("vvvvvvvvvvvvvvvv" + i).getBytes(), 1L));
rm.add(cf);
rm.apply();
// flush to disk
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
Row result;
@ -311,7 +349,7 @@ public class TableTest extends CleanupHelper
assertEquals(new String(cfres.getColumn("col1992").value()), "vvvvvvvvvvvvvvvv1992");
}
private void assertColumns(ColumnFamily columnFamily, String... columnFamilyNames)
private void assertColumns(ColumnFamily columnFamily, String... columnNames)
{
assertNotNull(columnFamily);
SortedSet<IColumn> columns = columnFamily.getAllColumns();
@ -320,7 +358,8 @@ public class TableTest extends CleanupHelper
{
L.add(column.name());
}
assert Arrays.equals(L.toArray(new String[columns.size()]), columnFamilyNames);
assert Arrays.equals(L.toArray(new String[columns.size()]), columnNames)
: "Columns [" + StringUtils.join(columns, ", ") + "] is not expected [" + StringUtils.join(columnNames, ", ") + "]";
}
private void validateGetSliceFromAdvanced(Table table, String row) throws Throwable
@ -333,7 +372,7 @@ public class TableTest extends CleanupHelper
assertColumns(cfres, "col2", "col3", "col4");
assertEquals(new String(cfres.getColumn("col2").value()), "valx");
assertEquals(new String(cfres.getColumn("col3").value()), "valx");
assertEquals(new String(cfres.getColumn("col4").value()), "val4");
assertEquals(new String(cfres.getColumn("col4").value()), "val4");
}
private void validateGetSliceFromBasic(Table table, String row) throws Throwable

6
test/unit/org/apache/cassandra/db/TimeSortTest.java
View File

@ -26,7 +26,9 @@ import java.util.Iterator;
import org.apache.commons.lang.ArrayUtils;
import org.junit.Test;
public class TimeSortTest extends ColumnFamilyStoreTest
import org.apache.cassandra.CleanupHelper;
public class TimeSortTest extends CleanupHelper
{
@Test
public void testTimeSort() throws IOException, ExecutionException, InterruptedException
@ -39,7 +41,7 @@ public class TimeSortTest extends ColumnFamilyStoreTest
RowMutation rm;
for (int j = 0; j < 8; ++j)
{
byte[] bytes = j % 2 == 0 ? bytes1 : bytes2;
byte[] bytes = j % 2 == 0 ? "a".getBytes() : "b".getBytes();
rm = new RowMutation("Table1", key);
rm.add("StandardByTime1:" + "Column-" + j, bytes, j * 2);
rm.apply();

16
test/unit/org/apache/cassandra/io/SSTableTest.java
View File

@ -49,8 +49,14 @@ public class SSTableTest extends CleanupHelper
ssTable.close(bf);
// verify
verifySingle(f, bytes, key);
SSTable.indexMetadataMap_.clear(); // force reloading the index
ssTable = new SSTable(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
verifySingle(f, bytes, key);
}
private void verifySingle(File f, byte[] bytes, String key) throws IOException
{
SSTable ssTable = new SSTable(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
FileStruct fs = new FileStruct(SequenceFile.bufferedReader(ssTable.dataFile_, 128 * 1024), new OrderPreservingPartitioner());
fs.seekTo(key);
int size = fs.getBufIn().readInt();
@ -80,10 +86,16 @@ public class SSTableTest extends CleanupHelper
ssTable.close(bf);
// verify
verifyMany(f, map);
SSTable.indexMetadataMap_.clear(); // force reloading the index
verifyMany(f, map);
}
private void verifyMany(File f, TreeMap<String, byte[]> map) throws IOException
{
List<String> keys = new ArrayList(map.keySet());
Collections.shuffle(keys);
ssTable = new SSTable(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
SSTable ssTable = new SSTable(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
FileStruct fs = new FileStruct(SequenceFile.bufferedReader(ssTable.dataFile_, 128 * 1024), new OrderPreservingPartitioner());
for (String key : keys)
{