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 2013/05/23 17:08:33 UTC
[1/4] git commit: Ignore pre-truncate hints patch by Alexey Zotov and
jbellis; reviewed by vijay for CASSANDRA-4655
Updated Branches:
refs/heads/cassandra-1.2 49f220cbe -> c4c9626f8
refs/heads/trunk 0d5ed281f -> 66aadda2e
Ignore pre-truncate hints
patch by Alexey Zotov and jbellis; reviewed by vijay for CASSANDRA-4655
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b7063918
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b7063918
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b7063918
Branch: refs/heads/trunk
Commit: b70639187e1ee5e3aebb489baa07aaa2771fb4ed
Parents: e394625
Author: Jonathan Ellis <jb...@apache.org>
Authored: Thu May 23 10:04:04 2013 -0500
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Thu May 23 10:04:04 2013 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../org/apache/cassandra/db/ColumnFamilyStore.java | 7 +++
.../apache/cassandra/db/HintedHandOffManager.java | 25 ++++++++++
src/java/org/apache/cassandra/db/RowMutation.java | 9 ++++
src/java/org/apache/cassandra/db/SystemTable.java | 35 ++++++++++-----
.../cassandra/db/commitlog/CommitLogReplayer.java | 10 ++---
.../cassandra/db/compaction/CompactionManager.java | 2 +-
7 files changed, 71 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7063918/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 6f1127a..66c5f04 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.2.6
+ * Ignore pre-truncate hints (CASSANDRA-4655)
* Move System.exit on OOM into a separate thread (CASSANDRA-5273)
* Write row markers when serializing schema (CASSANDRA-5572)
* Check only SSTables for the requested range when streaming (CASSANDRA-5569)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7063918/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 055c415..429859e 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -298,6 +298,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
valid = false;
unregisterMBean();
+ SystemTable.removeTruncationRecord(metadata.cfId);
data.unreferenceSSTables();
indexManager.invalidate();
}
@@ -2077,4 +2078,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
return getDataTracker().getDroppableTombstoneRatio();
}
+
+ public long getTruncationTime()
+ {
+ Pair<ReplayPosition, Long> truncationRecord = SystemTable.getTruncationRecords().get(metadata.cfId);
+ return truncationRecord == null ? Long.MIN_VALUE : truncationRecord.right;
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7063918/src/java/org/apache/cassandra/db/HintedHandOffManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
index 53411f5..9346fb3 100644
--- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java
+++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
@@ -30,6 +30,7 @@ import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.RateLimiter;
@@ -365,6 +366,30 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
throw new AssertionError(e);
}
+ Map<UUID, Long> truncationTimesCache = new HashMap<UUID, Long>();
+ for (UUID cfId : ImmutableSet.copyOf((rm.getColumnFamilyIds())))
+ {
+ Long truncatedAt = truncationTimesCache.get(cfId);
+ if (truncatedAt == null)
+ {
+ ColumnFamilyStore cfs = Table.open(rm.getTable()).getColumnFamilyStore(cfId);
+ truncatedAt = cfs.getTruncationTime();
+ truncationTimesCache.put(cfId, truncatedAt);
+ }
+
+ if (hint.maxTimestamp() < truncatedAt)
+ {
+ logger.debug("Skipping delivery of hint for truncated columnfamily {}" + cfId);
+ rm = rm.without(cfId);
+ }
+ }
+
+ if (rm.isEmpty())
+ {
+ deleteHint(hostIdBytes, hint.name(), hint.maxTimestamp());
+ continue;
+ }
+
MessageOut<RowMutation> message = rm.createMessage();
rateLimiter.acquire(message.serializedSize(MessagingService.current_version));
Runnable callback = new Runnable()
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7063918/src/java/org/apache/cassandra/db/RowMutation.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java
index 826f3e0..b85cfcd 100644
--- a/src/java/org/apache/cassandra/db/RowMutation.java
+++ b/src/java/org/apache/cassandra/db/RowMutation.java
@@ -355,6 +355,15 @@ public class RowMutation implements IMutation
}
}
+ public RowMutation without(UUID cfId)
+ {
+ RowMutation rm = new RowMutation(table, key);
+ for (Map.Entry<UUID, ColumnFamily> entry : modifications.entrySet())
+ if (!entry.getKey().equals(cfId))
+ rm.add(entry.getValue());
+ return rm;
+ }
+
public static class RowMutationSerializer implements IVersionedSerializer<RowMutation>
{
public void serialize(RowMutation rm, DataOutput dos, int version) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7063918/src/java/org/apache/cassandra/db/SystemTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java
index 2e36aeb..327f01b 100644
--- a/src/java/org/apache/cassandra/db/SystemTable.java
+++ b/src/java/org/apache/cassandra/db/SystemTable.java
@@ -47,7 +47,10 @@ import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.Constants;
-import org.apache.cassandra.utils.*;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.CounterId;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.cql3.QueryProcessor.processInternal;
@@ -178,19 +181,30 @@ public class SystemTable
}
}
- public static void saveTruncationPosition(ColumnFamilyStore cfs, ReplayPosition position)
+ public static void saveTruncationRecord(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
{
String req = "UPDATE system.%s SET truncated_at = truncated_at + %s WHERE key = '%s'";
- processInternal(String.format(req, LOCAL_CF, positionAsMapEntry(cfs, position), LOCAL_KEY));
+ processInternal(String.format(req, LOCAL_CF, truncationAsMapEntry(cfs, truncatedAt, position), LOCAL_KEY));
forceBlockingFlush(LOCAL_CF);
}
- private static String positionAsMapEntry(ColumnFamilyStore cfs, ReplayPosition position)
+ /**
+ * This method is used to remove information about truncation time for specified column family
+ */
+ public static void removeTruncationRecord(UUID cfId)
+ {
+ String req = "DELETE truncation_time['%s'] from system.%s WHERE key = '%s'";
+ processInternal(String.format(req, cfId, LOCAL_CF, LOCAL_KEY));
+ forceBlockingFlush(LOCAL_CF);
+ }
+
+ private static String truncationAsMapEntry(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
{
DataOutputBuffer out = new DataOutputBuffer();
try
{
ReplayPosition.serializer.serialize(position, out);
+ out.writeLong(truncatedAt);
}
catch (IOException e)
{
@@ -201,7 +215,7 @@ public class SystemTable
ByteBufferUtil.bytesToHex(ByteBuffer.wrap(out.getData(), 0, out.getLength())));
}
- public static Map<UUID, ReplayPosition> getTruncationPositions()
+ public static Map<UUID, Pair<ReplayPosition, Long>> getTruncationRecords()
{
String req = "SELECT truncated_at FROM system.%s WHERE key = '%s'";
UntypedResultSet rows = processInternal(String.format(req, LOCAL_CF, LOCAL_KEY));
@@ -213,19 +227,18 @@ public class SystemTable
if (rawMap == null)
return Collections.emptyMap();
- Map<UUID, ReplayPosition> positions = new HashMap<UUID, ReplayPosition>();
+ Map<UUID, Pair<ReplayPosition, Long>> positions = new HashMap<UUID, Pair<ReplayPosition, Long>>();
for (Map.Entry<UUID, ByteBuffer> entry : rawMap.entrySet())
- {
- positions.put(entry.getKey(), positionFromBlob(entry.getValue()));
- }
+ positions.put(entry.getKey(), truncationRecordFromBlob(entry.getValue()));
return positions;
}
- private static ReplayPosition positionFromBlob(ByteBuffer bytes)
+ private static Pair<ReplayPosition, Long> truncationRecordFromBlob(ByteBuffer bytes)
{
try
{
- return ReplayPosition.serializer.deserialize(new DataInputStream(ByteBufferUtil.inputStream(bytes)));
+ DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(bytes));
+ return Pair.create(ReplayPosition.serializer.deserialize(in), in.available() > 0 ? in.readLong() : Long.MIN_VALUE);
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7063918/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
index 2728970..e1fefa1 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
@@ -40,10 +40,8 @@ import org.apache.cassandra.io.IColumnSerializer;
import org.apache.cassandra.io.util.FastByteArrayInputStream;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RandomAccessReader;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.PureJavaCrc32;
-import org.apache.cassandra.utils.WrappedRunnable;
+import org.apache.cassandra.utils.*;
+
import org.cliffc.high_scale_lib.NonBlockingHashSet;
public class CommitLogReplayer
@@ -73,7 +71,7 @@ public class CommitLogReplayer
// compute per-CF and global replay positions
cfPositions = new HashMap<UUID, ReplayPosition>();
Ordering<ReplayPosition> replayPositionOrdering = Ordering.from(ReplayPosition.comparator);
- Map<UUID, ReplayPosition> truncationPositions = SystemTable.getTruncationPositions();
+ Map<UUID,Pair<ReplayPosition,Long>> truncationPositions = SystemTable.getTruncationRecords();
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// it's important to call RP.gRP per-cf, before aggregating all the positions w/ the Ordering.min call
@@ -82,7 +80,7 @@ public class CommitLogReplayer
ReplayPosition rp = ReplayPosition.getReplayPosition(cfs.getSSTables());
// but, if we've truncted the cf in question, then we need to need to start replay after the truncation
- ReplayPosition truncatedAt = truncationPositions.get(cfs.metadata.cfId);
+ ReplayPosition truncatedAt = truncationPositions.get(cfs.metadata.cfId).left;
if (truncatedAt != null)
rp = replayPositionOrdering.max(Arrays.asList(rp, truncatedAt));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b7063918/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 7ebbc7c..c9e1b79 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -866,7 +866,7 @@ public class CompactionManager implements CompactionManagerMBean
for (SecondaryIndex index : main.indexManager.getIndexes())
index.truncate(truncatedAt);
- SystemTable.saveTruncationPosition(main, replayAfter);
+ SystemTable.saveTruncationRecord(main, truncatedAt, replayAfter);
for (RowCacheKey key : CacheService.instance.rowCache.getKeySet())
{
[4/4] git commit: Merge remote-tracking branch 'origin/trunk' into
trunk
Posted by jb...@apache.org.
Merge remote-tracking branch 'origin/trunk' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/66aadda2
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/66aadda2
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/66aadda2
Branch: refs/heads/trunk
Commit: 66aadda2efeefebabb50128091378531b7c864c7
Parents: f620b34 0d5ed28
Author: Jonathan Ellis <jb...@apache.org>
Authored: Thu May 23 10:08:25 2013 -0500
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Thu May 23 10:08:25 2013 -0500
----------------------------------------------------------------------
.../cql3/statements/ModificationStatement.java | 1 -
1 files changed, 0 insertions(+), 1 deletions(-)
----------------------------------------------------------------------
[2/4] git commit: merge from 1.2
Posted by jb...@apache.org.
merge from 1.2
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f620b348
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f620b348
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f620b348
Branch: refs/heads/trunk
Commit: f620b348a064a55749ffcbc4d25e08c3fe71f1be
Parents: 0680372 b706391
Author: Jonathan Ellis <jb...@apache.org>
Authored: Thu May 23 10:08:06 2013 -0500
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Thu May 23 10:08:06 2013 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../org/apache/cassandra/db/ColumnFamilyStore.java | 9 ++++-
.../apache/cassandra/db/HintedHandOffManager.java | 25 +++++++++++
src/java/org/apache/cassandra/db/RowMutation.java | 9 ++++
src/java/org/apache/cassandra/db/SystemTable.java | 32 ++++++++++-----
.../cassandra/db/commitlog/CommitLogReplayer.java | 10 ++---
6 files changed, 68 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f620b348/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index b765896,66c5f04..d283b72
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,58 -1,5 +1,59 @@@
+2.0
+ * Removed on-heap row cache (CASSANDRA-5348)
+ * use nanotime consistently for node-local timeouts (CASSANDRA-5581)
+ * Avoid unnecessary second pass on name-based queries (CASSANDRA-5577)
+ * Experimental triggers (CASSANDRA-1311)
+ * JEMalloc support for off-heap allocation (CASSANDRA-3997)
+ * Single-pass compaction (CASSANDRA-4180)
+ * Removed token range bisection (CASSANDRA-5518)
+ * Removed compatibility with pre-1.2.5 sstables and network messages
+ (CASSANDRA-5511)
+ * removed PBSPredictor (CASSANDRA-5455)
+ * CAS support (CASSANDRA-5062, 5441, 5443)
+ * Leveled compaction performs size-tiered compactions in L0
+ (CASSANDRA-5371, 5439)
+ * Add yaml network topology snitch for mixed ec2/other envs (CASSANDRA-5339)
+ * Log when a node is down longer than the hint window (CASSANDRA-4554)
+ * Optimize tombstone creation for ExpiringColumns (CASSANDRA-4917)
+ * Improve LeveledScanner work estimation (CASSANDRA-5250, 5407)
+ * Replace compaction lock with runWithCompactionsDisabled (CASSANDRA-3430)
+ * Change Message IDs to ints (CASSANDRA-5307)
+ * Move sstable level information into the Stats component, removing the
+ need for a separate Manifest file (CASSANDRA-4872)
+ * avoid serializing to byte[] on commitlog append (CASSANDRA-5199)
+ * make index_interval configurable per columnfamily (CASSANDRA-3961)
+ * add default_time_to_live (CASSANDRA-3974)
+ * add memtable_flush_period_in_ms (CASSANDRA-4237)
+ * replace supercolumns internally by composites (CASSANDRA-3237, 5123)
+ * upgrade thrift to 0.9.0 (CASSANDRA-3719)
+ * drop unnecessary keyspace parameter from user-defined compaction API
+ (CASSANDRA-5139)
+ * more robust solution to incomplete compactions + counters (CASSANDRA-5151)
+ * Change order of directory searching for c*.in.sh (CASSANDRA-3983)
+ * Add tool to reset SSTable compaction level for LCS (CASSANDRA-5271)
+ * Allow custom configuration loader (CASSANDRA-5045)
+ * Remove memory emergency pressure valve logic (CASSANDRA-3534)
+ * Reduce request latency with eager retry (CASSANDRA-4705)
+ * cqlsh: Remove ASSUME command (CASSANDRA-5331)
+ * Rebuild BF when loading sstables if bloom_filter_fp_chance
+ has changed since compaction (CASSANDRA-5015)
+ * remove row-level bloom filters (CASSANDRA-4885)
+ * Change Kernel Page Cache skipping into row preheating (disabled by default)
+ (CASSANDRA-4937)
+ * Improve repair by deciding on a gcBefore before sending
+ out TreeRequests (CASSANDRA-4932)
+ * Add an official way to disable compactions (CASSANDRA-5074)
+ * Reenable ALTER TABLE DROP with new semantics (CASSANDRA-3919)
+ * Add binary protocol versioning (CASSANDRA-5436)
+ * Swap THshaServer for TThreadedSelectorServer (CASSANDRA-5530)
+ * Add alias support to SELECT statement (CASSANDRA-5075)
+ * Don't create empty RowMutations in CommitLogReplayer (CASSANDRA-5541)
+ * Use range tombstones when dropping cfs/columns from schema (CASSANDRA-5579)
+ * cqlsh: drop CQL2/CQL3-beta support (CASSANDRA-5585)
+
+
1.2.6
+ * Ignore pre-truncate hints (CASSANDRA-4655)
* Move System.exit on OOM into a separate thread (CASSANDRA-5273)
* Write row markers when serializing schema (CASSANDRA-5572)
* Check only SSTables for the requested range when streaming (CASSANDRA-5569)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f620b348/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 78825cf,429859e..dcd7814
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@@ -1830,116 -1793,11 +1831,116 @@@ public class ColumnFamilyStore implemen
}
}
- long truncatedAt = System.currentTimeMillis();
- if (DatabaseDescriptor.isAutoSnapshot())
- snapshot(Table.getTimestampedSnapshotName(columnFamily));
+ Runnable truncateRunnable = new Runnable()
+ {
+ public void run()
+ {
+ logger.debug("Discarding sstable data for truncated CF + indexes");
+
+ final long truncatedAt = System.currentTimeMillis();
+ if (DatabaseDescriptor.isAutoSnapshot())
+ snapshot(Table.getTimestampedSnapshotName(name));
+
+ ReplayPosition replayAfter = discardSSTables(truncatedAt);
+
+ for (SecondaryIndex index : indexManager.getIndexes())
+ index.truncateBlocking(truncatedAt);
+
- SystemTable.saveTruncationPosition(ColumnFamilyStore.this, replayAfter);
++ SystemTable.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter);
+
+ logger.debug("cleaning out row cache");
+ for (RowCacheKey key : CacheService.instance.rowCache.getKeySet())
+ {
+ if (key.cfId == metadata.cfId)
+ CacheService.instance.rowCache.remove(key);
+ }
+ }
+ };
+
+ runWithCompactionsDisabled(Executors.callable(truncateRunnable), true);
+ logger.debug("truncate complete");
+ }
+
+ public <V> V runWithCompactionsDisabled(Callable<V> callable, boolean interruptValidation)
+ {
+ // synchronize so that concurrent invocations don't re-enable compactions partway through unexpectedly,
+ // and so we only run one major compaction at a time
+ synchronized (this)
+ {
+ logger.debug("Cancelling in-progress compactions for {}", metadata.cfName);
+
+ Iterable<ColumnFamilyStore> selfWithIndexes = concatWithIndexes();
+ for (ColumnFamilyStore cfs : selfWithIndexes)
+ cfs.getCompactionStrategy().pause();
+ try
+ {
+ // interrupt in-progress compactions
+ Function<ColumnFamilyStore, CFMetaData> f = new Function<ColumnFamilyStore, CFMetaData>()
+ {
+ public CFMetaData apply(ColumnFamilyStore cfs)
+ {
+ return cfs.metadata;
+ }
+ };
+ Iterable<CFMetaData> allMetadata = Iterables.transform(selfWithIndexes, f);
+ CompactionManager.instance.interruptCompactionFor(allMetadata, interruptValidation);
+
+ // wait for the interruption to be recognized
+ long start = System.nanoTime();
+ long delay = TimeUnit.MINUTES.toNanos(1);
+ while (System.nanoTime() - start < delay)
+ {
+ if (CompactionManager.instance.isCompacting(selfWithIndexes))
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ else
+ break;
+ }
+
+ // doublecheck that we finished, instead of timing out
+ for (ColumnFamilyStore cfs : selfWithIndexes)
+ {
+ if (!cfs.getDataTracker().getCompacting().isEmpty())
+ {
+ logger.warn("Unable to cancel in-progress compactions for {}. Probably there is an unusually large row in progress somewhere. It is also possible that buggy code left some sstables compacting after it was done with them", metadata.cfName);
+ }
+ }
+ logger.debug("Compactions successfully cancelled");
+
+ // run our task
+ try
+ {
+ return callable.call();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ finally
+ {
+ for (ColumnFamilyStore cfs : selfWithIndexes)
+ cfs.getCompactionStrategy().resume();
+ }
+ }
+ }
+
+ public Iterable<SSTableReader> markAllCompacting()
+ {
+ Callable<Iterable<SSTableReader>> callable = new Callable<Iterable<SSTableReader>>()
+ {
+ public Iterable<SSTableReader> call() throws Exception
+ {
+ assert data.getCompacting().isEmpty() : data.getCompacting();
+ Iterable<SSTableReader> sstables = Lists.newArrayList(AbstractCompactionStrategy.filterSuspectSSTables(getSSTables()));
+ if (Iterables.isEmpty(sstables))
+ return null;
+ boolean success = data.markCompacting(sstables);
+ assert success : "something marked things compacting while compactions are disabled";
+ return sstables;
+ }
+ };
- return CompactionManager.instance.submitTruncate(this, truncatedAt);
+ return runWithCompactionsDisabled(callable, false);
}
public long getBloomFilterFalsePositives()
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f620b348/src/java/org/apache/cassandra/db/HintedHandOffManager.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f620b348/src/java/org/apache/cassandra/db/RowMutation.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/RowMutation.java
index b08055f,b85cfcd..d78247b
--- a/src/java/org/apache/cassandra/db/RowMutation.java
+++ b/src/java/org/apache/cassandra/db/RowMutation.java
@@@ -238,14 -311,65 +238,23 @@@ public class RowMutation implements IMu
return buff.append("])").toString();
}
- public void addColumnOrSuperColumn(String cfName, ColumnOrSuperColumn cosc)
- {
- if (cosc.super_column != null)
- {
- for (org.apache.cassandra.thrift.Column column : cosc.super_column.columns)
- {
- add(new QueryPath(cfName, cosc.super_column.name, column.name), column.value, column.timestamp, column.ttl);
- }
- }
- else if (cosc.column != null)
- {
- add(new QueryPath(cfName, null, cosc.column.name), cosc.column.value, cosc.column.timestamp, cosc.column.ttl);
- }
- else if (cosc.counter_super_column != null)
- {
- for (org.apache.cassandra.thrift.CounterColumn column : cosc.counter_super_column.columns)
- {
- addCounter(new QueryPath(cfName, cosc.counter_super_column.name, column.name), column.value);
- }
- }
- else // cosc.counter_column != null
- {
- addCounter(new QueryPath(cfName, null, cosc.counter_column.name), cosc.counter_column.value);
- }
- }
-
- public void deleteColumnOrSuperColumn(String cfName, Deletion del)
- {
- if (del.predicate != null && del.predicate.column_names != null)
- {
- for(ByteBuffer c : del.predicate.column_names)
- {
- if (del.super_column == null && Schema.instance.getColumnFamilyType(table, cfName) == ColumnFamilyType.Super)
- delete(new QueryPath(cfName, c), del.timestamp);
- else
- delete(new QueryPath(cfName, del.super_column, c), del.timestamp);
- }
- }
- else
- {
- delete(new QueryPath(cfName, del.super_column), del.timestamp);
- }
- }
-
+ public RowMutation without(UUID cfId)
+ {
+ RowMutation rm = new RowMutation(table, key);
+ for (Map.Entry<UUID, ColumnFamily> entry : modifications.entrySet())
+ if (!entry.getKey().equals(cfId))
+ rm.add(entry.getValue());
+ return rm;
+ }
+
public static class RowMutationSerializer implements IVersionedSerializer<RowMutation>
{
- public void serialize(RowMutation rm, DataOutput dos, int version) throws IOException
+ public void serialize(RowMutation rm, DataOutput out, int version) throws IOException
{
- dos.writeUTF(rm.getTable());
- ByteBufferUtil.writeWithShortLength(rm.key(), dos);
+ if (version < MessagingService.VERSION_20)
+ out.writeUTF(rm.getTable());
+
+ ByteBufferUtil.writeWithShortLength(rm.key(), out);
/* serialize the modifications in the mutation */
int size = rm.modifications.size();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f620b348/src/java/org/apache/cassandra/db/SystemTable.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/SystemTable.java
index dd818b2,327f01b..9662661
--- a/src/java/org/apache/cassandra/db/SystemTable.java
+++ b/src/java/org/apache/cassandra/db/SystemTable.java
@@@ -45,14 -43,14 +45,14 @@@ import org.apache.cassandra.db.marshal.
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
- import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.io.sstable.SSTableReader;
+ import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.thrift.Constants;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.CounterId;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Pair;
+import org.apache.cassandra.service.paxos.Commit;
+import org.apache.cassandra.service.paxos.PaxosState;
+import org.apache.cassandra.thrift.cassandraConstants;
+import org.apache.cassandra.utils.*;
import static org.apache.cassandra.cql3.QueryProcessor.processInternal;
@@@ -126,71 -140,61 +126,81 @@@ public class SystemTabl
DatabaseDescriptor.getPartitioner().getClass().getName()));
}
- /** if system data becomes incompatible across versions of cassandra, that logic (and associated purging) is managed here */
- private static void upgradeSystemData() throws ExecutionException, InterruptedException
+ /**
+ * Write compaction log, except columfamilies under system keyspace.
+ *
+ * @param cfs
+ * @param toCompact sstables to compact
+ * @return compaction task id or null if cfs is under system keyspace
+ */
+ public static UUID startCompaction(ColumnFamilyStore cfs, Iterable<SSTableReader> toCompact)
{
- Table table = Table.open(Table.SYSTEM_KS);
- ColumnFamilyStore oldStatusCfs = table.getColumnFamilyStore(OLD_STATUS_CF);
- if (oldStatusCfs.getSSTables().size() > 0)
+ if (Table.SYSTEM_KS.equals(cfs.table.getName()))
+ return null;
+
+ UUID compactionId = UUIDGen.getTimeUUID();
+ String req = "INSERT INTO system.%s (id, keyspace_name, columnfamily_name, inputs) VALUES (%s, '%s', '%s', {%s})";
+ Iterable<Integer> generations = Iterables.transform(toCompact, new Function<SSTableReader, Integer>()
{
- SortedSet<ByteBuffer> cols = new TreeSet<ByteBuffer>(BytesType.instance);
- cols.add(ByteBufferUtil.bytes("ClusterName"));
- cols.add(ByteBufferUtil.bytes("Token"));
- QueryFilter filter = QueryFilter.getNamesFilter(decorate(ByteBufferUtil.bytes("L")), new QueryPath(OLD_STATUS_CF), cols);
- ColumnFamily oldCf = oldStatusCfs.getColumnFamily(filter);
- Iterator<IColumn> oldColumns = oldCf.columns.iterator();
-
- String clusterName = null;
- try
- {
- clusterName = ByteBufferUtil.string(oldColumns.next().value());
- }
- catch (CharacterCodingException e)
+ public Integer apply(SSTableReader sstable)
{
- throw new RuntimeException(e);
+ return sstable.descriptor.generation;
}
- // serialize the old token as a collection of (one )tokens.
- Token token = StorageService.getPartitioner().getTokenFactory().fromByteArray(oldColumns.next().value());
- String tokenBytes = tokensAsSet(Collections.singleton(token));
- // (assume that any node getting upgraded was bootstrapped, since that was stored in a separate row for no particular reason)
- String req = "INSERT INTO system.%s (key, cluster_name, tokens, bootstrapped) VALUES ('%s', '%s', %s, '%s')";
- processInternal(String.format(req, LOCAL_CF, LOCAL_KEY, clusterName, tokenBytes, BootstrapState.COMPLETED.name()));
-
- oldStatusCfs.truncate();
- }
+ });
+ processInternal(String.format(req, COMPACTION_LOG, compactionId, cfs.table.getName(), cfs.name, StringUtils.join(Sets.newHashSet(generations), ',')));
+ forceBlockingFlush(COMPACTION_LOG);
+ return compactionId;
+ }
+
+ public static void finishCompaction(UUID taskId)
+ {
+ assert taskId != null;
- ColumnFamilyStore oldHintsCfs = table.getColumnFamilyStore(OLD_HINTS_CF);
- if (oldHintsCfs.getSSTables().size() > 0)
+ String req = "DELETE FROM system.%s WHERE id = %s";
+ processInternal(String.format(req, COMPACTION_LOG, taskId));
+ forceBlockingFlush(COMPACTION_LOG);
+ }
+
+ /**
+ * @return unfinished compactions, grouped by keyspace/columnfamily pair.
+ */
+ public static SetMultimap<Pair<String, String>, Integer> getUnfinishedCompactions()
+ {
+ String req = "SELECT * FROM system.%s";
+ UntypedResultSet resultSet = processInternal(String.format(req, COMPACTION_LOG));
+
+ SetMultimap<Pair<String, String>, Integer> unfinishedCompactions = HashMultimap.create();
+ for (UntypedResultSet.Row row : resultSet)
{
- logger.info("Possible old-format hints found. Truncating");
- oldHintsCfs.truncate();
+ String keyspace = row.getString("keyspace_name");
+ String columnfamily = row.getString("columnfamily_name");
+ Set<Integer> inputs = row.getSet("inputs", Int32Type.instance);
+
+ unfinishedCompactions.putAll(Pair.create(keyspace, columnfamily), inputs);
}
+ return unfinishedCompactions;
+ }
+
+ public static void discardCompactionsInProgress()
+ {
+ ColumnFamilyStore compactionLog = Table.open(Table.SYSTEM_KS).getColumnFamilyStore(COMPACTION_LOG);
+ compactionLog.truncateBlocking();
}
- public static void saveTruncationPosition(ColumnFamilyStore cfs, ReplayPosition position)
+ public static void saveTruncationRecord(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
{
String req = "UPDATE system.%s SET truncated_at = truncated_at + %s WHERE key = '%s'";
- processInternal(String.format(req, LOCAL_CF, positionAsMapEntry(cfs, position), LOCAL_KEY));
+ processInternal(String.format(req, LOCAL_CF, truncationAsMapEntry(cfs, truncatedAt, position), LOCAL_KEY));
+ forceBlockingFlush(LOCAL_CF);
+ }
+
+ /**
+ * This method is used to remove information about truncation time for specified column family
+ */
+ public static void removeTruncationRecord(UUID cfId)
+ {
+ String req = "DELETE truncation_time['%s'] from system.%s WHERE key = '%s'";
+ processInternal(String.format(req, cfId, LOCAL_CF, LOCAL_KEY));
forceBlockingFlush(LOCAL_CF);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f620b348/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
----------------------------------------------------------------------
[3/4] git commit: Ignore pre-truncate hints patch by Alexey Zotov and
jbellis; reviewed by vijay for CASSANDRA-4655
Posted by jb...@apache.org.
Ignore pre-truncate hints
patch by Alexey Zotov and jbellis; reviewed by vijay for CASSANDRA-4655
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/c4c9626f
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/c4c9626f
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/c4c9626f
Branch: refs/heads/cassandra-1.2
Commit: c4c9626f8b3ba941aa51860a7d10a6e686362f85
Parents: 49f220c
Author: Jonathan Ellis <jb...@apache.org>
Authored: Thu May 23 10:04:04 2013 -0500
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Thu May 23 10:08:17 2013 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../org/apache/cassandra/db/ColumnFamilyStore.java | 7 +++
.../apache/cassandra/db/HintedHandOffManager.java | 25 ++++++++++
src/java/org/apache/cassandra/db/RowMutation.java | 9 ++++
src/java/org/apache/cassandra/db/SystemTable.java | 35 ++++++++++-----
.../cassandra/db/commitlog/CommitLogReplayer.java | 10 ++---
.../cassandra/db/compaction/CompactionManager.java | 2 +-
7 files changed, 71 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c4c9626f/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 6f1127a..66c5f04 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
1.2.6
+ * Ignore pre-truncate hints (CASSANDRA-4655)
* Move System.exit on OOM into a separate thread (CASSANDRA-5273)
* Write row markers when serializing schema (CASSANDRA-5572)
* Check only SSTables for the requested range when streaming (CASSANDRA-5569)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c4c9626f/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 055c415..429859e 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -298,6 +298,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
valid = false;
unregisterMBean();
+ SystemTable.removeTruncationRecord(metadata.cfId);
data.unreferenceSSTables();
indexManager.invalidate();
}
@@ -2077,4 +2078,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
return getDataTracker().getDroppableTombstoneRatio();
}
+
+ public long getTruncationTime()
+ {
+ Pair<ReplayPosition, Long> truncationRecord = SystemTable.getTruncationRecords().get(metadata.cfId);
+ return truncationRecord == null ? Long.MIN_VALUE : truncationRecord.right;
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c4c9626f/src/java/org/apache/cassandra/db/HintedHandOffManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
index 53411f5..9346fb3 100644
--- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java
+++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
@@ -30,6 +30,7 @@ import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.RateLimiter;
@@ -365,6 +366,30 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
throw new AssertionError(e);
}
+ Map<UUID, Long> truncationTimesCache = new HashMap<UUID, Long>();
+ for (UUID cfId : ImmutableSet.copyOf((rm.getColumnFamilyIds())))
+ {
+ Long truncatedAt = truncationTimesCache.get(cfId);
+ if (truncatedAt == null)
+ {
+ ColumnFamilyStore cfs = Table.open(rm.getTable()).getColumnFamilyStore(cfId);
+ truncatedAt = cfs.getTruncationTime();
+ truncationTimesCache.put(cfId, truncatedAt);
+ }
+
+ if (hint.maxTimestamp() < truncatedAt)
+ {
+ logger.debug("Skipping delivery of hint for truncated columnfamily {}" + cfId);
+ rm = rm.without(cfId);
+ }
+ }
+
+ if (rm.isEmpty())
+ {
+ deleteHint(hostIdBytes, hint.name(), hint.maxTimestamp());
+ continue;
+ }
+
MessageOut<RowMutation> message = rm.createMessage();
rateLimiter.acquire(message.serializedSize(MessagingService.current_version));
Runnable callback = new Runnable()
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c4c9626f/src/java/org/apache/cassandra/db/RowMutation.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java
index 826f3e0..b85cfcd 100644
--- a/src/java/org/apache/cassandra/db/RowMutation.java
+++ b/src/java/org/apache/cassandra/db/RowMutation.java
@@ -355,6 +355,15 @@ public class RowMutation implements IMutation
}
}
+ public RowMutation without(UUID cfId)
+ {
+ RowMutation rm = new RowMutation(table, key);
+ for (Map.Entry<UUID, ColumnFamily> entry : modifications.entrySet())
+ if (!entry.getKey().equals(cfId))
+ rm.add(entry.getValue());
+ return rm;
+ }
+
public static class RowMutationSerializer implements IVersionedSerializer<RowMutation>
{
public void serialize(RowMutation rm, DataOutput dos, int version) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c4c9626f/src/java/org/apache/cassandra/db/SystemTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java
index 2e36aeb..327f01b 100644
--- a/src/java/org/apache/cassandra/db/SystemTable.java
+++ b/src/java/org/apache/cassandra/db/SystemTable.java
@@ -47,7 +47,10 @@ import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.Constants;
-import org.apache.cassandra.utils.*;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.CounterId;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Pair;
import static org.apache.cassandra.cql3.QueryProcessor.processInternal;
@@ -178,19 +181,30 @@ public class SystemTable
}
}
- public static void saveTruncationPosition(ColumnFamilyStore cfs, ReplayPosition position)
+ public static void saveTruncationRecord(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
{
String req = "UPDATE system.%s SET truncated_at = truncated_at + %s WHERE key = '%s'";
- processInternal(String.format(req, LOCAL_CF, positionAsMapEntry(cfs, position), LOCAL_KEY));
+ processInternal(String.format(req, LOCAL_CF, truncationAsMapEntry(cfs, truncatedAt, position), LOCAL_KEY));
forceBlockingFlush(LOCAL_CF);
}
- private static String positionAsMapEntry(ColumnFamilyStore cfs, ReplayPosition position)
+ /**
+ * This method is used to remove information about truncation time for specified column family
+ */
+ public static void removeTruncationRecord(UUID cfId)
+ {
+ String req = "DELETE truncation_time['%s'] from system.%s WHERE key = '%s'";
+ processInternal(String.format(req, cfId, LOCAL_CF, LOCAL_KEY));
+ forceBlockingFlush(LOCAL_CF);
+ }
+
+ private static String truncationAsMapEntry(ColumnFamilyStore cfs, long truncatedAt, ReplayPosition position)
{
DataOutputBuffer out = new DataOutputBuffer();
try
{
ReplayPosition.serializer.serialize(position, out);
+ out.writeLong(truncatedAt);
}
catch (IOException e)
{
@@ -201,7 +215,7 @@ public class SystemTable
ByteBufferUtil.bytesToHex(ByteBuffer.wrap(out.getData(), 0, out.getLength())));
}
- public static Map<UUID, ReplayPosition> getTruncationPositions()
+ public static Map<UUID, Pair<ReplayPosition, Long>> getTruncationRecords()
{
String req = "SELECT truncated_at FROM system.%s WHERE key = '%s'";
UntypedResultSet rows = processInternal(String.format(req, LOCAL_CF, LOCAL_KEY));
@@ -213,19 +227,18 @@ public class SystemTable
if (rawMap == null)
return Collections.emptyMap();
- Map<UUID, ReplayPosition> positions = new HashMap<UUID, ReplayPosition>();
+ Map<UUID, Pair<ReplayPosition, Long>> positions = new HashMap<UUID, Pair<ReplayPosition, Long>>();
for (Map.Entry<UUID, ByteBuffer> entry : rawMap.entrySet())
- {
- positions.put(entry.getKey(), positionFromBlob(entry.getValue()));
- }
+ positions.put(entry.getKey(), truncationRecordFromBlob(entry.getValue()));
return positions;
}
- private static ReplayPosition positionFromBlob(ByteBuffer bytes)
+ private static Pair<ReplayPosition, Long> truncationRecordFromBlob(ByteBuffer bytes)
{
try
{
- return ReplayPosition.serializer.deserialize(new DataInputStream(ByteBufferUtil.inputStream(bytes)));
+ DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(bytes));
+ return Pair.create(ReplayPosition.serializer.deserialize(in), in.available() > 0 ? in.readLong() : Long.MIN_VALUE);
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c4c9626f/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
index 2728970..e1fefa1 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
@@ -40,10 +40,8 @@ import org.apache.cassandra.io.IColumnSerializer;
import org.apache.cassandra.io.util.FastByteArrayInputStream;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RandomAccessReader;
-import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.PureJavaCrc32;
-import org.apache.cassandra.utils.WrappedRunnable;
+import org.apache.cassandra.utils.*;
+
import org.cliffc.high_scale_lib.NonBlockingHashSet;
public class CommitLogReplayer
@@ -73,7 +71,7 @@ public class CommitLogReplayer
// compute per-CF and global replay positions
cfPositions = new HashMap<UUID, ReplayPosition>();
Ordering<ReplayPosition> replayPositionOrdering = Ordering.from(ReplayPosition.comparator);
- Map<UUID, ReplayPosition> truncationPositions = SystemTable.getTruncationPositions();
+ Map<UUID,Pair<ReplayPosition,Long>> truncationPositions = SystemTable.getTruncationRecords();
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// it's important to call RP.gRP per-cf, before aggregating all the positions w/ the Ordering.min call
@@ -82,7 +80,7 @@ public class CommitLogReplayer
ReplayPosition rp = ReplayPosition.getReplayPosition(cfs.getSSTables());
// but, if we've truncted the cf in question, then we need to need to start replay after the truncation
- ReplayPosition truncatedAt = truncationPositions.get(cfs.metadata.cfId);
+ ReplayPosition truncatedAt = truncationPositions.get(cfs.metadata.cfId).left;
if (truncatedAt != null)
rp = replayPositionOrdering.max(Arrays.asList(rp, truncatedAt));
http://git-wip-us.apache.org/repos/asf/cassandra/blob/c4c9626f/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 7ebbc7c..c9e1b79 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -866,7 +866,7 @@ public class CompactionManager implements CompactionManagerMBean
for (SecondaryIndex index : main.indexManager.getIndexes())
index.truncate(truncatedAt);
- SystemTable.saveTruncationPosition(main, replayAfter);
+ SystemTable.saveTruncationRecord(main, truncatedAt, replayAfter);
for (RowCacheKey key : CacheService.instance.rowCache.getKeySet())
{