diff --git a/CHANGES.txt b/CHANGES.txt index 96e30bab5c..b2a7ae0d05 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java b/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java index 367c304b08..6a712276bb 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraOutgoingFile.java @@ -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) diff --git a/src/java/org/apache/cassandra/streaming/StreamOperation.java b/src/java/org/apache/cassandra/streaming/StreamOperation.java index 8151b47b22..98a4070d2b 100644 --- a/src/java/org/apache/cassandra/streaming/StreamOperation.java +++ b/src/java/org/apache/cassandra/streaming/StreamOperation.java @@ -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; + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/streaming/LCSStreamingKeepLevelTest.java b/test/distributed/org/apache/cassandra/distributed/test/streaming/LCSStreamingKeepLevelTest.java new file mode 100644 index 0000000000..4319768169 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/streaming/LCSStreamingKeepLevelTest.java @@ -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 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); + }); + } + } +}