natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add new upgradesstables nodetool command 

patch by slebresne; reviewed by jbellis for CASSANDRA-3406


git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-1.0@1203117 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Sylvain Lebresne 2011-11-17 08:57:38 +00:00
parent 3a2bdffe76
commit bafa3059cc
10 changed files with 87 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -2,6 +2,7 @@
* fix "liveSize" stat when sstables are removed (CASSANDRA-3496)
* add bloom filter FP rates to nodetool cfstats (CASSANDRA-3347)
* record partitioner in sstable metadata component (CASSANDRA-3407)
* add new upgradesstables nodetool command (CASSANDRA-3406)
1.0.3

12
NEWS.txt
View File

@ -9,6 +9,18 @@ upgrade, just in case you need to roll back to the previous version.
by version X, but the inverse is not necessarily the case.)
1.0.4
=====
Features
--------
- A new upgradesstables command has been added to nodetool. It is very
similar to scrub but without the ability to discard corrupted rows (and
as a consequence it does not snapshot automatically before). This new
command is to be prefered to scrub in all cases where sstables should be
rewritten to the current format for upgrade purposes.
1.0.3
=====

4
interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
View File

@ -17041,8 +17041,6 @@ public class Cassandra {
private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
try {
// it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
__isset_bit_vector = new BitSet(1);
read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
} catch (org.apache.thrift.TException te) {
throw new java.io.IOException(te);
@ -25752,6 +25750,8 @@ public class Cassandra {
private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
try {
// it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
__isset_bit_vector = new BitSet(1);
read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
} catch (org.apache.thrift.TException te) {
throw new java.io.IOException(te);

2
interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java
View File

@ -44,6 +44,6 @@ import org.slf4j.LoggerFactory;
public class Constants {
public static final String VERSION = "19.18.0";
public static final String VERSION = "19.19.0";
}

5
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -969,6 +969,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
CompactionManager.instance.performScrub(ColumnFamilyStore.this);
}
public void sstablesRewrite() throws ExecutionException, InterruptedException
{
CompactionManager.instance.performSSTableRewrite(ColumnFamilyStore.this);
}
public void markCompacted(Collection<SSTableReader> sstables)
{
data.markCompacted(sstables);

86
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -153,19 +153,24 @@ public class CompactionManager implements CompactionManagerMBean
return executor.submit(callable);
}
public void performCleanup(final ColumnFamilyStore cfStore, final NodeId.OneShotRenewer renewer) throws InterruptedException, ExecutionException
private static interface AllSSTablesOperation
{
public void perform(ColumnFamilyStore store, Collection<SSTableReader> sstables) throws IOException;
}
private void performAllSSTableOperation(final ColumnFamilyStore cfStore, final AllSSTablesOperation operation) throws InterruptedException, ExecutionException
{
Callable<Object> runnable = new Callable<Object>()
{
public Object call() throws IOException
{
compactionLock.writeLock().lock();
try
try
{
if (!cfStore.isValid())
return this;
Collection<SSTableReader> tocleanup = cfStore.getDataTracker().markCompacting(cfStore.getSSTables(), 1, Integer.MAX_VALUE);
if (tocleanup == null || tocleanup.isEmpty())
Collection<SSTableReader> sstables = cfStore.getDataTracker().markCompacting(cfStore.getSSTables(), 1, Integer.MAX_VALUE);
if (sstables == null || sstables.isEmpty())
return this;
try
{
@ -174,7 +179,7 @@ public class CompactionManager implements CompactionManagerMBean
compactionLock.writeLock().unlock();
try
{
doCleanupCompaction(cfStore, tocleanup, renewer);
operation.perform(cfStore, sstables);
}
finally
{
@ -183,7 +188,7 @@ public class CompactionManager implements CompactionManagerMBean
}
finally
{
cfStore.getDataTracker().unmarkCompacting(tocleanup);
cfStore.getDataTracker().unmarkCompacting(sstables);
}
return this;
}
@ -198,51 +203,44 @@ public class CompactionManager implements CompactionManagerMBean
executor.submit(runnable).get();
}
public void performScrub(final ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException
public void performScrub(ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException
{
Callable<Object> runnable = new Callable<Object>()
performAllSSTableOperation(cfStore, new AllSSTablesOperation()
{
public Object call() throws IOException
public void perform(ColumnFamilyStore store, Collection<SSTableReader> sstables) throws IOException
{
// acquire the write lock to schedule all sstables
compactionLock.writeLock().lock();
try
{
if (!cfStore.isValid())
return this;
doScrub(store, sstables);
}
});
}
Collection<SSTableReader> toscrub = cfStore.getDataTracker().markCompacting(cfStore.getSSTables(), 1, Integer.MAX_VALUE);
if (toscrub == null || toscrub.isEmpty())
return this;
try
{
// downgrade the lock acquisition
compactionLock.readLock().lock();
compactionLock.writeLock().unlock();
try
{
doScrub(cfStore, toscrub);
}
finally
{
compactionLock.readLock().unlock();
}
}
finally
{
cfStore.getDataTracker().unmarkCompacting(toscrub);
}
return this;
}
finally
public void performSSTableRewrite(ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException
{
performAllSSTableOperation(cfStore, new AllSSTablesOperation()
{
public void perform(ColumnFamilyStore cfs, Collection<SSTableReader> sstables) throws IOException
{
assert !cfs.isIndex();
for (final SSTableReader sstable : sstables)
{
// we probably already downgraded
if (compactionLock.writeLock().isHeldByCurrentThread())
compactionLock.writeLock().unlock();
// SSTables are marked by the caller
CompactionTask task = new CompactionTask(cfs, Collections.singletonList(sstable), Integer.MAX_VALUE);
task.isUserDefined(true);
task.execute(executor);
}
}
};
executor.submit(runnable).get();
});
}
public void performCleanup(ColumnFamilyStore cfStore, final NodeId.OneShotRenewer renewer) throws InterruptedException, ExecutionException
{
performAllSSTableOperation(cfStore, new AllSSTablesOperation()
{
public void perform(ColumnFamilyStore store, Collection<SSTableReader> sstables) throws IOException
{
doCleanupCompaction(store, sstables, renewer);
}
});
}
public void performMaximal(final ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException

6
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -1516,6 +1516,12 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
cfStore.scrub();
}
public void upgradeSSTables(String tableName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
{
for (ColumnFamilyStore cfStore : getValidColumnFamilies(tableName, columnFamilies))
cfStore.sstablesRewrite();
}
public void forceTableCompaction(String tableName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
{
for (ColumnFamilyStore cfStore : getValidColumnFamilies(tableName, columnFamilies))

6
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -203,6 +203,12 @@ public interface StorageServiceMBean
*/
public void scrub(String tableName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException;
/**
* Rewrite all sstables to the latest version.
* Unlike scrub, it doesn't skip bad rows and do not snapshot sstables first.
*/
public void upgradeSSTables(String tableName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException;
/**
* Flush all memtables for the given column families, or all columnfamilies for the given table
* if none are explicitly listed.

8
src/java/org/apache/cassandra/tools/NodeCmd.java
View File

@ -82,7 +82,7 @@ public class NodeCmd
SETCACHECAPACITY, GETCOMPACTIONTHRESHOLD, SETCOMPACTIONTHRESHOLD, NETSTATS, CFHISTOGRAMS,
COMPACTIONSTATS, DISABLEGOSSIP, ENABLEGOSSIP, INVALIDATEKEYCACHE, INVALIDATEROWCACHE,
DISABLETHRIFT, ENABLETHRIFT, STATUSTHRIFT, JOIN, SETCOMPACTIONTHROUGHPUT, GETENDPOINTS,
REFRESH, GOSSIPINFO
REFRESH, GOSSIPINFO, UPGRADESSTABLES
}
@ -125,6 +125,7 @@ public class NodeCmd
addCmdHelp(header, "cleanup [keyspace] [cfnames]", "Run cleanup on one or more column family");
addCmdHelp(header, "compact [keyspace] [cfnames]", "Force a (major) compaction on one or more column family");
addCmdHelp(header, "scrub [keyspace] [cfnames]", "Scrub (rebuild sstables for) one or more column family");
addCmdHelp(header, "upgradesstables [keyspace] [cfnames]", "Scrub (rebuild sstables for) one or more column family");
addCmdHelp(header, "invalidatekeycache [keyspace] [cfnames]", "Invalidate the key cache of one or more column family");
addCmdHelp(header, "invalidaterowcache [keyspace] [cfnames]", "Invalidate the key cache of one or more column family");
addCmdHelp(header, "getcompactionthreshold <keyspace> <cfname>", "Print min and max compaction thresholds for a given column family");
@ -672,6 +673,7 @@ public class NodeCmd
case REPAIR :
case FLUSH :
case SCRUB :
case UPGRADESSTABLES :
case INVALIDATEKEYCACHE :
case INVALIDATEROWCACHE :
optionalKSandCFs(command, cmd, arguments, probe);
@ -835,6 +837,10 @@ public class NodeCmd
try { probe.scrub(keyspace, columnFamilies); }
catch (ExecutionException ee) { err(ee, "Error occured while scrubbing keyspace " + keyspace); }
break;
case UPGRADESSTABLES :
try { probe.upgradeSSTables(keyspace, columnFamilies); }
catch (ExecutionException ee) { err(ee, "Error occured while upgrading the sstables for keyspace " + keyspace); }
break;
default:
throw new RuntimeException("Unreachable code.");
}

5
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -182,6 +182,11 @@ public class NodeProbe
ssProxy.scrub(tableName, columnFamilies);
}
public void upgradeSSTables(String tableName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
{
ssProxy.upgradeSSTables(tableName, columnFamilies);
}
public void forceTableCompaction(String tableName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException
{
ssProxy.forceTableCompaction(tableName, columnFamilies);