diff --git a/.build/build-rat.xml b/.build/build-rat.xml
index 65f79e2c24fc..39808eb2b226 100644
--- a/.build/build-rat.xml
+++ b/.build/build-rat.xml
@@ -60,6 +60,7 @@
+
diff --git a/build.xml b/build.xml
index cedd0c567dba..1366d13a9822 100644
--- a/build.xml
+++ b/build.xml
@@ -1939,7 +1939,7 @@
-
+
diff --git a/run_gen.sh b/run_gen.sh
new file mode 100755
index 000000000000..40c20682774a
--- /dev/null
+++ b/run_gen.sh
@@ -0,0 +1,31 @@
+#!/bin/bash
+java \
+ --add-exports java.base/jdk.internal.misc=ALL-UNNAMED \
+ --add-exports java.base/java.lang.ref=ALL-UNNAMED \
+ --add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED \
+ --add-exports java.management/com.sun.jmx.remote.security=ALL-UNNAMED \
+ --add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED \
+ --add-exports java.rmi/sun.rmi.server=ALL-UNNAMED \
+ --add-exports java.sql/java.sql=ALL-UNNAMED \
+ --add-exports jdk.unsupported/sun.misc=ALL-UNNAMED \
+ --add-opens java.base/java.io=ALL-UNNAMED \
+ --add-opens java.base/java.lang=ALL-UNNAMED \
+ --add-opens java.base/java.lang.module=ALL-UNNAMED \
+ --add-opens java.base/java.lang.reflect=ALL-UNNAMED \
+ --add-opens java.base/java.nio=ALL-UNNAMED \
+ --add-opens java.base/java.util=ALL-UNNAMED \
+ --add-opens java.base/jdk.internal.loader=ALL-UNNAMED \
+ --add-opens java.base/jdk.internal.math=ALL-UNNAMED \
+ --add-opens java.base/jdk.internal.module=ALL-UNNAMED \
+ --add-opens java.base/jdk.internal.ref=ALL-UNNAMED \
+ --add-opens java.base/jdk.internal.reflect=ALL-UNNAMED \
+ --add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED \
+ --add-opens java.base/sun.nio.ch=ALL-UNNAMED \
+ --add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED \
+ -cp 'build/dtest-7.0.jar' \
+ org.apache.cassandra.harry.stress.RangeAwareSSTableGenerator \
+ --contact-points 127.0.0.1 \
+ --schema test/resources/harry/stress/levelled-sstable-schema.yaml \
+ --output-dir /tmp/range-sstables \
+ --visits 10000000 \
+ --sstable-size-mib 64
diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
index 723c773920d7..8eec2cd93e41 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 TableMetadata metadata()
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/db/virtual/DataPlacementsTable.java b/src/java/org/apache/cassandra/db/virtual/DataPlacementsTable.java
new file mode 100644
index 000000000000..8fe3a7ec98ff
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/virtual/DataPlacementsTable.java
@@ -0,0 +1,212 @@
+/*
+ * 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.db.virtual;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.marshal.CompositeType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.SetType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.LocalPartitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.ReversedLongLocalPartitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.locator.EndpointsForRange;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.ReplicationParams;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.ownership.DataPlacement;
+import org.apache.cassandra.tcm.ownership.DataPlacements;
+import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
+
+final class DataPlacementsTable extends AbstractVirtualTable
+{
+ static final String TABLE_COMMENT = "Data placement information showing read and write endpoints per range";
+
+ static final String TABLE_NAME = "data_placements";
+
+ // Partition key columns
+ static final String KEYSPACE_NAME = "keyspace_name";
+ static final String TABLE_NAME_COLUMN = "table_name";
+ static final String RANGE_START = "range_start";
+ static final String RANGE_END = "range_end";
+
+ // Regular columns
+ static final String RANGE_START_BYTES = "range_start_bytes";
+ static final String RANGE_END_BYTES = "range_end_bytes";
+ static final String TOKEN_TYPE = "token_type";
+ static final String READ_ENDPOINTS = "read_endpoints";
+ static final String WRITE_ENDPOINTS = "write_endpoints";
+ static final String READ_REPLICAS = "read_replicas";
+ static final String WRITE_REPLICAS = "write_replicas";
+
+ DataPlacementsTable(String keyspace)
+ {
+ super(buildTableMetadata(keyspace));
+ }
+
+ @Override
+ public DataSet data()
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+ DataPlacements placements = metadata.placements;
+
+ for (KeyspaceMetadata ksm : metadata.schema.getKeyspaces())
+ {
+ ReplicationParams params = ksm.params.replication;
+ DataPlacement placement = placements.get(params);
+
+ for (TableMetadata table : ksm.tables)
+ addPlacementRows(result, ksm.name, table.name, params, placement, metadata, null, null);
+ }
+
+ return result;
+ }
+
+ @Override
+ public DataSet data(DecoratedKey partitionKey)
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+
+ ByteBuffer keyBytes = partitionKey.getKey();
+
+ Preconditions.checkArgument(metadata().partitionKeyType instanceof CompositeType,
+ "Expected CompositeType partition key, got %s", metadata().partitionKeyType.getClass());
+
+ ByteBuffer[] components = ((CompositeType) metadata().partitionKeyType).split(keyBytes);
+ Preconditions.checkArgument(components.length == 1 || components.length == 2 || components.length == 4,
+ "Expected 1, 2, or 4 partition key components (keyspace[, table[, range_start, range_end]]), got %d",
+ components.length);
+
+ String keyspaceName = UTF8Type.instance.compose(components[0]);
+ String tableName = components.length >= 2 ? UTF8Type.instance.compose(components[1]) : null;
+ String rangeStartFilter = components.length == 4 ? UTF8Type.instance.compose(components[2]) : null;
+ String rangeEndFilter = components.length == 4 ? UTF8Type.instance.compose(components[3]) : null;
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+ KeyspaceMetadata ksm = metadata.schema.getKeyspaces().getNullable(keyspaceName);
+ if (ksm == null)
+ return result;
+
+ ReplicationParams params = ksm.params.replication;
+ DataPlacement placement = metadata.placements.get(params);
+ Preconditions.checkState(placement != null, "Placements should never be null for %s", keyspaceName);
+
+ if (tableName != null)
+ {
+ TableMetadata table = ksm.getTableOrViewNullable(tableName);
+ Preconditions.checkArgument(table != null, "Couldn't find table %s", tableName);
+ addPlacementRows(result, ksm.name, table.name, params, placement, metadata, rangeStartFilter, rangeEndFilter);
+ }
+ else
+ {
+ for (TableMetadata table : ksm.tables)
+ addPlacementRows(result, ksm.name, table.name, params, placement, metadata, null, null);
+ }
+
+ return result;
+ }
+
+ private void addPlacementRows(SimpleDataSet result, String keyspaceName, String tableName, ReplicationParams params, DataPlacement placement, ClusterMetadata metadata, String rangeStartFilter, String rangeEndFilter)
+ {
+ Preconditions.checkState(placement.reads.ranges.equals(placement.writes.ranges),
+ "Read ranges (%s) are not the same as write ranges (%s)",
+ placement.reads.ranges, placement.writes.ranges);
+
+ List> ranges = placement.reads.ranges;
+ for (int i = 0; i < ranges.size(); i++)
+ {
+ Range range = ranges.get(i);
+ if (rangeStartFilter != null && !range.left.toString().equals(rangeStartFilter))
+ continue;
+ if (rangeEndFilter != null && !range.right.toString().equals(rangeEndFilter))
+ continue;
+
+ VersionedEndpoints.ForRange readEndpoints = placement.reads.forRange(range);
+ VersionedEndpoints.ForRange writeEndpoints = placement.writes.forRange(range);
+
+ Set readEndpointSet = toEndpointStrings(readEndpoints.get());
+ Set writeEndpointSet = toEndpointStrings(writeEndpoints.get());
+
+ Set readReplicaSet = toNodeIds(readEndpoints.get(), metadata);
+ Set writeReplicaSet = toNodeIds(writeEndpoints.get(), metadata);
+
+ IPartitioner partitioner = params.isMeta() ? ReversedLongLocalPartitioner.instance : metadata.partitioner;
+ result.row(keyspaceName, tableName, range.left.toString(), range.right.toString())
+ .column(TOKEN_TYPE, range.left.getClass().getCanonicalName())
+ .column(RANGE_START_BYTES, partitioner.getTokenFactory().toByteArray(range.left))
+ .column(RANGE_END_BYTES, partitioner.getTokenFactory().toByteArray(range.right))
+ .column(READ_ENDPOINTS, readEndpointSet)
+ .column(WRITE_ENDPOINTS, writeEndpointSet)
+ .column(READ_REPLICAS, readReplicaSet)
+ .column(WRITE_REPLICAS, writeReplicaSet);
+ }
+ }
+
+ static Set toEndpointStrings(EndpointsForRange endpoints)
+ {
+ return endpoints.stream()
+ .map(Replica::endpoint)
+ .map(Object::toString)
+ .collect(Collectors.toSet());
+ }
+
+ static Set toNodeIds(EndpointsForRange endpoints, ClusterMetadata metadata)
+ {
+ return endpoints.stream()
+ .map(Replica::endpoint)
+ .map(metadata.directory::peerId)
+ .map(NodeId::id)
+ .collect(Collectors.toSet());
+ }
+
+ private static TableMetadata buildTableMetadata(String keyspace)
+ {
+ return TableMetadata.builder(keyspace, TABLE_NAME)
+ .comment(TABLE_COMMENT)
+ .kind(TableMetadata.Kind.VIRTUAL)
+ .partitioner(new LocalPartitioner(CompositeType.getInstance(UTF8Type.instance, UTF8Type.instance, UTF8Type.instance, UTF8Type.instance)))
+ .addPartitionKeyColumn(KEYSPACE_NAME, UTF8Type.instance)
+ .addPartitionKeyColumn(TABLE_NAME_COLUMN, UTF8Type.instance)
+ .addPartitionKeyColumn(RANGE_START, UTF8Type.instance)
+ .addPartitionKeyColumn(RANGE_END, UTF8Type.instance)
+ .addRegularColumn(TOKEN_TYPE, UTF8Type.instance)
+ .addRegularColumn(RANGE_START_BYTES, BytesType.instance)
+ .addRegularColumn(RANGE_END_BYTES, BytesType.instance)
+ .addRegularColumn(READ_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false))
+ .addRegularColumn(WRITE_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false))
+ .addRegularColumn(READ_REPLICAS, SetType.getInstance(Int32Type.instance, false))
+ .addRegularColumn(WRITE_REPLICAS, SetType.getInstance(Int32Type.instance, false))
+ .build();
+ }
+}
diff --git a/src/java/org/apache/cassandra/db/virtual/PartitionLocationTable.java b/src/java/org/apache/cassandra/db/virtual/PartitionLocationTable.java
new file mode 100644
index 000000000000..af4d239b8828
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/virtual/PartitionLocationTable.java
@@ -0,0 +1,163 @@
+/*
+ * 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.db.virtual;
+
+import java.nio.ByteBuffer;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.marshal.CompositeType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.SetType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.LocalPartitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.ReversedLongLocalPartitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.ReplicationParams;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ownership.DataPlacement;
+import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
+
+import static com.google.common.base.Preconditions.*;
+
+final class PartitionLocationTable extends AbstractVirtualTable
+{
+ static final String TABLE_NAME = "partition_location";
+ static final String TABLE_COMMENT = "shows the token range and replicas (read and write) for a given partition";
+
+ // Partition keys
+ static final String COLUMN_KEYSPACE_NAME = "keyspace_name";
+ static final String COLUMN_TABLE_NAME = "table_name";
+ static final String COLUMN_KEY = "key";
+
+ // Regular columns
+ static final String COLUMN_TOKEN = "tkn"; // can't use "token" because this is reserved in CQL
+ static final String COLUMN_TOKEN_BYTES = "tkn_bytes";
+ static final String COLUMN_RANGE_START = "range_start";
+ static final String COLUMN_RANGE_END = "range_end";
+ static final String COLUMN_RANGE_START_BYTES = "range_start_bytes";
+ static final String COLUMN_RANGE_END_BYTES = "range_end_bytes";
+ static final String COLUMN_READ_ENDPOINTS = "read_endpoints";
+ static final String COLUMN_WRITE_ENDPOINTS = "write_endpoints";
+ static final String COLUMN_READ_REPLICAS = "read_replicas";
+ static final String COLUMN_WRITE_REPLICAS = "write_replicas";
+
+ PartitionLocationTable(String keyspace)
+ {
+ super(buildTableMetadata(keyspace));
+ }
+
+ @Override
+ public DataSet data()
+ {
+ throw new InvalidRequestException("Partition location table requires a partition key, for example: " +
+ "SELECT * FROM system_views.partition_location WHERE keyspace_name = 'ks' AND table_name = 'tbl' AND key = '1:a'");
+ }
+
+ @Override
+ public DataSet data(DecoratedKey partitionKey)
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+
+ // Partition key is (keyspace_name, table_name, key)
+ ByteBuffer keyBytes = partitionKey.getKey();
+
+ checkArgument(metadata().partitionKeyType instanceof CompositeType,
+ "PartitionLocationTable partition key type must be CompositeType, got %s",
+ metadata().partitionKeyType.getClass().getName());
+
+ ByteBuffer[] components = ((CompositeType) metadata().partitionKeyType).split(keyBytes);
+ checkArgument(components.length == 3,
+ "PartitionLocationTable partition key must have exactly 3 components: keyspace_name, table_name, key; got %d",
+ components.length);
+
+ String keyspaceName = UTF8Type.instance.compose(components[0]);
+ String tableName = UTF8Type.instance.compose(components[1]);
+ String keyString = UTF8Type.instance.compose(components[2]);
+
+ ClusterMetadata metadata = ClusterMetadata.current();
+ KeyspaceMetadata ksm = checkNotNull(metadata.schema.getKeyspaceMetadata(keyspaceName),
+ "Keyspace %s is not found in metadata", keyspaceName);
+ TableMetadata table = checkNotNull(ksm.getTableOrViewNullable(tableName),
+ "Table %s is not found in metadata (within keyspace %s)", tableName, keyspaceName);
+
+ ByteBuffer partitionKeyBytes = table.partitionKeyType.fromString(keyString);
+
+ DecoratedKey dk = table.partitioner.decorateKey(partitionKeyBytes);
+ Token token = dk.getToken();
+
+ ReplicationParams replicationParams = ksm.params.replication;
+ DataPlacement placement = metadata.placements.get(replicationParams);
+
+ VersionedEndpoints.ForRange readEndpoints = placement.reads.forRange(token);
+ VersionedEndpoints.ForRange writeEndpoints = placement.writes.forRange(token);
+
+ Range range = readEndpoints.get().range();
+
+ Set readEndpointSet = DataPlacementsTable.toEndpointStrings(readEndpoints.get());
+ Set writeEndpointSet = DataPlacementsTable.toEndpointStrings(writeEndpoints.get());
+
+ Set readReplicaSet = DataPlacementsTable.toNodeIds(readEndpoints.get(), metadata);
+ Set writeReplicaSet = DataPlacementsTable.toNodeIds(writeEndpoints.get(), metadata);
+
+ IPartitioner partitioner = replicationParams.isMeta() ? ReversedLongLocalPartitioner.instance : metadata.partitioner;
+ result.row(keyspaceName, tableName, keyString)
+ .column(COLUMN_TOKEN, token.toString())
+ .column(COLUMN_TOKEN_BYTES, partitioner.getTokenFactory().toByteArray(token))
+ .column(COLUMN_RANGE_START, range.left.toString())
+ .column(COLUMN_RANGE_END, range.right.toString())
+ .column(COLUMN_RANGE_START_BYTES, partitioner.getTokenFactory().toByteArray(range.left))
+ .column(COLUMN_RANGE_END_BYTES, partitioner.getTokenFactory().toByteArray(range.right))
+ .column(COLUMN_READ_ENDPOINTS, readEndpointSet)
+ .column(COLUMN_WRITE_ENDPOINTS, writeEndpointSet)
+ .column(COLUMN_READ_REPLICAS, readReplicaSet)
+ .column(COLUMN_WRITE_REPLICAS, writeReplicaSet);
+
+ return result;
+ }
+
+ private static TableMetadata buildTableMetadata(String keyspace) {
+ return TableMetadata.builder(keyspace, TABLE_NAME)
+ .comment(TABLE_COMMENT)
+ .kind(TableMetadata.Kind.VIRTUAL)
+ .partitioner(new LocalPartitioner(CompositeType.getInstance(UTF8Type.instance, UTF8Type.instance, UTF8Type.instance)))
+ .addPartitionKeyColumn(COLUMN_KEYSPACE_NAME, UTF8Type.instance)
+ .addPartitionKeyColumn(COLUMN_TABLE_NAME, UTF8Type.instance)
+ .addPartitionKeyColumn(COLUMN_KEY, UTF8Type.instance)
+ .addRegularColumn(COLUMN_TOKEN, UTF8Type.instance)
+ .addRegularColumn(COLUMN_TOKEN_BYTES, BytesType.instance)
+ .addRegularColumn(COLUMN_RANGE_START, UTF8Type.instance)
+ .addRegularColumn(COLUMN_RANGE_END, UTF8Type.instance)
+ .addRegularColumn(COLUMN_RANGE_START_BYTES, BytesType.instance)
+ .addRegularColumn(COLUMN_RANGE_END_BYTES, BytesType.instance)
+ .addRegularColumn(COLUMN_READ_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false))
+ .addRegularColumn(COLUMN_WRITE_ENDPOINTS, SetType.getInstance(UTF8Type.instance, false))
+ .addRegularColumn(COLUMN_READ_REPLICAS, SetType.getInstance(Int32Type.instance, false))
+ .addRegularColumn(COLUMN_WRITE_REPLICAS, SetType.getInstance(Int32Type.instance, false))
+ .build();
+ }
+}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
index 9a80d1e9b930..d953310165d7 100644
--- a/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
+++ b/src/java/org/apache/cassandra/db/virtual/SystemViewsKeyspace.java
@@ -69,6 +69,8 @@ private SystemViewsKeyspace()
.add(new LocalTable(VIRTUAL_VIEWS))
.add(new ClusterMetadataLogTable(VIRTUAL_VIEWS))
.add(new ClusterMetadataDirectoryTable(VIRTUAL_VIEWS))
+ .add(new DataPlacementsTable(VIRTUAL_VIEWS))
+ .add(new PartitionLocationTable(VIRTUAL_VIEWS))
.addAll(LocalRepairTables.getAll(VIRTUAL_VIEWS))
.addAll(CIDRFilteringMetricsTable.getAll(VIRTUAL_VIEWS))
.addAll(StorageAttachedIndexTables.getAll(VIRTUAL_VIEWS))
diff --git a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
index 2487e66fd280..19501fcc9b55 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.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 @@ protected AbstractSSTableSimpleWriter(File directory, TableMetadataRef metadata,
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 void setCompressionDictionary(CompressionDictionary compressionDictionary
}
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 @@ protected boolean shouldOpenSSTables()
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 @@ protected SSTableTxnWriter createWriter(SSTable.Owner owner) throws IOException
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 @@ protected SSTableTxnWriter createWriter(SSTable.Owner owner) throws IOException
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 8c9b2f979621..bf1f04c9b784 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 @@
*
* @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 @@ PartitionUpdate.Builder getUpdateFor(DecoratedKey key)
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 @@ private void countRow(Row row)
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 @@ private PartitionUpdate.Builder createPartitionUpdateBuilder(DecoratedKey key)
@Override
public void add(Row row)
{
+ int before = rowCount();
super.add(row);
- countRow(row);
+
+ if (rowCount() > before)
+ countRow(row);
maybeSync();
}
};
@@ -229,7 +236,7 @@ public void run()
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 7b5fb8a89e45..ed29e7db709d 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 @@
* 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 @@ private void maybeCloseWriter(SSTableTxnWriter writer)
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 3b43dcfdf4fb..51f66b0ddc62 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 static SSTableTxnWriter create(TableMetadataRef metadata,
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 fd1a2618db6b..f021b41bcd8c 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.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 DataPlacement(ReplicaGroups reads,
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 2d06c4db1520..fc3418cc3b61 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 Builder copy()
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/DataPlacementsTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/DataPlacementsTableTest.java
new file mode 100644
index 000000000000..304ffe358870
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/DataPlacementsTableTest.java
@@ -0,0 +1,376 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.ownership.DataPlacement;
+import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
+import org.apache.cassandra.tcm.transformations.Register;
+import org.apache.cassandra.tcm.transformations.UnsafeJoin;
+import org.apache.cassandra.utils.Shared;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class DataPlacementsTableTest extends TestBaseImpl
+{
+ private static final int NUM_FAKE_NODES = 60;
+ private static final String DC1 = "datacenter0";
+ private static final String DC2 = "datacenter1";
+ private static final String SIMPLE_KS = "test_ks_simple";
+ private static final String NTS_KS = "test_ks_nts";
+
+ private static final String SIMPLE_TABLE1 = "tbl1";
+ private static final String SIMPLE_TABLE2 = "tbl2";
+ private static final String NTS_TABLE1 = "nts_tbl1";
+ private static final String NTS_TABLE2 = "nts_tbl2";
+
+ private static final long REAL_NODE_TOKEN = Long.MAX_VALUE - 1;
+
+ private static Cluster CLUSTER;
+ private static Map> SIMPLE_EXPECTED;
+ private static Map> NTS_EXPECTED;
+
+ @BeforeClass
+ public static void setup() throws IOException
+ {
+ CLUSTER = Cluster.build(1)
+ .withConfig(c -> c.set("num_tokens", 1)
+ .set("initial_token", Long.toString(REAL_NODE_TOKEN)))
+ .start();
+
+ // Register 60 fake nodes split evenly across the two DCs, evenly spaced around the ring.
+ CLUSTER.get(1).runOnInstance(() -> {
+ try
+ {
+ for (int i = 0; i < NUM_FAKE_NODES; i++)
+ {
+ String dc = (i % 2 == 0) ? DC1 : DC2;
+ String address = "127.0.1." + (i + 2);
+
+ InetAddressAndPort addr = InetAddressAndPort.getByName(address);
+ NodeAddresses nodeAddresses = new NodeAddresses(addr);
+ Location location = new Location(dc, "rack1");
+
+ ClusterMetadata metadata = ClusterMetadataService.instance().commit(
+ new Register(nodeAddresses, location, NodeVersion.CURRENT));
+
+ NodeId nodeId = metadata.directory.peerId(addr);
+
+ long token = Long.MIN_VALUE + (long) i * (Long.MAX_VALUE / (NUM_FAKE_NODES / 2));
+ Set tokens = new HashSet<>();
+ tokens.add(new Murmur3Partitioner.LongToken(token));
+
+ UnsafeJoin.unsafeJoin(nodeId, tokens);
+ }
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException("Failed to register fake nodes", e);
+ }
+ });
+
+ CLUSTER.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3}", SIMPLE_KS));
+ CLUSTER.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'NetworkTopologyStrategy', '%s': 2, '%s': 2}", NTS_KS, DC1, DC2));
+
+ CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", SIMPLE_KS, SIMPLE_TABLE1));
+ CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", SIMPLE_KS, SIMPLE_TABLE2));
+ CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", NTS_KS, NTS_TABLE1));
+ CLUSTER.schemaChange(String.format("CREATE TABLE %s.%s (id int PRIMARY KEY, val text)", NTS_KS, NTS_TABLE2));
+
+ SIMPLE_EXPECTED = fetchExpectedPlacements(SIMPLE_KS);
+ NTS_EXPECTED = fetchExpectedPlacements(NTS_KS);
+ }
+
+
+ @AfterClass
+ public static void cleanup()
+ {
+ if (CLUSTER != null)
+ CLUSTER.close();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPlacements()
+ {
+ Map>> allExpected = new HashMap<>();
+ allExpected.put(SIMPLE_KS, SIMPLE_EXPECTED);
+ allExpected.put(NTS_KS, NTS_EXPECTED);
+
+ Object[][] result = CLUSTER.coordinator(1)
+ .execute("SELECT keyspace_name, table_name, range_start, range_end, token_type, " +
+ "range_start_bytes, range_end_bytes, read_endpoints, write_endpoints, " +
+ "read_replicas, write_replicas " +
+ "FROM system_views.data_placements",
+ ConsistencyLevel.ONE);
+
+ assertThat(result).isNotEmpty();
+
+ Map> tablesFound = new HashMap<>();
+ Map> rangesFound = new HashMap<>();
+ for (String ks : allExpected.keySet())
+ {
+ tablesFound.put(ks, new HashSet<>());
+ rangesFound.put(ks, new HashSet<>());
+ }
+
+ for (Object[] row : result)
+ {
+ String keyspace = (String) row[0];
+ Map> expected = allExpected.get(keyspace);
+ if (expected == null)
+ continue; // skip system keyspaces
+
+ String tableName = (String) row[1];
+ String rangeStart = (String) row[2];
+ String rangeEnd = (String) row[3];
+ String tokenType = (String) row[4];
+ ByteBuffer rangeStartBytes = (ByteBuffer) row[5];
+ ByteBuffer rangeEndBytes = (ByteBuffer) row[6];
+ Set readEndpoints = (Set) row[7];
+ Set writeEndpoints = (Set) row[8];
+ Set readReplicas = (Set) row[9];
+ Set writeReplicas = (Set) row[10];
+
+ assertThat(tokenType).as("token_type for %s.%s", keyspace, tableName)
+ .contains("Murmur3Partitioner");
+ assertThat(rangeStartBytes).as("range_start_bytes for %s.%s", keyspace, tableName).isNotNull();
+ assertThat(rangeEndBytes).as("range_end_bytes for %s.%s", keyspace, tableName).isNotNull();
+
+ Range range = new Range(Long.parseLong(rangeStart), Long.parseLong(rangeEnd));
+ Set exp = expected.get(range);
+ assertThat(exp).as("range (%s, %s] not in expected for %s", rangeStart, rangeEnd, keyspace)
+ .isNotNull();
+
+ Set expectedIps = exp.stream().map(e -> e.ip).collect(Collectors.toSet());
+ Set expectedNodeIds = exp.stream().map(e -> e.nodeId).collect(Collectors.toSet());
+
+ assertThat(readEndpoints).as("read endpoints for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd)
+ .isEqualTo(expectedIps);
+ assertThat(writeEndpoints).as("write endpoints for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd)
+ .isEqualTo(expectedIps);
+ assertThat(readReplicas).as("read replica node-ids for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd)
+ .isEqualTo(expectedNodeIds);
+ assertThat(writeReplicas).as("write replica node-ids for %s.%s (%s,%s]", keyspace, tableName, rangeStart, rangeEnd)
+ .isEqualTo(expectedNodeIds);
+
+ tablesFound.get(keyspace).add(tableName);
+ rangesFound.get(keyspace).add(range);
+ }
+
+ assertThat(tablesFound.get(SIMPLE_KS)).as("tables in " + SIMPLE_KS).contains(SIMPLE_TABLE1, SIMPLE_TABLE2);
+ assertThat(rangesFound.get(SIMPLE_KS)).as("unique ranges in " + SIMPLE_KS).hasSize(SIMPLE_EXPECTED.size());
+ assertThat(tablesFound.get(NTS_KS)).as("tables in " + NTS_KS).contains(NTS_TABLE1, NTS_TABLE2);
+ assertThat(rangesFound.get(NTS_KS)).as("unique ranges in " + NTS_KS).hasSize(NTS_EXPECTED.size());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testQueryByKeyspaceAndTable()
+ {
+ Object[][] result = CLUSTER.coordinator(1)
+ .execute("SELECT keyspace_name, table_name, range_start, range_end, " +
+ "read_endpoints, write_endpoints " +
+ "FROM system_views.data_placements WHERE keyspace_name = ? AND table_name = ?",
+ ConsistencyLevel.ONE, NTS_KS, NTS_TABLE1);
+
+ assertThat(result).isNotEmpty();
+
+ Set rangesFound = new HashSet<>();
+ for (Object[] row : result)
+ {
+ assertThat(row[0]).isEqualTo(NTS_KS);
+ assertThat(row[1]).isEqualTo(NTS_TABLE1);
+
+ String rangeStart = (String) row[2];
+ String rangeEnd = (String) row[3];
+ Set readEndpoints = (Set) row[4];
+ Set writeEndpoints = (Set) row[5];
+
+ Range range = new Range(Long.parseLong(rangeStart), Long.parseLong(rangeEnd));
+ Set expected = NTS_EXPECTED.get(range);
+ assertThat(expected).as("range (%s, %s] not found in simulation", rangeStart, rangeEnd)
+ .isNotNull();
+
+ Set expectedIps = expected.stream().map(e -> e.ip).collect(Collectors.toSet());
+ assertThat(readEndpoints).as("read endpoints for (%s,%s]", rangeStart, rangeEnd)
+ .isEqualTo(expectedIps);
+ assertThat(writeEndpoints).as("write endpoints for (%s,%s]", rangeStart, rangeEnd)
+ .isEqualTo(expectedIps);
+
+ rangesFound.add(range);
+ }
+
+ // The 2-component query must return ALL ranges for the table, not a subset.
+ assertThat(rangesFound).as("ranges for " + NTS_KS + "." + NTS_TABLE1)
+ .hasSize(NTS_EXPECTED.size());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testQueryByFullPartitionKey()
+ {
+ String rangeStart, rangeEnd;
+ {
+ Range range = NTS_EXPECTED.keySet().iterator().next();
+ rangeStart = String.valueOf(range.start);
+ rangeEnd = String.valueOf(range.end);
+ }
+
+ Object[][] result = CLUSTER.coordinator(1)
+ .execute("SELECT keyspace_name, table_name, range_start, range_end, " +
+ "read_endpoints, write_endpoints, read_replicas, write_replicas " +
+ "FROM system_views.data_placements " +
+ "WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?",
+ ConsistencyLevel.ONE, NTS_KS, NTS_TABLE1, rangeStart, rangeEnd);
+
+ assertThat(result).as("full-PK query must return exactly one row").hasNumberOfRows(1);
+
+ Object[] row = result[0];
+ assertThat(row[0]).as("keyspace_name").isEqualTo(NTS_KS);
+ assertThat(row[1]).as("table_name").isEqualTo(NTS_TABLE1);
+ assertThat(row[2]).as("range_start").isEqualTo(rangeStart);
+ assertThat(row[3]).as("range_end").isEqualTo(rangeEnd);
+
+ Set readEndpoints = (Set) row[4];
+ Set writeEndpoints = (Set) row[5];
+ Set readReplicas = (Set) row[6];
+ Set writeReplicas = (Set) row[7];
+
+ Range range = new Range(Long.parseLong(rangeStart), Long.parseLong(rangeEnd));
+ Set expected = NTS_EXPECTED.get(range);
+ assertThat(expected).as("range (%s, %s] not found in simulation", rangeStart, rangeEnd)
+ .isNotNull();
+
+ Set expectedIps = expected.stream().map(e -> e.ip).collect(Collectors.toSet());
+ Set expectedNodeIds = expected.stream().map(e -> e.nodeId).collect(Collectors.toSet());
+
+ assertThat(readEndpoints).as("read endpoints").isEqualTo(expectedIps);
+ assertThat(writeEndpoints).as("write endpoints").isEqualTo(expectedIps);
+ assertThat(readReplicas).as("read replica node-ids").isEqualTo(expectedNodeIds);
+ assertThat(writeReplicas).as("write replica node-ids").isEqualTo(expectedNodeIds);
+ }
+
+ @Shared
+ public static final class Range
+ {
+ public final long start;
+ public final long end;
+
+ public Range(long start, long end)
+ {
+ this.start = start;
+ this.end = end;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof Range)) return false;
+ Range r = (Range) o;
+ return r.start == start && r.end == end;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Long.hashCode(start) * 31 + Long.hashCode(end);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "(" + start + ", " + end + "]";
+ }
+ }
+
+ @Shared
+ public static final class Endpoint
+ {
+ public final String ip;
+ public final int nodeId;
+
+ public Endpoint(String ip, int nodeId)
+ {
+ this.ip = ip;
+ this.nodeId = nodeId;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof Endpoint)) return false;
+ Endpoint e = (Endpoint) o;
+ return e.nodeId == nodeId && e.ip.equals(ip);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return ip.hashCode() * 31 + nodeId;
+ }
+ }
+
+ private static Map> fetchExpectedPlacements(String keyspaceName)
+ {
+ return CLUSTER.get(1).callOnInstance(() -> {
+ Map> result = new HashMap<>();
+ ClusterMetadata metadata = ClusterMetadata.current();
+ KeyspaceMetadata ksm = metadata.schema.getKeyspaceMetadata(keyspaceName);
+ DataPlacement placement = metadata.placements.get(ksm.params.replication);
+ for (org.apache.cassandra.dht.Range range : placement.reads.ranges)
+ {
+ VersionedEndpoints.ForRange re = placement.reads.forRange(range);
+ Set endpoints = new HashSet<>();
+ for (Replica r : re.get())
+ {
+ NodeId nid = metadata.directory.peerId(r.endpoint());
+ endpoints.add(new Endpoint(r.endpoint().toString(), nid.id()));
+ }
+ result.put(new Range(Long.parseLong(range.left.toString()),
+ Long.parseLong(range.right.toString())),
+ endpoints);
+ }
+ return result;
+ });
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/PartitionLocationTableTest.java b/test/distributed/org/apache/cassandra/distributed/test/PartitionLocationTableTest.java
new file mode 100644
index 000000000000..86c39dc6ca90
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/PartitionLocationTableTest.java
@@ -0,0 +1,211 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.ownership.DataPlacement;
+import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
+import org.apache.cassandra.utils.Shared;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class PartitionLocationTableTest extends TestBaseImpl
+{
+ private static final int NUM_NODES = 3;
+ private static final String KEYSPACE = "test_ks";
+
+ private static Cluster CLUSTER;
+
+ @BeforeClass
+ public static void setup() throws IOException
+ {
+ CLUSTER = Cluster.build(NUM_NODES).start();
+ }
+
+ @AfterClass
+ public static void cleanup()
+ {
+ if (CLUSTER != null)
+ CLUSTER.close();
+ }
+
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testSimplePartitionKey()
+ {
+ CLUSTER.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3}");
+ CLUSTER.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl1 (col1 int PRIMARY KEY, col2 text)");
+
+ ExpectedLocation expected = computeExpectedLocation(KEYSPACE, "tbl1", "123");
+
+ Object[][] result = CLUSTER.coordinator(1).execute(
+ "SELECT tkn, range_start, range_end, range_start_bytes, range_end_bytes, " +
+ "read_endpoints, write_endpoints, read_replicas, write_replicas " +
+ "FROM system_views.partition_location " +
+ "WHERE keyspace_name = ? AND table_name = ? AND key = ?",
+ ConsistencyLevel.ONE,
+ KEYSPACE, "tbl1", "123");
+
+ Assert.assertEquals(1, result.length);
+ Object[] row = result[0];
+ String token = (String) row[0];
+ String rangeStart = (String) row[1];
+ String rangeEnd = (String) row[2];
+ ByteBuffer rangeStartBytes = (ByteBuffer) row[3];
+ ByteBuffer rangeEndBytes = (ByteBuffer) row[4];
+ Set readEndpoints = (Set) row[5];
+ Set writeEndpoints = (Set) row[6];
+ Set readReplicas = (Set) row[7];
+ Set writeReplicas = (Set) row[8];
+
+ assertThat(token).as("token").isEqualTo(expected.token);
+ assertThat(rangeStart).as("range_start").isEqualTo(expected.rangeStart);
+ assertThat(rangeEnd).as("range_end").isEqualTo(expected.rangeEnd);
+ assertThat(rangeStartBytes).as("range_start_bytes").isNotNull();
+ assertThat(rangeEndBytes).as("range_end_bytes").isNotNull();
+
+ assertThat(readEndpoints).as("read endpoints").isEqualTo(expected.readEndpoints);
+ assertThat(writeEndpoints).as("write endpoints").isEqualTo(expected.writeEndpoints);
+ assertThat(readReplicas).as("read replica node-ids").isEqualTo(expected.readReplicas);
+ assertThat(writeReplicas).as("write replica node-ids").isEqualTo(expected.writeReplicas);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCompositePartitionKey()
+ {
+ CLUSTER.schemaChange("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3}");
+ CLUSTER.schemaChange("CREATE TABLE IF NOT EXISTS " + KEYSPACE + ".tbl2 (col1 int, col2 text, col3 int, PRIMARY KEY ((col1, col2), col3))");
+
+ ExpectedLocation expected = computeExpectedLocation(KEYSPACE, "tbl2", "123:value1");
+
+ Object[][] result = CLUSTER.coordinator(1).execute(
+ "SELECT tkn, range_start, range_end, range_start_bytes, range_end_bytes, " +
+ "read_endpoints, write_endpoints, read_replicas, write_replicas " +
+ "FROM system_views.partition_location " +
+ "WHERE keyspace_name = ? AND table_name = ? AND key = ?",
+ ConsistencyLevel.ONE,
+ KEYSPACE, "tbl2", "123:value1");
+
+ Assert.assertEquals(1, result.length);
+ Object[] row = result[0];
+ String token = (String) row[0];
+ String rangeStart = (String) row[1];
+ String rangeEnd = (String) row[2];
+ ByteBuffer rangeStartBytes = (ByteBuffer) row[3];
+ ByteBuffer rangeEndBytes = (ByteBuffer) row[4];
+ Set readEndpoints = (Set) row[5];
+ Set writeEndpoints = (Set) row[6];
+ Set readReplicas = (Set) row[7];
+ Set writeReplicas = (Set) row[8];
+
+ assertThat(token).as("token").isEqualTo(expected.token);
+ assertThat(rangeStart).as("range_start").isEqualTo(expected.rangeStart);
+ assertThat(rangeEnd).as("range_end").isEqualTo(expected.rangeEnd);
+ assertThat(rangeStartBytes).as("range_start_bytes").isNotNull();
+ assertThat(rangeEndBytes).as("range_end_bytes").isNotNull();
+
+ assertThat(readEndpoints).as("read endpoints").isEqualTo(expected.readEndpoints);
+ assertThat(writeEndpoints).as("write endpoints").isEqualTo(expected.writeEndpoints);
+ assertThat(readReplicas).as("read replica node-ids").isEqualTo(expected.readReplicas);
+ assertThat(writeReplicas).as("write replica node-ids").isEqualTo(expected.writeReplicas);
+ }
+
+ @Shared
+ public static final class ExpectedLocation // class needs to be public since it's shared
+ {
+ public final String token;
+ public final String rangeStart;
+ public final String rangeEnd;
+ public final Set readEndpoints;
+ public final Set writeEndpoints;
+ public final Set readReplicas;
+ public final Set writeReplicas;
+
+ public ExpectedLocation(String token, String rangeStart, String rangeEnd,
+ Set readEndpoints, Set writeEndpoints,
+ Set readReplicas, Set writeReplicas)
+ {
+ this.token = token;
+ this.rangeStart = rangeStart;
+ this.rangeEnd = rangeEnd;
+ this.readEndpoints = readEndpoints;
+ this.writeEndpoints = writeEndpoints;
+ this.readReplicas = readReplicas;
+ this.writeReplicas = writeReplicas;
+ }
+ }
+
+ private static ExpectedLocation computeExpectedLocation(String ks, String tbl, String key)
+ {
+ return CLUSTER.get(1).callOnInstance(() -> {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ KeyspaceMetadata ksm = metadata.schema.getKeyspaceMetadata(ks);
+ TableMetadata table = ksm.getTableOrViewNullable(tbl);
+
+ ByteBuffer partitionKeyBytes = table.partitionKeyType.fromString(key);
+ DecoratedKey dk = table.partitioner.decorateKey(partitionKeyBytes);
+ Token token = dk.getToken();
+
+ DataPlacement placement = metadata.placements.get(ksm.params.replication);
+ VersionedEndpoints.ForRange readEps = placement.reads.forRange(token);
+ VersionedEndpoints.ForRange writeEps = placement.writes.forRange(token);
+ Range range = readEps.get().range();
+
+ Set readEndpoints = readEps.get().stream()
+ .map(r -> r.endpoint().toString())
+ .collect(Collectors.toSet());
+ Set writeEndpoints = writeEps.get().stream()
+ .map(r -> r.endpoint().toString())
+ .collect(Collectors.toSet());
+ Set readReplicas = readEps.get().stream()
+ .map(r -> metadata.directory.peerId(r.endpoint()).id())
+ .collect(Collectors.toSet());
+ Set writeReplicas = writeEps.get().stream()
+ .map(r -> metadata.directory.peerId(r.endpoint()).id())
+ .collect(Collectors.toSet());
+
+ return new ExpectedLocation(token.toString(),
+ range.left.toString(),
+ range.right.toString(),
+ readEndpoints,
+ writeEndpoints,
+ readReplicas,
+ writeReplicas);
+ });
+ }
+}
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 ceebb3933dca..e8587ff56606 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.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 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 void hostReplace() throws IOException
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 void hostReplace() throws IOException
}
}
- 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 6d05fc981748..b2114a4cd15e 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.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 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 void testTopologyChanges() throws Exception
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 void testTopologyChanges() throws Exception
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 ed6faf26aa95..100dd856274f 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.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.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 void bounceTest() throws Exception
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 void bounceTest() throws Exception
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 void run()
{
// 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 9246a7df000c..727848fca708 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.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 void writeConsistencyTest() throws Throwable
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 void writeConsistencyTest() throws Throwable
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 d64a1bdff968..d9a780254bf2 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.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.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 @@ private void failedLeaveTest(BiFunction {
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 @@ private void failedLeaveTest(BiFunction {
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 07ec054d54ae..b97c1fae6b43 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.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 void bootstrapWithDeferredJoinTest() throws Throwable
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 void bootstrapWithDeferredJoinTest() throws Throwable
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 void bootstrapWithDeferredJoinTest() throws Throwable
// 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 void bootstrapWithDeferredJoinTest() throws Throwable
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 64810654d5cf..486adcf8000a 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 void simpleUpgradeTest()
.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 void simpleUpgradeTest()
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 void simpleUpgradeTest()
}
})
.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 8607c2551ddb..ad432fd2bd55 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 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 void rangeTombstoneBurnTest()
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 void rangeTombstoneBurnTest()
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 65b1bf444140..e37f13a49750 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.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 static void before() throws Throwable
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 void repairBurnTest()
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 000000000000..ef8a6f93c592
--- /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 000000000000..e612a4920d8d
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/fuzz/harry/sstable/RangeAwareSSTableLoadAndStressTest.java
@@ -0,0 +1,210 @@
+/*
+ * 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.HashMap;
+import java.util.List;
+import java.util.Map;
+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.TokenIndex;
+import org.apache.cassandra.harry.stress.TokenIndexGenerator;
+import org.apache.cassandra.harry.stress.VisitGenerator;
+import org.apache.cassandra.harry.stress.config.StressSchemaConfig;
+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();
+ Map weights = new HashMap<>();
+ weights.put(VisitGenerator.VisitType.MUTATE, 10);
+ weights.put(VisitGenerator.VisitType.VALIDATE, 10);
+ Generator visitTypeGen = Generators.weighted(weights);
+
+
+ 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");
+
+ // 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());
+ if (run != null)
+ run.run();
+ return rs;
+ },
+ CONCURRENCY,
+ RATE_PER_SECOND,
+ /*minPartitionIdx=*/ 0,
+ /*maxPartitionIdx=*/ Long.MAX_VALUE,
+ /*initialLts=*/ INITIAL_LTS);
+
+ stress.replay(INITIAL_LTS, END_LTS);
+ tokenIndex.close();
+
+ 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 000000000000..f44b142a540f
--- /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 08af4f4d4b9e..465a848c7fd6 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.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.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 void bootstrapFuzzTest() throws Throwable
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 void bootstrapFuzzTest() throws Throwable
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 void coordinatorIsBehindTest() throws Throwable
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 void coordinatorIsBehindTest() throws Throwable
.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 void coordinatorIsBehindTest() throws Throwable
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 void coordinatorIsBehindTest() throws Throwable
{
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 bb9fd4867c18..d6803e959909 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.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 void decommissionTest() throws Throwable
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 7f4d3ad5df4f..3e7e597258f1 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.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 void moveTest() throws Throwable
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 bbe74c34a127..a441651f2bb6 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.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 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 @@ protected int rf()
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 void mixedFilteringSaiTest()
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 @@ private void saiTest(EntropySource rng, SchemaSpec schema, Supplier cre
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 @@ private void saiTest(EntropySource rng, SchemaSpec schema, Supplier cre
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 @@ private void saiTest(EntropySource rng, SchemaSpec schema, Supplier cre
}
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 @@ else if (i % repairSkip == 0)
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 @@ else if (i % repairSkip == 0)
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 static Consumer defaultConfig()
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 07844b045e31..d445fd5da40f 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.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 void staticsTortureTest()
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 void staticsTortureTest(List> cks, int idx)
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 void staticsTortureTest(List> cks, int idx)
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 4ddcd03d45a7..4891cf31e354 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.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 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 void bootstrapFuzzTest() throws Throwable
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 void bootstrapFuzzTest() throws Throwable
}
})
- .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 69d12bf9ee31..5b5ab244ef71 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.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.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 String get()
{
return "bootstrap_fuzz" + (i++);
}
- }, POPULATION,
+ },
SchemaSpec.optionsBuilder()
.addWriteTimestamps(false)
.withTransactionalMode(TransactionalMode.full)
@@ -78,12 +80,13 @@ public String get()
{
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 String get()
{
return "bootstrap_fuzz" + (i++);
}
- }, POPULATION,
+ },
SchemaSpec.optionsBuilder()
.addWriteTimestamps(false)
.withTransactionalMode(TransactionalMode.full)
@@ -136,12 +139,13 @@ public String get()
{
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 000000000000..623c0ed65369
--- /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 000000000000..cf40abbba51d
--- /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 d54ffc336769..8f00fdf4715d 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.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 @@ private static BiFunction createSchemaSpec(AccordMo
schemaGen = SchemaGenerators.schemaSpecGen("harry", "table", 1000);
schema = schemaGen.generate(rng);
-
- HistoryBuilder harry = new ReplayingHistoryBuilder(schema.valueGenerators,
+ HistoryBuilder harry = new ReplayingHistoryBuilder(HistoryBuilder.valueGenerators(schema, rng.next(), 1000),
hb -> {
InJvmDTestVisitExecutor.Builder builder = InJvmDTestVisitExecutor.builder();
if (mode.kind == AccordMode.Kind.Direct)
@@ -182,7 +181,7 @@ public int select(long lts)
}
return false;
})
- .build(schema, hb, cluster);
+ .build(schema, hb.valueGenerators(), cluster);
});
cluster.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};", schema.keyspace));
cluster.schemaChange(schema.compile());
@@ -239,7 +238,7 @@ private static CommandGen cqlOperations(Spec spec)
TransactionalMode transationalMode = spec.schema.options.transactionalMode();
if (TransactionalMode.full == transationalMode)
- reads.add(new HarryCommand(s -> 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 Spec(HistoryBuilder harry, SchemaSpec schema)
{
this.harry = harry;
this.schema = schema;
- this.pkGen = Generators.tracking(Generators.int32(0, schema.valueGenerators.pkPopulation()));
+ this.pkGen = Generators.tracking(Generators.adaptLongToInt(harry.valueGenerators().pkIdxGen()));
}
@Override
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/IdenticalTopologyTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/IdenticalTopologyTest.java
index 91b6527801c6..90aed519b692 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/IdenticalTopologyTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/IdenticalTopologyTest.java
@@ -27,6 +27,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.IndexedValueGenerators;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
@@ -38,6 +39,7 @@
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 IdenticalTopologyTest extends FuzzTestBase
{
@@ -64,12 +66,13 @@ public void identicalTopologyTest() throws Throwable
SchemaSpec schema = schemaGen.generate(rng);
cluster.schemaChange(schema.compile());
- HistoryBuilder history = new ReplayingHistoryBuilder(schema.valueGenerators,
+ IndexedValueGenerators valueGenerators = valueGenerators(schema, rng.next(), POPULATION);
+ 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 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 d70192f13677..1f208a334698 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.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 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 void journalGCTest() throws Throwable
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 9e7137fccc37..2d5a6245d728 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 String toString()
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 f6f5b8bf224b..8884b1ccebc2 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.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