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

Keep sstable level when streaming for decommission and move 

Patch by marcuse; reviewed by David Capwell for CASSANDRA-17969
This commit is contained in:
Marcus Eriksson 2022-10-18 08:26:21 +02:00
parent 89e01ea2eb
commit 7e6799a4b9
4 changed files with 154 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.2
* Keep sstable level when streaming for decommission and move (CASSANDRA-17969)
* Add Unavailables metric for CASWrite in the docs (CASSANDRA-16357)
* Make Cassandra logs able to be viewed in the virtual table system_views.system_logs (CASSANDRA-17946)
* IllegalArgumentException in Gossiper#order due to concurrent mutations to elements being applied (CASSANDRA-17908)

4
src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java
View File

@ -76,8 +76,6 @@ public class CassandraOutgoingFile implements OutgoingStream
boolean shouldStreamEntireSSTable,
ComponentManifest manifest)
{
boolean keepSSTableLevel = operation == StreamOperation.BOOTSTRAP || operation == StreamOperation.REBUILD;
CompressionInfo compressionInfo = sstable.compression
? CompressionInfo.newLazyInstance(sstable.getCompressionMetadata(), sections)
: null;
@ -85,7 +83,7 @@ public class CassandraOutgoingFile implements OutgoingStream
return CassandraStreamHeader.builder()
.withSSTableFormat(sstable.descriptor.formatType)
.withSSTableVersion(sstable.descriptor.version)
.withSSTableLevel(keepSSTableLevel ? sstable.getSSTableLevel() : 0)
.withSSTableLevel(operation.keepSSTableLevel() ? sstable.getSSTableLevel() : 0)
.withEstimatedKeys(estimatedKeys)
.withSections(sections)
.withCompressionInfo(compressionInfo)

43
src/java/org/apache/cassandra/streaming/StreamOperation.java
View File

@ -19,43 +19,43 @@ package org.apache.cassandra.streaming;
public enum StreamOperation
{
OTHER("Other"), // Fallback to avoid null types when deserializing from string
RESTORE_REPLICA_COUNT("Restore replica count", false), // Handles removeNode
DECOMMISSION("Unbootstrap", false),
RELOCATION("Relocation", false),
BOOTSTRAP("Bootstrap", false),
REBUILD("Rebuild", false),
BULK_LOAD("Bulk Load"),
REPAIR("Repair");
OTHER("Other", true, false), // Fallback to avoid null types when deserializing from string
RESTORE_REPLICA_COUNT("Restore replica count", false, false), // Handles removeNode
DECOMMISSION("Unbootstrap", false, true),
RELOCATION("Relocation", false, true),
BOOTSTRAP("Bootstrap", false, true),
REBUILD("Rebuild", false, true),
BULK_LOAD("Bulk Load", true, false),
REPAIR("Repair", true, false);
private final String description;
private final boolean requiresViewBuild;
StreamOperation(String description) {
this(description, true);
}
private final boolean keepSSTableLevel;
/**
* @param description The operation description
* @param requiresViewBuild Whether this operation requires views to be updated if it involves a base table
*/
StreamOperation(String description, boolean requiresViewBuild) {
StreamOperation(String description, boolean requiresViewBuild, boolean keepSSTableLevel)
{
this.description = description;
this.requiresViewBuild = requiresViewBuild;
this.keepSSTableLevel = keepSSTableLevel;
}
public static StreamOperation fromString(String text) {
for (StreamOperation b : StreamOperation.values()) {
if (b.description.equalsIgnoreCase(text)) {
public static StreamOperation fromString(String text)
{
for (StreamOperation b : StreamOperation.values())
{
if (b.description.equalsIgnoreCase(text))
return b;
}
}
return OTHER;
}
public String getDescription() {
public String getDescription()
{
return description;
}
@ -66,4 +66,9 @@ public enum StreamOperation
{
return this.requiresViewBuild;
}
public boolean keepSSTableLevel()
{
return keepSSTableLevel;
}
}

128
test/distributed/org/apache/cassandra/distributed/test/streaming/LCSStreamingKeepLevelTest.java Normal file
View File

@ -0,0 +1,128 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.distributed.test.streaming;
import java.io.IOException;
import java.util.Set;
import org.junit.Test;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.junit.Assert.assertTrue;
public class LCSStreamingKeepLevelTest extends TestBaseImpl
{
@Test
public void testDecom() throws IOException
{
try (Cluster cluster = builder().withNodes(4)
.withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL))
.withoutVNodes()
.withDataDirCount(1)
.start())
{
populate(cluster);
cluster.get(4).nodetoolResult("decommission").asserts().success();
assertEmptyL0(cluster);
}
}
@Test
public void testMove() throws IOException
{
try (Cluster cluster = builder().withNodes(4)
.withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL))
.withoutVNodes()
.withDataDirCount(1)
.start())
{
populate(cluster);
long tokenVal = ((Murmur3Partitioner.LongToken)cluster.tokens().get(3).getToken()).token;
long prevTokenVal = ((Murmur3Partitioner.LongToken)cluster.tokens().get(2).getToken()).token;
// move node 4 to the middle point between its current position and the previous node
long newToken = (tokenVal + prevTokenVal) / 2;
cluster.get(4).nodetoolResult("move", String.valueOf(newToken)).asserts().success();
assertEmptyL0(cluster);
}
}
private static void populate(Cluster cluster)
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2}"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s.decom_test (id int PRIMARY KEY, value int) with compaction = { 'class':'LeveledCompactionStrategy', 'enabled':'false' }"));
for (int i = 0; i < 500; i++)
{
cluster.coordinator(1).execute(withKeyspace("insert into %s.decom_test (id, value) VALUES (?, ?)"), ConsistencyLevel.ALL, i, i);
if (i % 100 == 0)
cluster.forEach((inst) -> inst.flush(KEYSPACE));
}
cluster.forEach((i) -> i.flush(KEYSPACE));
relevel(cluster);
}
private static void relevel(Cluster cluster)
{
for (IInvokableInstance i : cluster)
{
i.runOnInstance(() -> {
Set<SSTableReader> sstables = Keyspace.open(KEYSPACE).getColumnFamilyStore("decom_test").getLiveSSTables();
int lvl = 1;
for (SSTableReader sstable : sstables)
{
try
{
sstable.mutateLevelAndReload(lvl++);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
});
}
assertEmptyL0(cluster);
}
private static void assertEmptyL0(Cluster cluster)
{
for (IInvokableInstance i : cluster)
{
i.runOnInstance(() -> {
for (SSTableReader sstable : Keyspace.open(KEYSPACE).getColumnFamilyStore("decom_test").getLiveSSTables())
assertTrue(sstable.getSSTableLevel() > 0);
});
}
}
}