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

Merge branch 'cassandra-3.11' into cassandra-4.0

This commit is contained in:
Andrés de la Peña 2022-08-12 10:28:42 +01:00
parent e5e13c02cc 4e410fc46d
commit 174e0dc399
7 changed files with 57 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -6,6 +6,8 @@
* Fix Settings Virtual Table - index_summary_resize_interval and index_summary_capacity were not updated after startup (CASSANDRA-17735)
* Clean up ScheduledExecutors, CommitLog, and MessagingService shutdown for in-JVM dtests (CASSANDRA-17731)
* Remove extra write to system table for prepared statements (CASSANDRA-17764)
Merged from 3.11:
* Document usage of closed token intervals in manual compaction (CASSANDRA-17575)
Merged from 3.0:
* Fix restarting of services on gossipping-only member (CASSANDRA-17752)

7
src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java
View File

@ -51,7 +51,12 @@ public interface ColumnFamilyStoreMBean
public void forceMajorCompaction(boolean splitOutput) throws ExecutionException, InterruptedException;
/**
* force a major compaction of specified key range in this column family
* Forces a major compaction of specified token ranges in this column family.
* <p>
* The token ranges will be interpreted as closed intervals to match the closed interval defined by the first and
* last keys of a sstable, even though the {@link Range} class is suppossed to be half-open by definition.
*
* @param tokenRanges The token ranges to be compacted, interpreted as closed intervals.
*/
public void forceCompactionForTokenRange(Collection<Range<Token>> tokenRanges) throws ExecutionException, InterruptedException;
/**

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

@ -889,6 +889,15 @@ public class CompactionManager implements CompactionManagerMBean
return futures;
}
/**
* Forces a major compaction of specified token ranges of the specified column family.
* <p>
* The token ranges will be interpreted as closed intervals to match the closed interval defined by the first and
* last keys of a sstable, even though the {@link Range} class is suppossed to be half-open by definition.
*
* @param cfStore The column family store to be compacted.
* @param ranges The token ranges to be compacted, interpreted as closed intervals.
*/
public void forceCompactionForTokenRange(ColumnFamilyStore cfStore, Collection<Range<Token>> ranges)
{
Callable<CompactionTasks> taskCreator = () -> {
@ -924,6 +933,12 @@ public class CompactionManager implements CompactionManagerMBean
}
}
/**
* Returns the sstables of the specified column family store that intersect with the specified token ranges.
* <p>
* The token ranges will be interpreted as closed intervals to match the closed interval defined by the first and
* last keys of a sstable, even though the {@link Range} class is suppossed to be half-open by definition.
*/
private static Collection<SSTableReader> sstablesInBounds(ColumnFamilyStore cfs, Collection<Range<Token>> tokenRangeCollection)
{
final Set<SSTableReader> sstables = new HashSet<>();

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

@ -316,8 +316,14 @@ public interface StorageServiceMBean extends NotificationEmitter
@Deprecated
public int relocateSSTables(String keyspace, String ... cfnames) throws IOException, ExecutionException, InterruptedException;
public int relocateSSTables(int jobs, String keyspace, String ... cfnames) throws IOException, ExecutionException, InterruptedException;
/**
* Forces major compaction of specified token range in a single keyspace
* Forces major compaction of specified token range in a single keyspace.
*
* @param keyspaceName the name of the keyspace to be compacted
* @param startToken the token at which the compaction range starts (inclusive)
* @param endToken the token at which compaction range ends (inclusive)
* @param tableNames the names of the tables to be compacted
*/
public void forceKeyspaceCompactionForTokenRange(String keyspaceName, String startToken, String endToken, String... tableNames) throws IOException, ExecutionException, InterruptedException;

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

@ -409,6 +409,14 @@ public class NodeProbe implements AutoCloseable
ssProxy.relocateSSTables(jobs, keyspace, cfnames);
}
/**
* Forces major compaction of specified token range in a single keyspace.
*
* @param keyspaceName the name of the keyspace to be compacted
* @param startToken the token at which the compaction range starts (inclusive)
* @param endToken the token at which compaction range ends (inclusive)
* @param tableNames the names of the tables to be compacted
*/
public void forceKeyspaceCompactionForTokenRange(String keyspaceName, final String startToken, final String endToken, String... tableNames) throws IOException, ExecutionException, InterruptedException
{
ssProxy.forceKeyspaceCompactionForTokenRange(keyspaceName, startToken, endToken, tableNames);

4
src/java/org/apache/cassandra/tools/nodetool/Compact.java
View File

@ -41,10 +41,10 @@ public class Compact extends NodeToolCmd
@Option(title = "user-defined", name = {"--user-defined"}, description = "Use --user-defined to submit listed files for user-defined compaction")
private boolean userDefined = false;
@Option(title = "start_token", name = {"-st", "--start-token"}, description = "Use -st to specify a token at which the compaction range starts")
@Option(title = "start_token", name = {"-st", "--start-token"}, description = "Use -st to specify a token at which the compaction range starts (inclusive)")
private String startToken = EMPTY;
@Option(title = "end_token", name = {"-et", "--end-token"}, description = "Use -et to specify a token at which compaction range ends")
@Option(title = "end_token", name = {"-et", "--end-token"}, description = "Use -et to specify a token at which compaction range ends (inclusive)")
private String endToken = EMPTY;

40
test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
View File

@ -67,6 +67,7 @@ import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.awaitility.Awaitility;
import static java.util.Collections.singleton;
import static org.assertj.core.api.Assertions.assertThat;
@ -390,12 +391,10 @@ public class LeveledCompactionStrategyTest
assertFalse(repaired.manifest.getLevel(1).contains(sstable2));
}
@Test
public void testTokenRangeCompaction() throws Exception
{
// Remove any existing data so we can start out clean with predictable number of sstables
// Remove any existing data, so we can start out clean with predictable number of sstables
cfs.truncateBlocking();
// Disable auto compaction so cassandra does not compact
@ -405,15 +404,17 @@ public class LeveledCompactionStrategyTest
DecoratedKey key1 = Util.dk(String.valueOf(1));
DecoratedKey key2 = Util.dk(String.valueOf(2));
List<DecoratedKey> keys = new ArrayList<>(Arrays.asList(key1, key2));
List<DecoratedKey> keys = Arrays.asList(key1, key2);
int numIterations = 10;
int columns = 2;
// Add enough data to trigger multiple sstables.
// create 10 sstables that contain data for both key1 and key2
for (int i = 0; i < numIterations; i++) {
for (DecoratedKey key : keys) {
for (int i = 0; i < numIterations; i++)
{
for (DecoratedKey key : keys)
{
UpdateBuilder update = UpdateBuilder.create(cfs.metadata(), key);
for (int c = 0; c < columns; c++)
update.newRow("column" + c).add("val", value);
@ -436,28 +437,23 @@ public class LeveledCompactionStrategyTest
// We should have a total of 30 sstables by now
assertEquals(30, cfs.getLiveSSTables().size());
// Compact just the tables with key2
// Bit hackish to use the key1.token as the prior key but works in BytesToken
// Compact just the tables with key2. The token ranges for compaction are interpreted as closed intervals,
// so we can use [token, token] to select a single token.
Range<Token> tokenRange = new Range<>(key2.getToken(), key2.getToken());
Collection<Range<Token>> tokenRanges = new ArrayList<>(Arrays.asList(tokenRange));
Collection<Range<Token>> tokenRanges = singleton(tokenRange);
cfs.forceCompactionForTokenRange(tokenRanges);
while(CompactionManager.instance.isCompacting(Arrays.asList(cfs), (sstable) -> true)) {
Thread.sleep(100);
}
Awaitility.await().until(() -> !CompactionManager.instance.isCompacting(singleton(cfs), sstable -> true));
// 20 tables that have key2 should have been compacted in to 1 table resulting in 11 (30-20+1)
assertEquals(11, cfs.getLiveSSTables().size());
// Compact just the tables with key1. At this point all 11 tables should have key1
Range<Token> tokenRange2 = new Range<>(key1.getToken(), key1.getToken());
Collection<Range<Token>> tokenRanges2 = new ArrayList<>(Arrays.asList(tokenRange2));
Collection<Range<Token>> tokenRanges2 = singleton(tokenRange2);
cfs.forceCompactionForTokenRange(tokenRanges2);
while(CompactionManager.instance.isCompacting(Arrays.asList(cfs), (sstable) -> true)) {
Thread.sleep(100);
}
Awaitility.await().until(() -> !CompactionManager.instance.isCompacting(singleton(cfs), sstable -> true));
// the 11 tables containing key1 should all compact to 1 table
assertEquals(1, cfs.getLiveSSTables().size());
@ -493,7 +489,8 @@ public class LeveledCompactionStrategyTest
// We should have a total of 30 sstables again
assertEquals(30, cfs.getLiveSSTables().size());
// This time, we're going to make sure the token range wraps around, to cover the full range
// This time, we're going to make sure the token range wraps around, to cover the full range.
// Note that the ranges used by compaction are interpreted as closed intervals, so it will be [32, 31].
Range<Token> wrappingRange;
if (key1.getToken().compareTo(key2.getToken()) < 0)
{
@ -503,13 +500,10 @@ public class LeveledCompactionStrategyTest
{
wrappingRange = new Range<>(key1.getToken(), key2.getToken());
}
Collection<Range<Token>> wrappingRanges = new ArrayList<>(Arrays.asList(wrappingRange));
Collection<Range<Token>> wrappingRanges = singleton(wrappingRange);
cfs.forceCompactionForTokenRange(wrappingRanges);
while(CompactionManager.instance.isCompacting(Arrays.asList(cfs), (sstable) -> true))
{
Thread.sleep(100);
}
Awaitility.await().until(() -> !CompactionManager.instance.isCompacting(singleton(cfs), sstable -> true));
// should all compact to 1 table
assertEquals(1, cfs.getLiveSSTables().size());