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

Allow nodetool garbagecollect to take a user defined list of SSTables 

patch by Scott Carey; reviewed by Stefan Miklosovic, Brandon Williams for CASSANDRA-16767
This commit is contained in:
Scott Carey 2021-06-25 23:17:50 -07:00 committed by Stefan Miklosovic
parent eabd2a27f5
commit cbbd401244
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
10 changed files with 202 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

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

@ -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<Descriptor> sstables)
{
return CompactionManager.instance.performGarbageCollection(this, tombstoneOption, jobs, sstables);
}
public void markObsolete(Collection<SSTableReader> sstables, OperationType compactionType)
{
assert !sstables.isEmpty();

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

@ -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<SSTableReader> sstables = Lists.newArrayList(operation.filterSSTables(compacting));
if (Iterables.isEmpty(sstables))
List<SSTableReader> 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<Descriptor> dataFiles)
{
Set<Descriptor> files = new HashSet<>(dataFiles);
return performGarbageCollection(cfStore, tombstoneOption, jobs, files::contains);
}
@VisibleForTesting
AllSSTableOpStatus performGarbageCollection(ColumnFamilyStore cfStore, TombstoneOption tombstoneOption, int jobs, Predicate<Descriptor> 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<ColumnFamilyStore, Descriptor> 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<ColumnFamilyStore, Descriptor> descriptors = Descriptor.fromFilenamesGrouped(Arrays.asList(dataFiles.split(",")));
List<Future<?>> futures = new ArrayList<>(descriptors.size());
long nowInSec = FBUtilities.nowInSeconds();

30
src/java/org/apache/cassandra/io/sstable/Descriptor.java
View File

@ -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 <SSTable Data.db>} 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<ColumnFamilyStore, Descriptor> fromFilenamesGrouped(Collection<String> filenames)
{
Multimap<ColumnFamilyStore, Descriptor> 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();

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

@ -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<String> 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<ColumnFamilyStore, Collection<Descriptor>> entry : Descriptor.fromFilenamesGrouped(userDefinedTables).asMap().entrySet())
{
ColumnFamilyStore cfs = entry.getKey();
Collection<Descriptor> 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
{

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

@ -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<String> 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.

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

@ -418,6 +418,11 @@ public class NodeProbe implements AutoCloseable
return ssProxy.recompressSSTables(keyspaceName, jobs, tableNames);
}
public int userDefinedGarbageCollect(String tombstoneOption, int jobs, List<String> 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<String> 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);

23
src/java/org/apache/cassandra/tools/nodetool/GarbageCollect.java
View File

@ -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 = "[<keyspace> <tables>...]", description = "The keyspace followed by one or many tables")
@CassandraUsage(usage = "[<keyspace> <tables>...] or <SSTable file>...",
description = "The keyspace followed by one or many tables, " +
"or a list of SSTable data files when using --user-defined")
private List<String> 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<String> keyspaces = parseOptionalKeyspace(args, probe);
String[] tableNames = parseOptionalTables(args);

13
test/resources/nodetool/help/garbagecollect
View File

@ -7,7 +7,8 @@ SYNOPSIS
[(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
[(-u <username> | --username <username>)] garbagecollect
[(-g <granularity> | --granularity <granularity>)]
[(-j <jobs> | --jobs <jobs>)] [--] [<keyspace> <tables>...]
[(-j <jobs> | --jobs <jobs>)] [--user-defined] [--] [<keyspace>
<tables>...] or <SSTable file>...
OPTIONS
-g <granularity>, --granularity <granularity>
@ -34,10 +35,16 @@ OPTIONS
-u <username>, --username <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
[<keyspace> <tables>...]
The keyspace followed by one or many tables
[<keyspace> <tables>...] or <SSTable file>...
The keyspace followed by one or many tables, or a list of SSTable
data files when using --user-defined

66
test/unit/org/apache/cassandra/cql3/GcCompactionTest.java
View File

@ -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<SSTableReader> sorted = new ArrayList<>(cfs.getLiveSSTables());
sorted.sort((a, b) -> SSTableIdFactory.COMPARATOR.compare(a.descriptor.id, b.descriptor.id));
ArrayList<Descriptor> toGc = new ArrayList<>();
Set<SSTableId> 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<SSTableId> 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
{