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 2011/05/08 03:43:00 UTC
svn commit: r1100671 - in /cassandra/branches/cassandra-0.8: ./
src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/db/commitlog/
src/java/org/apache/cassandra/io/sstable/
test/unit/org/apache/cassandra/db/ test/unit/org/apache/cassandra/db...
Author: jbellis
Date: Sun May 8 01:43:00 2011
New Revision: 1100671
URL: http://svn.apache.org/viewvc?rev=1100671&view=rev
Log:
fix potential double-application of counter updates on commitlog replay
patch by slebresne and jbellis for CASSANDRA-2419
Added:
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java
Removed:
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java
Modified:
cassandra/branches/cassandra-0.8/CHANGES.txt
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/BinaryMemtable.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CompactionManager.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/IFlushable.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Memtable.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Table.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/Descriptor.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTable.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java
Modified: cassandra/branches/cassandra-0.8/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/CHANGES.txt?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.8/CHANGES.txt Sun May 8 01:43:00 2011
@@ -6,6 +6,8 @@
* snapshot_before_compaction directory name fix (CASSANDRA-2598)
* Disable compaction throttling during bootstrap (CASSANDRA-2612)
* fix CQL treatment of > and < operators in range slices (CASSANDRA-2592)
+ * fix potential double-application of counter updates on commitlog replay
+ (CASSANDRA-2419)
0.8.0-beta2
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/BinaryMemtable.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/BinaryMemtable.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/BinaryMemtable.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/BinaryMemtable.java Sun May 8 01:43:00 2011
@@ -35,6 +35,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
@@ -83,7 +84,7 @@ public class BinaryMemtable implements I
if (!isFrozen)
{
isFrozen = true;
- cfs.submitFlush(this, new CountDownLatch(1));
+ cfs.submitFlush(this, new CountDownLatch(1), null);
cfs.switchBinaryMemtable(key, buffer);
}
else
@@ -122,10 +123,10 @@ public class BinaryMemtable implements I
return keys;
}
- private SSTableReader writeSortedContents(List<DecoratedKey> sortedKeys) throws IOException
+ private SSTableReader writeSortedContents(List<DecoratedKey> sortedKeys, ReplayPosition context) throws IOException
{
logger.info("Writing " + this);
- SSTableWriter writer = cfs.createFlushWriter(sortedKeys.size(), DatabaseDescriptor.getBMTThreshold());
+ SSTableWriter writer = cfs.createFlushWriter(sortedKeys.size(), DatabaseDescriptor.getBMTThreshold(), context);
for (DecoratedKey key : sortedKeys)
{
@@ -138,7 +139,7 @@ public class BinaryMemtable implements I
return sstable;
}
- public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer)
+ public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context)
{
sorter.execute(new Runnable()
{
@@ -149,7 +150,7 @@ public class BinaryMemtable implements I
{
public void runMayThrow() throws IOException
{
- cfs.addSSTable(writeSortedContents(sortedKeys));
+ cfs.addSSTable(writeSortedContents(sortedKeys, context));
latch.countDown();
}
});
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamilyStore.java Sun May 8 01:43:00 2011
@@ -52,7 +52,7 @@ import org.apache.cassandra.config.Confi
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.columniterator.IColumnIterator;
import org.apache.cassandra.db.commitlog.CommitLog;
-import org.apache.cassandra.db.commitlog.CommitLogSegment;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
@@ -645,7 +645,7 @@ public class ColumnFamilyStore implement
assert getMemtableThreadSafe() == oldMemtable;
oldMemtable.freeze();
- final CommitLogSegment.CommitLogContext ctx = writeCommitLog ? CommitLog.instance.getContext() : null;
+ final ReplayPosition ctx = writeCommitLog ? CommitLog.instance.getContext() : null;
// submit the memtable for any indexed sub-cfses, and our own.
List<ColumnFamilyStore> icc = new ArrayList<ColumnFamilyStore>(indexedColumns.size());
@@ -657,7 +657,7 @@ public class ColumnFamilyStore implement
}
final CountDownLatch latch = new CountDownLatch(icc.size());
for (ColumnFamilyStore cfs : icc)
- submitFlush(cfs.data.switchMemtable(), latch);
+ submitFlush(cfs.data.switchMemtable(), latch, ctx);
// we marked our memtable as frozen as part of the concurrency control,
// so even if there was nothing to flush we need to switch it out
@@ -739,7 +739,7 @@ public class ColumnFamilyStore implement
if (binaryMemtable.get().isClean())
return;
- submitFlush(binaryMemtable.get(), new CountDownLatch(1));
+ submitFlush(binaryMemtable.get(), new CountDownLatch(1), null);
}
public void updateRowCache(DecoratedKey key, ColumnFamily columnFamily)
@@ -1006,10 +1006,10 @@ public class ColumnFamilyStore implement
* flushing thread finishes sorting, which will almost always be longer than any of the flushSorter threads proper
* (since, by definition, it started last).
*/
- void submitFlush(IFlushable flushable, CountDownLatch latch)
+ void submitFlush(IFlushable flushable, CountDownLatch latch, ReplayPosition context)
{
logger.info("Enqueuing flush of {}", flushable);
- flushable.flushAndSignal(latch, flushSorter, flushWriter);
+ flushable.flushAndSignal(latch, flushSorter, flushWriter, context);
}
public long getMemtableColumnsCount()
@@ -2116,14 +2116,15 @@ public class ColumnFamilyStore implement
return intern(name);
}
- public SSTableWriter createFlushWriter(long estimatedRows, long estimatedSize) throws IOException
+ public SSTableWriter createFlushWriter(long estimatedRows, long estimatedSize, ReplayPosition context) throws IOException
{
- return new SSTableWriter(getFlushPath(estimatedSize, Descriptor.CURRENT_VERSION), estimatedRows, metadata, partitioner);
+ return new SSTableWriter(getFlushPath(estimatedSize, Descriptor.CURRENT_VERSION), estimatedRows, metadata, partitioner, context);
}
- public SSTableWriter createCompactionWriter(long estimatedRows, String location) throws IOException
+ public SSTableWriter createCompactionWriter(long estimatedRows, String location, Collection<SSTableReader> sstables) throws IOException
{
- return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner);
+ ReplayPosition rp = ReplayPosition.getReplayPosition(sstables);
+ return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner, rp);
}
public Iterable<ColumnFamilyStore> concatWithIndexes()
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CompactionManager.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CompactionManager.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CompactionManager.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CompactionManager.java Sun May 8 01:43:00 2011
@@ -564,7 +564,7 @@ public class CompactionManager implement
return 0;
}
- writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation);
+ writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, sstables);
while (nni.hasNext())
{
AbstractCompactedRow row = nni.next();
@@ -652,7 +652,7 @@ public class CompactionManager implement
assert firstRowPositionFromIndex == 0 : firstRowPositionFromIndex;
}
- SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null);
+ SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null, Collections.singletonList(sstable));
executor.beginCompaction(new ScrubInfo(dataFile, sstable));
int goodRows = 0, badRows = 0, emptyRows = 0;
@@ -840,7 +840,7 @@ public class CompactionManager implement
SSTableIdentityIterator row = (SSTableIdentityIterator) scanner.next();
if (Range.isTokenInRanges(row.getKey().token, ranges))
{
- writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer);
+ writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer, Collections.singletonList(sstable));
writer.append(getCompactedRow(row, sstable.descriptor, false));
totalkeysWritten++;
}
@@ -921,13 +921,13 @@ public class CompactionManager implement
: new PrecompactedRow(CompactionController.getBasicController(forceDeserialize), Arrays.asList(row));
}
- private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer)
+ private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer, Collection<SSTableReader> sstables)
throws IOException
{
if (writer == null)
{
FileUtils.createDirectory(compactionFileLocation);
- writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation);
+ writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, sstables);
}
return writer;
}
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/IFlushable.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/IFlushable.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/IFlushable.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/IFlushable.java Sun May 8 01:43:00 2011
@@ -24,7 +24,9 @@ package org.apache.cassandra.db;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
+
public interface IFlushable
{
- public void flushAndSignal(CountDownLatch condition, ExecutorService sorter, ExecutorService writer);
+ public void flushAndSignal(CountDownLatch condition, ExecutorService sorter, ExecutorService writer, ReplayPosition context);
}
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Memtable.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Memtable.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Memtable.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Memtable.java Sun May 8 01:43:00 2011
@@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.db.columniterator.IColumnIterator;
import org.apache.cassandra.db.columniterator.SimpleAbstractColumnIterator;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.AbstractColumnIterator;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.SliceQueryFilter;
@@ -231,7 +232,7 @@ public class Memtable implements Compara
}
- private SSTableReader writeSortedContents() throws IOException
+ private SSTableReader writeSortedContents(ReplayPosition context) throws IOException
{
logger.info("Writing " + this);
@@ -242,7 +243,7 @@ public class Memtable implements Compara
+ keySize // keys in data file
+ currentThroughput.get()) // data
* 1.2); // bloom filter and row index overhead
- SSTableWriter writer = cfs.createFlushWriter(columnFamilies.size(), estimatedSize);
+ SSTableWriter writer = cfs.createFlushWriter(columnFamilies.size(), estimatedSize, context);
// (we can't clear out the map as-we-go to free up memory,
// since the memtable is being used for queries in the "pending flush" category)
@@ -255,7 +256,7 @@ public class Memtable implements Compara
return ssTable;
}
- public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer)
+ public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context)
{
writer.execute(new WrappedRunnable()
{
@@ -266,7 +267,7 @@ public class Memtable implements Compara
{
if (!cfs.isDropped())
{
- SSTableReader sstable = writeSortedContents();
+ SSTableReader sstable = writeSortedContents(context);
cfs.replaceFlushed(Memtable.this, sstable);
}
}
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Table.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Table.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Table.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/Table.java Sun May 8 01:43:00 2011
@@ -36,6 +36,7 @@ import org.apache.cassandra.config.Confi
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.LocalToken;
@@ -748,7 +749,8 @@ public class Table
}
@Override
- public String toString() {
+ public String toString()
+ {
return getClass().getSimpleName() + "(name='" + name + "')";
}
}
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLog.java Sun May 8 01:43:00 2011
@@ -27,6 +27,11 @@ import java.util.concurrent.atomic.Atomi
import java.util.zip.CRC32;
import java.util.zip.Checksum;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Ordering;
+
+import org.apache.cassandra.db.*;
+import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.net.MessagingService;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
@@ -37,10 +42,6 @@ import org.apache.cassandra.concurrent.S
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.ColumnFamily;
-import org.apache.cassandra.db.RowMutation;
-import org.apache.cassandra.db.Table;
-import org.apache.cassandra.db.UnserializableColumnFamilyException;
import org.apache.cassandra.io.DeletionService;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.FileUtils;
@@ -133,7 +134,7 @@ public class CommitLog
return false;
}
- public static void recover() throws IOException
+ public static int recover() throws IOException
{
String directory = DatabaseDescriptor.getCommitLogLocation();
File[] files = new File(directory).listFiles(new FilenameFilter()
@@ -149,49 +150,63 @@ public class CommitLog
if (files.length == 0)
{
logger.info("No commitlog files found; skipping replay");
- return;
+ return 0;
}
Arrays.sort(files, new FileUtils.FileComparator());
logger.info("Replaying " + StringUtils.join(files, ", "));
- recover(files);
+ int replayed = recover(files);
for (File f : files)
{
- FileUtils.delete(CommitLogHeader.getHeaderPathFromSegmentPath(f.getAbsolutePath())); // may not actually exist
if (!f.delete())
logger.error("Unable to remove " + f + "; you should remove it manually or next restart will replay it again (harmless, but time-consuming)");
}
- logger.info("Log replay complete");
+ logger.info("Log replay complete, " + replayed + " replayed mutations");
+ return replayed;
}
- public static void recover(File[] clogs) throws IOException
+ // returns the number of replayed mutation (useful for tests in particular)
+ public static int recover(File[] clogs) throws IOException
{
Set<Table> tablesRecovered = new HashSet<Table>();
List<Future<?>> futures = new ArrayList<Future<?>>();
byte[] bytes = new byte[4096];
Map<Integer, AtomicInteger> invalidMutations = new HashMap<Integer, AtomicInteger>();
- for (File file : clogs)
+ // count the number of replayed mutation. We don't really care about atomicity, but we need it to be a reference.
+ final AtomicInteger replayedCount = new AtomicInteger();
+
+ // compute per-CF and global replay positions
+ final Map<Integer, ReplayPosition> cfPositions = new HashMap<Integer, ReplayPosition>();
+ for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
+ // it's important to call RP.gRP per-cf, before aggregating all the positions w/ the Ordering.min call
+ // below: gRP will return NONE if there are no flushed sstables, which is important to have in the
+ // list (otherwise we'll just start replay from the first flush position that we do have, which is not correct).
+ ReplayPosition rp = ReplayPosition.getReplayPosition(cfs.getSSTables());
+ cfPositions.put(cfs.metadata.cfId, rp);
+ }
+ final ReplayPosition globalPosition = Ordering.from(ReplayPosition.comparator).min(cfPositions.values());
+
+ for (final File file : clogs)
+ {
+ final long segment = CommitLogSegment.idFromFilename(file.getName());
+
int bufferSize = (int) Math.min(Math.max(file.length(), 1), 32 * 1024 * 1024);
BufferedRandomAccessFile reader = new BufferedRandomAccessFile(new File(file.getAbsolutePath()), "r", bufferSize, true);
+ assert reader.length() <= Integer.MAX_VALUE;
try
{
- CommitLogHeader clHeader = null;
- int replayPosition = 0;
- String headerPath = CommitLogHeader.getHeaderPathFromSegmentPath(file.getAbsolutePath());
- try
- {
- clHeader = CommitLogHeader.readCommitLogHeader(headerPath);
- replayPosition = clHeader.getReplayPosition();
- }
- catch (IOException ioe)
- {
- logger.info(headerPath + " incomplete, missing or corrupt. Everything is ok, don't panic. CommitLog will be replayed from the beginning");
- logger.debug("exception was", ioe);
- }
- if (replayPosition < 0 || replayPosition > reader.length())
+ int replayPosition;
+ if (globalPosition.segment < segment)
+ replayPosition = 0;
+ else if (globalPosition.segment == segment)
+ replayPosition = globalPosition.position;
+ else
+ replayPosition = (int) reader.length();
+
+ if (replayPosition < 0 || replayPosition >= reader.length())
{
// replayPosition > reader.length() can happen if some data gets flushed before it is written to the commitlog
// (see https://issues.apache.org/jira/browse/CASSANDRA-2285)
@@ -277,7 +292,6 @@ public class CommitLog
tablesRecovered.add(table);
final Collection<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>(rm.getColumnFamilies());
final long entryLocation = reader.getFilePointer();
- final CommitLogHeader finalHeader = clHeader;
final RowMutation frm = rm;
Runnable runnable = new WrappedRunnable()
{
@@ -294,8 +308,15 @@ public class CommitLog
// null means the cf has been dropped
continue;
- if (finalHeader == null || (finalHeader.isDirty(columnFamily.id()) && entryLocation > finalHeader.getPosition(columnFamily.id())))
+ ReplayPosition rp = cfPositions.get(columnFamily.id());
+
+ // replay if current segment is newer than last flushed one or, if it is the last known
+ // segment, if we are after the replay position
+ if (segment > rp.segment || (segment == rp.segment && entryLocation > rp.position))
+ {
newRm.add(columnFamily);
+ replayedCount.incrementAndGet();
+ }
}
if (!newRm.isEmpty())
{
@@ -330,6 +351,8 @@ public class CommitLog
for (Table table : tablesRecovered)
futures.addAll(table.flush());
FBUtilities.waitOnFutures(futures);
+
+ return replayedCount.get();
}
private CommitLogSegment currentSegment()
@@ -337,11 +360,11 @@ public class CommitLog
return segments.getLast();
}
- public CommitLogSegment.CommitLogContext getContext()
+ public ReplayPosition getContext()
{
- Callable<CommitLogSegment.CommitLogContext> task = new Callable<CommitLogSegment.CommitLogContext>()
+ Callable<ReplayPosition> task = new Callable<ReplayPosition>()
{
- public CommitLogSegment.CommitLogContext call() throws Exception
+ public ReplayPosition call() throws Exception
{
return currentSegment().getContext();
}
@@ -377,7 +400,7 @@ public class CommitLog
* The bit flag associated with this column family is set in the
* header and this is used to decide if the log file can be deleted.
*/
- public void discardCompletedSegments(final Integer cfId, final CommitLogSegment.CommitLogContext context) throws IOException
+ public void discardCompletedSegments(final Integer cfId, final ReplayPosition context) throws IOException
{
Callable task = new Callable()
{
@@ -408,7 +431,7 @@ public class CommitLog
* param @ id id of the columnFamily being flushed to disk.
*
*/
- private void discardCompletedSegmentsInternal(CommitLogSegment.CommitLogContext context, Integer id) throws IOException
+ private void discardCompletedSegmentsInternal(ReplayPosition context, Integer id) throws IOException
{
if (logger.isDebugEnabled())
logger.debug("discard completed log segments for " + context + ", column family " + id + ".");
@@ -423,26 +446,20 @@ public class CommitLog
while (iter.hasNext())
{
CommitLogSegment segment = iter.next();
- CommitLogHeader header = segment.getHeader();
- if (segment.equals(context.getSegment()))
+ if (segment.id == context.segment)
{
// we can't just mark the segment where the flush happened clean,
// since there may have been writes to it between when the flush
- // started and when it finished. so mark the flush position as
- // the replay point for this CF, instead.
- if (logger.isDebugEnabled())
- logger.debug("Marking replay position " + context.position + " on commit log " + segment);
- header.turnOn(id, context.position);
- segment.writeHeader();
+ // started and when it finished.
+ segment.turnOn(id);
break;
}
- header.turnOff(id);
- if (header.isSafeToDelete() && iter.hasNext())
+ segment.turnOff(id);
+ if (segment.isSafeToDelete() && iter.hasNext())
{
logger.info("Discarding obsolete commit log:" + segment);
segment.close();
- DeletionService.executeDelete(segment.getHeaderPath());
DeletionService.executeDelete(segment.getPath());
// usually this will be the first (remaining) segment, but not always, if segment A contains
// writes to a CF that is unflushed but is followed by segment B whose CFs are all flushed.
@@ -451,8 +468,7 @@ public class CommitLog
else
{
if (logger.isDebugEnabled())
- logger.debug("Not safe to delete commit log " + segment + "; dirty is " + header.dirtyString());
- segment.writeHeader();
+ logger.debug("Not safe to delete commit log " + segment + "; dirty is " + segment.dirtyString() + "; hasNext: " + iter.hasNext());
}
}
}
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java Sun May 8 01:43:00 2011
@@ -1,198 +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.db.commitlog;
-
-import java.io.*;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.zip.CRC32;
-import java.util.zip.Checksum;
-
-import org.apache.cassandra.io.ICompactSerializer2;
-import org.apache.cassandra.io.util.FileUtils;
-
-public class CommitLogHeader
-{
- public static String getHeaderPathFromSegment(CommitLogSegment segment)
- {
- return getHeaderPathFromSegmentPath(segment.getPath());
- }
-
- public static String getHeaderPathFromSegmentPath(String segmentPath)
- {
- return segmentPath + ".header";
- }
-
- public static CommitLogHeaderSerializer serializer = new CommitLogHeaderSerializer();
-
- private Map<Integer, Integer> cfDirtiedAt; // position at which each CF was last flushed
-
- CommitLogHeader()
- {
- this(new HashMap<Integer, Integer>());
- }
-
- /*
- * This ctor is used while deserializing. This ctor
- * also builds an index of position to column family
- * Id.
- */
- private CommitLogHeader(Map<Integer, Integer> cfDirtiedAt)
- {
- this.cfDirtiedAt = cfDirtiedAt;
- }
-
- boolean isDirty(Integer cfId)
- {
- return cfDirtiedAt.containsKey(cfId);
- }
-
- int getPosition(Integer cfId)
- {
- Integer x = cfDirtiedAt.get(cfId);
- return x == null ? 0 : x;
- }
-
- void turnOn(Integer cfId, long position)
- {
- assert position >= 0 && position <= Integer.MAX_VALUE;
- cfDirtiedAt.put(cfId, (int)position);
- }
-
- void turnOff(Integer cfId)
- {
- cfDirtiedAt.remove(cfId);
- }
-
- boolean isSafeToDelete() throws IOException
- {
- return cfDirtiedAt.isEmpty();
- }
-
- // we use cf ids. getting the cf names would be pretty pretty expensive.
- public String toString()
- {
- StringBuilder sb = new StringBuilder("");
- sb.append("CLH(dirty+flushed={");
- for (Map.Entry<Integer, Integer> entry : cfDirtiedAt.entrySet())
- {
- sb.append(entry.getKey()).append(": ").append(entry.getValue()).append(", ");
- }
- sb.append("})");
- return sb.toString();
- }
-
- public String dirtyString()
- {
- StringBuilder sb = new StringBuilder();
- for (Map.Entry<Integer, Integer> entry : cfDirtiedAt.entrySet())
- sb.append(entry.getKey()).append(", ");
- return sb.toString();
- }
-
- static void writeCommitLogHeader(CommitLogHeader header, String headerFile) throws IOException
- {
- DataOutputStream out = null;
- try
- {
- /*
- * FileOutputStream doesn't sync on flush/close.
- * As headers are "optional" now there is no reason to sync it.
- * This provides nearly double the performance of BRAF, more under heavey load.
- */
- out = new DataOutputStream(new FileOutputStream(headerFile));
- serializer.serialize(header, out);
- }
- finally
- {
- if (out != null)
- out.close();
- }
- }
-
- static CommitLogHeader readCommitLogHeader(String headerFile) throws IOException
- {
- DataInputStream reader = null;
- try
- {
- reader = new DataInputStream(new BufferedInputStream(new FileInputStream(headerFile)));
- return serializer.deserialize(reader);
- }
- finally
- {
- FileUtils.closeQuietly(reader);
- }
- }
-
- int getReplayPosition()
- {
- return cfDirtiedAt.isEmpty() ? -1 : Collections.min(cfDirtiedAt.values());
- }
-
- static class CommitLogHeaderSerializer implements ICompactSerializer2<CommitLogHeader>
- {
- public void serialize(CommitLogHeader clHeader, DataOutput dos) throws IOException
- {
- Checksum checksum = new CRC32();
-
- // write the first checksum after the fixed-size part, so we won't read garbage lastFlushedAt data.
- dos.writeInt(clHeader.cfDirtiedAt.size()); // 4
- checksum.update(clHeader.cfDirtiedAt.size());
- dos.writeLong(checksum.getValue());
-
- // write the 2nd checksum after the lastflushedat map
- for (Map.Entry<Integer, Integer> entry : clHeader.cfDirtiedAt.entrySet())
- {
- dos.writeInt(entry.getKey()); // 4
- checksum.update(entry.getKey());
- dos.writeInt(entry.getValue()); // 4
- checksum.update(entry.getValue());
- }
- dos.writeLong(checksum.getValue());
- }
-
- public CommitLogHeader deserialize(DataInput dis) throws IOException
- {
- Checksum checksum = new CRC32();
-
- int lastFlushedAtSize = dis.readInt();
- checksum.update(lastFlushedAtSize);
- if (checksum.getValue() != dis.readLong())
- {
- throw new IOException("Invalid or corrupt commitlog header");
- }
- Map<Integer, Integer> lastFlushedAt = new HashMap<Integer, Integer>();
- for (int i = 0; i < lastFlushedAtSize; i++)
- {
- int key = dis.readInt();
- checksum.update(key);
- int value = dis.readInt();
- checksum.update(value);
- lastFlushedAt.put(key, value);
- }
- if (checksum.getValue() != dis.readLong())
- {
- throw new IOException("Invalid or corrupt commitlog header");
- }
-
- return new CommitLogHeader(lastFlushedAt);
- }
- }
-}
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java Sun May 8 01:43:00 2011
@@ -23,6 +23,10 @@ package org.apache.cassandra.db.commitlo
import java.io.File;
import java.io.IOError;
import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
@@ -30,30 +34,30 @@ import org.apache.cassandra.net.Messagin
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
public class CommitLogSegment
{
private static final Logger logger = LoggerFactory.getLogger(CommitLogSegment.class);
+ private static Pattern COMMIT_LOG_FILE_PATTERN = Pattern.compile("CommitLog-(\\d+).log");
+ public final long id;
private final BufferedRandomAccessFile logWriter;
- private final CommitLogHeader header;
+
+ // cache which cf is dirty in this segment to avoid having to lookup all ReplayPositions to decide if we could delete this segment
+ private Set<Integer> cfDirty = new HashSet<Integer>();
public CommitLogSegment()
{
- this.header = new CommitLogHeader();
- String logFile = DatabaseDescriptor.getCommitLogLocation() + File.separator + "CommitLog-" + System.currentTimeMillis() + ".log";
+ id = System.currentTimeMillis();
+ String logFile = DatabaseDescriptor.getCommitLogLocation() + File.separator + "CommitLog-" + id + ".log";
logger.info("Creating new commitlog segment " + logFile);
try
{
logWriter = createWriter(logFile);
-
- writeHeader();
}
catch (IOException e)
{
@@ -61,14 +65,26 @@ public class CommitLogSegment
}
}
- public static boolean possibleCommitLogFile(String filename)
+ // assume filename is a 'possibleCommitLogFile()'
+ public static long idFromFilename(String filename)
{
- return filename.matches("CommitLog-\\d+.log");
+ Matcher matcher = COMMIT_LOG_FILE_PATTERN.matcher(filename);
+ try
+ {
+ if (matcher.matches())
+ return Long.valueOf(matcher.group(1));
+ else
+ return -1L;
+ }
+ catch (NumberFormatException e)
+ {
+ return -1L;
+ }
}
- public void writeHeader() throws IOException
+ public static boolean possibleCommitLogFile(String filename)
{
- CommitLogHeader.writeCommitLogHeader(header, getHeaderPath());
+ return COMMIT_LOG_FILE_PATTERN.matcher(filename).matches();
}
private static BufferedRandomAccessFile createWriter(String file) throws IOException
@@ -76,35 +92,14 @@ public class CommitLogSegment
return new BufferedRandomAccessFile(new File(file), "rw", 128 * 1024, true);
}
- public CommitLogSegment.CommitLogContext write(RowMutation rowMutation) throws IOException
+ public ReplayPosition write(RowMutation rowMutation) throws IOException
{
long currentPosition = -1L;
try
{
currentPosition = logWriter.getFilePointer();
- CommitLogSegment.CommitLogContext cLogCtx = new CommitLogSegment.CommitLogContext(currentPosition);
-
- // update header
- for (ColumnFamily columnFamily : rowMutation.getColumnFamilies())
- {
- // we can ignore the serialized map in the header (and avoid deserializing it) since we know we are
- // writing the cfs as they exist now. check for null cfm in case a cl write goes through after the cf is
- // defined but before a new segment is created.
- CFMetaData cfm = DatabaseDescriptor.getCFMetaData(columnFamily.id());
- if (cfm == null)
- {
- logger.error("Attempted to write commit log entry for unrecognized column family: " + columnFamily.id());
- }
- else
- {
- Integer id = cfm.cfId;
- if (!header.isDirty(id))
- {
- header.turnOn(id, logWriter.getFilePointer());
- writeHeader();
- }
- }
- }
+ assert currentPosition <= Integer.MAX_VALUE;
+ ReplayPosition cLogCtx = new ReplayPosition(id, (int) currentPosition);
// write mutation, w/ checksum on the size and data
Checksum checksum = new CRC32();
@@ -131,14 +126,11 @@ public class CommitLogSegment
logWriter.sync();
}
- public CommitLogContext getContext()
- {
- return new CommitLogContext(logWriter.getFilePointer());
- }
-
- public CommitLogHeader getHeader()
+ public ReplayPosition getContext()
{
- return header;
+ long position = logWriter.getFilePointer();
+ assert position <= Integer.MAX_VALUE;
+ return new ReplayPosition(id, (int) position);
}
public String getPath()
@@ -146,9 +138,9 @@ public class CommitLogSegment
return logWriter.getPath();
}
- public String getHeaderPath()
+ public String getName()
{
- return CommitLogHeader.getHeaderPathFromSegment(this);
+ return logWriter.getPath().substring(logWriter.getPath().lastIndexOf(File.separator) + 1);
}
public long length()
@@ -175,34 +167,34 @@ public class CommitLogSegment
}
}
- @Override
- public String toString()
+ void turnOn(Integer cfId)
{
- return "CommitLogSegment(" + logWriter.getPath() + ')';
+ cfDirty.add(cfId);
}
- public class CommitLogContext
+ void turnOff(Integer cfId)
{
- public final long position;
+ cfDirty.remove(cfId);
+ }
- public CommitLogContext(long position)
- {
- assert position >= 0;
- this.position = position;
- }
+ // For debugging, not fast
+ String dirtyString()
+ {
+ StringBuilder sb = new StringBuilder();
+ for (Integer cfId : cfDirty)
+ sb.append(DatabaseDescriptor.getCFMetaData(cfId).cfName).append(" (").append(cfId).append("), ");
+ return sb.toString();
+ }
- public CommitLogSegment getSegment()
- {
- return CommitLogSegment.this;
- }
+ boolean isSafeToDelete()
+ {
+ return cfDirty.isEmpty();
+ }
- @Override
- public String toString()
- {
- return "CommitLogContext(" +
- "file='" + logWriter.getPath() + '\'' +
- ", position=" + position +
- ')';
- }
+ @Override
+ public String toString()
+ {
+ return "CommitLogSegment(" + logWriter.getPath() + ')';
}
+
}
Added: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java?rev=1100671&view=auto
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java (added)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java Sun May 8 01:43:00 2011
@@ -0,0 +1,115 @@
+package org.apache.cassandra.db.commitlog;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Comparator;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Ordering;
+
+import org.apache.cassandra.io.ICompactSerializer2;
+import org.apache.cassandra.io.sstable.SSTable;
+
+public class ReplayPosition implements Comparable<ReplayPosition>
+{
+ public static final ReplayPositionSerializer serializer = new ReplayPositionSerializer();
+
+ // NONE is used for SSTables that are streamed from other nodes and thus have no relationship
+ // with our local commitlog. The values satisfy the critera that
+ // - no real commitlog segment will have the given id
+ // - it will sort before any real replayposition, so it will be effectively ignored by getReplayPosition
+ public static final ReplayPosition NONE = new ReplayPosition(-1, 0);
+
+ /**
+ * Convenience method to compute the replay position for a group of SSTables.
+ * @param sstables
+ * @return the most recent (highest) replay position
+ */
+ public static ReplayPosition getReplayPosition(Iterable<? extends SSTable> sstables)
+ {
+ if (Iterables.isEmpty(sstables))
+ return NONE;
+
+ Function<SSTable, ReplayPosition> f = new Function<SSTable, ReplayPosition>()
+ {
+ public ReplayPosition apply(SSTable sstable)
+ {
+ return sstable.replayPosition;
+ }
+ };
+ Ordering<ReplayPosition> ordering = Ordering.from(ReplayPosition.comparator);
+ return ordering.max(Iterables.transform(sstables, f));
+ }
+
+
+ public final long segment;
+ public final int position;
+
+ public static final Comparator<ReplayPosition> comparator = new Comparator<ReplayPosition>()
+ {
+ public int compare(ReplayPosition o1, ReplayPosition o2)
+ {
+ if (o1.segment != o2.segment)
+ return new Long(o1.segment).compareTo(o2.segment);
+
+ return new Integer(o1.position).compareTo(o2.position);
+ }
+ };
+
+ public ReplayPosition(long segment, int position)
+ {
+ this.segment = segment;
+ assert position >= 0;
+ this.position = position;
+ }
+
+ public int compareTo(ReplayPosition other)
+ {
+ return comparator.compare(this, other);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ReplayPosition that = (ReplayPosition) o;
+
+ if (position != that.position) return false;
+ return segment == that.segment;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = (int) (segment ^ (segment >>> 32));
+ result = 31 * result + position;
+ return result;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "ReplayPosition(" +
+ "segmentId=" + segment +
+ ", position=" + position +
+ ')';
+ }
+
+ public static class ReplayPositionSerializer implements ICompactSerializer2<ReplayPosition>
+ {
+ public void serialize(ReplayPosition rp, DataOutput dos) throws IOException
+ {
+ dos.writeLong(rp.segment);
+ dos.writeInt(rp.position);
+ }
+
+ public ReplayPosition deserialize(DataInput dis) throws IOException
+ {
+ return new ReplayPosition(dis.readLong(), dis.readInt());
+ }
+ }
+}
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/Descriptor.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/Descriptor.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/Descriptor.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/Descriptor.java Sun May 8 01:43:00 2011
@@ -38,7 +38,7 @@ import org.apache.cassandra.utils.Pair;
public class Descriptor
{
public static final String LEGACY_VERSION = "a";
- public static final String CURRENT_VERSION = "f";
+ public static final String CURRENT_VERSION = "g";
public final File directory;
public final String version;
@@ -80,6 +80,11 @@ public class Descriptor
isLatestVersion = version.compareTo(CURRENT_VERSION) == 0;
}
+ public boolean hasReplayPosition()
+ {
+ return version.compareTo("g") >= 0;
+ }
+
public String filenameFor(Component component)
{
return filenameFor(component.name());
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTable.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTable.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTable.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTable.java Sun May 8 01:43:00 2011
@@ -31,6 +31,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.FileUtils;
@@ -67,42 +68,53 @@ public abstract class SSTable
public final CFMetaData metadata;
public final IPartitioner partitioner;
+ public final ReplayPosition replayPosition;
+
protected final EstimatedHistogram estimatedRowSize;
protected final EstimatedHistogram estimatedColumnCount;
- protected SSTable(Descriptor descriptor, CFMetaData metadata, IPartitioner partitioner)
+ protected SSTable(Descriptor descriptor, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner)
{
- this(descriptor, new HashSet<Component>(), metadata, partitioner);
+ this(descriptor, new HashSet<Component>(), metadata, replayPosition, partitioner);
}
- protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, IPartitioner partitioner)
- {
- this(descriptor, components, metadata, partitioner, defaultRowHistogram(), defaultColumnHistogram());
- }
-
- static EstimatedHistogram defaultColumnHistogram()
+ protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner)
{
- return new EstimatedHistogram(114);
+ this(descriptor, components, metadata, replayPosition, partitioner, defaultRowHistogram(), defaultColumnHistogram());
}
- static EstimatedHistogram defaultRowHistogram()
+ protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner, EstimatedHistogram rowSizes, EstimatedHistogram columnCounts)
{
- return new EstimatedHistogram(150);
- }
+ assert descriptor != null;
+ assert components != null;
+ assert metadata != null;
+ assert replayPosition != null;
+ assert partitioner != null;
+ assert rowSizes != null;
+ assert columnCounts != null;
- protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, IPartitioner partitioner, EstimatedHistogram rowSizes, EstimatedHistogram columnCounts)
- {
this.descriptor = descriptor;
Set<Component> dataComponents = new HashSet<Component>(components);
for (Component component : components)
assert component.type != Component.Type.COMPACTED_MARKER;
this.components = Collections.unmodifiableSet(dataComponents);
this.metadata = metadata;
+ this.replayPosition = replayPosition;
this.partitioner = partitioner;
estimatedRowSize = rowSizes;
estimatedColumnCount = columnCounts;
}
+ static EstimatedHistogram defaultColumnHistogram()
+ {
+ return new EstimatedHistogram(114);
+ }
+
+ static EstimatedHistogram defaultRowHistogram()
+ {
+ return new EstimatedHistogram(150);
+ }
+
public EstimatedHistogram getEstimatedRowSize()
{
return estimatedRowSize;
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableReader.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableReader.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableReader.java Sun May 8 01:43:00 2011
@@ -34,6 +34,7 @@ import org.apache.cassandra.cache.Instru
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.AbstractBounds;
@@ -156,15 +157,18 @@ public class SSTableReader extends SSTab
EstimatedHistogram rowSizes;
EstimatedHistogram columnCounts;
File statsFile = new File(descriptor.filenameFor(SSTable.COMPONENT_STATS));
+ ReplayPosition rp = ReplayPosition.NONE;
if (statsFile.exists())
{
DataInputStream dis = null;
try
{
- logger.debug("Load statistics for {}", descriptor);
+ logger.debug("Load metadata for {}", descriptor);
dis = new DataInputStream(new BufferedInputStream(new FileInputStream(statsFile)));
rowSizes = EstimatedHistogram.serializer.deserialize(dis);
columnCounts = EstimatedHistogram.serializer.deserialize(dis);
+ if (descriptor.hasReplayPosition())
+ rp = ReplayPosition.serializer.deserialize(dis);
}
finally
{
@@ -178,7 +182,7 @@ public class SSTableReader extends SSTab
columnCounts = SSTable.defaultColumnHistogram();
}
- SSTableReader sstable = new SSTableReader(descriptor, components, metadata, partitioner, null, null, null, null, System.currentTimeMillis(), rowSizes, columnCounts);
+ SSTableReader sstable = new SSTableReader(descriptor, components, metadata, rp, partitioner, null, null, null, null, System.currentTimeMillis(), rowSizes, columnCounts);
sstable.setTrackedBy(tracker);
// versions before 'c' encoded keys as utf-16 before hashing to the filter
@@ -203,16 +207,17 @@ public class SSTableReader extends SSTab
/**
* Open a RowIndexedReader which already has its state initialized (by SSTableWriter).
*/
- static SSTableReader internalOpen(Descriptor desc, Set<Component> components, CFMetaData metadata, IPartitioner partitioner, SegmentedFile ifile, SegmentedFile dfile, IndexSummary isummary, Filter bf, long maxDataAge, EstimatedHistogram rowsize,
+ static SSTableReader internalOpen(Descriptor desc, Set<Component> components, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner, SegmentedFile ifile, SegmentedFile dfile, IndexSummary isummary, Filter bf, long maxDataAge, EstimatedHistogram rowsize,
EstimatedHistogram columncount) throws IOException
{
assert desc != null && partitioner != null && ifile != null && dfile != null && isummary != null && bf != null;
- return new SSTableReader(desc, components, metadata, partitioner, ifile, dfile, isummary, bf, maxDataAge, rowsize, columncount);
+ return new SSTableReader(desc, components, metadata, replayPosition, partitioner, ifile, dfile, isummary, bf, maxDataAge, rowsize, columncount);
}
private SSTableReader(Descriptor desc,
Set<Component> components,
CFMetaData metadata,
+ ReplayPosition replayPosition,
IPartitioner partitioner,
SegmentedFile ifile,
SegmentedFile dfile,
@@ -223,7 +228,7 @@ public class SSTableReader extends SSTab
EstimatedHistogram columnCounts)
throws IOException
{
- super(desc, components, metadata, partitioner, rowSizes, columnCounts);
+ super(desc, components, metadata, replayPosition, partitioner, rowSizes, columnCounts);
this.maxDataAge = maxDataAge;
this.ifile = ifile;
Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java Sun May 8 01:43:00 2011
@@ -28,6 +28,7 @@ import java.util.Set;
import com.google.common.collect.Sets;
+import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.io.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.slf4j.Logger;
@@ -62,14 +63,15 @@ public class SSTableWriter extends SSTab
public SSTableWriter(String filename, long keyCount) throws IOException
{
- this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner());
+ this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner(), ReplayPosition.NONE);
}
- public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner) throws IOException
+ public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner, ReplayPosition replayPosition) throws IOException
{
super(Descriptor.fromFilename(filename),
new HashSet<Component>(Arrays.asList(Component.DATA, Component.FILTER, Component.PRIMARY_INDEX, Component.STATS)),
metadata,
+ replayPosition,
partitioner,
SSTable.defaultRowHistogram(),
SSTable.defaultColumnHistogram());
@@ -182,7 +184,7 @@ public class SSTableWriter extends SSTab
FileUtils.truncate(dataFile.getPath(), position);
// write sstable statistics
- writeStatistics(descriptor, estimatedRowSize, estimatedColumnCount);
+ writeMetadata(descriptor, estimatedRowSize, estimatedColumnCount, replayPosition);
// remove the 'tmp' marker from all components
final Descriptor newdesc = rename(descriptor, components);
@@ -190,20 +192,21 @@ public class SSTableWriter extends SSTab
// finalize in-memory state for the reader
SegmentedFile ifile = iwriter.builder.complete(newdesc.filenameFor(SSTable.COMPONENT_INDEX));
SegmentedFile dfile = dbuilder.complete(newdesc.filenameFor(SSTable.COMPONENT_DATA));
- SSTableReader sstable = SSTableReader.internalOpen(newdesc, components, metadata, partitioner, ifile, dfile, iwriter.summary, iwriter.bf, maxDataAge, estimatedRowSize, estimatedColumnCount);
+ SSTableReader sstable = SSTableReader.internalOpen(newdesc, components, metadata, replayPosition, partitioner, ifile, dfile, iwriter.summary, iwriter.bf, maxDataAge, estimatedRowSize, estimatedColumnCount);
iwriter = null;
dbuilder = null;
return sstable;
}
- private static void writeStatistics(Descriptor desc, EstimatedHistogram rowSizes, EstimatedHistogram columnnCounts) throws IOException
+ private static void writeMetadata(Descriptor desc, EstimatedHistogram rowSizes, EstimatedHistogram columnCounts, ReplayPosition rp) throws IOException
{
BufferedRandomAccessFile out = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_STATS)),
"rw",
BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE,
true);
EstimatedHistogram.serializer.serialize(rowSizes, out);
- EstimatedHistogram.serializer.serialize(columnnCounts, out);
+ EstimatedHistogram.serializer.serialize(columnCounts, out);
+ ReplayPosition.serializer.serialize(rp, out);
out.close();
}
@@ -457,7 +460,7 @@ public class SSTableWriter extends SSTab
rows++;
}
- writeStatistics(desc, rowSizes, columnCounts);
+ writeMetadata(desc, rowSizes, columnCounts, ReplayPosition.NONE);
return rows;
}
}
@@ -527,7 +530,7 @@ public class SSTableWriter extends SSTab
rows++;
}
- writeStatistics(desc, rowSizes, columnCounts);
+ writeMetadata(desc, rowSizes, columnCounts, ReplayPosition.NONE);
if (writerDfile.getFilePointer() != dfile.getFilePointer())
{
Modified: cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java (original)
+++ cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CommitLogTest.java Sun May 8 01:43:00 2011
@@ -27,36 +27,15 @@ import org.junit.Test;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.db.commitlog.CommitLog;
-import org.apache.cassandra.db.commitlog.CommitLogHeader;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.utils.Pair;
public class CommitLogTest extends CleanupHelper
{
@Test
- public void testRecoveryWithEmptyHeader() throws Exception
- {
- testRecovery(new byte[0], new byte[10]);
- }
-
- @Test
- public void testRecoveryWithShortHeader() throws Exception
- {
- testRecovery(new byte[2], new byte[10]);
- }
-
- @Test
- public void testRecoveryWithGarbageHeader() throws Exception
- {
- byte[] garbage = new byte[100];
- (new java.util.Random()).nextBytes(garbage);
- testRecovery(garbage, garbage);
- }
-
- @Test
public void testRecoveryWithEmptyLog() throws Exception
{
- CommitLog.recover(new File[] {tmpFiles().right});
+ CommitLog.recover(new File[] {tmpFile()});
}
@Test
@@ -69,13 +48,13 @@ public class CommitLogTest extends Clean
@Test
public void testRecoveryWithShortSize() throws Exception
{
- testRecovery(new byte[0], new byte[2]);
+ testRecovery(new byte[2]);
}
@Test
public void testRecoveryWithShortCheckSum() throws Exception
{
- testRecovery(new byte[0], new byte[6]);
+ testRecovery(new byte[6]);
}
@Test
@@ -83,7 +62,7 @@ public class CommitLogTest extends Clean
{
byte[] garbage = new byte[100];
(new java.util.Random()).nextBytes(garbage);
- testRecovery(new byte[0], garbage);
+ testRecovery(garbage);
}
@Test
@@ -108,30 +87,6 @@ public class CommitLogTest extends Clean
testRecoveryWithBadSizeArgument(-10, 10); // negative size, but no EOF
}
- @Test
- public void testRecoveryWithHeaderPositionGreaterThanLogLength() throws Exception
- {
- // Note: this can actually happen (in periodic mode) when data is flushed
- // before it had time to hit the commitlog (since the header is flushed by the system)
- // see https://issues.apache.org/jira/browse/CASSANDRA-2285
- ByteArrayOutputStream out = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(out);
- Checksum checksum = new CRC32();
-
- // write the first checksum after the fixed-size part, so we won't read garbage lastFlushedAt data.
- dos.writeInt(1);
- checksum.update(1);
- dos.writeLong(checksum.getValue());
- dos.writeInt(0);
- checksum.update(0);
- dos.writeInt(200);
- checksum.update(200);
- dos.writeLong(checksum.getValue());
- dos.close();
-
- testRecovery(out.toByteArray(), new byte[0]);
- }
-
protected void testRecoveryWithBadSizeArgument(int size, int dataSize) throws Exception
{
Checksum checksum = new CRC32();
@@ -147,29 +102,22 @@ public class CommitLogTest extends Clean
dout.writeLong(checksum);
dout.write(new byte[dataSize]);
dout.close();
- testRecovery(new byte[0], out.toByteArray());
+ testRecovery(out.toByteArray());
}
- protected Pair<File, File> tmpFiles() throws IOException
+ protected File tmpFile() throws IOException
{
File logFile = File.createTempFile("testRecoveryWithPartiallyWrittenHeaderTestFile", null);
- File headerFile = new File(CommitLogHeader.getHeaderPathFromSegmentPath(logFile.getAbsolutePath()));
logFile.deleteOnExit();
- headerFile.deleteOnExit();
assert logFile.length() == 0;
- assert headerFile.length() == 0;
- return new Pair<File, File>(headerFile, logFile);
+ return logFile;
}
- protected void testRecovery(byte[] headerData, byte[] logData) throws Exception
+ protected void testRecovery(byte[] logData) throws Exception
{
- Pair<File, File> tmpFiles = tmpFiles();
- File logFile = tmpFiles.right;
- File headerFile = tmpFiles.left;
+ File logFile = tmpFile();
OutputStream lout = new FileOutputStream(logFile);
- OutputStream hout = new FileOutputStream(headerFile);
lout.write(logData);
- hout.write(headerData);
//statics make it annoying to test things correctly
CommitLog.recover(new File[] {logFile}); //CASSANDRA-1119 / CASSANDRA-1179 throw on failure*/
}
Modified: cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java (original)
+++ cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java Sun May 8 01:43:00 2011
@@ -223,7 +223,7 @@ public class CompactionsPurgeTest extend
// Check that the second insert did went in
ColumnFamily cf = cfs.getColumnFamily(QueryFilter.getIdentityFilter(key, new QueryPath(cfName)));
- assert cf.getColumnCount() == 10;
+ assertEquals(10, cf.getColumnCount());
for (IColumn c : cf)
assert !c.isMarkedForDelete();
}
Modified: cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java?rev=1100671&r1=1100670&r2=1100671&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java (original)
+++ cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java Sun May 8 01:43:00 2011
@@ -59,17 +59,12 @@ public class RecoveryManager2Test extend
logger.debug("forcing flush");
cfs.forceBlockingFlush();
- // remove Standard1 SSTable/MemTables
- cfs.clearUnsafe();
-
logger.debug("begin manual replay");
- // replay the commit log (nothing should be replayed since everything was flushed)
+ // replay the commit log (nothing on Standard1 should be replayed since everything was flushed, so only the row on Standard2
+ // will be replayed)
CommitLog.instance.resetUnsafe();
- CommitLog.recover();
-
- // since everything that was flushed was removed (i.e. clearUnsafe)
- // and the commit shouldn't have replayed anything, there should be no data
- assert Util.getRangeSlice(cfs).isEmpty();
+ int replayed = CommitLog.recover();
+ assert replayed == 1 : "Expecting only 1 replayed mutation, got " + replayed;
}
private void insertRow(String cfname, String key) throws IOException