diff --git a/.build/build-rat.xml b/.build/build-rat.xml
index 65f79e2c24..39808eb2b2 100644
--- a/.build/build-rat.xml
+++ b/.build/build-rat.xml
@@ -60,6 +60,7 @@
+
diff --git a/build.xml b/build.xml
index cedd0c567d..1366d13a98 100644
--- a/build.xml
+++ b/build.xml
@@ -1939,7 +1939,7 @@
-
+
diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
index 723c773920..8eec2cd93e 100644
--- a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
+++ b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
@@ -1097,6 +1097,17 @@ public class PartitionUpdate extends AbstractBTreePartition
return metadata;
}
+ /**
+ * Returns the number of non-static rows accumulated so far.
+ * Accurate when rows are added in clustering order; may overcount with unordered input.
+ */
+ public int rowCount()
+ {
+ if (rowBuilder != null)
+ return rowBuilder.count();
+ return firstRow != null ? 1 : 0;
+ }
+
private static final UpdateFunction ROWS_MERGE_FUNCTION = UpdateFunction.Simple.of(Rows::merge);
public PartitionUpdate build()
diff --git a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
index 2487e66fd2..19501fcc9b 100644
--- a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
@@ -27,6 +27,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Stream;
@@ -56,10 +57,12 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
protected static final AtomicReference id = new AtomicReference<>(SSTableIdFactory.instance.defaultBuilder().generator(Stream.empty()).get());
protected boolean makeRangeAware = false;
protected final Collection indexGroups;
- protected Consumer> sstableProducedListener;
+ protected BiConsumer> sstableProducedListener;
protected boolean openSSTableOnProduced = false;
protected CompressionDictionary compressionDictionary;
protected SSTable.Owner owner;
+ protected int sstableLevel = 0;
+ protected long reparedAtMillis = ActiveRepairService.UNREPAIRED_SSTABLE;
protected AbstractSSTableSimpleWriter(File directory, TableMetadataRef metadata, RegularAndStaticColumns columns)
{
@@ -69,11 +72,21 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
indexGroups = new ArrayList<>();
}
+ protected void setReparedAtMillis(long reparedAtMillis)
+ {
+ this.reparedAtMillis = reparedAtMillis;
+ }
+
protected void setSSTableFormatType(SSTableFormat, ?> type)
{
this.format = type;
}
+ protected void setSSTableLevel(int level)
+ {
+ this.sstableLevel = level;
+ }
+
protected void setRangeAwareWriting(boolean makeRangeAware)
{
this.makeRangeAware = makeRangeAware;
@@ -90,6 +103,12 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
}
protected void setSSTableProducedListener(Consumer> listener)
+ {
+ Objects.requireNonNull(listener, "sstableProducedListener cannot be null");
+ this.sstableProducedListener = (writer, readers) -> listener.accept(readers);
+ }
+
+ protected void setSSTableProducedListener(BiConsumer> listener)
{
this.sstableProducedListener = Objects.requireNonNull(listener, "sstableProducedListener cannot be null");
}
@@ -107,12 +126,12 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
return openSSTableOnProduced;
}
- protected void notifySSTableProduced(Collection sstables)
+ protected void notifySSTableProduced(SSTableTxnWriter writer, Collection sstables)
{
if (sstableProducedListener == null)
return;
- sstableProducedListener.accept(sstables);
+ sstableProducedListener.accept(writer, sstables);
}
protected SSTableTxnWriter createWriter(SSTable.Owner owner) throws IOException
@@ -120,7 +139,7 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
SerializationHeader header = new SerializationHeader(true, metadata.get(), columns, EncodingStats.NO_STATS);
if (makeRangeAware)
- return SSTableTxnWriter.createRangeAware(metadata, 0, ActiveRepairService.UNREPAIRED_SSTABLE, ActiveRepairService.NO_PENDING_REPAIR, false, format, header);
+ return SSTableTxnWriter.createRangeAware(metadata, 0, reparedAtMillis, ActiveRepairService.NO_PENDING_REPAIR, false, format, header);
SSTable.Owner effectiveOwner;
@@ -139,9 +158,10 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
return SSTableTxnWriter.create(metadata,
createDescriptor(directory, metadata.keyspace, metadata.name, format),
0,
- ActiveRepairService.UNREPAIRED_SSTABLE,
+ reparedAtMillis,
ActiveRepairService.NO_PENDING_REPAIR,
false,
+ sstableLevel,
header,
indexGroups,
effectiveOwner);
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
index 8c9b2f9796..bf1f04c9b7 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
@@ -54,13 +54,14 @@ import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQu
*
* @see SSTableSimpleWriter
*/
-class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
+public class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
{
private static final Buffer SENTINEL = new Buffer();
private Buffer buffer = new Buffer();
private final long maxSStableSizeInBytes;
private long currentSize;
+ private boolean syncPending;
// Used to compute the row serialized size
private final SerializationHeader header;
@@ -91,6 +92,18 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
PartitionUpdate.Builder previous = buffer.get(key);
if (previous == null)
{
+ if (syncPending)
+ {
+ syncPending = false;
+ try
+ {
+ sync();
+ }
+ catch (IOException e)
+ {
+ throw new SyncException(e);
+ }
+ }
// todo: inefficient - we create and serialize a PU just to get its size, then recreate it
// todo: either allow PartitionUpdateBuilder to have .build() called several times or pre-calculate the size
currentSize += PartitionUpdate.serializer.serializedSize(createPartitionUpdateBuilder(key).build(), format.getLatestVersion().correspondingMessagingVersion());
@@ -110,19 +123,10 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
currentSize += UnfilteredSerializer.serializer.serializedSize(row, helper, 0, format.getLatestVersion().correspondingMessagingVersion());
}
- private void maybeSync() throws SyncException
+ private void maybeSync()
{
- try
- {
- if (currentSize > maxSStableSizeInBytes)
- sync();
- }
- catch (IOException e)
- {
- // addColumn does not throw IOException but we want to report this to the user,
- // so wrap it in a temporary RuntimeException that we'll catch in rawAddRow above.
- throw new SyncException(e);
- }
+ if (currentSize > maxSStableSizeInBytes)
+ syncPending = true;
}
private PartitionUpdate.Builder createPartitionUpdateBuilder(DecoratedKey key)
@@ -132,8 +136,11 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
@Override
public void add(Row row)
{
+ int before = rowCount();
super.add(row);
- countRow(row);
+
+ if (rowCount() > before)
+ countRow(row);
maybeSync();
}
};
@@ -229,7 +236,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
for (Map.Entry entry : b.entrySet())
writer.append(entry.getValue().build().unfilteredIterator());
Collection finished = writer.finish(shouldOpenSSTables());
- notifySSTableProduced(finished);
+ notifySSTableProduced(writer, finished);
}
}
catch (Throwable e)
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java
index 7b5fb8a89e..ed29e7db70 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java
@@ -44,7 +44,7 @@ import org.apache.cassandra.schema.TableMetadataRef;
* The output will be a series of SSTables that do not exceed a specified size.
* By default, all sorted data are written into a single SSTable.
*/
-class SSTableSimpleWriter extends AbstractSSTableSimpleWriter
+public class SSTableSimpleWriter extends AbstractSSTableSimpleWriter
{
private final long maxSSTableSizeInBytes;
@@ -161,7 +161,7 @@ class SSTableSimpleWriter extends AbstractSSTableSimpleWriter
return;
Collection finished = writer.finish(shouldOpenSSTables());
- notifySSTableProduced(finished);
+ notifySSTableProduced(writer, finished);
}
catch (Throwable t)
{
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
index 3b43dcfdf4..51f66b0ddc 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
@@ -151,10 +151,25 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
SerializationHeader header,
Collection indexGroups,
SSTable.Owner owner)
+ {
+ return create(metadata, descriptor, keyCount, repairedAt, pendingRepair, isTransient, 0, header, indexGroups, owner);
+ }
+
+ @SuppressWarnings({"resource", "RedundantSuppression"}) // log and writer closed during doPostCleanup
+ public static SSTableTxnWriter create(TableMetadataRef metadata,
+ Descriptor descriptor,
+ long keyCount,
+ long repairedAt,
+ TimeUUID pendingRepair,
+ boolean isTransient,
+ int sstableLevel,
+ SerializationHeader header,
+ Collection indexGroups,
+ SSTable.Owner owner)
{
// if the column family store does not exist, we create a new default SSTableMultiWriter to use:
LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.WRITE);
- SSTableMultiWriter writer = SimpleSSTableMultiWriter.create(descriptor, keyCount, repairedAt, pendingRepair, isTransient, metadata, null, 0, header, indexGroups, txn, owner);
+ SSTableMultiWriter writer = SimpleSSTableMultiWriter.create(descriptor, keyCount, repairedAt, pendingRepair, isTransient, metadata, null, sstableLevel, header, indexGroups, txn, owner);
return new SSTableTxnWriter(txn, writer);
}
}
diff --git a/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java b/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java
index fd1a2618db..f021b41bcd 100644
--- a/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java
+++ b/src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Objects;
import java.util.Set;
+import com.google.common.base.Preconditions;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@@ -62,6 +63,9 @@ public class DataPlacement
this.writes = reads; // performance optimization for a typical case, to not search for endpoints twice
else
this.writes = writes;
+
+ Preconditions.checkArgument(reads.ranges.equals(writes.ranges),
+ "Read ranges (%s) are not the same as write ranges (%s)", reads.ranges, writes.ranges);
}
/**
diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java
index 2d06c4db15..fc3418cc3b 100644
--- a/src/java/org/apache/cassandra/utils/btree/BTree.java
+++ b/src/java/org/apache/cassandra/utils/btree/BTree.java
@@ -1611,6 +1611,11 @@ public class BTree
return new Builder<>(this);
}
+ public int count()
+ {
+ return count;
+ }
+
public Builder setQuickResolver(QuickResolver quickResolver)
{
this.quickResolver = quickResolver;
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordHostReplacementTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordHostReplacementTest.java
index ceebb3933d..e8587ff566 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordHostReplacementTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordHostReplacementTest.java
@@ -32,6 +32,7 @@ import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.RingAwareInJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
@@ -43,6 +44,7 @@ import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.distributed.shared.ClusterUtils.stopUnchecked;
import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.*;
public class AccordHostReplacementTest extends TestBaseImpl
{
@@ -72,9 +74,10 @@ public class AccordHostReplacementTest extends TestBaseImpl
SchemaSpec.optionsBuilder().withTransactionalMode(transactionalModeGen.generate(rng))
.withSpeculativeRetry("ALWAYS"));
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(Generators.int64(0, Math.min(valueGenerators.pkGen().population(), 1000))));
- HistoryBuilder history = historyBuilder(schema, cluster);
+ HistoryBuilder history = historyBuilder(schema, valueGenerators, cluster);
waitForCMSToQuiesce(cluster, cluster.get(1));
for (int i = 0; i < 1000; i++)
@@ -93,13 +96,13 @@ public class AccordHostReplacementTest extends TestBaseImpl
}
}
- private static HistoryBuilder historyBuilder(SchemaSpec schema, Cluster cluster)
+ private static HistoryBuilder historyBuilder(SchemaSpec schema, IndexedValueGenerators valueGenerators, Cluster cluster)
{
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
hb -> RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(new TokenPlacementModel.SimpleReplicationFactor(3))
.consistencyLevel(ConsistencyLevel.ALL)
- .build(schema, hb, cluster));
+ .build(schema, hb.valueGenerators(), cluster));
history.customThrowing(() -> cluster.schemaChange(schema.compile()), "Setup");
return history;
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/AlterTopologyTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/AlterTopologyTest.java
index 6d05fc9817..b2114a4cd1 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/AlterTopologyTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/AlterTopologyTest.java
@@ -34,6 +34,7 @@ import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.exceptions.ExceptionCode;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
@@ -54,6 +55,7 @@ import static java.time.Duration.ofSeconds;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
import static org.junit.Assert.assertEquals;
public class AlterTopologyTest extends FuzzTestBase
@@ -75,13 +77,13 @@ public class AlterTopologyTest extends FuzzTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(Generators.int64(0, Math.min(valueGenerators.pkGen().population(), 1000))));
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
(hb) -> InJvmDTestVisitExecutor.builder()
.nodeSelector(i -> 1)
- .build(schema, hb, cluster));
+ .build(schema, valueGenerators, cluster));
history.custom(() -> {
cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE +
" WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1' : 3 };");
@@ -92,7 +94,10 @@ public class AlterTopologyTest extends FuzzTestBase
Runnable writeAndValidate = () -> {
for (int i = 0; i < 2000; i++)
- history.insert(pkGen.generate(rng), ckGen.generate(rng));
+ {
+ int pdIdx = pkGen.generate(rng);
+ history.insert(pdIdx, valueGenerators.forPdIdx(pdIdx).ckIdxGen().generate(rng));
+ }
for (int pk : pkGen.generated())
history.selectPartition(pk);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/BounceGossipTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/BounceGossipTest.java
index ed6faf26aa..100dd85627 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/BounceGossipTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/BounceGossipTest.java
@@ -40,7 +40,7 @@ import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
-import org.apache.cassandra.harry.dsl.HistoryBuilderHelper;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.execution.CQLVisitExecutor;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
@@ -53,6 +53,7 @@ import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
import static org.junit.Assert.fail;
import static org.psjava.util.AssertStatus.assertTrue;
@@ -75,8 +76,9 @@ public class BounceGossipTest extends TestBaseImpl
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
+
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(valueGenerators.pkIdxGen()));
cluster.schemaChange("CREATE KEYSPACE " + schema.keyspace +
" WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};");
@@ -84,12 +86,12 @@ public class BounceGossipTest extends TestBaseImpl
Future> f = es.submit(new Runnable()
{
- final HistoryBuilder history = new HistoryBuilder(schema.valueGenerators);
+ final HistoryBuilder history = new HistoryBuilder(valueGenerators);
final Iterator iterator = history.iterator();
final CQLVisitExecutor executor = InJvmDTestVisitExecutor.builder()
.nodeSelector(lts -> 1)
.retryPolicy(InJvmDTestVisitExecutor.RetryPolicy.RETRY_ON_TIMEOUT)
- .build(schema, history, cluster);
+ .build(schema, history.valueGenerators(), cluster);
@Override
public void run()
{
@@ -97,7 +99,7 @@ public class BounceGossipTest extends TestBaseImpl
{
// Rate limit to ~10 per second
LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(500));
- HistoryBuilderHelper.insertRandomData(schema, pkGen, ckGen, rng, history);
+ history.insert();
history.selectPartition(pkGen.generate(rng));
while (iterator.hasNext() && !stop.get())
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTest.java
index 9246a7df00..727848fca7 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/CoordinatorPathTest.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.distributed.test.log;
-
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Random;
@@ -38,11 +37,13 @@ import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.ValueGeneratorHelper;
import org.apache.cassandra.harry.cql.WriteHelper;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.execution.CompiledStatement;
import org.apache.cassandra.harry.gen.Generator;
import org.apache.cassandra.harry.gen.SchemaGenerators;
import org.apache.cassandra.harry.model.TokenPlacementModel;
import org.apache.cassandra.harry.model.TokenPlacementModel.Replica;
+import org.apache.cassandra.harry.op.Kind;
import org.apache.cassandra.harry.op.Operations;
import org.apache.cassandra.harry.util.ByteUtils;
import org.apache.cassandra.harry.util.TokenUtil;
@@ -77,12 +78,12 @@ public class CoordinatorPathTest extends CoordinatorPathTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
+ IndexedValueGenerators valueGenerators = HistoryBuilder.valueGenerators(schema, rng.next(), 1000);
cluster.schemaChange("CREATE KEYSPACE " + schema.keyspace +
" WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};");
cluster.schemaChange(schema.compile());
- HistoryBuilder.IndexedValueGenerators valueGenerators = (HistoryBuilder.IndexedValueGenerators) schema.valueGenerators;
- for (int i = 0; i < valueGenerators.pkPopulation(); i++)
+ for (int i = 0; i < valueGenerators.pkGen().population(); i++)
{
long pd = valueGenerators.pkGen().descriptorAt(i);
@@ -102,12 +103,12 @@ public class CoordinatorPathTest extends CoordinatorPathTestBase
long lts = 1L;
Future> writeQuery = async(() -> {
-
CompiledStatement s = WriteHelper.inflateInsert(new Operations.WriteOp(lts, pd, 0,
- ValueGeneratorHelper.randomDescriptors(rng, valueGenerators::regularColumnGen, valueGenerators.regularColumnCount()),
- ValueGeneratorHelper.randomDescriptors(rng, valueGenerators::staticColumnGen, valueGenerators.staticColumnCount()),
- Operations.Kind.INSERT),
+ ValueGeneratorHelper.randomDescriptors(rng, valueGenerators.forPd(pd)::regularColumnGen, valueGenerators.forPd(pd).regularColumnCount()),
+ ValueGeneratorHelper.randomDescriptors(rng, valueGenerators.forPd(pd)::staticColumnGen, valueGenerators.forPd(pd).staticColumnCount()),
+ Kind.INSERT),
schema,
+ valueGenerators,
lts);
cluster.coordinator(1).execute(s.cql(), ConsistencyLevel.QUORUM, s.bindings());
return null;
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/FailedLeaveTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/FailedLeaveTest.java
index d64a1bdff9..d9a780254b 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/FailedLeaveTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/FailedLeaveTest.java
@@ -41,6 +41,7 @@ import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.shared.ClusterUtils.SerializableBiPredicate;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
@@ -60,6 +61,7 @@ import static org.apache.cassandra.distributed.shared.ClusterUtils.decommission;
import static org.apache.cassandra.distributed.shared.ClusterUtils.getClusterMetadataVersion;
import static org.apache.cassandra.distributed.shared.ClusterUtils.getSequenceAfterCommit;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
public class FailedLeaveTest extends FuzzTestBase
{
@@ -98,13 +100,13 @@ public class FailedLeaveTest extends FuzzTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(Generators.int64(0, Math.min(valueGenerators.pkGen().population(), 1000))));
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
(hb) -> InJvmDTestVisitExecutor.builder()
.nodeSelector(i -> 1)
- .build(schema, hb, cluster));
+ .build(schema, hb.valueGenerators(), cluster));
history.custom(() -> {
cluster.schemaChange("CREATE KEYSPACE " + schema.keyspace +
" WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 2};");
@@ -113,7 +115,10 @@ public class FailedLeaveTest extends FuzzTestBase
Runnable writeAndValidate = () -> {
for (int i = 0; i < WRITES; i++)
- history.insert(pkGen.generate(rng), ckGen.generate(rng));
+ {
+ int pkIdx = pkGen.generate(rng);
+ history.insert(pkIdx, valueGenerators.forPdIdx(pkIdx).ckIdxGen().generate(rng));
+ }
for (int pk : pkGen.generated())
history.selectPartition(pk);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/ResumableStartupTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/ResumableStartupTest.java
index 07ec054d54..b97c1fae6b 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/log/ResumableStartupTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/log/ResumableStartupTest.java
@@ -38,6 +38,7 @@ import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.execution.DataTracker;
import org.apache.cassandra.harry.execution.RingAwareInJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
@@ -79,10 +80,10 @@ public class ResumableStartupTest extends FuzzTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = HistoryBuilder.valueGenerators(schema, rng.next(), 1000);
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(Generators.int64(0, Math.min(valueGenerators.pkGen().population(), 1000))));
- HistoryBuilder history = new HistoryBuilder(schema.valueGenerators);
+ HistoryBuilder history = new HistoryBuilder(valueGenerators);
history.customThrowing(() -> {
cluster.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 2};", schema.keyspace));
cluster.schemaChange(schema.compile());
@@ -91,7 +92,10 @@ public class ResumableStartupTest extends FuzzTestBase
Runnable writeAndValidate = () -> {
for (int i = 0; i < WRITES; i++)
- history.insert(pkGen.generate(rng), ckGen.generate(rng));
+ {
+ int pdIdx = pkGen.generate(rng);
+ history.insert(pdIdx, history.valueGenerators().forPdIdx(pdIdx).ckIdxGen().generate(rng));
+ }
for (int pk : pkGen.generated())
history.selectPartition(pk);
@@ -100,15 +104,15 @@ public class ResumableStartupTest extends FuzzTestBase
// First write with ONE, as we only have 1 node
TokenPlacementModel.ReplicationFactor rf = new TokenPlacementModel.SimpleReplicationFactor(1);
- DataTracker tracker = new DataTracker.SequentialDataTracker();
- QuiescentChecker checker = new QuiescentChecker(schema.valueGenerators, tracker, history);
+ DataTracker.SequentialDataTracker tracker = new DataTracker.SequentialDataTracker();
+ QuiescentChecker checker = new QuiescentChecker(valueGenerators, tracker);
RingAwareInJvmDTestVisitExecutor executor;
// RF is ONE here since we have no pending nodes
executor = RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(rf)
.consistencyLevel(ConsistencyLevel.ONE)
- .build(schema, tracker, checker, cluster);
+ .build(schema, valueGenerators, tracker, checker, cluster);
Iterator iterator = history.iterator();
while (iterator.hasNext())
executor.execute(iterator.next());
@@ -130,7 +134,7 @@ public class ResumableStartupTest extends FuzzTestBase
executor = RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(rf)
.consistencyLevel(ConsistencyLevel.ONE)
- .build(schema, tracker, checker, cluster);
+ .build(schema, valueGenerators, tracker, checker, cluster);
while (iterator.hasNext())
executor.execute(iterator.next());
diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeHarryTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeHarryTest.java
index 64810654d5..486adcf800 100644
--- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeHarryTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeHarryTest.java
@@ -89,9 +89,7 @@ public class ClusterMetadataUpgradeHarryTest extends UpgradeTestBase
.upgradesToCurrentFrom(v41)
.withUpgradeListener(listener)
.setup((cluster) -> {
- SchemaSpec schema = new SchemaSpec(rng.next(),
- 10_000,
- "harry", "test_table",
+ SchemaSpec schema = new SchemaSpec("harry", "test_table",
asList(pk("pk1", asciiType), pk("pk2", int64Type)),
asList(ck("ck1", asciiType, false), ck("ck2", int64Type, false)),
asList(regularColumn("regular1", asciiType), regularColumn("regular2", int64Type)),
@@ -99,7 +97,7 @@ public class ClusterMetadataUpgradeHarryTest extends UpgradeTestBase
cluster.schemaChange(String.format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': %d};", schema.keyspace, 3));
cluster.schemaChange(schema.compile());
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(HistoryBuilder.valueGenerators(schema, rng.next()),
hb -> InJvmDTestVisitExecutor.builder()
.retryPolicy(retry -> true)
.nodeSelector(lts -> {
@@ -112,9 +110,9 @@ public class ClusterMetadataUpgradeHarryTest extends UpgradeTestBase
}
})
.consistencyLevel(ConsistencyLevel.QUORUM)
- .build(schema, hb, cluster));
+ .build(schema, hb.valueGenerators(), cluster));
- Generator pkIdxGen = Generators.int32(0, Math.min(10_000, schema.valueGenerators.ckPopulation()));
+ Generator pkIdxGen = Generators.adaptLongToInt(Generators.int64(0, Math.min(10_000, history.valueGenerators().pkGen().population())));
executor.set(executorFactory().infiniteLoop("R/W Worload",
() -> {
diff --git a/test/distributed/org/apache/cassandra/fuzz/harry/examples/RangeTombstoneBurnTest.java b/test/distributed/org/apache/cassandra/fuzz/harry/examples/RangeTombstoneBurnTest.java
index 8607c2551d..ad432fd2bd 100644
--- a/test/distributed/org/apache/cassandra/fuzz/harry/examples/RangeTombstoneBurnTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/harry/examples/RangeTombstoneBurnTest.java
@@ -33,10 +33,14 @@ import org.apache.cassandra.harry.gen.SchemaGenerators;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
+
public class RangeTombstoneBurnTest extends IntegrationTestBase
{
private final int ITERATIONS = 10;
private final int STEPS_PER_ITERATION = 1000;
+ private final int POPULATION = 1000;
@Test
public void rangeTombstoneBurnTest()
@@ -47,8 +51,9 @@ public class RangeTombstoneBurnTest extends IntegrationTestBase
cluster.get(1).nodetool("disableautocompaction");
cluster.schemaChange(schema.compile());
- int perIteration = Math.min(10, schema.valueGenerators.pkPopulation());;
- int maxPartitions = Math.max(perIteration, schema.valueGenerators.pkPopulation());
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), POPULATION);
+ int perIteration = 10;
+ int maxPartitions = Math.toIntExact(Math.max(perIteration, valueGenerators.pkGen().population()));
for (int iteration = 0; iteration < ITERATIONS; iteration++)
{
@@ -61,13 +66,13 @@ public class RangeTombstoneBurnTest extends IntegrationTestBase
float deleteColumnsChance = rng.nextFloat(0.95f, 1.0f);
float deleteRangeChance = rng.nextFloat(0.95f, 1.0f);
float flushChance = rng.nextFloat(0.999f, 1.0f);
- int maxPartitionSize = Math.min(rng.nextInt(1, 1 << rng.nextInt(5, 11)), schema.valueGenerators.ckPopulation());
+ int maxPartitionSize = Math.min(rng.nextInt(1, 1 << rng.nextInt(5, 11)), POPULATION);
Generator partitionPicker = Generators.pick(partitions);
Generator rowPicker = Generators.int32(0, maxPartitionSize);
ModelChecker model = new ModelChecker<>();
- ReplayingHistoryBuilder historyBuilder = new ReplayingHistoryBuilder(schema.valueGenerators,
- (hb) -> InJvmDTestVisitExecutor.builder().build(schema, hb, cluster));
+ ReplayingHistoryBuilder historyBuilder = new ReplayingHistoryBuilder(valueGenerators,
+ (hb) -> InJvmDTestVisitExecutor.builder().build(schema, hb.valueGenerators(), cluster));
model.init(historyBuilder)
.step((history, rng_) -> {
diff --git a/test/distributed/org/apache/cassandra/fuzz/harry/examples/RepairBurnTest.java b/test/distributed/org/apache/cassandra/fuzz/harry/examples/RepairBurnTest.java
index 65b1bf4441..e37f13a497 100644
--- a/test/distributed/org/apache/cassandra/fuzz/harry/examples/RepairBurnTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/harry/examples/RepairBurnTest.java
@@ -26,7 +26,7 @@ import org.apache.cassandra.distributed.test.IntegrationTestBase;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.checker.ModelChecker;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
-import org.apache.cassandra.harry.dsl.HistoryBuilderHelper;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
import org.apache.cassandra.harry.gen.Generators;
@@ -47,19 +47,19 @@ public class RepairBurnTest extends IntegrationTestBase
public void repairBurnTest()
{
int maxPartitionSize = 10;
- int partitions = 1000;
Generator schemaGen = SchemaGenerators.schemaSpecGen(KEYSPACE, "repair_burn", 1000);
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
+ IndexedValueGenerators valueGenerators = HistoryBuilder.valueGenerators(schema, rng.next(), 1000);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), partitions)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), maxPartitionSize));
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(valueGenerators.pkIdxGen()));
+ Generator ckGen = Generators.int32(0, maxPartitionSize);
ModelChecker modelChecker = new ModelChecker<>();
- modelChecker.init(new HistoryBuilder(schema.valueGenerators))
- .step((history, rng_) -> HistoryBuilderHelper.insertRandomData(schema, pkGen, ckGen, rng, history))
+ modelChecker.init(new HistoryBuilder(valueGenerators))
+ .step((history, rng_) -> history.insert(pkGen.generate(rng)))
.step((history, rng_) -> history.deleteRow(pkGen.generate(rng), ckGen.generate(rng)))
.exitCondition((history) -> {
if (history.size() < 10_000)
@@ -73,7 +73,7 @@ public class RepairBurnTest extends IntegrationTestBase
cluster.schemaChange(schema.compile());
- InJvmDTestVisitExecutor.replay(InJvmDTestVisitExecutor.builder().build(schema, history, cluster),
+ InJvmDTestVisitExecutor.replay(InJvmDTestVisitExecutor.builder().build(schema, history.valueGenerators(), cluster),
history);
return true;
diff --git a/test/distributed/org/apache/cassandra/fuzz/harry/sstable/LevelledSSTableGeneratorTest.java b/test/distributed/org/apache/cassandra/fuzz/harry/sstable/LevelledSSTableGeneratorTest.java
new file mode 100644
index 0000000000..ef8a6f93c5
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/fuzz/harry/sstable/LevelledSSTableGeneratorTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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.fuzz.harry.sstable;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.harry.SchemaSpec;
+import org.apache.cassandra.harry.execution.DataTracker;
+import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
+import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
+import org.apache.cassandra.harry.gen.Generators;
+import org.apache.cassandra.harry.model.QuiescentChecker;
+import org.apache.cassandra.harry.op.ClusteringOrderBy;
+import org.apache.cassandra.harry.op.Operations;
+import org.apache.cassandra.harry.op.Visit;
+import org.apache.cassandra.harry.stress.ActivePartition;
+import org.apache.cassandra.harry.stress.LevelledSStableGenerator;
+import org.apache.cassandra.harry.stress.RotationStrategy;
+import org.apache.cassandra.harry.stress.config.StressSchemaConfig;
+import org.apache.cassandra.harry.stress.TokenIndex;
+import org.apache.cassandra.harry.stress.TokenIndexGenerator;
+import org.apache.cassandra.harry.stress.VisitGenerator;
+import org.apache.cassandra.harry.stress.distribution.Distribution;
+import org.apache.cassandra.harry.stress.distribution.Distributions;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+
+public class LevelledSSTableGeneratorTest extends TestBaseImpl
+{
+ private static final Logger logger = LoggerFactory.getLogger(LevelledSSTableGeneratorTest.class);
+
+ private static final String SCHEMA_CONFIG = "test/resources/harry/stress/levelled-sstable-schema.yaml";
+
+ private static final long INITIAL_LTS = 1;
+ private static final long VISITS = 100_000; // total writes; with visitSize=1, one op per LTS
+ private static final long END_LTS = INITIAL_LTS + VISITS; // first LTS past the written history; reads use it
+ private static final int SSTABLE_SIZE_MIB = 1;
+ private static final int[] LEVEL_WEIGHTS = { 1, 2, 4, 8, 16 }; // index == LCS level; weights spread writes across levels
+
+ // Initialize static state for offline tool usage, since we are generating SSTables on the main class loader rather
+ // than in-jvm dtest nodes
+ public static void initForOfflineTool()
+ {
+ DatabaseDescriptor.toolInitialization(false);
+ DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+ ClusterMetadataService.initializeForClients();
+ }
+
+ @Test
+ public void generateAndValidateLevelledSSTables() throws Throwable
+ {
+ initForOfflineTool();
+
+ StressSchemaConfig config = StressSchemaConfig.load(Paths.get(SCHEMA_CONFIG));
+ SchemaSpec schema = config.schema();
+
+ Distribution visitSize = Distributions.fixed(1);
+ VisitGenerator.OpKindGenFactory opKindGen = new VisitGenerator.RandomOpKindGenFactory();
+
+ try (Cluster cluster = init(Cluster.build(1).start(), 1))
+ {
+ cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + schema.keyspace +
+ " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}");
+ cluster.schemaChange(schema.compile());
+ cluster.get(1).nodetool("disableautocompaction", schema.keyspace, schema.table);
+
+
+ // Build TokenIndex
+ File tokenDir = new File(Files.createTempDirectory("harry-token-index"));
+ TokenIndexGenerator.generate(tokenDir.toJavaIOFile(), schema, config.rotationStrategy(),
+ config.rowPopulation(), Generators.constant(VisitGenerator.VisitType.MUTATE),
+ visitSize, INITIAL_LTS, VISITS);
+ TokenIndex tokenIndex = new TokenIndex(new File(tokenDir, "merged_tokens"),
+ new File(tokenDir, "merged_tokens.idx"));
+
+ // Generate levelled SSTables
+ File sstableDir = new File(Files.createDirectories(Files.createTempDirectory("harry-levelled-sstables")
+ .resolve(schema.keyspace)
+ .resolve(schema.table + '-' + "0".repeat(32))));
+ LevelledSStableGenerator generator =
+ new LevelledSStableGenerator(schema, config.rowPopulation(), config.columnPopulation(), visitSize, opKindGen,
+ false, SSTABLE_SIZE_MIB,
+ new LevelledSStableGenerator.SSTableLevelPicker(LEVEL_WEIGHTS),
+ tokenIndex, sstableDir);
+ generator.generate(Long.MIN_VALUE, Long.MAX_VALUE);
+
+ // Import SSTables; -l to keeps generated levels
+ cluster.get(1).nodetoolResult("import", "-l", schema.keyspace, schema.table, sstableDir.absolutePath())
+ .asserts().success();
+
+ TestOracle model = replay(schema, config, tokenIndex, visitSize, opKindGen);
+ tokenIndex.close();
+
+ // Validate
+ InJvmDTestVisitExecutor validator = new InJvmDTestVisitExecutor(schema, model.partitions, new DataTracker.NoOpDataTracker(),
+ new QuiescentChecker(model.partitions, model.tracker), cluster,
+ lts -> 1,
+ v -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING,
+ InJvmDTestVisitExecutor.RetryPolicy.NO_RETRY,
+ v -> ConsistencyLevel.NODE_LOCAL,
+ QueryBuildingVisitExecutor.WrapQueries.EMPTY);
+ for (long pd : model.generatedPds)
+ validator.execute(new Visit(END_LTS, new Operations.Operation[]{ new Operations.SelectPartition(END_LTS, pd, ClusteringOrderBy.ASC) }));
+ logger.info("Validated {} partitions", model.generatedPds.size());
+ }
+ }
+
+ private static TestOracle replay(SchemaSpec schema, StressSchemaConfig config, TokenIndex tokenIndex,
+ Distribution visitSize, VisitGenerator.OpKindGenFactory opKindGen)
+ {
+ RotationStrategy rotation = config.rotationStrategy();
+ ActivePartition.Partitions partitions = new ActivePartition.Partitions(schema, config.rowPopulation(), config.columnPopulation(),
+ rotation, 0, rotation.targetSize(), INITIAL_LTS);
+ partitions.populate();
+ DataTracker.SimpleDataTracker modelTracker = new DataTracker.SimpleDataTracker();
+ Set generatedPds = new TreeSet<>();
+ TokenIndex.EntryIterator iter = tokenIndex.range(Long.MIN_VALUE, Long.MAX_VALUE);
+ while (iter.hasNext())
+ {
+ long pd = iter.pd();
+ generatedPds.add(pd);
+ for (long lts : iter.readLts())
+ {
+ Visit visit = VisitGenerator.mutatingVisit(lts, visitSize, opKindGen, rng -> partitions.forPd(pd));
+ modelTracker.begin(visit);
+ modelTracker.end(visit);
+ }
+ iter.advance();
+ }
+ return new TestOracle(partitions, modelTracker, generatedPds);
+ }
+
+ private static final class TestOracle
+ {
+ final ActivePartition.Partitions partitions;
+ final DataTracker.SimpleDataTracker tracker;
+ final Set generatedPds;
+
+ TestOracle(ActivePartition.Partitions partitions, DataTracker.SimpleDataTracker tracker, Set generatedPds)
+ {
+ this.partitions = partitions;
+ this.tracker = tracker;
+ this.generatedPds = generatedPds;
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/fuzz/harry/sstable/RangeAwareSSTableLoadAndStressTest.java b/test/distributed/org/apache/cassandra/fuzz/harry/sstable/RangeAwareSSTableLoadAndStressTest.java
new file mode 100644
index 0000000000..41b45250ac
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/fuzz/harry/sstable/RangeAwareSSTableLoadAndStressTest.java
@@ -0,0 +1,205 @@
+/*
+ * 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.fuzz.harry.sstable;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.harry.SchemaSpec;
+import org.apache.cassandra.harry.gen.Generator;
+import org.apache.cassandra.harry.gen.Generators;
+import org.apache.cassandra.harry.stress.HarryStress;
+import org.apache.cassandra.harry.stress.LevelledSStableGenerator;
+import org.apache.cassandra.harry.stress.RotationStrategy;
+import org.apache.cassandra.harry.stress.config.StressSchemaConfig;
+import org.apache.cassandra.harry.stress.TokenIndex;
+import org.apache.cassandra.harry.stress.TokenIndexGenerator;
+import org.apache.cassandra.harry.stress.VisitGenerator;
+import org.apache.cassandra.harry.stress.distribution.Distribution;
+import org.apache.cassandra.harry.stress.distribution.Distributions;
+import org.apache.cassandra.io.util.File;
+
+public class RangeAwareSSTableLoadAndStressTest extends TestBaseImpl
+{
+ private static final Logger logger = LoggerFactory.getLogger(RangeAwareSSTableLoadAndStressTest.class);
+
+ private static final String SCHEMA_CONFIG = "test/resources/harry/stress/levelled-sstable-schema.yaml";
+
+ private static final int NUM_NODES = 4;
+ private static final int RF = 3;
+
+ private static final long INITIAL_LTS = 1;
+ private static final long VISITS = 100_000;
+ private static final long END_LTS = INITIAL_LTS + VISITS;
+ private static final long STRESS_VISITS = 16_000;
+ private static final int CONCURRENCY = 2;
+ private static final int RATE_PER_SECOND = 20_000;
+ private static final int SSTABLE_SIZE_MIB = 1;
+ private static final int[] LEVEL_WEIGHTS = { 1, 2, 4, 8, 16 };
+
+ @Test
+ public void rangeAwareLoadAndStress() throws Throwable
+ {
+ LevelledSSTableGeneratorTest.initForOfflineTool();
+
+ StressSchemaConfig config = StressSchemaConfig.load(Paths.get(SCHEMA_CONFIG));
+ SchemaSpec schema = config.schema();
+ RotationStrategy rotation = config.rotationStrategy();
+
+ Distribution visitSize = Distributions.fixed(1);
+ VisitGenerator.OpKindGenFactory opKindGen = new VisitGenerator.RandomOpKindGenFactory();
+ Generator visitTypeGen = Generators.constant(VisitGenerator.VisitType.MUTATE);
+
+ try (Cluster cluster = init(Cluster.build(NUM_NODES)
+ .withConfig(c -> c.set("num_tokens", 1))
+ .start(), RF))
+ {
+ cluster.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + schema.keyspace +
+ " WITH replication = {'class': 'NetworkTopologyStrategy', 'datacenter0': " + RF + "}");
+ cluster.schemaChange(schema.compile());
+ cluster.forEach(n -> n.nodetool("disableautocompaction", schema.keyspace, schema.table));
+
+ List ranges = getTopology(cluster, schema);
+ for (RangeReplicas r : ranges)
+ System.out.println(String.format(" range (%d, %d] -> nodes %s", r.startToken, r.endToken, java.util.Arrays.toString(r.replicas)));
+ logger.info("data_placements returned {} ranges", ranges.size());
+
+ File tokenDir = new File(Files.createTempDirectory("harry-token-index"));
+ TokenIndexGenerator.generate(tokenDir.toJavaIOFile(), schema, rotation,
+ config.rowPopulation(), visitTypeGen,
+ visitSize, INITIAL_LTS, VISITS);
+ TokenIndex tokenIndex = new TokenIndex(new File(tokenDir, "merged_tokens"),
+ new File(tokenDir, "merged_tokens.idx"));
+
+ // Ror each range, generate levelled SSTables and import into replicas
+ for (RangeReplicas range : ranges)
+ {
+ File sstableDir = new File(Files.createDirectories(Files.createTempDirectory("range-aware-sstables")
+ .resolve(String.format("%d-%d", range.startToken, range.endToken))
+ .resolve(schema.keyspace)
+ .resolve(schema.table + '-' + "0".repeat(32))));
+ LevelledSStableGenerator generator = new LevelledSStableGenerator(schema, config.rowPopulation(), config.columnPopulation(), visitSize, opKindGen,
+ false, SSTABLE_SIZE_MIB,
+ new LevelledSStableGenerator.SSTableLevelPicker(LEVEL_WEIGHTS),
+ tokenIndex, sstableDir);
+ // Handle wraparound ranges
+ if (range.startToken <= range.endToken)
+ {
+ generator.generate(range.startToken, range.endToken);
+ }
+ else
+ {
+ generator.generate(Long.MIN_VALUE, range.endToken);
+ generator.generate(range.startToken, Long.MAX_VALUE);
+ }
+
+ for (int node : range.replicas)
+ cluster.get(node).nodetoolResult("import", "-cd", "-t", schema.keyspace, schema.table, sstableDir.absolutePath())
+ .asserts().success();
+ }
+ logger.info("Generated and imported SSTables for all ranges");
+ tokenIndex.close();
+
+ // Continue HarryStress live from where SSTableGenerator has finished
+ HarryStress stress = new HarryStress(schema,
+ config.rowPopulation(),
+ config.columnPopulation(),
+ visitTypeGen,
+ visitSize,
+ opKindGen,
+ config.rotationStrategy(),
+ /*metricsOut=*/ null,
+ /*reportIntervalSeconds=*/ 30,
+ () -> (statement, run) -> {
+ Object[][] rs = cluster.coordinator(1).execute(statement.cql(),
+ ConsistencyLevel.QUORUM,
+ statement.bindings());
+ System.out.println("rs = " + Arrays.toString(rs));
+ if (run != null)
+ run.run();
+ return rs;
+ },
+ CONCURRENCY,
+ RATE_PER_SECOND,
+ /*minPartitionIdx=*/ 0,
+ /*maxPartitionIdx=*/ Long.MAX_VALUE,
+ /*initialLts=*/ END_LTS);
+
+ stress.replay(INITIAL_LTS, END_LTS);
+ stress.start(END_LTS + STRESS_VISITS, Long.MAX_VALUE);
+ logger.info("HarryStress completed an additional {} visits past the loaded SSTable history", STRESS_VISITS);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List getTopology(Cluster cluster, SchemaSpec schema)
+ {
+ Object[][] rows = cluster.coordinator(1).execute(
+ "SELECT range_start, range_end, write_replicas FROM system_views.data_placements " +
+ "WHERE keyspace_name = ? AND table_name = ?",
+ ConsistencyLevel.ONE, schema.keyspace, schema.table);
+ List result = new ArrayList<>();
+
+ for (Object[] row : rows)
+ {
+ long startToken = Long.parseLong((String) row[0]);
+ long endToken = Long.parseLong((String) row[1]);
+ int[] replicas = toIdx((Set) row[2]);
+ result.add(new RangeReplicas(startToken, endToken, replicas));
+ }
+ return result;
+ }
+
+ private static int[] toIdx(Set nodeIds)
+ {
+ // In dtest clusters, TCM node IDs are assigned 1..N matching the instance indices
+ int[] idxs = new int[nodeIds.size()];
+ int i = 0;
+ for (int nodeId : nodeIds)
+ idxs[i++] = nodeId;
+ return idxs;
+ }
+
+ // A token range and its write-replica set
+ private static final class RangeReplicas
+ {
+ final long startToken;
+ final long endToken;
+ final int[] replicas;
+
+ RangeReplicas(long startToken, long endToken, int[] replicas)
+ {
+ this.startToken = startToken;
+ this.endToken = endToken;
+ this.replicas = replicas;
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/fuzz/harry/stress/HarryStressValidationSmokeTest.java b/test/distributed/org/apache/cassandra/fuzz/harry/stress/HarryStressValidationSmokeTest.java
new file mode 100644
index 0000000000..f44b142a54
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/fuzz/harry/stress/HarryStressValidationSmokeTest.java
@@ -0,0 +1,103 @@
+/*
+ * 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.fuzz.harry.stress;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.harry.SchemaSpec;
+import org.apache.cassandra.harry.gen.Generators;
+import org.apache.cassandra.harry.stress.HarryStress;
+import org.apache.cassandra.harry.stress.RotationStrategy;
+import org.apache.cassandra.harry.stress.VisitGenerator;
+import org.apache.cassandra.harry.stress.distribution.Distributions;
+
+import static java.util.Arrays.asList;
+import static org.apache.cassandra.harry.ColumnSpec.asciiType;
+import static org.apache.cassandra.harry.ColumnSpec.ck;
+import static org.apache.cassandra.harry.ColumnSpec.int64Type;
+import static org.apache.cassandra.harry.ColumnSpec.pk;
+import static org.apache.cassandra.harry.ColumnSpec.regularColumn;
+
+public class HarryStressValidationSmokeTest extends TestBaseImpl
+{
+ private static final Logger logger = LoggerFactory.getLogger(HarryStressValidationSmokeTest.class);
+
+ private static final String TABLE = "stress_smoke";
+
+ private static final long ITERATIONS = 100_000;
+ private static final int NUM_PARTITIONS = 100;
+ private static final int REPLACE_WITH_NEW = 10;
+ private static final int ROW_POPULATION = 100;
+ private static final int COLUMN_POPULATION = 100;
+ private static final int CONCURRENCY = 2;
+ private static final int RATE_PER_SECOND = 20_000;
+
+ @Test
+ public void validatingStressSmokeTest() throws Throwable
+ {
+ try (Cluster cluster = init(Cluster.build(1).start(), 1))
+ {
+ SchemaSpec schema = schema();
+ cluster.schemaChange(schema.compile());
+
+ HarryStress stress =
+ new HarryStress(schema,
+ Distributions.fixed(ROW_POPULATION),
+ column -> Distributions.fixed(COLUMN_POPULATION),
+ Generators.pick(VisitGenerator.VisitType.values()),
+ Distributions.fixed(1),
+ new VisitGenerator.RandomOpKindGenFactory(),
+ new RotationStrategy.FixedRotationStrategy(NUM_PARTITIONS, REPLACE_WITH_NEW, 0),
+ null,
+ 30,
+ () -> (statement, onComplete) -> {
+ Object[][] rows = cluster.coordinator(1).execute(statement.cql(),
+ ConsistencyLevel.QUORUM,
+ statement.bindings());
+ if (onComplete != null)
+ onComplete.run();
+ return rows;
+ },
+ CONCURRENCY,
+ RATE_PER_SECOND,
+ 0,
+ Long.MAX_VALUE,
+ 0);
+
+ stress.start(ITERATIONS, Long.MAX_VALUE);
+ logger.info("Validating stress smoke test ran ~{} visits across {} partitions with no validation failures",
+ ITERATIONS, NUM_PARTITIONS);
+ }
+ }
+
+ private static SchemaSpec schema()
+ {
+ return new SchemaSpec(KEYSPACE, TABLE,
+ asList(pk("pk1", asciiType), pk("pk2", int64Type)),
+ asList(ck("ck1", asciiType, false), ck("ck2", int64Type, false)),
+ asList(regularColumn("v1", asciiType), regularColumn("v2", int64Type), regularColumn("v3", asciiType)),
+ asList(),
+ SchemaSpec.optionsBuilder().ifNotExists(true).addWriteTimestamps(true));
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java
index 08af4f4d4b..465a848c7f 100644
--- a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java
@@ -39,7 +39,7 @@ import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.distributed.test.log.FuzzTestBase;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
-import org.apache.cassandra.harry.dsl.HistoryBuilderHelper;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.DataTracker;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
@@ -59,6 +59,7 @@ import static org.apache.cassandra.distributed.shared.ClusterUtils.pauseBeforeCo
import static org.apache.cassandra.distributed.shared.ClusterUtils.unpauseCommits;
import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
public class ConsistentBootstrapTest extends FuzzTestBase
{
@@ -83,13 +84,16 @@ public class ConsistentBootstrapTest extends FuzzTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(valueGenerators.pkIdxGen()));
- HistoryBuilder history = new HistoryBuilder(schema.valueGenerators);
+ HistoryBuilder history = new HistoryBuilder(valueGenerators);
Runnable writeAndValidate = () -> {
for (int i = 0; i < WRITES; i++)
- HistoryBuilderHelper.insertRandomData(schema, pkGen, ckGen, rng, history);
+ {
+ int pkIdx = pkGen.generate(rng);
+ history.insert(pkIdx, valueGenerators.forPdIdx(pkIdx).ckIdxGen().generate(rng));
+ }
for (int pk : pkGen.generated())
history.selectPartition(pk);
@@ -134,7 +138,7 @@ public class ConsistentBootstrapTest extends FuzzTestBase
RingAwareInJvmDTestVisitExecutor.replay(RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(new TokenPlacementModel.SimpleReplicationFactor(2))
.consistencyLevel(ConsistencyLevel.ALL)
- .build(schema, history, cluster),
+ .build(schema, history.valueGenerators(), cluster),
history);
});
}
@@ -178,13 +182,13 @@ public class ConsistentBootstrapTest extends FuzzTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
hb -> RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(new TokenPlacementModel.SimpleReplicationFactor(3))
.consistencyLevel(ConsistencyLevel.ALL)
- .build(schema, hb, cluster));
+ .build(schema, hb.valueGenerators(), cluster));
cluster.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};", KEYSPACE));
cluster.schemaChange(schema.compile());
@@ -198,15 +202,16 @@ public class ConsistentBootstrapTest extends FuzzTestBase
.drop()
.on();
- DataTracker tracker = new DataTracker.SequentialDataTracker();
+ DataTracker.SequentialDataTracker tracker = new DataTracker.SequentialDataTracker();
RingAwareInJvmDTestVisitExecutor executor = RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(new TokenPlacementModel.SimpleReplicationFactor(2))
.nodeSelector(i -> 2)
.consistencyLevel(ConsistencyLevel.ALL)
.retryPolicy(InJvmDTestVisitExecutor.RetryPolicy.NO_RETRY)
.build(schema,
+ history.valueGenerators(),
tracker,
- new QuiescentChecker(schema.valueGenerators, tracker, history),
+ new QuiescentChecker(valueGenerators, tracker),
cluster);
// Prime the CMS node to pause before the finish join event is committed
@@ -231,7 +236,7 @@ public class ConsistentBootstrapTest extends FuzzTestBase
boolean triggered = false;
outer:
- for (int i = 0; i < history.valueGenerators().pkPopulation(); i++)
+ for (int i = 0; i < history.valueGenerators().pkGen().population(); i++)
{
long pd = history.valueGenerators().pkGen().descriptorAt(i);
for (TokenPlacementModel.Replica replica : executor.getReplicasFor(pd))
@@ -240,7 +245,7 @@ public class ConsistentBootstrapTest extends FuzzTestBase
{
try
{
- HistoryBuilderHelper.insertRandomData(schema, i, ckGen.generate(rng), rng, history);
+ history.insert(i, valueGenerators.forPdIdx(i).ckIdxGen().generate(rng));
}
catch (Throwable t)
{
diff --git a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentLeaveTest.java b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentLeaveTest.java
index bb9fd4867c..d6803e9599 100644
--- a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentLeaveTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentLeaveTest.java
@@ -57,6 +57,8 @@ import static org.apache.cassandra.distributed.shared.ClusterUtils.pauseBeforeCo
import static org.apache.cassandra.distributed.shared.ClusterUtils.unpauseCommits;
import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import static org.junit.Assert.assertFalse;
public class ConsistentLeaveTest extends FuzzTestBase
@@ -82,19 +84,22 @@ public class ConsistentLeaveTest extends FuzzTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = HistoryBuilder.valueGenerators(schema, rng.next(), 1000);
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(valueGenerators.pkIdxGen()));
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
(hb) -> RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(new TokenPlacementModel.SimpleReplicationFactor(2))
.consistencyLevel(ConsistencyLevel.ALL)
.retryPolicy(InJvmDTestVisitExecutor.RetryPolicy.RETRY_ON_TIMEOUT)
.nodeSelector(lts -> 1)
- .build(schema, hb, cluster));
+ .build(schema, hb.valueGenerators(), cluster));
Runnable writeAndValidate = () -> {
for (int i = 0; i < WRITES; i++)
- history.insert(pkGen.generate(rng), ckGen.generate(rng));
+ {
+ int pkIdx = pkGen.generate(rng);
+ history.insert(pkIdx, valueGenerators.forPdIdx(pkIdx).ckIdxGen().generate(rng));
+ }
for (int pk : pkGen.generated())
history.selectPartition(pk);
diff --git a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentMoveTest.java b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentMoveTest.java
index 7f4d3ad5df..3e7e597258 100644
--- a/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentMoveTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/ring/ConsistentMoveTest.java
@@ -40,6 +40,7 @@ import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.RingAwareInJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
@@ -83,17 +84,20 @@ public class ConsistentMoveTest extends FuzzTestBase
withRandom(rng -> {
SchemaSpec schema = schemaGen.generate(rng);
- Generators.TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), 1000)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), 1000));
+ IndexedValueGenerators valueGenerators = HistoryBuilder.valueGenerators(schema, rng.next(), 1000);
+ Generators.TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(valueGenerators.pkIdxGen()));
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
(hb) -> RingAwareInJvmDTestVisitExecutor.builder()
.replicationFactor(new TokenPlacementModel.SimpleReplicationFactor(2))
.consistencyLevel(ConsistencyLevel.ALL)
- .build(schema, hb, cluster));
+ .build(schema, hb.valueGenerators(), cluster));
Runnable writeAndValidate = () -> {
for (int i = 0; i < WRITES; i++)
- history.insert(pkGen.generate(rng), ckGen.generate(rng));
+ {
+ int pkIdx = pkGen.generate(rng);
+ history.insert(pkIdx, valueGenerators.forPdIdx(pkIdx).ckIdxGen().generate(rng));
+ }
for (int pk : pkGen.generated())
history.selectPartition(pk);
diff --git a/test/distributed/org/apache/cassandra/fuzz/sai/SingleNodeSAITestBase.java b/test/distributed/org/apache/cassandra/fuzz/sai/SingleNodeSAITestBase.java
index bbe74c34a1..a441651f2b 100644
--- a/test/distributed/org/apache/cassandra/fuzz/sai/SingleNodeSAITestBase.java
+++ b/test/distributed/org/apache/cassandra/fuzz/sai/SingleNodeSAITestBase.java
@@ -18,7 +18,6 @@
package org.apache.cassandra.fuzz.sai;
-
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -43,6 +42,7 @@ import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
import org.apache.cassandra.harry.dsl.HistoryBuilderHelper;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.EntropySource;
@@ -55,6 +55,7 @@ import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
import static org.apache.cassandra.harry.dsl.SingleOperationBuilder.IdxRelation;
// TODO: "WITH OPTIONS = {'case_sensitive': 'false', 'normalize': 'true', 'ascii': 'true'};",
@@ -135,7 +136,7 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
public void simplifiedSaiTest()
{
withRandom(rng -> saiTest(rng,
- SchemaGenerators.trivialSchema(KEYSPACE, "simplified", 1000).generate(rng),
+ SchemaGenerators.trivialSchema(KEYSPACE, "simplified").generate(rng),
() -> true,
DEFAULT_REPAIR_SKIP));
}
@@ -169,9 +170,8 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
private void saiTest(EntropySource rng, SchemaSpec schema, Supplier createIndex, int repairSkip)
{
logger.info(schema.compile());
-
- Generator globalPkGen = Generators.int32(0, Math.min(NUM_PARTITIONS, schema.valueGenerators.pkPopulation()));
- Generator ckGen = Generators.int32(0, schema.valueGenerators.ckPopulation());
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), MAX_PARTITION_SIZE);
+ Generator globalPkGen = Generators.adaptLongToInt(Generators.int64(0, Math.min(NUM_PARTITIONS, valueGenerators.pkGen().population())));
beforeEach();
cluster.forEach(i -> i.nodetool("disableautocompaction"));
@@ -201,11 +201,11 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
CassandraRelevantProperties.SAI_INTERSECTION_CLAUSE_LIMIT.setInt(indexCount.get());
waitForIndexesQueryable(schema);
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
(hb) -> InJvmDTestVisitExecutor.builder()
.pageSizeSelector(pageSizeSelector(rng))
.consistencyLevel(consistencyLevelSelector())
- .doubleWriting(schema, hb, cluster, "debug_table"));
+ .doubleWriting(schema, hb.valueGenerators(), cluster, "debug_table"));
Set partitions = new HashSet<>();
int attempts = 0;
while (partitions.size() < NUM_VISITED_PARTITIONS && attempts < NUM_VISITED_PARTITIONS * 10)
@@ -237,14 +237,15 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
for (int i = 0; i < OPERATIONS_PER_RUN; i++)
{
- int partitionIndex = pkGen.generate(rng);
- HistoryBuilderHelper.insertRandomData(schema, partitionIndex, ckGen.generate(rng), rng, 0.5d, history);
+ int pdIdx = pkGen.generate(rng);
+ IndexedValueGenerators.IndexedPartitionValues partitionValues = valueGenerators.forPdIdx(pdIdx);
+ HistoryBuilderHelper.insertRandomData(schema, pdIdx, partitionValues.ckIdxGen().generate(rng), rng, 0.5d, history);
if (rng.nextFloat() > 0.99f)
{
- int row1 = ckGen.generate(rng);
- int row2 = ckGen.generate(rng);
- history.deleteRowRange(partitionIndex,
+ int row1 = partitionValues.ckIdxGen().generate(rng);
+ int row2 = partitionValues.ckIdxGen().generate(rng);
+ history.deleteRowRange(pdIdx,
Math.min(row1, row2),
Math.max(row1, row2),
rng.nextInt(schema.clusteringKeys.size()),
@@ -253,10 +254,10 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
}
if (rng.nextFloat() > 0.995f)
- HistoryBuilderHelper.deleteRandomColumns(schema, partitionIndex, ckGen.generate(rng), rng, history);
+ HistoryBuilderHelper.deleteRandomColumns(schema, pdIdx, partitionValues.ckIdxGen().generate(rng), rng, history);
if (rng.nextFloat() > 0.9995f)
- history.deletePartition(partitionIndex);
+ history.deletePartition(pdIdx);
if (i % FLUSH_SKIP == 0)
history.custom(() -> flush(schema), "Flush");
@@ -272,13 +273,13 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
List regularRelations =
HistoryBuilderHelper.generateValueRelations(rng,
schema.regularColumns.size(),
- column -> Math.min(schema.valueGenerators.regularPopulation(column), UNIQUE_CELL_VALUES),
+ column -> Math.min(partitionValues.regularPopulation(column), UNIQUE_CELL_VALUES),
eqOnlyRegularColumns::contains);
List staticRelations =
HistoryBuilderHelper.generateValueRelations(rng,
schema.staticColumns.size(),
- column -> Math.min(schema.valueGenerators.staticPopulation(column), UNIQUE_CELL_VALUES),
+ column -> Math.min(partitionValues.staticPopulation(column), UNIQUE_CELL_VALUES),
eqOnlyStaticColumns::contains);
Integer pk = pkGen.generate(rng);
@@ -286,7 +287,7 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
IdxRelation[] ckRelations =
HistoryBuilderHelper.generateClusteringRelations(rng,
schema.clusteringKeys.size(),
- ckGen,
+ partitionValues.ckIdxGen(),
eqOnlyClusteringColumns).toArray(new IdxRelation[0]);
IdxRelation[] regularRelationsArray = regularRelations.toArray(new IdxRelation[regularRelations.size()]);
@@ -353,7 +354,7 @@ public abstract class SingleNodeSAITestBase extends TestBaseImpl
protected InJvmDTestVisitExecutor.ConsistencyLevelSelector consistencyLevelSelector()
{
return visit -> {
- if (visit.selectOnly)
+ if (visit.validating)
return ConsistencyLevel.ALL;
// The goal here is to make replicas as out of date as possible, modulo the efforts of repair
diff --git a/test/distributed/org/apache/cassandra/fuzz/sai/StaticsTortureTest.java b/test/distributed/org/apache/cassandra/fuzz/sai/StaticsTortureTest.java
index 07844b045e..d445fd5da4 100644
--- a/test/distributed/org/apache/cassandra/fuzz/sai/StaticsTortureTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/sai/StaticsTortureTest.java
@@ -36,6 +36,8 @@ import org.apache.cassandra.harry.gen.Generators;
import org.apache.cassandra.harry.gen.rng.JdkRandomEntropySource;
import org.apache.cassandra.harry.util.BitSet;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
import static org.apache.cassandra.harry.dsl.HistoryBuilderHelper.generateClusteringRelations;
import static org.apache.cassandra.harry.dsl.HistoryBuilderHelper.generateValueRelations;
import static org.apache.cassandra.harry.dsl.SingleOperationBuilder.IdxRelation;
@@ -45,7 +47,7 @@ public class StaticsTortureTest extends IntegrationTestBase
private static final int MAX_PARTITION_SIZE = 10_000;
private static final int NUM_PARTITIONS = 100;
private static final int UNIQUE_CELL_VALUES = 5;
-
+ private static final int POPULATION = 1000;
@Test
public void staticsTortureTest()
{
@@ -64,9 +66,7 @@ public class StaticsTortureTest extends IntegrationTestBase
public void staticsTortureTest(List> cks, int idx)
{
- SchemaSpec schema = new SchemaSpec(idx,
- 10_000,
- KEYSPACE,
+ SchemaSpec schema = new SchemaSpec(KEYSPACE,
"tbl" + idx,
Arrays.asList(ColumnSpec.pk("pk1", ColumnSpec.int64Type, Generators.int64()),
ColumnSpec.pk("pk2", ColumnSpec.asciiType, Generators.ascii(4, 1000)),
@@ -124,8 +124,9 @@ public class StaticsTortureTest extends IntegrationTestBase
Generator staticColumnBitSet = Generators.bitSet(schema.staticColumns.size());
EntropySource rng = new JdkRandomEntropySource(1l);
- ReplayingHistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators, hb -> {
- return InJvmDTestVisitExecutor.builder().pageSizeSelector(i -> rng.nextInt(1, 10)).build(schema, hb, cluster);
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), POPULATION);
+ ReplayingHistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators, hb -> {
+ return InJvmDTestVisitExecutor.builder().pageSizeSelector(i -> rng.nextInt(1, 10)).build(schema, hb.valueGenerators(), cluster);
});
@@ -175,10 +176,8 @@ public class StaticsTortureTest extends IntegrationTestBase
for (int i = 0; i < 10; i++)
{
List ckRelations = generateClusteringRelations(rng, schema.clusteringKeys.size(), ckIdxGen);
- List regularRelations = generateValueRelations(rng, schema.regularColumns.size(),
- column -> Math.min(schema.valueGenerators.regularPopulation(column), MAX_PARTITION_SIZE));
- List staticRelations = generateValueRelations(rng, schema.staticColumns.size(),
- column -> Math.min(schema.valueGenerators.staticPopulation(column), MAX_PARTITION_SIZE));
+ List regularRelations = generateValueRelations(rng, schema.regularColumns.size(),column -> POPULATION);
+ List staticRelations = generateValueRelations(rng, schema.staticColumns.size(),column -> POPULATION);
history.select(pdx,
ckRelations.toArray(new IdxRelation[0]),
regularRelations.toArray(new IdxRelation[0]),
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/AccordBootstrapTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/AccordBootstrapTest.java
index 4ddcd03d45..4891cf31e3 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/AccordBootstrapTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/AccordBootstrapTest.java
@@ -36,6 +36,7 @@ import org.apache.cassandra.distributed.test.log.FuzzTestBase;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
import org.apache.cassandra.harry.dsl.HistoryBuilderHelper;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
@@ -48,6 +49,7 @@ import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.distributed.shared.ClusterUtils.unpauseCommits;
import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
public class AccordBootstrapTest extends FuzzTestBase
{
@@ -77,16 +79,17 @@ public class AccordBootstrapTest extends FuzzTestBase
HashSet downInstances = new HashSet<>();
withRandom(rng -> {
- Generator schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, () -> "bootstrap_fuzz", POPULATION,
+ Generator schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, () -> "bootstrap_fuzz",
SchemaSpec.optionsBuilder()
.addWriteTimestamps(false)
.withTransactionalMode(TransactionalMode.full)
);
SchemaSpec schema = schemaGen.generate(rng);
- TrackingGenerator pkGen = Generators.tracking(Generators.int32(0, Math.min(schema.valueGenerators.pkPopulation(), POPULATION)));
- Generator ckGen = Generators.int32(0, Math.min(schema.valueGenerators.ckPopulation(), POPULATION));
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
+ TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(Generators.int64(0, Math.min(valueGenerators.pkGen().population(), POPULATION))));
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
hb -> InJvmDTestVisitExecutor.builder()
.consistencyLevel(ConsistencyLevel.QUORUM)
.wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
@@ -100,11 +103,15 @@ public class AccordBootstrapTest extends FuzzTestBase
}
})
- .build(schema, hb, cluster));
+ .build(schema, valueGenerators, cluster));
Runnable writeAndValidate = () -> {
for (int i = 0; i < WRITES; i++)
- HistoryBuilderHelper.insertRandomData(schema, pkGen, ckGen, rng, history);
+ {
+ int pdIdx = pkGen.generate(rng);
+ IndexedValueGenerators.IndexedPartitionValues partitionValues = valueGenerators.forPdIdx(pdIdx);
+ HistoryBuilderHelper.insertRandomData(schema, pdIdx, partitionValues.ckIdxGen.generate(rng), rng, 0.5d, history);
+ }
for (int pk : pkGen.generated())
history.selectPartition(pk);
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/AccordBounceTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/AccordBounceTest.java
index 69d12bf9ee..5b5ab244ef 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/AccordBounceTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/AccordBounceTest.java
@@ -36,6 +36,7 @@ import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.test.log.FuzzTestBase;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
@@ -45,6 +46,7 @@ import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
public class AccordBounceTest extends FuzzTestBase
{
@@ -67,7 +69,7 @@ public class AccordBounceTest extends FuzzTestBase
{
return "bootstrap_fuzz" + (i++);
}
- }, POPULATION,
+ },
SchemaSpec.optionsBuilder()
.addWriteTimestamps(false)
.withTransactionalMode(TransactionalMode.full)
@@ -78,12 +80,13 @@ public class AccordBounceTest extends FuzzTestBase
{
SchemaSpec schema = schemaGen.generate(rng);
cluster.schemaChange(schema.compile());
- historyBuilders.add(new ReplayingHistoryBuilder(schema.valueGenerators,
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), POPULATION);
+ historyBuilders.add(new ReplayingHistoryBuilder(valueGenerators,
hb -> InJvmDTestVisitExecutor.builder()
.consistencyLevel(ConsistencyLevel.QUORUM)
.wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
.pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
- .build(schema, hb, cluster)));
+ .build(schema, valueGenerators, cluster)));
}
for (HistoryBuilder hb : historyBuilders)
@@ -125,7 +128,7 @@ public class AccordBounceTest extends FuzzTestBase
{
return "bootstrap_fuzz" + (i++);
}
- }, POPULATION,
+ },
SchemaSpec.optionsBuilder()
.addWriteTimestamps(false)
.withTransactionalMode(TransactionalMode.full)
@@ -136,12 +139,13 @@ public class AccordBounceTest extends FuzzTestBase
{
SchemaSpec schema = schemaGen.generate(rng);
cluster.schemaChange(schema.compile());
- historyBuilders.add(new ReplayingHistoryBuilder(schema.valueGenerators,
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), POPULATION);
+ historyBuilders.add(new ReplayingHistoryBuilder(valueGenerators,
hb -> InJvmDTestVisitExecutor.builder()
.consistencyLevel(ConsistencyLevel.QUORUM)
.wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
.pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
- .build(schema, hb, cluster)));
+ .build(schema, valueGenerators, cluster)));
}
Runnable writeAndValidate = () -> {
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/AccordHardCatchupTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/AccordHardCatchupTest.java
new file mode 100644
index 0000000000..623c0ed653
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/AccordHardCatchupTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.fuzz.topology;
+
+import java.util.HashSet;
+
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
+import org.junit.Test;
+
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.log.FuzzTestBase;
+import org.apache.cassandra.harry.SchemaSpec;
+import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.HistoryBuilderHelper;
+import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
+import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
+import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
+import org.apache.cassandra.harry.gen.Generator;
+import org.apache.cassandra.harry.gen.Generators;
+import org.apache.cassandra.harry.gen.Generators.TrackingGenerator;
+import org.apache.cassandra.harry.gen.SchemaGenerators;
+import org.apache.cassandra.service.consensus.TransactionalMode;
+
+import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
+import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
+
+public class AccordHardCatchupTest extends FuzzTestBase
+{
+ private static final int WRITES = 10;
+ private static final int POPULATION = 1000;
+
+ @Test
+ public void hardCatchupFuzzTest() throws Throwable
+ {
+ CassandraRelevantProperties.SYSTEM_TRACES_DEFAULT_RF.setInt(3);
+ Cluster.Builder builder = builder();
+ try (Cluster cluster = builder.withNodes(3)
+ .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(100))
+ .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(100, "dc0", "rack0"))
+ .withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP))
+ .start())
+ {
+ IInvokableInstance cmsInstance = cluster.get(1);
+ waitForCMSToQuiesce(cluster, cmsInstance);
+
+ HashSet downInstances = new HashSet<>();
+ withRandom(rng -> {
+ Generator schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, () -> "bootstrap_fuzz",
+ SchemaSpec.optionsBuilder()
+ .addWriteTimestamps(false)
+ .withTransactionalMode(TransactionalMode.full)
+ );
+
+ SchemaSpec schema = schemaGen.generate(rng);
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), POPULATION);
+ TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(Generators.int64(0, Math.min(valueGenerators.pkGen().population(), POPULATION))));
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
+ hb -> InJvmDTestVisitExecutor.builder()
+ .consistencyLevel(ConsistencyLevel.QUORUM)
+ .wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
+ .pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
+ .nodeSelector(lts -> {
+ while (true)
+ {
+ int pick = rng.nextInt(1, cluster.size() + 1);
+ if (!downInstances.contains(pick))
+ return pick;
+
+ }
+ })
+ .build(schema, valueGenerators, cluster));
+
+ Runnable writeAndValidate = () -> {
+ for (int i = 0; i < WRITES; i++)
+ {
+ int pdIdx = pkGen.generate(rng);
+ IndexedValueGenerators.IndexedPartitionValues partitionValues = valueGenerators.forPdIdx(pdIdx);
+ HistoryBuilderHelper.insertRandomData(schema, pdIdx, partitionValues.ckIdxGen.generate(rng), rng, 0.5d, history);
+ }
+
+ for (int pk : pkGen.generated())
+ history.selectPartition(pk);
+ };
+
+ history.customThrowing(() -> {
+ cluster.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};", KEYSPACE));
+ cluster.schemaChange(schema.compile());
+ waitForCMSToQuiesce(cluster, cmsInstance);
+ }, "Setup");
+ Thread.sleep(1000);
+ writeAndValidate.run();
+
+ history.customThrowing(() -> {
+ downInstances.add(2);
+ ClusterUtils.stopUnchecked(cluster.get(2));
+ cluster.get(1).logs().watchFor("/127.0.0.2:.* is now DOWN");
+ }, "Shut down node 2");
+
+ writeAndValidate.run();
+
+ history.customThrowing(() -> {
+ cluster.get(2).config().set("accord.catchup_on_start", "HARD");
+ cluster.get(2).startup();
+ cluster.get(2).logs().watchFor(".*Catchup.*");
+ cluster.get(1).logs().watchFor("/127.0.0.2:.* is now UP");
+ downInstances.remove(2);
+ }, "Start down node 2");
+
+ writeAndValidate.run();
+
+ history.customThrowing(() -> {
+ downInstances.add(1);
+ ClusterUtils.stopUnchecked(cluster.get(1));
+ cluster.get(2).logs().watchFor("/127.0.0.1:.* is now DOWN");
+ }, "Shut down node 1");
+
+ writeAndValidate.run();
+ });
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java
new file mode 100644
index 0000000000..cf40abbba5
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/AccordRebootstrapTest.java
@@ -0,0 +1,175 @@
+/*
+ * 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.fuzz.topology;
+
+import java.nio.file.Path;
+import java.util.HashSet;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.IntPredicate;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.CassandraRelevantProperties;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.ClusterUtils;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.log.FuzzTestBase;
+import org.apache.cassandra.harry.SchemaSpec;
+import org.apache.cassandra.harry.dsl.HistoryBuilder;
+import org.apache.cassandra.harry.dsl.HistoryBuilderHelper;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
+import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
+import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
+import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
+import org.apache.cassandra.harry.gen.EntropySource;
+import org.apache.cassandra.harry.gen.Generator;
+import org.apache.cassandra.harry.gen.Generators;
+import org.apache.cassandra.harry.gen.Generators.TrackingGenerator;
+import org.apache.cassandra.harry.gen.SchemaGenerators;
+import org.apache.cassandra.harry.util.ThrowingRunnable;
+import org.apache.cassandra.io.util.PathUtils;
+import org.apache.cassandra.service.consensus.TransactionalMode;
+
+import static org.apache.cassandra.distributed.shared.ClusterUtils.waitForCMSToQuiesce;
+import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
+
+public class AccordRebootstrapTest extends FuzzTestBase
+{
+ private static final int WRITES = 10;
+ private static final int POPULATION = 1000;
+
+ @Test
+ public void rebootstrapFuzzTest() throws Throwable
+ {
+ CassandraRelevantProperties.SYSTEM_TRACES_DEFAULT_RF.setInt(3);
+ Cluster.Builder builder = builder();
+ try (Cluster cluster = builder.withNodes(3)
+ .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(100))
+ .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(100, "dc0", "rack0"))
+ .withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP))
+ .start())
+ {
+ IInvokableInstance cmsInstance = cluster.get(1);
+ waitForCMSToQuiesce(cluster, cmsInstance);
+
+ HashSet downInstances = new HashSet<>();
+ AtomicInteger nextId = new AtomicInteger();
+ withRandom(rng -> {
+ Generator schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, () -> "bootstrap_fuzz" + (nextId.incrementAndGet()),
+ SchemaSpec.optionsBuilder()
+ .addWriteTimestamps(false)
+ .withTransactionalMode(TransactionalMode.full)
+ );
+
+ History history1 = createNewSchemaWithWriteAndValidate(schemaGen, rng, cluster, downInstances::contains);
+ history1.writeAndValidate();
+
+ history1.run(() -> {
+ downInstances.add(2);
+ ClusterUtils.stopUnchecked(cluster.get(2));
+ cluster.get(1).logs().watchFor("/127.0.0.2:.* is now DOWN");
+ }, "Shut down node 2");
+
+ history1.writeAndValidate();
+ History history2 = createNewSchemaWithWriteAndValidate(schemaGen, rng, cluster, downInstances::contains);
+ history2.writeAndValidate();
+ History history3 = createNewSchemaWithWriteAndValidate(schemaGen, rng, cluster, downInstances::contains);
+ history3.writeAndValidate();
+
+ history1.run(() -> {
+ cluster.get(2).config().set("accord.journal.stop_marker_failure_policy", "REBOOTSTRAP");
+ Path journalDir = Path.of(cluster.get(2).config().get("accord.journal_directory").toString());
+ Path stopMarker = journalDir.resolve("stopped");
+ PathUtils.delete(stopMarker);
+ cluster.get(2).startup();
+ cluster.get(2).logs().watchFor(".*Rebootstrapping.*");
+ cluster.get(1).logs().watchFor("/127.0.0.2:.* is now UP");
+ downInstances.remove(2);
+ }, "Start down node 2");
+
+ history1.writeAndValidate();
+ history2.writeAndValidate();
+ history3.writeAndValidate();
+ });
+ }
+ }
+
+ interface History
+ {
+ void writeAndValidate();
+ void run(ThrowingRunnable run, String tag);
+ }
+
+ private History createNewSchemaWithWriteAndValidate(Generator schemaGen, EntropySource rng, Cluster cluster, IntPredicate downInstances) throws InterruptedException
+ {
+ IInvokableInstance cmsInstance = cluster.get(1);
+ SchemaSpec schema = schemaGen.generate(rng);
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), POPULATION);
+ TrackingGenerator pkGen = Generators.tracking(Generators.adaptLongToInt(Generators.int64(0, Math.min(valueGenerators.pkGen().population(), POPULATION))));
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
+ hb -> InJvmDTestVisitExecutor.builder()
+ .consistencyLevel(ConsistencyLevel.QUORUM)
+ .wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
+ .pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
+ .nodeSelector(lts -> {
+ while (true)
+ {
+ int pick = rng.nextInt(1, cluster.size() + 1);
+ if (!downInstances.test(pick))
+ return pick;
+ }
+ })
+ .build(schema, valueGenerators, cluster));
+
+ history.customThrowing(() -> {
+ cluster.schemaChangeIgnoringStoppedInstances(String.format("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};", KEYSPACE));
+ cluster.schemaChangeIgnoringStoppedInstances(schema.compile());
+ waitForCMSToQuiesce(cluster, cmsInstance);
+ }, "Setup");
+ Thread.sleep(1000);
+
+ return new History()
+ {
+ @Override
+ public void writeAndValidate()
+ {
+ for (int i = 0; i < WRITES; i++)
+ {
+ int pdIdx = pkGen.generate(rng);
+ IndexedValueGenerators.IndexedPartitionValues partitionValues = valueGenerators.forPdIdx(pdIdx);
+ HistoryBuilderHelper.insertRandomData(schema, pdIdx, partitionValues.ckIdxGen.generate(rng), rng, 0.5d, history);
+ }
+
+ for (int pk : pkGen.generated())
+ history.selectPartition(pk);
+ }
+
+ @Override
+ public void run(ThrowingRunnable run, String tag)
+ {
+ history.customThrowing(run, tag);
+ }
+ };
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/HarryTopologyMixupTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/HarryTopologyMixupTest.java
index d54ffc3367..8f00fdf471 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/HarryTopologyMixupTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/HarryTopologyMixupTest.java
@@ -55,7 +55,7 @@ import org.apache.cassandra.harry.gen.Generator;
import org.apache.cassandra.harry.gen.Generators;
import org.apache.cassandra.harry.gen.SchemaGenerators;
import org.apache.cassandra.harry.gen.rng.JdkRandomEntropySource;
-import org.apache.cassandra.harry.op.Operations;
+import org.apache.cassandra.harry.op.ClusteringOrderBy;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.utils.AssertionUtils;
@@ -135,8 +135,7 @@ public class HarryTopologyMixupTest extends TopologyMixupTestBase {
InJvmDTestVisitExecutor.Builder builder = InJvmDTestVisitExecutor.builder();
if (mode.kind == AccordMode.Kind.Direct)
@@ -182,7 +181,7 @@ public class HarryTopologyMixupTest extends TopologyMixupTestBase String.format("Harry Reverse Validate pd=%d%s", pd, state.commandNamePostfix()), s -> spec.harry.selectPartition(pkIdx, Operations.ClusteringOrderBy.DESC)));
+ reads.add(new HarryCommand(s -> String.format("Harry Reverse Validate pd=%d%s", pd, state.commandNamePostfix()), s -> spec.harry.selectPartition(pkIdx, ClusteringOrderBy.DESC)));
}
reads.add(new HarryCommand(s -> "Reset Harry Write State" + state.commandNamePostfix(), s -> ((HarryState) s).numInserts = 0));
return Property.multistep(reads);
@@ -255,7 +254,7 @@ public class HarryTopologyMixupTest extends TopologyMixupTestBase InJvmDTestVisitExecutor.builder()
.consistencyLevel(ConsistencyLevel.QUORUM)
.wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
.pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
- .build(schema, hb, cluster));
+ .build(schema, valueGenerators, cluster));
for (int i = 0; i <= 100; i++)
{
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
index d70192f136..1f208a3346 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
@@ -22,6 +22,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.cassandra.harry.dsl.IndexedValueGenerators;
import org.junit.Assert;
import org.junit.Test;
@@ -47,11 +48,10 @@ import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
+import static org.apache.cassandra.harry.dsl.HistoryBuilder.valueGenerators;
public class JournalGCTest extends FuzzTestBase
{
- private static final int POPULATION = 1000;
-
@Test
public void journalGCTest() throws Throwable
{
@@ -70,19 +70,20 @@ public class JournalGCTest extends FuzzTestBase
Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL).disableAutoCompaction();
});
- Generator schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, () -> "bootstrap_fuzz", POPULATION,
+ Generator schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, () -> "bootstrap_fuzz",
SchemaSpec.optionsBuilder()
.addWriteTimestamps(false)
.withTransactionalMode(TransactionalMode.full));
SchemaSpec schema = schemaGen.generate(rng);
cluster.schemaChange(schema.compile());
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), 1000);
+ HistoryBuilder history = new ReplayingHistoryBuilder(valueGenerators,
hb -> InJvmDTestVisitExecutor.builder()
.consistencyLevel(ConsistencyLevel.QUORUM)
.wrapQueries(QueryBuildingVisitExecutor.WrapQueries.TRANSACTION)
.pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
- .build(schema, hb, cluster));
+ .build(schema, valueGenerators, cluster));
for (int pk = 0; pk <= 500; pk++) {
for (int i = 0; i < 100; i++)
diff --git a/test/harry/main/org/apache/cassandra/harry/MagicConstants.java b/test/harry/main/org/apache/cassandra/harry/MagicConstants.java
index 9e7137fccc..2d5a6245d7 100644
--- a/test/harry/main/org/apache/cassandra/harry/MagicConstants.java
+++ b/test/harry/main/org/apache/cassandra/harry/MagicConstants.java
@@ -55,6 +55,11 @@ public class MagicConstants
public static final long UNSET_DESCR = Long.MIN_VALUE + 3;
public static final long NIL_DESCR = Long.MIN_VALUE;
public static final Set MAGIC_DESCRIPTOR_VALS = Set.of(UNKNOWN_DESCR, EMPTY_VALUE_DESCR, UNSET_DESCR, NIL_DESCR);
+
+ public static boolean isMagicDescriptor(long val)
+ {
+ return val <= UNSET_DESCR;
+ }
/**
* For LTS
*/
diff --git a/test/harry/main/org/apache/cassandra/harry/SchemaSpec.java b/test/harry/main/org/apache/cassandra/harry/SchemaSpec.java
index f6f5b8bf22..8884b1cceb 100644
--- a/test/harry/main/org/apache/cassandra/harry/SchemaSpec.java
+++ b/test/harry/main/org/apache/cassandra/harry/SchemaSpec.java
@@ -25,10 +25,8 @@ import java.util.Objects;
import java.util.function.Consumer;
import org.apache.cassandra.cql3.ast.Symbol;
-import org.apache.cassandra.harry.dsl.HistoryBuilder;
import org.apache.cassandra.harry.gen.Generator;
import org.apache.cassandra.harry.gen.Generators;
-import org.apache.cassandra.harry.gen.ValueGenerators;
import org.apache.cassandra.harry.util.IteratorsUtil;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.utils.ByteArrayUtil;
@@ -46,25 +44,20 @@ public class SchemaSpec
public final List> staticColumns;
public final List> allColumnInSelectOrder;
- public final ValueGenerators