You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2009/04/09 23:40:50 UTC
svn commit: r763820 - in /incubator/cassandra/trunk/src/org/apache/cassandra:
db/ColumnFamilyStore.java db/FileStruct.java db/FileStructComparator.java
io/Coordinate.java io/SSTable.java
Author: jbellis
Date: Thu Apr 9 21:40:49 2009
New Revision: 763820
URL: http://svn.apache.org/viewvc?rev=763820&view=rev
Log:
Clean up FileStruct API and fix key decoration bug in anticompaction. patch by jbellis; reviewed by Jun Rao for #67
Also fix calls to isKeyInRanges in anticompaction.
Modified:
incubator/cassandra/trunk/src/org/apache/cassandra/db/ColumnFamilyStore.java
incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStruct.java
incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStructComparator.java
incubator/cassandra/trunk/src/org/apache/cassandra/io/Coordinate.java
incubator/cassandra/trunk/src/org/apache/cassandra/io/SSTable.java
Modified: incubator/cassandra/trunk/src/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/db/ColumnFamilyStore.java?rev=763820&r1=763819&r2=763820&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/db/ColumnFamilyStore.java Thu Apr 9 21:40:49 2009
@@ -46,6 +46,7 @@
import org.apache.cassandra.io.SequenceFile;
import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.IPartitioner;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.utils.LogUtil;
@@ -662,13 +663,9 @@
{
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.advance();
+ if(fs.isExhausted())
continue;
pq.add(fs);
}
@@ -677,9 +674,9 @@
ex.printStackTrace();
try
{
- if(fs != null)
+ if (fs != null)
{
- fs.reader_.close();
+ fs.close();
}
}
catch(Exception e)
@@ -880,38 +877,6 @@
}
- /*
- * 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);
@@ -1001,6 +966,7 @@
long totalkeysWritten = 0;
String rangeFileLocation;
String mergedFileName;
+ IPartitioner p = StorageService.getPartitioner();
try
{
// Calculate the expected compacted filesize
@@ -1040,11 +1006,11 @@
fs = pq.poll();
}
if (fs != null
- && (lastkey == null || lastkey.compareTo(fs.key_) == 0))
+ && (lastkey == null || lastkey.equals(fs.getKey())))
{
// 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
@@ -1059,9 +1025,9 @@
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)
{
@@ -1069,7 +1035,7 @@
merge(columnFamilies);
}
// deserialize into column families
- columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.bufIn_));
+ columnFamilies.add(ColumnFamily.serializer().deserialize(filestruct.getBufIn()));
}
catch ( Exception ex)
{
@@ -1091,17 +1057,17 @@
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;
}
}
- if ( Range.isKeyInRanges(ranges, lastkey) )
+ if ( Range.isKeyInRanges(ranges, p.undecorateKey(lastkey)) )
{
if(ssTableRange == null )
{
@@ -1125,16 +1091,16 @@
{
try
{
- filestruct = getNextKey ( filestruct );
- if(filestruct == null)
+ filestruct.advance();
+ if (filestruct.isExhausted())
{
continue;
}
/* keep on looping until we find a key in the range */
- while ( !Range.isKeyInRanges(ranges, filestruct.key_ ) )
+ while ( !Range.isKeyInRanges(ranges, p.undecorateKey(filestruct.getKey())) )
{
- filestruct = getNextKey ( filestruct );
- if(filestruct == null)
+ filestruct.advance();
+ if (filestruct.isExhausted())
{
break;
}
@@ -1146,7 +1112,7 @@
//break;
//}
}
- if ( filestruct != null)
+ if (!filestruct.isExhausted())
{
pq.add(filestruct);
}
@@ -1158,7 +1124,7 @@
// 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();
@@ -1253,11 +1219,11 @@
fs = pq.poll();
}
if (fs != null
- && (lastkey == null || lastkey.compareTo(fs.key_) == 0))
+ && (lastkey == null || lastkey.equals(fs.getKey())))
{
// 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
@@ -1272,16 +1238,16 @@
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)
{
@@ -1303,13 +1269,13 @@
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;
}
}
@@ -1321,14 +1287,14 @@
ssTable.append(lastkey, bufOut);
/* Fill the bloom filter with the key */
- doFill(compactedBloomFilter, lastkey);
+ doFill(compactedBloomFilter, lastkey);
totalkeysWritten++;
for (FileStruct filestruct : lfs)
{
try
{
- filestruct = getNextKey(filestruct);
- if(filestruct == null)
+ filestruct.advance();
+ if (filestruct.isExhausted())
{
continue;
}
@@ -1340,7 +1306,7 @@
// 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();
Modified: incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStruct.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStruct.java?rev=763820&r1=763819&r2=763820&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStruct.java (original)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStruct.java Thu Apr 9 21:40:49 2009
@@ -24,87 +24,89 @@
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<FileStruct>
{
- IFileReader reader_;
- String key_; // decorated!
- DataInputBuffer bufIn_;
- DataOutputBuffer bufOut_;
-
- public FileStruct()
- {
- }
-
- public FileStruct(String file, int bufSize) throws IOException
- {
- 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.");
+ private String key = null; // decorated!
+ 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 String getFileName()
+ {
+ 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_;
+ return key;
}
-
- public DataOutputBuffer getBuffer()
+
+ public int compareTo(FileStruct f)
{
- return bufOut_;
+ return key.compareTo(f.key);
}
-
- public long advance() throws IOException
- {
- long bytesRead = -1L;
- bufOut_.reset();
- /* advance and read the next key in the file. */
- if (reader_.isEOF())
+
+ /*
+ * 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 advance() throws IOException
+ {
+ if (exhausted)
{
- reader_.close();
- return bytesRead;
+ throw new IndexOutOfBoundsException();
}
-
- bytesRead = reader_.next(bufOut_);
- if (bytesRead == -1)
+
+ bufOut.reset();
+ if (reader.isEOF())
{
- reader_.close();
- return bytesRead;
+ reader.close();
+ exhausted = true;
+ return;
}
- bufIn_.reset(bufOut_.getData(), bufOut_.getLength());
- key_ = bufIn_.readUTF();
+ 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_) )
+ 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();
+ reader.close();
+ exhausted = true;
}
-
- return bytesRead;
}
- public int compareTo(FileStruct f)
- {
- return StorageService.getPartitioner().getDecoratedKeyComparator().compare(key_, f.key_);
- }
-
- public void close() throws IOException
- {
- bufIn_.close();
- bufOut_.close();
- reader_.close();
- }
}
Modified: incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStructComparator.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStructComparator.java?rev=763820&r1=763819&r2=763820&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStructComparator.java (original)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/db/FileStructComparator.java Thu Apr 9 21:40:49 2009
@@ -6,13 +6,6 @@
{
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
Modified: incubator/cassandra/trunk/src/org/apache/cassandra/io/Coordinate.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/io/Coordinate.java?rev=763820&r1=763819&r2=763820&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/io/Coordinate.java (original)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/io/Coordinate.java Thu Apr 9 21:40:49 2009
@@ -21,10 +21,10 @@
* 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)
{
Modified: incubator/cassandra/trunk/src/org/apache/cassandra/io/SSTable.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/io/SSTable.java?rev=763820&r1=763819&r2=763820&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/io/SSTable.java (original)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/io/SSTable.java Thu Apr 9 21:40:49 2009
@@ -644,9 +644,9 @@
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<KeyPositionInfo> indexInfo = indexMetadataMap_.get(dataFile_);
+ List<KeyPositionInfo> indexInfo = indexMetadataMap_.get(dataReader.getFileName());
int size = (indexInfo == null) ? 0 : indexInfo.size();
long start = 0L;
long end = dataReader.getEOF();