diff --git a/CHANGES.txt b/CHANGES.txt index ea694a56ee..e35e8ff420 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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 diff --git a/NEWS.txt b/NEWS.txt index ee82e287e1..b8016ee7f9 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -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 ===== diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java index dc3114b0cb..10baa32b88 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java @@ -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); diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java index 9ecf8c6a5d..9f6d9faf6d 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java @@ -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"; } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 044e7a0f54..136b3f42a6 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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 sstables) { data.markCompacted(sstables); diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index dba4cd75bd..38c1b7dbdd 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -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 sstables) throws IOException; + } + + private void performAllSSTableOperation(final ColumnFamilyStore cfStore, final AllSSTablesOperation operation) throws InterruptedException, ExecutionException { Callable runnable = new Callable() { public Object call() throws IOException { compactionLock.writeLock().lock(); - try + try { if (!cfStore.isValid()) return this; - Collection tocleanup = cfStore.getDataTracker().markCompacting(cfStore.getSSTables(), 1, Integer.MAX_VALUE); - if (tocleanup == null || tocleanup.isEmpty()) + Collection 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 runnable = new Callable() + performAllSSTableOperation(cfStore, new AllSSTablesOperation() { - public Object call() throws IOException + public void perform(ColumnFamilyStore store, Collection sstables) throws IOException { - // acquire the write lock to schedule all sstables - compactionLock.writeLock().lock(); - try - { - if (!cfStore.isValid()) - return this; + doScrub(store, sstables); + } + }); + } - Collection 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 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 sstables) throws IOException + { + doCleanupCompaction(store, sstables, renewer); + } + }); } public void performMaximal(final ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 0da6824253..ffdecbff95 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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)) diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index db8658ac5f..5012c09dc8 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -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. diff --git a/src/java/org/apache/cassandra/tools/NodeCmd.java b/src/java/org/apache/cassandra/tools/NodeCmd.java index 4f21959015..d646d7775c 100644 --- a/src/java/org/apache/cassandra/tools/NodeCmd.java +++ b/src/java/org/apache/cassandra/tools/NodeCmd.java @@ -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 ", "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."); } diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index bf07ee5384..8a87349d2c 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -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);