Fix truncate to always call flush on table

Patch by Jeremiah Jordan; reviewed by tjake for (CASSANDRA-7511)
This commit is contained in:
Jake Luciani 2014-08-01 10:30:48 -04:00
parent 1879d9928b
commit 60eab4e45e
6 changed files with 44 additions and 42 deletions

View File

@ -1,4 +1,5 @@
2.0.10
* Fix truncate to always flush (CASSANDRA-7511)
* Remove shuffle and taketoken (CASSANDRA-7601)
* Switch liveRatio-related log messages to DEBUG (CASSANDRA-7467)
* (cqlsh) Add tab-completion for CREATE/DROP USER IF [NOT] EXISTS (CASSANDRA-7611)

View File

@ -1131,6 +1131,11 @@ public class DatabaseDescriptor
return conf.auto_snapshot;
}
@VisibleForTesting
public static void setAutoSnapshot(boolean autoSnapshot) {
conf.auto_snapshot = autoSnapshot;
}
public static boolean isAutoBootstrap()
{
return conf.auto_bootstrap;

View File

@ -2002,31 +2002,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// position in the System keyspace.
logger.debug("truncating {}", name);
if (DatabaseDescriptor.isAutoSnapshot())
{
// flush the CF being truncated before forcing the new segment
forceBlockingFlush();
// flush the CF being truncated before forcing the new segment
forceBlockingFlush();
// sleep a little to make sure that our truncatedAt comes after any sstable
// that was part of the flushed we forced; otherwise on a tie, it won't get deleted.
Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);
}
// nuke the memtable data w/o writing to disk first
Keyspace.switchLock.writeLock().lock();
try
{
for (ColumnFamilyStore cfs : concatWithIndexes())
{
Memtable mt = cfs.getMemtableThreadSafe();
if (!mt.isClean())
mt.cfs.data.renewMemtable();
}
}
finally
{
Keyspace.switchLock.writeLock().unlock();
}
// sleep a little to make sure that our truncatedAt comes after any sstable
// that was part of the flushed we forced; otherwise on a tie, it won't get deleted.
Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);
Runnable truncateRunnable = new Runnable()
{

View File

@ -123,24 +123,6 @@ public class DataTracker
return toFlushMemtable;
}
/**
* Renew the current memtable without putting the old one for a flush.
* Used when we flush but a memtable is clean (in which case we must
* change it because it was frozen).
*/
public void renewMemtable()
{
Memtable newMemtable = new Memtable(cfstore, view.get().memtable);
View currentView, newView;
do
{
currentView = view.get();
newView = currentView.renewMemtable(newMemtable);
}
while (!view.compareAndSet(currentView, newView));
notifyRenewed(currentView.memtable);
}
public void replaceFlushed(Memtable memtable, SSTableReader sstable)
{
// sstable may be null if we flushed batchlog and nothing needed to be retained

View File

@ -6,6 +6,7 @@ cluster_name: Test Cluster
in_memory_compaction_limit_in_mb: 1
commitlog_sync: batch
commitlog_sync_batch_window_in_ms: 1.0
commitlog_segment_size_in_mb: 1
partitioner: org.apache.cassandra.dht.ByteOrderedPartitioner
listen_address: 127.0.0.1
storage_port: 7010

View File

@ -22,6 +22,7 @@ package org.apache.cassandra.db;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
@ -35,8 +36,10 @@ import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogDescriptor;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
@ -257,4 +260,33 @@ public class CommitLogTest extends SchemaLoader
}
}
@Test
public void testTruncateWithoutSnapshot() throws ExecutionException, InterruptedException
{
CommitLog.instance.resetUnsafe();
boolean prev = DatabaseDescriptor.isAutoSnapshot();
DatabaseDescriptor.setAutoSnapshot(false);
ColumnFamilyStore cfs1 = Keyspace.open("Keyspace1").getColumnFamilyStore("Standard1");
ColumnFamilyStore cfs2 = Keyspace.open("Keyspace1").getColumnFamilyStore("Standard2");
final RowMutation rm1 = new RowMutation("Keyspace1", bytes("k"));
rm1.add("Standard1", bytes("c1"), ByteBuffer.allocate(100), 0);
rm1.apply();
cfs1.truncateBlocking();
DatabaseDescriptor.setAutoSnapshot(prev);
final RowMutation rm2 = new RowMutation("Keyspace1", bytes("k"));
rm2.add("Standard2", bytes("c1"), ByteBuffer.allocate(DatabaseDescriptor.getCommitLogSegmentSize() / 4), 0);
for (int i = 0 ; i < 5 ; i++)
CommitLog.instance.add(rm2);
Assert.assertEquals(2, CommitLog.instance.activeSegments());
ReplayPosition position = CommitLog.instance.getContext().get();
for (Keyspace ks : Keyspace.system())
for (ColumnFamilyStore syscfs : ks.getColumnFamilyStores())
CommitLog.instance.discardCompletedSegments(syscfs.metadata.cfId, position);
CommitLog.instance.discardCompletedSegments(cfs2.metadata.cfId, position);
Assert.assertEquals(1, CommitLog.instance.activeSegments());
}
}