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

zero copy streaming allocates direct memory that isnt used, but does help to fragment the memory space 

patch by David Capwell; reviewed by Yifan Cai for CASSANDRA-20577
This commit is contained in:
David Capwell 2025-05-02 17:00:20 -07:00
parent e8dc9f1863
commit 0a4777dae4
13 changed files with 458 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -186,6 +186,7 @@
* Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
* Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
Merged from 5.0:
* zero copy streaming allocates direct memory that isn't used, but does help to fragment the memory space (CASSANDRA-20577)
* CQLSSTableWriter supports setting the format (BTI or Big) (CASSANDRA-20609)
* Don't allocate in ThreadLocalReadAheadBuffer#close() (CASSANDRA-20551)
* Ensure RowFilter#isMutableIntersection() properly evaluates numeric ranges on a single column (CASSANDRA-20566)

2
modules/accord

@ -1 +1 @@
Subproject commit 3825403cc50ef7897d5dfb4cbdca5efbc432e8ee
Subproject commit 7f95490b1390b7fc68a4ff4ced7f161bafd8776b

9
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -5711,6 +5711,15 @@ public class DatabaseDescriptor
return Objects.requireNonNull(selectedSSTableFormat, "Forgot to initialize DatabaseDescriptor?");
}
@VisibleForTesting
public static void setSelectedSSTableFormat(String name)
{
SSTableFormat<?, ?> format = getSSTableFormats().get(name);
if (format == null)
throw new IllegalArgumentException("Unknown sstable format: " + name);
setSelectedSSTableFormat(format);
}
@VisibleForTesting
public static void setSelectedSSTableFormat(SSTableFormat<?, ?> format)
{

7
test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkBase.java
View File

@ -66,6 +66,7 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest
@Override
protected void clusterConfig(IInstanceConfig c)
{
super.clusterConfig(c);
c.set("range_request_timeout", "180s")
.set("read_request_timeout", "180s")
.set("write_request_timeout", "180s")
@ -100,6 +101,12 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest
return true;
}
@Override
protected boolean allowRepair()
{
return hasEnoughMemtableForRepair() || hasEnoughSSTablesForRepair();
}
@Override
protected IInvokableInstance selectInstance(RandomSource rs)
{

3
test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithReadRepairTest.java
View File

@ -18,13 +18,10 @@
package org.apache.cassandra.distributed.test.cql3;
import org.junit.Ignore;
import accord.utils.Property;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
@Ignore("In order to stay stable RR tests are ignored for now. Once Single node and multi node w/o RR are stable, then this test should be enabled to include RR testing")
public class MultiNodeTableWalkWithReadRepairTest extends MultiNodeTableWalkBase
{
public MultiNodeTableWalkWithReadRepairTest()

1
test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTokenConflictTest.java
View File

@ -63,6 +63,7 @@ public class MultiNodeTokenConflictTest extends SingleNodeTokenConflictTest
@Override
protected void clusterConfig(IInstanceConfig c)
{
super.clusterConfig(c);
c.set("range_request_timeout", "180s")
.set("read_request_timeout", "180s")
.set("write_request_timeout", "180s")

14
test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
View File

@ -60,6 +60,7 @@ import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.test.sai.SAIUtil;
import org.apache.cassandra.utils.LoggingCommand;
import org.apache.cassandra.harry.model.BytesPartitionState;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
@ -322,7 +323,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
Select select = builder.build();
String annotate = cols.stream().map(symbol -> {
var indexed = state.indexes.get(symbol);
return symbol.detailedName() + (indexed == null ? "" : " (indexed with " + indexed.indexDDL.indexer.name() + ")");
return symbol.detailedName() + (indexed == null ? "" : " (indexed with " + indexed.indexDDL.indexer.name() + ')');
}).collect(Collectors.joining(", "));
return state.command(rs, select, annotate);
}
@ -367,15 +368,20 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
.add(StatefulASTBase::insert)
.add(StatefulASTBase::fullTableScan)
.addIf(State::hasPartitions, this::selectExisting)
.addAllIf(State::supportTokens, b -> b.add(this::selectToken)
.add(this::selectTokenRange)
.add(StatefulASTBase::selectMinTokenRange))
.addAllIf(State::supportTokens,
this::selectToken,
this::selectTokenRange,
StatefulASTBase::selectMinTokenRange)
.addIf(State::hasEnoughMemtable, StatefulASTBase::flushTable)
.addIf(State::hasEnoughSSTables, StatefulASTBase::compactTable)
.addAllIf(BaseState::allowRepair,
StatefulASTBase::incrementalRepair,
StatefulASTBase::previewRepair)
.addIf(State::allowNonPartitionQuery, this::nonPartitionQuery)
.addIf(State::allowNonPartitionMultiColumnQuery, this::multiColumnQuery)
.addIf(State::allowPartitionQuery, this::partitionRestrictedQuery)
.destroyState(State::close)
.commandsTransformer(LoggingCommand.factory())
.onSuccess(onSuccess(logger))
.build());
}

174
test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java
View File

@ -22,9 +22,9 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
@ -34,6 +34,7 @@ import java.util.stream.Stream;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import org.slf4j.Logger;
@ -69,6 +70,7 @@ import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
@ -80,6 +82,9 @@ import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.harry.model.ASTSingleTableModel;
import org.apache.cassandra.harry.util.StringUtils;
import org.apache.cassandra.repair.RepairGenerators;
import org.apache.cassandra.repair.RepairGenerators.PreviewType;
import org.apache.cassandra.repair.RepairGenerators.RepairType;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.AbstractTypeGenerators;
import org.apache.cassandra.utils.CassandraGenerators;
@ -118,6 +123,8 @@ public class StatefulASTBase extends TestBaseImpl
.collect(Collectors.toList()));
protected static final Gen<Gen.IntGen> FETCH_SIZE_DISTRO = Gens.mixedDistribution(new int[] {1, 10, 100, 1000, 5000});
protected static final Gen<Gen.IntGen> LIMIT_DISTRO = Gens.mixedDistribution(1, 1001);
protected static final Gen<Gen.IntGen> REPAIR_TYPE_EMPTY_MODEL_DISTRO = Gens.mixedDistribution(0, 2);
protected static final Gen<Gen.IntGen> REPAIR_TYPE_DISTRO = Gens.mixedDistribution(0, 3);
static
{
@ -145,7 +152,7 @@ public class StatefulASTBase extends TestBaseImpl
protected void clusterConfig(IInstanceConfig config)
{
config.set("repair.retries.max_attempts", Integer.MAX_VALUE);
}
protected void clusterInitializer(ClassLoader cl, int node)
@ -185,7 +192,7 @@ public class StatefulASTBase extends TestBaseImpl
protected static <S extends BaseState> Property.Command<S, Void, ?> flushTable(RandomSource rs, S state)
{
return new Property.SimpleCommand<>("nodetool flush " + state.metadata.keyspace + " " + state.metadata.name, s2 -> {
return new Property.SimpleCommand<>("nodetool flush " + state.metadata.keyspace + ' ' + state.metadata.name, s2 -> {
s2.cluster.forEach(i -> i.nodetoolResult("flush", s2.metadata.keyspace, s2.metadata.name).asserts().success());
s2.flush();
});
@ -193,7 +200,7 @@ public class StatefulASTBase extends TestBaseImpl
protected static <S extends BaseState> Property.Command<S, Void, ?> compactTable(RandomSource rs, S state)
{
return new Property.SimpleCommand<>("nodetool compact " + state.metadata.keyspace + " " + state.metadata.name, s2 -> {
return new Property.SimpleCommand<>("nodetool compact " + state.metadata.keyspace + ' ' + state.metadata.name, s2 -> {
state.cluster.forEach(i -> i.nodetoolResult("compact", s2.metadata.keyspace, s2.metadata.name).asserts().success());
s2.compact();
});
@ -211,6 +218,59 @@ public class StatefulASTBase extends TestBaseImpl
state.commandSafeRandomHistory(selectForMutation(state, mutation), "Select for Mutation Validation"));
}
protected static <S extends BaseState> Property.Command<S, Void, ?> incrementalRepair(RandomSource rs, S state)
{
return repair(rs, state, state.repairArgsBuilder().withType(i -> RepairType.IR).withPreviewType(i -> PreviewType.NONE), null);
}
protected static <S extends BaseState> Property.Command<S, Void, ?> previewRepair(RandomSource rs, S state)
{
return repair(rs, state, state.repairArgsBuilder().withType(i -> RepairType.FULL).withPreviewType(i -> PreviewType.REPAIRED), null);
}
protected static <S extends BaseState> Property.Command<S, Void, ?> repair(RandomSource rs, S state, RepairGenerators.Builder argsBuilder, @Nullable String annotate)
{
IInvokableInstance inst = state.selectInstance(rs);
Gen<List<String>> argsGen = argsBuilder.build();
List<String> args = ImmutableList.<String>builder()
.add("repair")
.addAll(argsGen.next(rs))
.build();
boolean preview = RepairGenerators.isPreview(args);
// mimic org.apache.cassandra.repair.state.CoordinatorState.getType
String type;
if (preview)
{
// mimic org.apache.cassandra.tools.nodetool.Repair.getPreviewKind
PreviewType previewType = RepairGenerators.previewType(args);
switch (previewType)
{
case REPAIRED:
type = "preview repaired";
break;
case UNREPAIRED:
type = RepairGenerators.isFull(args) ? "preview full" : "preview unrepaired";
break;
default:
throw new UnsupportedOperationException(previewType.name());
}
}
else
{
type = RepairGenerators.isFull(args) ? "full" : "incremental";
}
String postfix = "type " + type + ", on " + inst;
if (annotate == null) annotate = postfix;
else annotate += ", " + postfix;
return new Property.SimpleCommand<>("nodetool " + String.join(" ", args) + " -- " + annotate, s2 -> {
inst.nodetoolResult(args.toArray(String[]::new)).asserts().success();
if (!preview)
s2.repair();
});
}
private static <S extends CommonState> Select selectForMutation(S state, Mutation mutation)
{
var select = Select.builder(state.metadata).allowFiltering();
@ -288,16 +348,19 @@ public class StatefulASTBase extends TestBaseImpl
protected final Gen<Conditional.Where.Inequality> lessThanGen;
protected final Gen<Conditional.Where.Inequality> greaterThanGen;
protected final Gen<Conditional.Where.Inequality> rangeInequalityGen;
protected final Gen.IntGen repairTypeEmptyModelGen, repairTypeGen;
protected final Gen.IntGen fetchSizeGen;
protected final TableMetadata metadata;
protected final TableReference tableRef;
protected final ASTSingleTableModel model;
private final String sstableFormatName;
private final Visitor debug;
private final int enoughMemtables;
private final int enoughSSTables;
private final int enoughMemtables, enoughMemtablesForRepair;
private final int enoughSSTables, enoughSSTablesForRepair;
protected int numMutations, mutationsSinceLastFlush;
protected int numFlushes, flushesSinceLastCompaction;
protected int numFlushes, flushesSinceLastCompaction, flushesSinceLastRepair;
protected int numCompact;
protected int numRepairs;
protected int operations;
protected BaseState(RandomSource rs, Cluster cluster, TableMetadata metadata)
@ -322,13 +385,21 @@ public class StatefulASTBase extends TestBaseImpl
this.perPartitionLimitGen = LIMIT_DISTRO.next(rs);
this.limitGen = LIMIT_DISTRO.next(rs);
this.enoughMemtables = rs.pickInt(3, 10, 50);
this.repairTypeEmptyModelGen = REPAIR_TYPE_EMPTY_MODEL_DISTRO.next(rs);
this.repairTypeGen = REPAIR_TYPE_DISTRO.next(rs);
this.enoughMemtables = rs.pickInt(1, 3, 10, 50);
this.enoughMemtablesForRepair = rs.pickInt(1, 3, 10, 50);
this.enoughSSTables = rs.pickInt(3, 10, 50);
this.enoughSSTablesForRepair = rs.pickInt(1, 3, 10, 50);
this.metadata = metadata;
this.tableRef = TableReference.from(metadata);
this.model = new ASTSingleTableModel(metadata, IGNORED_ISSUES);
createTable(metadata);
String sstableFormatName = this.sstableFormatName = Generators.toGen(CassandraGenerators.sstableFormatNames()).next(rs);
cluster.forEach(i -> i.runOnInstance(() -> DatabaseDescriptor.setSelectedSSTableFormat(sstableFormatName)));
}
public boolean hasPartitions()
@ -364,6 +435,35 @@ public class StatefulASTBase extends TestBaseImpl
return command(rs, select, null);
}
protected boolean allowRepair()
{
return false;
}
protected RepairGenerators.Builder repairArgsBuilder()
{
return new RepairGenerators.Builder(i -> Arrays.asList(metadata.keyspace, metadata.name))
// paxos cleanup's finish prepare is delayed based off CAS/Write timeout, but these tests make that 3 minutes (so CI is stable)
// which means this step is delayed 3 minutes, making repairs suppppper slow...
// see org.apache.cassandra.service.paxos.cleanup.PaxosCleanup#finishPrepare
.withSkipPaxosGen(i -> true)
.withRanges(rs -> {
switch (model.isEmpty() ? repairTypeEmptyModelGen.next(rs) : repairTypeGen.next(rs))
{
case 0: return RepairGenerators.LOCAL_RANGE;
case 1: return RepairGenerators.PRIMARY_RANGE;
case 2:
{
Token a = rs.pickOrderedSet(model.partitionKeys()).token;
return List.of("--start-token", Long.toString(a.getLongValue() - 1),
"--end-token", a.toString());
}
default: throw new UnsupportedOperationException();
}
})
;
}
protected boolean allowLimit(Select select)
{
//TODO (coverage): allow this in the model!
@ -467,11 +567,23 @@ public class StatefulASTBase extends TestBaseImpl
return mutationsSinceLastFlush > enoughMemtables;
}
protected boolean hasEnoughMemtableForRepair()
{
// use last flush rather than last repair as this method cares about data in the memtable
// and not amount of mutations since repair
return mutationsSinceLastFlush > enoughMemtablesForRepair;
}
protected boolean hasEnoughSSTables()
{
return flushesSinceLastCompaction > enoughSSTables;
}
protected boolean hasEnoughSSTablesForRepair()
{
return flushesSinceLastRepair > enoughSSTablesForRepair;
}
protected void mutation()
{
numMutations++;
@ -483,6 +595,7 @@ public class StatefulASTBase extends TestBaseImpl
mutationsSinceLastFlush = 0;
numFlushes++;
flushesSinceLastCompaction++;
flushesSinceLastRepair++;
}
protected void compact()
@ -491,6 +604,15 @@ public class StatefulASTBase extends TestBaseImpl
numCompact++;
}
protected void repair()
{
if (mutationsSinceLastFlush > 0)
flush();
numRepairs++;
flushesSinceLastRepair = 0;
}
protected Value value(RandomSource rs, ByteBuffer bb, AbstractType<?> type)
{
return bindOrLiteralGen.next(rs) ? new Bind(bb, type) : new Literal(bb, type);
@ -599,7 +721,8 @@ public class StatefulASTBase extends TestBaseImpl
protected void toString(StringBuilder sb)
{
sb.append(createKeyspaceCQL(metadata.keyspace));
sb.append("Config:\nsstable:\n\tselected_format: ").append(sstableFormatName);
sb.append('\n').append(createKeyspaceCQL(metadata.keyspace));
CassandraGenerators.visitUDTs(metadata, udt -> sb.append('\n').append(udt.toCqlString(false, false, true)).append(';'));
sb.append('\n').append(metadata.toCqlString(false, false, false));
}
@ -620,39 +743,6 @@ public class StatefulASTBase extends TestBaseImpl
toString(sb);
return sb.toString();
}
private static final class ValueWithType
{
final ByteBuffer value;
final AbstractType type;
private ValueWithType(ByteBuffer value, AbstractType<?> type)
{
this.value = value;
this.type = type;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ValueWithType value1 = (ValueWithType) o;
return value.equals(value1.value) && type.equals(value1.type);
}
@Override
public int hashCode()
{
return Objects.hash(value, type);
}
@Override
public String toString()
{
return type.toCQLString(value);
}
}
}
protected static abstract class CommonState extends BaseState

42
test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java
View File

@ -84,8 +84,8 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tools.nodetool.formatter.TableBuilder;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.ConfigGenBuilder;
import org.apache.cassandra.utils.LoggingCommand;
import org.apache.cassandra.utils.Retry;
import static accord.utils.Property.commands;
@ -491,35 +491,6 @@ public abstract class TopologyMixupTestBase<S extends TopologyMixupTestBase.Sche
Command<State<S>, Void, ?> apply(RandomSource rs, State<S> state);
}
private static class LoggingCommand<State, SystemUnderTest, Result> extends Property.ForwardingCommand<State, SystemUnderTest, Result>
{
private static final Logger logger = LoggerFactory.getLogger(LoggingCommand.class);
private LoggingCommand(Command<State, SystemUnderTest, Result> delegate)
{
super(delegate);
}
@Override
public Result apply(State s) throws Throwable
{
String name = detailed(s);
long startNanos = Clock.Global.nanoTime();
try
{
logger.info("Starting command: {}", name);
Result o = super.apply(s);
logger.info("Command {} was success after {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos));
return o;
}
catch (Throwable t)
{
logger.warn("Command {} failed after {}: {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos), t.toString()); // don't want stack trace, just type/msg
throw t;
}
}
}
protected static class State<S extends Schema> implements AutoCloseable
{
final TopologyHistory topologyHistory;
@ -656,16 +627,7 @@ public abstract class TopologyMixupTestBase<S extends TopologyMixupTestBase.Sche
};
}
});
commandsTransformers.add((state, commandGen) -> rs2 -> {
Command<State<S>, Void, ?> c = commandGen.next(rs2);
if (!(c instanceof Property.MultistepCommand))
return new LoggingCommand<>(c);
Property.MultistepCommand<State<S>, Void> multistep = (Property.MultistepCommand<State<S>, Void>) c;
List<Command<State<S>, Void, ?>> subcommands = new ArrayList<>();
for (var sub : multistep)
subcommands.add(new LoggingCommand<>(sub));
return multistep(subcommands);
});
commandsTransformers.add(LoggingCommand.factory());
preActions.add(() -> {
int[] up = topologyHistory.up();
// use the most recent node just in case the cluster isn't in-sync

63
test/unit/org/apache/cassandra/repair/FuzzTestBase.java
View File

@ -47,6 +47,7 @@ import java.util.function.Function;
import java.util.function.Supplier;
import javax.annotation.Nullable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@ -103,6 +104,8 @@ import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.SimulatedMessageDelivery;
import org.apache.cassandra.net.SimulatedMessageDelivery.SimulatedMessageReceiver;
import org.apache.cassandra.repair.RepairGenerators.PreviewType;
import org.apache.cassandra.repair.RepairGenerators.RepairType;
import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.repair.messages.ValidationResponse;
@ -534,12 +537,6 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
}
}
private enum RepairType
{FULL, IR}
private enum PreviewType
{NONE, REPAIRED, UNREPAIRED}
static RepairOption repairOption(RandomSource rs, Cluster.Node coordinator, String ks, List<String> tableNames)
{
return repairOption(rs, coordinator, ks, Gens.lists(Gens.pick(tableNames)).ofSizeBetween(1, tableNames.size()), Gens.enums().all(RepairType.class), Gens.enums().all(PreviewType.class), Gens.enums().all(RepairParallelism.class));
@ -557,53 +554,13 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
private static RepairOption repairOption(RandomSource rs, Cluster.Node coordinator, String ks, Gen<List<String>> tablesGen, Gen<RepairType> repairTypeGen, Gen<PreviewType> previewTypeGen, Gen<RepairParallelism> repairParallelismGen)
{
RepairType type = repairTypeGen.next(rs);
PreviewType previewType = previewTypeGen.next(rs);
List<String> args = new ArrayList<>();
args.add(ks);
List<String> tables = tablesGen.next(rs);
args.addAll(tables);
args.add("-pr");
switch (type)
{
case IR:
// default
break;
case FULL:
args.add("--full");
break;
default:
throw new AssertionError("Unsupported repair type: " + type);
}
switch (previewType)
{
case NONE:
break;
case REPAIRED:
args.add("--validate");
break;
case UNREPAIRED:
args.add("--preview");
break;
default:
throw new AssertionError("Unsupported preview type: " + previewType);
}
RepairParallelism parallelism = repairParallelismGen.next(rs);
switch (parallelism)
{
case SEQUENTIAL:
args.add("--sequential");
break;
case PARALLEL:
// default
break;
case DATACENTER_AWARE:
args.add("--dc-parallel");
break;
default:
throw new AssertionError("Unknown parallelism: " + parallelism);
}
if (rs.nextBoolean()) args.add("--optimise-streams");
List<String> args = new RepairGenerators.Builder(tablesGen.map(l -> ImmutableList.<String>builderWithExpectedSize(l.size() + 1).add(ks).addAll(l).build()))
.withType(repairTypeGen)
.withPreviewType(previewTypeGen)
.withParallelism(repairParallelismGen)
.withRanges(i -> RepairGenerators.PRIMARY_RANGE)
.build()
.next(rs);
RepairOption options = RepairOption.parse(Repair.parseOptionMap(() -> "test", args), DatabaseDescriptor.getPartitioner());
if (options.getRanges().isEmpty())
{

201
test/unit/org/apache/cassandra/repair/RepairGenerators.java Normal file
View File

@ -0,0 +1,201 @@
/*
* 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.repair;
import java.util.ArrayList;
import java.util.List;
import accord.utils.Gen;
import accord.utils.Gens;
public class RepairGenerators
{
public static final List<String> LOCAL_RANGE = List.of();
public static final List<String> PRIMARY_RANGE = List.of("-pr"); // repair calls this partition range, but StorageService calls this primary
public enum RepairType
{
FULL("--full"),
IR("");
public final String arg;
RepairType(String s)
{
this.arg = s;
}
}
public enum PreviewType
{
NONE(""),
REPAIRED("--validate"),
UNREPAIRED("--preview");
public final String arg;
PreviewType(String s)
{
this.arg = s;
}
}
public static boolean isPreview(List<String> args)
{
return args.stream().anyMatch(s -> PreviewType.REPAIRED.arg.equals(s)
|| PreviewType.UNREPAIRED.arg.equals(s));
}
public static PreviewType previewType(List<String> args)
{
for (String s : args)
{
if (PreviewType.REPAIRED.arg.equals(s))
return PreviewType.REPAIRED;
if (PreviewType.UNREPAIRED.arg.equals(s))
return PreviewType.UNREPAIRED;
}
return PreviewType.NONE;
}
public static boolean isFull(List<String> args)
{
return args.stream().anyMatch(s -> RepairType.FULL.arg.equals(s));
}
public static boolean isIncremental(List<String> args)
{
return !isFull(args);
}
public static class Builder
{
final Gen<List<String>> tablesGen;
Gen<RepairType> typeGen = Gens.enums().all(RepairType.class);
Gen<PreviewType> previewTypeGen = Gens.enums().all(PreviewType.class);
Gen<List<String>> ranges = Gens.pick(List.of(), PRIMARY_RANGE);
Gen<Boolean> optimizeStreamsGen = Gens.bools().all();
Gen<RepairParallelism> parallelismGen = Gens.enums().all(RepairParallelism.class);
Gen<Boolean> skipPaxosGen = i -> false;
Gen<Boolean> skipAccordGen = i -> false;
public Builder(Gen<List<String>> tablesGen)
{
this.tablesGen = tablesGen;
}
public Builder withType(Gen<RepairType> typeGen)
{
this.typeGen = typeGen;
return this;
}
public Builder withPreviewType(Gen<PreviewType> previewTypeGen)
{
this.previewTypeGen = previewTypeGen;
return this;
}
public Builder withRanges(Gen<List<String>> ranges)
{
this.ranges = ranges;
return this;
}
public Builder withOptimizeStreams(Gen<Boolean> optimizeStreamsGen)
{
this.optimizeStreamsGen = optimizeStreamsGen;
return this;
}
public Builder withParallelism(Gen<RepairParallelism> parallelismGen)
{
this.parallelismGen = parallelismGen;
return this;
}
public Builder withSkipPaxosGen(Gen<Boolean> skipPaxosGen)
{
this.skipPaxosGen = skipPaxosGen;
return this;
}
public Builder withSkipAccordGen(Gen<Boolean> skipAccordGen)
{
this.skipAccordGen = skipAccordGen;
return this;
}
public Gen<List<String>> build()
{
return rs -> {
RepairType type = typeGen.next(rs);
PreviewType previewType = previewTypeGen.next(rs);
List<String> args = new ArrayList<>();
args.addAll(tablesGen.next(rs));
args.addAll(ranges.next(rs));
if (skipPaxosGen.next(rs))
args.add("--skip-paxos");
if (skipAccordGen.next(rs))
args.add("--skip-accord");
switch (type)
{
case IR:
// default
break;
case FULL:
args.add(type.arg);
break;
default:
throw new AssertionError("Unsupported repair type: " + type);
}
switch (previewType)
{
case NONE:
break;
case REPAIRED:
case UNREPAIRED:
args.add(previewType.arg);
break;
default:
throw new AssertionError("Unsupported preview type: " + previewType);
}
RepairParallelism parallelism = parallelismGen.next(rs);
switch (parallelism)
{
case SEQUENTIAL:
args.add("--sequential");
break;
case PARALLEL:
// default
break;
case DATACENTER_AWARE:
args.add("--dc-parallel");
break;
default:
throw new AssertionError("Unknown parallelism: " + parallelism);
}
if (optimizeStreamsGen.next(rs))
args.add("--optimise-streams");
return args;
};
}
}
}

8
test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java
View File

@ -62,6 +62,14 @@ public class ImmutableUniqueList<T> extends AbstractList<T> implements RandomAcc
return (ImmutableUniqueList<T>) EMPTY;
}
public static <T> ImmutableUniqueList<T> of(T... values)
{
Builder<T> builder = builder(values.length);
for (T v : values)
builder.add(v);
return builder.build();
}
public AsSet asSet()
{
if (asSet != null) return asSet;

76
test/unit/org/apache/cassandra/utils/LoggingCommand.java Normal file
View File

@ -0,0 +1,76 @@
/*
* 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.utils;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.function.BiFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.Gen;
import accord.utils.Property;
import accord.utils.Property.Command;
import static accord.utils.Property.multistep;
public class LoggingCommand<State, SystemUnderTest, Result> extends Property.ForwardingCommand<State, SystemUnderTest, Result>
{
private static final Logger logger = LoggerFactory.getLogger(LoggingCommand.class);
public LoggingCommand(Command<State, SystemUnderTest, Result> delegate)
{
super(delegate);
}
public static <State, SystemUnderTest> BiFunction<State, Gen<Command<State, SystemUnderTest, ?>>, Gen<Command<State, SystemUnderTest, ?>>> factory()
{
return (state, commandGen) -> rs -> {
Command<State, SystemUnderTest, ?> c = commandGen.next(rs);
if (!(c instanceof Property.MultistepCommand))
return new LoggingCommand<>(c);
Property.MultistepCommand<State, SystemUnderTest> multistep = (Property.MultistepCommand<State, SystemUnderTest>) c;
List<Command<State, SystemUnderTest, ?>> subcommands = new ArrayList<>();
for (var sub : multistep)
subcommands.add(new LoggingCommand<>(sub));
return multistep(subcommands);
};
}
@Override
public Result apply(State s) throws Throwable
{
String name = detailed(s);
long startNanos = Clock.Global.nanoTime();
try
{
logger.info("Starting command: {}", name);
Result o = super.apply(s);
logger.info("Command {} was success after {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos));
return o;
}
catch (Throwable t)
{
logger.warn("Command {} failed after {}: {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos), t.toString()); // don't want stack trace, just type/msg
throw t;
}
}
}