diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index e88b7ac7c9..9724ec827e 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -246,6 +246,7 @@ public final class ColumnFamily void clear() { + logger_.debug("clearing"); columns_.clear(); size_.set(0); } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 0be08b7b1d..e690e63d3b 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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() diff --git a/src/java/org/apache/cassandra/db/FileStruct.java b/src/java/org/apache/cassandra/db/FileStruct.java index d6639b2afd..0e0f9a77fc 100644 --- a/src/java/org/apache/cassandra/db/FileStruct.java +++ b/src/java/org/apache/cassandra/db/FileStruct.java @@ -85,26 +85,14 @@ public class FileStruct implements Comparable, Iterator { 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, Iterator 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, Iterator protected String computeNext() { - if (key.equals(SSTable.blockIndexKey_)) + if (isExhausted()) { return endOfData(); } diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index beeca0bce1..806e6d721a 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -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; diff --git a/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java b/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java index 4e9c4ba144..0f2e6086b9 100644 --- a/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java +++ b/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java @@ -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_) diff --git a/src/java/org/apache/cassandra/io/IFileReader.java b/src/java/org/apache/cassandra/io/IFileReader.java index 4e4de3a7b8..bd6475940f 100644 --- a/src/java/org/apache/cassandra/io/IFileReader.java +++ b/src/java/org/apache/cassandra/io/IFileReader.java @@ -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 columnNames, IndexHelper.TimeRange timeRange, Coordinate section) throws IOException; + public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, List columnNames, IndexHelper.TimeRange timeRange, long position) throws IOException; /** * Close the file after reading. diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index abd9520ef0..d3fa04e0df 100644 --- a/src/java/org/apache/cassandra/io/SSTable.java +++ b/src/java/org/apache/cassandra/io/SSTable.java @@ -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: - *

- * ------------------------- - * |------------------------|<-------| - * | | | BLOCK-INDEX PTR - * | | | - * |------------------------|-------- - * |------------------------|<-------| - * | | | - * | | | BLOCK-INDEX PTR - * | | | - * |------------------------|--------- - * |------------------------|<--------| - * | | | - * | | | - * | | | BLOCK-INDEX PTR - * | | | - * |------------------------| | - * |------------------------|---------- - * |------------------------|-----------------> BLOOM-FILTER - * version-info <--|----------|-------------|-------> relative offset to last block index. - *

- * 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 index = indexMetadataMap_.get(dataFile); + List index = indexMetadataMap_.get(dataFile); if (index != null) { int indexKeyCount = index.size(); @@ -154,17 +130,17 @@ public class SSTable public static List getIndexedKeys() { Set indexFiles = indexMetadataMap_.keySet(); - List KeyPositions = new ArrayList(); + List positions = new ArrayList(); for (String indexFile : indexFiles) { - KeyPositions.addAll(indexMetadataMap_.get(indexFile)); + positions.addAll(indexMetadataMap_.get(indexFile)); } List indexedKeys = new ArrayList(); - 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 blockIndex_; - /* Holds all the block indicies for this SSTable */ - private List> 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(partitioner_.getReverseDecoratedKeyComparator()); - blockIndexes_ = new ArrayList>(); - 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 KeyPositions = new ArrayList(); - 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 indexEntries = new ArrayList(); + + 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(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 block : blockIndexes_) - { - dumpBlockIndex(block); - } - } - - private void dumpBlockIndex(SortedMap 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 keys = blockIndex.keySet(); - /* Number of keys in this block */ - bufOut.writeInt(keys.size()); - for (String decoratedKey : keys) + List 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(); + 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 KeyPositions = SSTable.indexMetadataMap_.get(dataFile_); - if (KeyPositions == null) - { - KeyPositions = new ArrayList(); - 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 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 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 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 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 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 * * All keys are decorated. */ -class KeyPositionInfo implements Comparable +class KeyPosition implements Comparable { - 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 */ class IndexMap { - private final Hashtable> hashtable = new Hashtable>(); + private final Hashtable> hashtable = new Hashtable>(); private String cannonicalize(String filename) { @@ -780,12 +657,12 @@ class IndexMap } } - public List get(String filename) + public List get(String filename) { return hashtable.get(cannonicalize(filename)); } - public List put(String filename, List value) + public List put(String filename, List value) { return hashtable.put(cannonicalize(filename), value); } @@ -800,7 +677,7 @@ class IndexMap return hashtable.keySet(); } - public List remove(String filename) + public List remove(String filename) { return hashtable.remove(cannonicalize(filename)); } diff --git a/src/java/org/apache/cassandra/io/SSTableIndex.java b/src/java/org/apache/cassandra/io/SSTableIndex.java deleted file mode 100644 index cd47100139..0000000000 --- a/src/java/org/apache/cassandra/io/SSTableIndex.java +++ /dev/null @@ -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; - } - } -} diff --git a/src/java/org/apache/cassandra/io/SequenceFile.java b/src/java/org/apache/cassandra/io/SequenceFile.java index dcf3981dc0..e2c06538e7 100644 --- a/src/java/org/apache/cassandra/io/SequenceFile.java +++ b/src/java/org/apache/cassandra/io/SequenceFile.java @@ -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 columnNames, IndexHelper.TimeRange timeRange, Coordinate section) throws IOException + public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, List 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 diff --git a/src/java/org/apache/cassandra/loader/Loader.java b/src/java/org/apache/cassandra/loader/Loader.java index 8c468ad9b3..8055e22ccf 100644 --- a/src/java/org/apache/cassandra/loader/Loader.java +++ b/src/java/org/apache/cassandra/loader/Loader.java @@ -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 indexedKeys = SSTable.getIndexedKeys(); diff --git a/src/java/org/apache/cassandra/loader/PreLoad.java b/src/java/org/apache/cassandra/loader/PreLoad.java index b9f950791f..861f0b9246 100644 --- a/src/java/org/apache/cassandra/loader/PreLoad.java +++ b/src/java/org/apache/cassandra/loader/PreLoad.java @@ -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 indexedKeys = SSTable.getIndexedKeys(); diff --git a/src/java/org/apache/cassandra/tools/IndexBuilder.java b/src/java/org/apache/cassandra/tools/IndexBuilder.java deleted file mode 100644 index e9fbc0793d..0000000000 --- a/src/java/org/apache/cassandra/tools/IndexBuilder.java +++ /dev/null @@ -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 "); - 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(); - } - } - -} diff --git a/src/java/org/apache/cassandra/tools/KeyExtracter.java b/src/java/org/apache/cassandra/tools/KeyExtracter.java deleted file mode 100644 index ce2bf6eb39..0000000000 --- a/src/java/org/apache/cassandra/tools/KeyExtracter.java +++ /dev/null @@ -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 "); - 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()); - } -} diff --git a/src/java/org/apache/cassandra/utils/BloomFilter.java b/src/java/org/apache/cassandra/utils/BloomFilter.java index 44a9604595..931d56c502 100644 --- a/src/java/org/apache/cassandra/utils/BloomFilter.java +++ b/src/java/org/apache/cassandra/utils/BloomFilter.java @@ -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 diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java index d6bd2a72cf..b05fe6a466 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -115,10 +115,9 @@ public class ColumnFamilyStoreTest extends CleanupHelper store.forceBlockingFlush(); List 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); } } diff --git a/test/unit/org/apache/cassandra/db/NameSortTest.java b/test/unit/org/apache/cassandra/db/NameSortTest.java index ebe30a227e..14e658f319 100644 --- a/test/unit/org/apache/cassandra/db/NameSortTest.java +++ b/test/unit/org/apache/cassandra/db/NameSortTest.java @@ -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()); } } diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java index 2f0b3b2870..3b9bf2e4a8 100644 --- a/test/unit/org/apache/cassandra/db/TableTest.java +++ b/test/unit/org/apache/cassandra/db/TableTest.java @@ -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 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 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 diff --git a/test/unit/org/apache/cassandra/db/TimeSortTest.java b/test/unit/org/apache/cassandra/db/TimeSortTest.java index 71c9fc07bc..c96d1d515d 100644 --- a/test/unit/org/apache/cassandra/db/TimeSortTest.java +++ b/test/unit/org/apache/cassandra/db/TimeSortTest.java @@ -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(); diff --git a/test/unit/org/apache/cassandra/io/SSTableTest.java b/test/unit/org/apache/cassandra/io/SSTableTest.java index 0c13afcd34..873465085c 100644 --- a/test/unit/org/apache/cassandra/io/SSTableTest.java +++ b/test/unit/org/apache/cassandra/io/SSTableTest.java @@ -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 map) throws IOException + { List 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) {