diff --git a/src/org/apache/cassandra/db/ColumnFamilyStore.java b/src/org/apache/cassandra/db/ColumnFamilyStore.java index baf7675fef..be7c245c76 100644 --- a/src/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/org/apache/cassandra/db/ColumnFamilyStore.java @@ -675,13 +675,9 @@ public class ColumnFamilyStore { try { - fs = new FileStruct(); - fs.bufIn_ = new DataInputBuffer(); - fs.bufOut_ = new DataOutputBuffer(); - fs.reader_ = SequenceFile.bufferedReader(file, bufferSize); - fs.key_ = null; - fs = getNextKey(fs); - if(fs == null) + fs = new FileStruct(SequenceFile.bufferedReader(file, bufferSize)); + fs.getNextKey(); + if(fs.isExhausted()) continue; pq.add(fs); } @@ -690,9 +686,9 @@ public class ColumnFamilyStore ex.printStackTrace(); try { - if(fs != null) + if (fs != null) { - fs.reader_.close(); + fs.close(); } } catch(Exception e) @@ -897,38 +893,6 @@ public class ColumnFamilyStore } - /* - * Read the next key from the data file , this fn will skip teh block index - * and read teh next available key into the filestruct that is passed. - * If it cannot read or a end of file is reached it will return null. - */ - FileStruct getNextKey(FileStruct filestruct) throws IOException - { - filestruct.bufOut_.reset(); - if (filestruct.reader_.isEOF()) - { - filestruct.reader_.close(); - return null; - } - - long bytesread = filestruct.reader_.next(filestruct.bufOut_); - if (bytesread == -1) - { - filestruct.reader_.close(); - return null; - } - - filestruct.bufIn_.reset(filestruct.bufOut_.getData(), filestruct.bufOut_.getLength()); - filestruct.key_ = filestruct.bufIn_.readUTF(); - /* If the key we read is the Block Index Key then we are done reading the keys so exit */ - if ( filestruct.key_.equals(SSTable.blockIndexKey_) ) - { - filestruct.reader_.close(); - return null; - } - return filestruct; - } - void forceCleanup() { MinorCompactionManager.instance().submitCleanup(ColumnFamilyStore.this); @@ -1057,11 +1021,11 @@ public class ColumnFamilyStore fs = pq.poll(); } if (fs != null - && (lastkey == null || lastkey.compareTo(fs.key_) == 0)) + && (lastkey == null || lastkey.compareTo(fs.getKey()) == 0)) { // The keys are the same so we need to add this to the // ldfs list - lastkey = fs.key_; + lastkey = fs.getKey(); lfs.add(fs); } else @@ -1076,9 +1040,9 @@ public class ColumnFamilyStore try { /* read the length although we don't need it */ - filestruct.bufIn_.readInt(); + filestruct.getBufIn().readInt(); // Skip the Index - IndexHelper.skipBloomFilterAndIndex(filestruct.bufIn_); + IndexHelper.skipBloomFilterAndIndex(filestruct.getBufIn()); // We want to add only 2 and resolve them right there in order to save on memory footprint if(columnFamilies.size() > 1) { @@ -1086,7 +1050,7 @@ public class ColumnFamilyStore merge(columnFamilies); } // deserialize into column families - columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.bufIn_)); + columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.getBufIn())); } catch ( Exception ex) { @@ -1108,13 +1072,13 @@ public class ColumnFamilyStore try { /* read the length although we don't need it */ - int size = filestruct.bufIn_.readInt(); - bufOut.write(filestruct.bufIn_, size); + int size = filestruct.getBufIn().readInt(); + bufOut.write(filestruct.getBufIn(), size); } catch ( Exception ex) { logger_.warn(LogUtil.throwableToString(ex)); - filestruct.reader_.close(); + filestruct.close(); continue; } } @@ -1142,16 +1106,16 @@ public class ColumnFamilyStore { try { - filestruct = getNextKey ( filestruct ); - if(filestruct == null) + filestruct.getNextKey(); + if (filestruct.isExhausted()) { continue; } /* keep on looping until we find a key in the range */ - while ( !Range.isKeyInRanges(filestruct.key_, ranges) ) + while ( !Range.isKeyInRanges(filestruct.getKey(), ranges) ) { - filestruct = getNextKey ( filestruct ); - if(filestruct == null) + filestruct.getNextKey(); + if (filestruct.isExhausted()) { break; } @@ -1163,7 +1127,7 @@ public class ColumnFamilyStore //break; //} } - if ( filestruct != null) + if (!filestruct.isExhausted()) { pq.add(filestruct); } @@ -1175,7 +1139,7 @@ public class ColumnFamilyStore // in any case we have read as far as possible from it // and it will be deleted after compaction. logger_.warn(LogUtil.throwableToString(ex)); - filestruct.reader_.close(); + filestruct.close(); } } lfs.clear(); @@ -1270,11 +1234,11 @@ public class ColumnFamilyStore fs = pq.poll(); } if (fs != null - && (lastkey == null || lastkey.compareTo(fs.key_) == 0)) + && (lastkey == null || lastkey.compareTo(fs.getKey()) == 0)) { // The keys are the same so we need to add this to the // ldfs list - lastkey = fs.key_; + lastkey = fs.getKey(); lfs.add(fs); } else @@ -1289,16 +1253,16 @@ public class ColumnFamilyStore try { /* read the length although we don't need it */ - filestruct.bufIn_.readInt(); + filestruct.getBufIn().readInt(); // Skip the Index - IndexHelper.skipBloomFilterAndIndex(filestruct.bufIn_); + IndexHelper.skipBloomFilterAndIndex(filestruct.getBufIn()); // We want to add only 2 and resolve them right there in order to save on memory footprint if(columnFamilies.size() > 1) { merge(columnFamilies); } // deserialize into column families - columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.bufIn_)); + columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.getBufIn())); } catch ( Exception ex) { @@ -1320,13 +1284,13 @@ public class ColumnFamilyStore try { /* read the length although we don't need it */ - int size = filestruct.bufIn_.readInt(); - bufOut.write(filestruct.bufIn_, size); + int size = filestruct.getBufIn().readInt(); + bufOut.write(filestruct.getBufIn(), size); } catch ( Exception ex) { ex.printStackTrace(); - filestruct.reader_.close(); + filestruct.close(); continue; } } @@ -1344,8 +1308,8 @@ public class ColumnFamilyStore { try { - filestruct = getNextKey(filestruct); - if(filestruct == null) + filestruct.getNextKey(); + if (filestruct.isExhausted()) { continue; } @@ -1357,7 +1321,7 @@ public class ColumnFamilyStore // Ignore the exception as it might be a corrupted file // in any case we have read as far as possible from it // and it will be deleted after compaction. - filestruct.reader_.close(); + filestruct.close(); } } lfs.clear(); diff --git a/src/org/apache/cassandra/db/FileStruct.java b/src/org/apache/cassandra/db/FileStruct.java index f231b05730..6b3b365fd6 100644 --- a/src/org/apache/cassandra/db/FileStruct.java +++ b/src/org/apache/cassandra/db/FileStruct.java @@ -19,92 +19,189 @@ package org.apache.cassandra.db; import java.io.IOException; +import java.util.Iterator; +import org.apache.cassandra.io.Coordinate; import org.apache.cassandra.io.DataInputBuffer; import org.apache.cassandra.io.DataOutputBuffer; import org.apache.cassandra.io.IFileReader; import org.apache.cassandra.io.SSTable; -import org.apache.cassandra.io.SequenceFile; -import org.apache.cassandra.service.StorageService; -public class FileStruct implements Comparable +public class FileStruct implements Comparable, Iterable { - IFileReader reader_; - String key_; - DataInputBuffer bufIn_; - DataOutputBuffer bufOut_; - public FileStruct() + private String key = null; + private boolean exhausted = false; + private IFileReader reader; + private DataInputBuffer bufIn; + private DataOutputBuffer bufOut; + + public FileStruct(IFileReader reader) { + this.reader = reader; + bufIn = new DataInputBuffer(); + bufOut = new DataOutputBuffer(); } - - public FileStruct(String file, int bufSize) throws IOException + + public String getFileName() { - bufIn_ = new DataInputBuffer(); - bufOut_ = new DataOutputBuffer(); - reader_ = SequenceFile.bufferedReader(file, bufSize); - long bytesRead = advance(); - if ( bytesRead == -1L ) - throw new IOException("Either the file is empty or EOF has been reached."); + return reader.getFileName(); } - + + public void close() throws IOException + { + reader.close(); + } + + public boolean isExhausted() + { + return exhausted; + } + + public DataInputBuffer getBufIn() + { + return bufIn; + } + public String getKey() { - return key_; - } - - public DataOutputBuffer getBuffer() - { - return bufOut_; - } - - public long advance() throws IOException - { - long bytesRead = -1L; - bufOut_.reset(); - /* advance and read the next key in the file. */ - if (reader_.isEOF()) - { - reader_.close(); - return bytesRead; - } - - bytesRead = reader_.next(bufOut_); - if (bytesRead == -1) - { - reader_.close(); - return bytesRead; - } - - 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_) ) - { - bufOut_.reset(); - bytesRead = reader_.next(bufOut_); - if (bytesRead == -1) - { - reader_.close(); - return bytesRead; - } - bufIn_.reset(bufOut_.getData(), bufOut_.getLength()); - key_ = bufIn_.readUTF(); - } - - return bytesRead; + return key; } public int compareTo(FileStruct f) { - return StorageService.getPartitioner().getDecoratedKeyComparator().compare(key_, f.key_); + return key.compareTo(f.key); } - - public void close() throws IOException + + // we don't use SequenceReader.seekTo, since that (sometimes) throws an exception + // if the key is not found. unsure if this behavior is desired. + public void seekTo(String seekKey) { - bufIn_.close(); - bufOut_.close(); - reader_.close(); + try + { + Coordinate range = SSTable.getCoordinates(seekKey, reader); + reader.seek(range.end_); + long position = reader.getPositionFromBlockIndex(seekKey); + if (position == -1) + { + reader.seek(range.start_); + } + else + { + reader.seek(position); + } + + while (!exhausted) + { + getNextKey(); + if (key.compareTo(seekKey) >= 0) + { + break; + } + } + } + catch (IOException e) + { + throw new RuntimeException("corrupt sstable", e); + } + } + + /* + * Read the next key from the data file, skipping block indexes. + * Caller must check isExhausted after each call to see if further + * reads are valid. + */ + public void getNextKey() + { + if (exhausted) + { + throw new IndexOutOfBoundsException(); + } + + try + { + bufOut.reset(); + if (reader.isEOF()) + { + reader.close(); + exhausted = true; + return; + } + + long bytesread = reader.next(bufOut); + if (bytesread == -1) + { + reader.close(); + exhausted = true; + return; + } + + 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_)) + { + bufOut.reset(); + bytesread = reader.next(bufOut); + if (bytesread == -1) + { + reader.close(); + exhausted = true; + return; + } + bufIn.reset(bufOut.getData(), bufOut.getLength()); + key = bufIn.readUTF(); + } + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + public Iterator iterator() + { + return new FileStructIterator(); + } + + private class FileStructIterator implements Iterator + { + String saved; + + public FileStructIterator() + { + if (getKey() == null && !isExhausted()) + { + forward(); + } + } + + private void forward() + { + getNextKey(); + saved = isExhausted() ? null : getKey(); + } + + public boolean hasNext() + { + return saved != null; + } + + public String next() + { + if (saved == null) + { + throw new IndexOutOfBoundsException(); + } + String key = saved; + forward(); + return key; + } + + public void remove() + { + throw new UnsupportedOperationException(); + } } } diff --git a/src/org/apache/cassandra/db/FileStructComparator.java b/src/org/apache/cassandra/db/FileStructComparator.java index 3a5580e3ed..ae3608f69c 100644 --- a/src/org/apache/cassandra/db/FileStructComparator.java +++ b/src/org/apache/cassandra/db/FileStructComparator.java @@ -6,13 +6,6 @@ class FileStructComparator implements Comparator { public int compare(FileStruct f, FileStruct f2) { - return f.reader_.getFileName().compareTo(f2.reader_.getFileName()); - } - - public boolean equals(Object o) - { - if (!(o instanceof FileStructComparator)) - return false; - return true; + return f.getFileName().compareTo(f2.getFileName()); } } \ No newline at end of file diff --git a/src/org/apache/cassandra/io/Coordinate.java b/src/org/apache/cassandra/io/Coordinate.java index 77858adff4..7a544b948d 100644 --- a/src/org/apache/cassandra/io/Coordinate.java +++ b/src/org/apache/cassandra/io/Coordinate.java @@ -21,10 +21,10 @@ package org.apache.cassandra.io; * Section of a file that needs to be scanned * is represented by this class. */ -class Coordinate +public class Coordinate { - long start_; - long end_; + public final long start_; + public final long end_; Coordinate(long start, long end) { diff --git a/src/org/apache/cassandra/io/SSTable.java b/src/org/apache/cassandra/io/SSTable.java index 914b8eff68..59d9cb766b 100644 --- a/src/org/apache/cassandra/io/SSTable.java +++ b/src/org/apache/cassandra/io/SSTable.java @@ -653,9 +653,9 @@ public class SSTable afterAppend(decoratedKey, currentPosition, value.length ); } - private Coordinate getCoordinates(String decoratedKey, IFileReader dataReader) throws IOException + public static Coordinate getCoordinates(String decoratedKey, IFileReader dataReader) throws IOException { - List indexInfo = indexMetadataMap_.get(dataFile_); + List indexInfo = indexMetadataMap_.get(dataReader.getFileName()); int size = (indexInfo == null) ? 0 : indexInfo.size(); long start = 0L; long end = dataReader.getEOF();