From cbbd401244aa5b2e2222b1c80f47c5cf0e5f2c23 Mon Sep 17 00:00:00 2001 From: Scott Carey Date: Fri, 25 Jun 2021 23:17:50 -0700 Subject: [PATCH] Allow nodetool garbagecollect to take a user defined list of SSTables patch by Scott Carey; reviewed by Stefan Miklosovic, Brandon Williams for CASSANDRA-16767 --- CHANGES.txt | 1 + .../cassandra/db/ColumnFamilyStore.java | 5 ++ .../db/compaction/CompactionManager.java | 54 ++++++++------- .../cassandra/io/sstable/Descriptor.java | 30 +++++++++ .../cassandra/service/StorageService.java | 18 +++++ .../service/StorageServiceMBean.java | 6 ++ .../org/apache/cassandra/tools/NodeProbe.java | 14 ++++ .../tools/nodetool/GarbageCollect.java | 23 ++++++- test/resources/nodetool/help/garbagecollect | 13 +++- .../cassandra/cql3/GcCompactionTest.java | 66 +++++++++++++++++++ 10 files changed, 202 insertions(+), 28 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index b1def717a3..7cfce353c7 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 7.0 + * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767) * Add a guardrail for misprepared statements (CASSANDRA-21139) Merged from 6.0: * Ensure schema created before 2.1 without tableId in folder name can be loaded in SnapshotLoader (CASSANDRA-21246) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index a60d4a01a7..aa62443a48 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1857,6 +1857,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner return CompactionManager.instance.performGarbageCollection(this, tombstoneOption, jobs); } + public CompactionManager.AllSSTableOpStatus partialGarbageCollect(TombstoneOption tombstoneOption, int jobs, Collection sstables) + { + return CompactionManager.instance.performGarbageCollection(this, tombstoneOption, jobs, sstables); + } + public void markObsolete(Collection sstables, OperationType compactionType) { assert !sstables.isEmpty(); diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index ea370b4601..65a3357636 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -47,7 +47,6 @@ import com.codahale.metrics.Meter; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.base.Predicates; -import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.Collections2; import com.google.common.collect.ConcurrentHashMultiset; import com.google.common.collect.ImmutableList; @@ -577,8 +576,10 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan if (compacting == null) return AllSSTableOpStatus.UNABLE_TO_CANCEL; - Iterable sstables = Lists.newArrayList(operation.filterSSTables(compacting)); - if (Iterables.isEmpty(sstables)) + List sstables = Lists.newArrayList(operation.filterSSTables(compacting)); + int originalCount = compacting.originals().size(); + int processedCount = sstables.size(); + if (sstables.isEmpty()) { logger.info("No sstables to {} for {}.{}", operationName, keyspace, table); return AllSSTableOpStatus.SUCCESSFUL; @@ -610,7 +611,13 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan } } FBUtilities.waitOnFutures(futures); - assert compacting.originals().isEmpty(); + // For full-set operations processedCount == originalCount, so compacting.originals() + // is empty here; for user-defined / partial operations the un-touched sstables stay + // in compacting.originals(). Either way the count must shrink by exactly what we + // split out. + assert compacting.originals().size() == originalCount - processedCount + : String.format("originals=%d, expected %d after processing %d sstables", + compacting.originals().size(), originalCount - processedCount, processedCount); logger.info("Finished {} for {}.{} successfully", operationType, keyspace, table); return AllSSTableOpStatus.SUCCESSFUL; } @@ -839,7 +846,19 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan }, jobs, OperationType.CLEANUP); } - public AllSSTableOpStatus performGarbageCollection(final ColumnFamilyStore cfStore, TombstoneOption tombstoneOption, int jobs) throws InterruptedException, ExecutionException + public AllSSTableOpStatus performGarbageCollection(ColumnFamilyStore cfStore, TombstoneOption tombstoneOption, int jobs) + { + return performGarbageCollection(cfStore, tombstoneOption, jobs, descriptor -> true); + } + + public AllSSTableOpStatus performGarbageCollection(ColumnFamilyStore cfStore, TombstoneOption tombstoneOption, int jobs, Collection dataFiles) + { + Set files = new HashSet<>(dataFiles); + return performGarbageCollection(cfStore, tombstoneOption, jobs, files::contains); + } + + @VisibleForTesting + AllSSTableOpStatus performGarbageCollection(ColumnFamilyStore cfStore, TombstoneOption tombstoneOption, int jobs, Predicate allowedFile) { assert !cfStore.isIndex(); @@ -877,12 +896,14 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan filteredSSTables.addAll(transaction.originals()); } - filteredSSTables.sort(SSTableReader.maxTimestampAscending); - return filteredSSTables; + return filteredSSTables.stream() + .filter(reader -> allowedFile.test(reader.descriptor)) + .sorted(SSTableReader.maxTimestampAscending) + .collect(Collectors.toList()); } @Override - public void execute(LifecycleTransaction txn) throws IOException + public void execute(LifecycleTransaction txn) { logger.debug("Garbage collecting {}", txn.originals()); CompactionTask task = new CompactionTask(cfStore, txn, getDefaultGcBefore(cfStore, FBUtilities.nowInSeconds())) @@ -1334,22 +1355,7 @@ public class CompactionManager implements CompactionManagerMBean, ICompactionMan @Override public void forceUserDefinedCompaction(String dataFiles) { - String[] filenames = dataFiles.split(","); - Multimap descriptors = ArrayListMultimap.create(); - - for (String filename : filenames) - { - // extract keyspace and columnfamily name from filename - Descriptor desc = Descriptor.fromFileWithComponent(new File(filename.trim()), false).left; - if (Schema.instance.getTableMetadataRef(desc) == null) - { - logger.warn("Schema does not exist for file {}. Skipping.", filename); - continue; - } - // group by keyspace/columnfamily - ColumnFamilyStore cfs = Keyspace.open(desc.ksname).getColumnFamilyStore(desc.cfname); - descriptors.put(cfs, cfs.getDirectories().find(new File(filename.trim()).name())); - } + Multimap descriptors = Descriptor.fromFilenamesGrouped(Arrays.asList(dataFiles.split(","))); List> futures = new ArrayList<>(descriptors.size()); long nowInSec = FBUtilities.nowInSeconds(); diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java index 68c85ccfa5..4def223793 100644 --- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java +++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java @@ -18,6 +18,7 @@ package org.apache.cassandra.io.sstable; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.Set; import java.util.regex.Matcher; @@ -26,19 +27,24 @@ import java.util.regex.Pattern; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Objects; import com.google.common.base.Splitter; +import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Multimap; import com.google.common.collect.Sets; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Directories; +import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.io.sstable.format.SSTableFormat; import org.apache.cassandra.io.sstable.format.Version; import org.apache.cassandra.io.sstable.metadata.IMetadataSerializer; import org.apache.cassandra.io.sstable.metadata.MetadataSerializer; import org.apache.cassandra.io.util.File; +import org.apache.cassandra.schema.Schema; import org.apache.cassandra.utils.Pair; import static com.google.common.base.Preconditions.checkNotNull; @@ -265,6 +271,30 @@ public class Descriptor return fromFileWithComponent(file).left; } + /** + * Resolve each {@code } filename in {@code filenames} to its on-disk + * {@link Descriptor} and group the results by owning {@link ColumnFamilyStore}. Filenames + * whose keyspace/table no longer exists in the schema are logged and skipped. + */ + public static Multimap fromFilenamesGrouped(Collection filenames) + { + Multimap descriptors = ArrayListMultimap.create(); + for (String filename : filenames) + { + Descriptor desc = Descriptor.fromFileWithComponent(new File(filename.trim()), false).left; + if (Schema.instance.getTableMetadataRef(desc) == null) + { + logger.warn("Schema does not exist for file {}. Skipping.", filename); + continue; + } + ColumnFamilyStore cfs = Keyspace.open(desc.ksname).getColumnFamilyStore(desc.cfname); + Descriptor onDisk = cfs.getDirectories().find(new File(filename.trim()).name()); + if (onDisk != null) + descriptors.put(cfs, onDisk); + } + return descriptors; + } + public static Component componentFromFile(File file) { String name = file.name(); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index b7460b1be5..09527b5d23 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -134,6 +134,7 @@ import org.apache.cassandra.gms.VersionedValue.VersionedValueFactory; import org.apache.cassandra.hints.Hint; import org.apache.cassandra.hints.HintsService; import org.apache.cassandra.index.IndexStatusManager; +import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.IScrubber; import org.apache.cassandra.io.sstable.IVerifier; import org.apache.cassandra.io.sstable.SSTableLoader; @@ -2838,6 +2839,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return status.statusCode; } + public int userDefinedGarbageCollect(String tombstoneOptionString, int jobs, List userDefinedTables) throws ExecutionException, InterruptedException + { + TombstoneOption tombstoneOption = TombstoneOption.valueOf(tombstoneOptionString); + CompactionManager.AllSSTableOpStatus status = CompactionManager.AllSSTableOpStatus.SUCCESSFUL; + logger.info("Starting {} on {}", OperationType.GARBAGE_COLLECT, userDefinedTables); + for (Map.Entry> entry : Descriptor.fromFilenamesGrouped(userDefinedTables).asMap().entrySet()) + { + ColumnFamilyStore cfs = entry.getKey(); + Collection sstables = entry.getValue(); + CompactionManager.AllSSTableOpStatus oneStatus = cfs.partialGarbageCollect(tombstoneOption, jobs, sstables); + if (oneStatus != CompactionManager.AllSSTableOpStatus.SUCCESSFUL) + status = oneStatus; + } + logger.info("Completed {} with status {}", OperationType.GARBAGE_COLLECT, status); + return status.statusCode; + } + @Override public void forceKeyspaceCompactionForTokenRange(String keyspaceName, String startToken, String endToken, String... tableNames) throws IOException, ExecutionException, InterruptedException { diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index b9f47b3b2d..e6f7a18c8e 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -487,6 +487,12 @@ public interface StorageServiceMBean extends NotificationEmitter */ public int garbageCollect(String tombstoneOption, int jobs, String keyspaceName, String... tableNames) throws IOException, ExecutionException, InterruptedException; + /** + * Rewrites all sstables from the given list of SSTable Data.db files. + * The tombstone option defines the granularity of the procedure: ROW removes deleted partitions and rows, CELL also removes overwritten or deleted cells. + */ + public int userDefinedGarbageCollect(String tombstoneOption, int jobs, List userDefinedTables) throws IOException, ExecutionException, InterruptedException; + /** * Flush all memtables for the given column families, or all columnfamilies for the given keyspace * if none are explicitly listed. diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 567df09283..5e9fee058d 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -418,6 +418,11 @@ public class NodeProbe implements AutoCloseable return ssProxy.recompressSSTables(keyspaceName, jobs, tableNames); } + public int userDefinedGarbageCollect(String tombstoneOption, int jobs, List userDefinedTables) throws IOException, ExecutionException, InterruptedException + { + return ssProxy.userDefinedGarbageCollect(tombstoneOption, jobs, userDefinedTables); + } + private void checkJobs(PrintStream out, int jobs) { int compactors = ssProxy.getConcurrentCompactors(); @@ -492,6 +497,15 @@ public class NodeProbe implements AutoCloseable } } + public void userDefinedGarbageCollect(PrintStream out, String tombstoneOption, int jobs, List userDefinedTables) throws IOException, ExecutionException, InterruptedException + { + if (userDefinedGarbageCollect(tombstoneOption, jobs, userDefinedTables) != 0) + { + out.println("Aborted garbage collection for at least one table, check server logs for more information."); + throw new RuntimeException("Aborted garbage collection for at least one table, check server logs for more information."); + } + } + public void forceUserDefinedCompaction(String datafiles) throws IOException, ExecutionException, InterruptedException { compactionProxy.forceUserDefinedCompaction(datafiles); diff --git a/src/java/org/apache/cassandra/tools/nodetool/GarbageCollect.java b/src/java/org/apache/cassandra/tools/nodetool/GarbageCollect.java index 5d285bb7bf..a4c13ee5aa 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/GarbageCollect.java +++ b/src/java/org/apache/cassandra/tools/nodetool/GarbageCollect.java @@ -32,10 +32,13 @@ import static org.apache.cassandra.tools.nodetool.CommandUtils.concatArgs; import static org.apache.cassandra.tools.nodetool.CommandUtils.parseOptionalKeyspace; import static org.apache.cassandra.tools.nodetool.CommandUtils.parseOptionalTables; + @Command(name = "garbagecollect", description = "Remove deleted data from one or more tables") public class GarbageCollect extends AbstractCommand { - @CassandraUsage(usage = "[ ...]", description = "The keyspace followed by one or many tables") + @CassandraUsage(usage = "[ ...] or ...", + description = "The keyspace followed by one or many tables, " + + "or a list of SSTable data files when using --user-defined") private List args = new ArrayList<>(); @Parameters(index = "0", description = "The keyspace followed by one or many tables to garbage collect", arity = "0..1") @@ -56,11 +59,29 @@ public class GarbageCollect extends AbstractCommand "and also remove tombstones.") private int jobs = 1; + @Option(names = { "--user-defined" }, + description = "Submit the listed Data.db files for user-defined garbagecollect instead of " + + "interpreting the positional arguments as keyspace + tables.") + private boolean userDefined = false; + @Override public void execute(NodeProbe probe) { args = concatArgs(keyspace, tables); + if (userDefined) + { + try + { + probe.userDefinedGarbageCollect(probe.output().out, tombstoneOption.toString(), jobs, args); + } + catch (Exception e) + { + throw new RuntimeException("Error occurred during user defined garbage collection", e); + } + return; + } + List keyspaces = parseOptionalKeyspace(args, probe); String[] tableNames = parseOptionalTables(args); diff --git a/test/resources/nodetool/help/garbagecollect b/test/resources/nodetool/help/garbagecollect index 3004ff777a..872f9c5f1a 100644 --- a/test/resources/nodetool/help/garbagecollect +++ b/test/resources/nodetool/help/garbagecollect @@ -7,7 +7,8 @@ SYNOPSIS [(-pwf | --password-file )] [(-u | --username )] garbagecollect [(-g | --granularity )] - [(-j | --jobs )] [--] [ ...] + [(-j | --jobs )] [--user-defined] [--] [ + ...] or ... OPTIONS -g , --granularity @@ -34,10 +35,16 @@ OPTIONS -u , --username Remote jmx agent username + --user-defined + Submit the listed Data.db files for user-defined garbagecollect + instead of interpreting the positional arguments as keyspace + + tables. + -- This option can be used to separate command-line options from the list of argument, (useful when arguments might be mistaken for command-line options - [ ...] - The keyspace followed by one or many tables + [ ...] or ... + The keyspace followed by one or many tables, or a list of SSTable + data files when using --user-defined diff --git a/test/unit/org/apache/cassandra/cql3/GcCompactionTest.java b/test/unit/org/apache/cassandra/cql3/GcCompactionTest.java index 67795ccb5e..d62934b7bd 100644 --- a/test/unit/org/apache/cassandra/cql3/GcCompactionTest.java +++ b/test/unit/org/apache/cassandra/cql3/GcCompactionTest.java @@ -18,6 +18,7 @@ package org.apache.cassandra.cql3; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; @@ -35,6 +36,7 @@ import org.apache.cassandra.db.rows.ComplexColumnData; import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.db.rows.Unfiltered; import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.ISSTableScanner; import org.apache.cassandra.io.sstable.SSTableId; import org.apache.cassandra.io.sstable.SSTableIdFactory; @@ -258,6 +260,70 @@ public class GcCompactionTest extends CQLTester assertEquals(1, collected2.getSSTableLevel()); // garbagecollect should leave the LCS level where it was } + @Test + public void testGarbageCollectPartial() + { + createTable("CREATE TABLE %s(" + + " key int," + + " column int," + + " data int," + + " PRIMARY KEY ((key), column)" + + ");"); + + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + cfs.disableAutoCompaction(); + + execute("INSERT INTO %s (key, column, data) VALUES (1, 1, 1)"); + flush(); + execute("INSERT INTO %s (key, column, data) VALUES (1, 2, 1)"); + flush(); + execute("INSERT INTO %s (key, column, data) VALUES (2, 1, 2)"); + flush(); + execute("INSERT INTO %s (key, column, data) VALUES (2, 2, 2)"); + flush(); + execute("DELETE FROM %s where key = 1 and column = 1"); // removes (1, 1, 1) + flush(); + execute("DELETE FROM %s where key = 2 and column = 2"); // removes (2, 2, 2) + flush(); + + assertEquals(6, cfs.getLiveSSTables().size()); + + // Sort live sstables by their SSTableId (oldest first) and pick the first three for + // user-defined garbage collection. SSTableIdFactory.COMPARATOR works for both + // sequence-based and UUID-based ids. + ArrayList sorted = new ArrayList<>(cfs.getLiveSSTables()); + sorted.sort((a, b) -> SSTableIdFactory.COMPARATOR.compare(a.descriptor.id, b.descriptor.id)); + + ArrayList toGc = new ArrayList<>(); + Set originalIds = new HashSet<>(); + for (int i = 0; i < sorted.size(); i++) + { + SSTableReader table = sorted.get(i); + assertEquals(1, countRows(table) + countTombstoneMarkers(table)); + originalIds.add(table.descriptor.id); + if (i < 3) + toGc.add(table.descriptor); + } + assertEquals(6, originalIds.size()); + + CompactionManager.AllSSTableOpStatus status = + CompactionManager.instance.performGarbageCollection(cfs, TombstoneOption.ROW, 1, toGc); + assertEquals(CompactionManager.AllSSTableOpStatus.SUCCESSFUL, status); + + // The three oldest sstables we asked for should have been compacted away; one of the + // outputs was empty, so 5 sstables remain (3 untouched + 2 newly written). + assertEquals(5, cfs.getLiveSSTables().size()); + Set remainingIds = new HashSet<>(); + for (SSTableReader table : cfs.getLiveSSTables()) + { + remainingIds.add(table.descriptor.id); + assertEquals(1, countRows(table) + countTombstoneMarkers(table)); + } + // None of the three GC'd descriptors should still be live. + for (Descriptor desc : toGc) + assertTrue("GC'd descriptor " + desc.id + " unexpectedly still live", !remainingIds.contains(desc.id)); + } + @Test public void testGarbageCollectOrder() throws Throwable {